简介交通流量数据集.zip 面向交通工程、城市规划与智能交通方向的研究者及算法学习者提供用于流量预测、拥堵识别与信号控制建模的基础数据。压缩包共12243个文件以6121个txt标注文件和6121个jpg图像文件为主另有1个yaml配置文件整体约162.31MBtxt多用于记录车辆计数、速度、时间戳等结构化字段jpg对应各监测点位的实景画面yaml则用于组织类别与路径信息便于直接接入检测或回归任务。内容覆盖多个路口与路段命名中可见不同方向与时段适合做时序分析、瓶颈定位及多源数据融合实验。目前已有1890人学习下载可作为课程设计、论文实验或模型预训练的现成素材帮助读者省去采集与清洗成本快速验证交通流量预测、智能信号灯调度等思路。1. 交通流量数据集.zip从压缩包到可训练张量的最短路径你拿到一个叫「交通流量数据集.zip」的压缩包双击解压里面大概率躺着几十个 CSV、若干 JSON 元数据、可能还有一两个 README。真正让人头疼的不是解压而是接下来三件事字段含义对不上、时间戳时区混乱、不同检测器采样频率不一致。我见过太多人卡在这一步直接把数据丢进模型结果 MAE 高得离谱回头查半天才发现是某几列单位搞错了。这个标题背后要解决的核心问题很明确把一份来源不明、格式混杂的交通流量压缩包变成能喂给时序模型LSTM、Transformer、STGNN或统计模型ARIMA、Prophet的规整张量。适合谁做智慧交通、城市规划、出行预测的算法工程师和数据分析师以及需要快速验证「这条路到底堵不堵」的产品同学。读完你应该能独立完成从解压到滑窗切片的完整链路并且知道每一步哪里容易翻车。2. 先搞清楚压缩包里到底有什么字段、粒度与拓扑2.1 交通流量数据的三种常见形态交通流量数据集不是只有一种长相。根据采集方式不同压缩包里的内容通常落在三类里第一类是线圈/地磁检测器数据典型字段是detector_id、timestamp、flow辆/小时或辆/5分钟、occupancy占有率、speed。这类数据采样间隔固定常见 5 分钟或 15 分钟缺失值用 -1 或空字符串表示。第二类是卡口/电警过车数据每条记录是一辆车经过一个点位字段包括plate脱敏后可能是哈希、cross_time、device_id。这种数据需要先做聚合才能得到流量粒度取决于你按分钟还是按小时 rollup。第三类是浮动车/GPS 轨迹数据字段是vehicle_id、lon、lat、timestamp。它不直接给流量需要先做地图匹配再统计路段通过量处理成本最高。拿到压缩包先别急着写代码用命令行扫一眼文件结构和编码# 查看压缩包内文件列表不解压 unzip -l 交通流量数据集.zip # 解压到指定目录 unzip 交通流量数据集.zip -d ./traffic_raw # 查看文件编码和前几行处理中文乱码 file -i ./traffic_raw/*.csv head -n 5 ./traffic_raw/$(ls ./traffic_raw | head -1)unzip -l先看有没有嵌套压缩包和目录层级file -i确认是 UTF-8 还是 GBK交通领域很多老系统导出的是 GBK直接pd.read_csv会报UnicodeDecodeError。head看表头是否带 BOM\ufeff带 BOM 的话第一列列名会多出不可见字符后面按列名取值会 KeyError。2.2 用 pandas 做一次字段体检确认编码后用一段脚本把每个文件的形状、列名、缺失率、时间范围打出来。这一步是后面所有处理的依据不要跳过。import pandas as pd import glob import os def inspect_traffic_files(raw_dir): 扫描目录下所有 CSV输出字段体检报告 report [] for fp in glob.glob(os.path.join(raw_dir, *.csv)): # 先尝试 utf-8失败回退 gbk try: df pd.read_csv(fp, nrows5000, encodingutf-8) except UnicodeDecodeError: df pd.read_csv(fp, nrows5000, encodinggbk) # 去掉列名可能的 BOM df.columns [c.strip().replace(\ufeff, ) for c in df.columns] info { file: os.path.basename(fp), rows_sampled: len(df), cols: list(df.columns), dtypes: df.dtypes.astype(str).to_dict(), missing_rate: (df.isna().mean().round(4)).to_dict(), } # 尝试识别时间列 time_cols [c for c in df.columns if time in c.lower() or date in c.lower()] if time_cols: tc time_cols[0] info[time_col] tc info[time_min] str(df[tc].min()) info[time_max] str(df[tc].max()) report.append(info) return report for r in inspect_traffic_files(./traffic_raw): print(r[file], r[rows_sampled], r[cols]) print( 缺失率:, r[missing_rate]) if time_col in r: print( 时间范围:, r[time_min], -, r[time_max])这段脚本的关键点nrows5000只采样避免大文件一次性读爆内存encoding回退策略覆盖国内交通系统常见的 GBK 导出missing_rate用字典输出方便你快速定位哪一列缺失严重。如果某列缺失率超过 30%要么放弃这列要么在后续插值前想清楚缺失机制——是设备离线还是真实流量为零这两者处理方式完全不同。2.3 检测器拓扑别把 ID 当普通字符串交通流量数据里detector_id或device_id往往隐含空间关系。比如D001和D002可能在同一条路段上下游R01_D03表示 1 号路第 3 个检测器。如果你要做 STGNN 这类需要邻接矩阵的模型必须从 ID 命名规则或配套的元数据文件里恢复拓扑。常见做法是压缩包里如果有一个meta.json或detector_info.csv里面会有detector_id、road_id、direction、lane、mileage字段。用mileage排序就能得到同路段检测器顺序用road_id分组就能构建路段级图。如果没有元数据只能靠 ID 前缀猜这时候要在论文或报告里明确说明拓扑是推断的别当成 ground truth。提示拓扑恢复错误是交通预测模型「离线指标好看、上线就崩」的常见原因之一。宁可先用纯时序模型跑 baseline也不要拿错误邻接矩阵硬上 STGNN。3. 清洗与重采样把脏数据变成规整时间序列3.1 时间戳统一时区、粒度与夏令时交通数据的时间戳有三种坑一是 Unix 秒/毫秒混用二是本地时间没带时区三是跨天数据里出现 24:00:00 这种非法表示。统一策略是全部转成 UTC 的datetime64[ns]再按需转回目标时区。import pandas as pd def normalize_timestamp(df, col): 把各种形态的时间戳统一成 UTC datetime s df[col] # 情况1纯数字判断秒还是毫秒 if pd.api.types.is_numeric_dtype(s): unit ms if s.max() 1e12 else s dt pd.to_datetime(s, unitunit, utcTrue) else: # 情况2字符串先尝试标准解析 dt pd.to_datetime(s, utcTrue, errorscoerce) # 情况3处理 24:00:00 mask s.astype(str).str.contains(24:00:00) if mask.any(): fixed s[mask].astype(str).str.replace(24:00:00, 00:00:00) dt.loc[mask] pd.to_datetime(fixed, utcTrue) pd.Timedelta(days1) df[col] dt return dfunit判断用1e12作为阈值因为秒级时间戳到 2286 年才到 1e11 量级毫秒级现在已经是 1e12 量级。errorscoerce把解析失败的置为 NaT后面统一统计丢弃或插值。24:00:00 的处理是很多交通系统导出 CSV 的通病不处理的话pd.to_datetime直接报错。3.2 缺失值插值不是万能药交通流量的缺失分两种随机缺失设备偶发离线和连续缺失设备故障数小时。随机缺失可以用线性插值或前向填充连续缺失必须标记出来否则插值会制造出虚假的平滑曲线模型学到的是幻觉。def fill_flow_gaps(df, group_col, time_col, value_col, max_gap30min): 按检测器分组短缺口插值长缺口标记 df df.sort_values([group_col, time_col]) df[is_gap] False filled [] for gid, g in df.groupby(group_col): g g.set_index(time_col) # 重采样到固定频率暴露隐式缺失 g g.resample(5min).agg({value_col: mean}) # 计算连续缺失长度 na_mask g[value_col].isna() gap_id (na_mask ! na_mask.shift()).cumsum() gap_len na_mask.groupby(gap_id).transform(sum) # 短缺口线性插值 short na_mask (gap_len pd.Timedelta(max_gap) / pd.Timedelta(5min)) g.loc[short, value_col] g[value_col].interpolate(methodlinear) # 长缺口保留 NaN 并标记 g[is_gap] na_mask ~short g[group_col] gid filled.append(g.reset_index()) return pd.concat(filled, ignore_indexTrue)resample(5min)的作用是把不规则采样对齐到固定网格同时把「时间戳跳跃」变成显式 NaN。gap_len用分组累计的方式算连续缺失长度比循环快得多。max_gap30min是经验值5 分钟粒度下 6 个点以内的缺口插值风险可控超过就标记。is_gap列要保留到模型阶段可以作为 mask 输入让模型知道哪些位置是补出来的。3.3 异常值流量不可能为负也不该突然翻十倍交通流量的物理约束很明确非负、有上限受车道数和饱和流率限制。常见异常是传感器漂移导致数值缓慢偏移或者通信故障导致某一刻数值突变。def clip_flow_outliers(df, value_col, lane_colNone, cap_per_lane2000): 按物理约束裁剪异常流量值 # 负值直接置零 df.loc[df[value_col] 0, value_col] 0 # 按车道数计算上限5分钟粒度单车道饱和约 2000 辆/小时 - 约 167 辆/5min if lane_col and lane_col in df.columns: cap df[lane_col] * cap_per_lane / 12 else: cap cap_per_lane / 12 # 用分位数辅助判断避免硬编码上限误伤 q999 df[value_col].quantile(0.999) upper min(cap, q999 * 1.5) if isinstance(cap, pd.Series) else min(cap, q999 * 1.5) df[value_col] df[value_col].clip(upperupper) return dfcap_per_lane2000是城市道路单车道小时饱和流量的常见量级换算到 5 分钟要除以 12。q999 * 1.5是兜底如果数据整体流量偏低硬用理论上限会削掉真实高峰。clip而不是删除是因为删除会破坏时间连续性后续滑窗会断。注意异常值处理没有银弹。我一般会先把处理前后的分布画出来对比确认高峰没被削平、低谷没被抬高再往下走。4. 滑窗切片与数据集封装让模型能直接吃4.1 从长表到三维张量时序模型需要的输入形状通常是(样本数, 时间步, 特征数)或(样本数, 节点数, 时间步, 特征数)。原始长表是(记录数, 字段数)中间要做透视和滑窗两步。import numpy as np def build_sliding_windows(df, time_col, node_col, value_cols, input_len12, pred_len3, stride1): 构建滑窗样本返回 X, y 及对应时间索引 # 透视成 (时间, 节点, 特征) pivot df.pivot_table( indextime_col, columnsnode_col, valuesvalue_cols, aggfuncmean ) # 确保时间连续 full_idx pd.date_range(pivot.index.min(), pivot.index.max(), freq5min) pivot pivot.reindex(full_idx) values pivot.values # shape: (T, N*F) times pivot.index X, y, t_idx [], [], [] total len(values) for i in range(0, total - input_len - pred_len 1, stride): X.append(values[i : i input_len]) y.append(values[i input_len : i input_len pred_len]) t_idx.append(times[i input_len]) return np.array(X), np.array(y), np.array(t_idx)pivot_table把长表转成「时间 × 节点」的宽表aggfuncmean处理同一时刻同一节点的重复记录。reindex到完整时间索引把缺失时刻显式补成 NaN避免滑窗跨过缺口。input_len12对应 5 分钟粒度下 1 小时历史pred_len3预测未来 15 分钟这是短时交通预测的常见设定。stride1逐点滑动样本量最大如果数据量太大可以调大 stride 降采样。4.2 训练/验证/测试划分别按随机划分时间序列不能随机划分否则未来信息泄漏到训练集指标虚高。正确做法是按时间切分并且验证集和测试集之间留 gap避免滑窗重叠导致泄漏。def temporal_split(X, y, t_idx, train_ratio0.7, val_ratio0.15, gap12): 按时间顺序划分验证/测试间留 gap 个时间步 n len(X) train_end int(n * train_ratio) val_end int(n * (train_ratio val_ratio)) X_train, y_train X[:train_end], y[:train_end] X_val, y_val X[train_end gap : val_end], y[train_end gap : val_end] X_test, y_test X[val_end gap :], y[val_end gap :] return (X_train, y_train), (X_val, y_val), (X_test, y_test)gap12对应 1 小时确保验证集第一个样本的输入窗口不与训练集最后一个样本的输出窗口重叠。这个细节在论文里经常被忽略但审稿人如果较真泄漏问题足以让结果不可信。4.3 标准化按节点还是全局交通流量不同检测器的量级差异很大主干道和支路可能差一个数量级。标准化方式有两种全局 z-score 和按节点 z-score。全局标准化简单但量级小的节点会被压缩到接近零模型难以学习按节点标准化保留每个节点的相对变化但推理时需要保存每个节点的均值和方差。def normalize_per_node(X_train, X_val, X_test): 按节点维度计算均值和方差避免量级差异导致的学习困难 # X shape: (样本, 时间, 节点*特征) mean X_train.mean(axis(0, 1), keepdimsTrue) std X_train.std(axis(0, 1), keepdimsTrue) 1e-8 return (X_train - mean) / std, (X_val - mean) / std, (X_test - mean) / std, mean, stdaxis(0, 1)是在样本和时间维度上聚合保留节点×特征维度。1e-8防止除零。保存mean和std是为了推理时反标准化这一步经常有人忘记导致预测值量级完全不对。提示如果某些节点在训练集中长期为零比如新装设备其 std 接近零标准化后会放大噪声。建议在标准化前把这类节点过滤掉或者单独处理。5. 避坑与排查交通流量数据集处理中的五个血泪教训5.1 现象模型训练 loss 正常下降但验证集 MAE 始终在 30 以上原因时间戳没统一时区训练集和验证集实际覆盖的时间段错位模型学到的是错误的时间模式。常见于跨系统拼接的数据一部分是 UTC一部分是本地时间。解决在处理第一步就强制utcTrue并在划分后打印训练/验证/测试的时间范围确认没有重叠且顺序正确。5.2 现象插值后流量曲线过于平滑高峰被削平原因对连续缺失做了线性插值而连续缺失往往发生在设备故障期间真实流量可能为零或极低插值制造了虚假的中间值。解决用is_gap标记长缺口插值只作用于短缺口。长缺口要么丢弃对应样本要么在 loss 里加 mask 忽略。5.3 现象按节点标准化后某些节点预测值恒为零原因这些节点在训练集中流量极低或长期缺失std 接近零标准化后数值被放大到极端值模型无法收敛。解决统计每个节点的有效样本比例低于阈值如 50%的节点直接排除或在模型中加节点嵌入区分。5.4 现象滑窗样本量远小于预期原因pivot_table后reindex引入了大量 NaN滑窗时如果做了dropna样本被大量丢弃。解决先统计缺失率决定是插值还是丢弃。如果缺失集中在少数节点可以按节点过滤后再透视而不是全局 dropna。5.5 现象推理时预测值量级完全不对原因训练时做了标准化推理时忘记用保存的mean和std反标准化或者反标准化时维度对不上。解决把mean和std跟模型一起保存推理脚本里显式调用反标准化并写单元测试验证「标准化再反标准化」能还原原始值。6. 进阶技巧用时间序列交叉验证替代单次划分单次时间划分的评估结果方差很大换一个切分点指标可能差 10% 以上。更稳的做法是滚动交叉验证rolling origin cross-validation在多个时间起点上训练和评估取平均指标。def rolling_cv(X, y, n_splits5, input_len12, pred_len3): 滚动交叉验证每次用历史训练预测下一段 n len(X) fold_size n // (n_splits 1) results [] for k in range(1, n_splits 1): train_end fold_size * k test_end min(train_end fold_size, n) X_tr, y_tr X[:train_end], y[:train_end] X_te, y_te X[train_end:test_end], y[train_end:test_end] # 这里替换成你的模型训练和评估 # model.fit(X_tr, y_tr); pred model.predict(X_te) # results.append(mae(y_te, pred)) results.append((k, train_end, test_end)) return resultsfold_size把数据均分每次训练集向前扩展一个 fold测试集是紧接着的下一个 fold。这样每个样本最多被预测一次且始终用历史预测未来。实际使用时把注释部分替换成你的模型调用收集每折的 MAE/RMSE报告均值和标准差。如果某折指标明显偏离回去查那段时间是否有节假日或特殊事件这往往能发现数据里的隐藏问题。我自己的习惯是任何交通流量数据集先跑通「解压 → 字段体检 → 时间统一 → 短缺口插值 → 滑窗 → 单次划分 baseline」这条链路再考虑上复杂模型。baseline 用历史均值或上周同期值如果复杂模型跑不过 baseline问题一定在数据处理而不是模型结构。这个顺序帮我省过很多次返工。希望帮到你。本文还有配套的精品资源点击获取