尧图网页设计 
简介交通流量数据集.zip 面向交通工程、城市规划与人工智能方向的研究者及算法学习者用于交通流量预测、拥堵识别与智能交通系统建模等场景。压缩包共含12243个文件以6121个txt标注文件和6121个jpg图像文件为主另有1个yaml配置文件整体约162.31MBtxt文件通常承载车辆计数、速度、时间戳等结构化记录jpg图像对应各监测点位的交通场景yaml则用于组织数据路径与类别信息便于直接接入目标检测或时序分析流程。内容覆盖多个路口与路段的流量记录可支撑流量预测、瓶颈识别、信号控制优化及环境影响评估等任务。目前已有1890人学习下载适合需要真实交通数据开展实验、复现模型或完成课程与竞赛项目的读者参考使用。1. 交通流量数据集从时序预测到拥堵识别的落地起点做交通流量预测的工程师大概都有过这种经历模型结构调了一周注意力机制换了三版结果一跑验证集误差比上周还大。回头一查训练集里节假日和雨天的样本几乎没标早晚高峰的周期特征被随机打散模型学到的全是噪声。问题往往不在网络本身而在数据这一层。这份交通流量数据集.zip 就是冲着这个痛点来的——它把某城市路网的多时段流量观测整理成结构化表格包含时间戳、路段编号、车道方向、流量计数和平均车速等字段适合直接喂给 LSTM、Transformer 或 XGBoost 做时序回归与拥堵分类。如果你正在找一份能跑通完整流程、又不用花两周做清洗的交通数据集它值得先拆开看看。2. 拆包与字段解析先搞清楚每一列能不能用2.1 压缩包结构与文件格式拿到一个数据集压缩包我习惯先不急着写代码而是用命令行把目录树和文件大小过一遍。这一步能快速判断数据是单表还是分片、有没有缺失的日期文件、编码是不是 UTF-8。# 查看压缩包内文件列表不实际解压 unzip -l traffic_flow_dataset.zip # 解压到指定目录保留原始结构 unzip traffic_flow_dataset.zip -d ./traffic_data # 查看解压后各文件大小和行数 ls -lh ./traffic_data/ wc -l ./traffic_data/*.csvunzip -l只列目录适合在解压前确认文件数量是否符合预期。-d参数指定输出目录避免文件散落在当前路径。wc -l统计行数时要注意如果 CSV 字段内含有换行符行数会虚高所以它只能作为粗略参考精确行数还是得用 pandas 读一遍。常见做法是解压后先看有没有 README 或字段说明文件。如果压缩包里只有 CSV 没有说明文档那就得靠列名和取值分布反推含义。这一步别偷懒字段理解错了后面特征工程全是白做。2.2 字段含义与数据类型推断用 pandas 读进来之后第一件事是看 dtypes 和 head。交通流量数据最容易翻车的地方是时间列被当成字符串、路段编号被当成数值。下面这段代码把字段类型和缺失情况一次性打出来。import pandas as pd # 读取时先把所有列当字符串避免自动类型推断出错 df pd.read_csv(./traffic_data/traffic_flow.csv, dtypestr) # 查看前五行和字段类型 print(df.head()) print(df.dtypes) # 统计每列缺失率和唯一值数量 for col in df.columns: missing df[col].isna().mean() unique df[col].nunique() print(f{col}: 缺失率{missing:.2%}, 唯一值{unique}) # 尝试解析时间列记录失败的行 df[timestamp] pd.to_datetime(df[timestamp], errorscoerce) bad_time df[df[timestamp].isna()] print(f时间解析失败行数: {len(bad_time)})dtypestr是为了防止 pandas 把路段编号001读成整数1也防止时间列被误判。errorscoerce会把解析失败的值变成 NaT方便你统计有多少脏时间戳。如果失败行数超过总行数的 1%说明时间格式不统一可能需要指定format参数或先做字符串清洗。字段类型确认后重点看三列时间戳、路段编号、流量值。时间戳决定你能不能做周期特征路段编号决定能不能做空间聚合流量值决定回归还是分类。平均车速那一列如果缺失严重可以考虑只用流量做单变量预测别硬塞进模型。2.3 时间粒度与采样频率确认交通流量数据的采样频率直接决定模型输入窗口怎么切。5 分钟粒度和 15 分钟粒度的处理方式完全不同。下面这段代码用来确认实际采样间隔。# 按路段分组后计算时间差的中位数 df_sorted df.sort_values([road_id, timestamp]) df_sorted[time_diff] df_sorted.groupby(road_id)[timestamp].diff() # 统计各时间差的出现次数 diff_counts df_sorted[time_diff].value_counts().head(10) print(diff_counts) # 如果中位数是 5 分钟说明是 5 分钟粒度 median_diff df_sorted[time_diff].median() print(f采样间隔中位数: {median_diff})groupby(road_id)是关键因为不同路段的时间序列是独立的直接对全表做 diff 会跨路段计算出错误间隔。value_counts()看分布比只看中位数更可靠——如果出现多个间隔值说明数据里有缺采或重复采集。常见情况是白天 5 分钟、夜间 15 分钟这种非均匀采样在建模前要么重采样到统一粒度要么在模型里加时间间隔特征。提示如果采样间隔不统一别急着用resample填充。先确认缺失时段是设备离线还是正常无车前者该补零后者该补 NaN。3. 特征工程实战把原始流量表变成模型能吃的张量3.1 时间周期特征构造交通流量最核心的规律就是周期性早高峰、晚高峰、工作日与周末。把时间戳拆成小时、星期、是否节假日是最基础也最有效的特征。import numpy as np df[hour] df[timestamp].dt.hour df[dayofweek] df[timestamp].dt.dayofweek # 0周一, 6周日 df[is_weekend] (df[dayofweek] 5).astype(int) # 用正弦余弦编码小时保留周期性 df[hour_sin] np.sin(2 * np.pi * df[hour] / 24) df[hour_cos] np.cos(2 * np.pi * df[hour] / 24) # 构造高峰标记7-9 点和 17-19 点 df[is_peak] df[hour].isin([7, 8, 9, 17, 18, 19]).astype(int) print(df[[timestamp, hour, hour_sin, hour_cos, is_peak]].head())正弦余弦编码的作用是让 23 点和 0 点在特征空间里距离更近而不是像原始小时数那样相差 23。is_peak这种硬标记适合树模型神经网络里用 sin/cos 更平滑。两种都保留让模型自己选。如果数据跨月还可以加dayofmonth和is_month_start。但要注意别加太多时间特征否则模型容易记住训练集的具体日期而不是学周期规律。我一般控制在 6 个以内。3.2 滑动窗口与滞后特征时序预测的核心是把过去 N 个时刻的流量作为输入预测未来 M 个时刻。下面这段代码构造滞后特征和滑动统计量。# 按路段分组后构造滞后特征 for lag in [1, 2, 3, 6, 12]: df[fflow_lag_{lag}] df.groupby(road_id)[flow].shift(lag) # 滑动窗口均值和标准差 df[flow_roll_mean_6] df.groupby(road_id)[flow].transform( lambda x: x.rolling(window6, min_periods1).mean() ) df[flow_roll_std_6] df.groupby(road_id)[flow].transform( lambda x: x.rolling(window6, min_periods1).std() ) # 去掉因滞后产生的空值行 df_model df.dropna().reset_index(dropTrue) print(f建模可用行数: {len(df_model)})shift(lag)必须放在groupby之后否则会把上一个路段的最后几条记录混进当前路段。rolling的min_periods1保证序列开头不会全变 NaN但代价是前几行的统计量不稳定。如果对精度要求高可以把前 12 行直接丢掉。滞后阶数的选择看采样间隔5 分钟粒度下lag 1 到 12 覆盖过去一小时15 分钟粒度下lag 1 到 4 就够了。别盲目堆滞后特征每多一列就多一份过拟合风险。3.3 路段编码与空间特征路段编号是类别变量直接扔进模型会被当成数值大小。常见做法是 One-Hot 或 Embedding。路段数量少就用 One-Hot多就用 Embedding。# 路段数量少于 50 时用 One-Hot road_count df_model[road_id].nunique() print(f路段数量: {road_count}) if road_count 50: df_model pd.get_dummies(df_model, columns[road_id], prefixroad) else: # 路段多时用频率编码替代 road_freq df_model[road_id].value_counts(normalizeTrue) df_model[road_freq] df_model[road_id].map(road_freq) # 构造路段平均流量作为静态特征 road_mean df_model.groupby(road_id)[flow].mean().to_dict() df_model[road_mean_flow] df_model[road_id].map(road_mean)get_dummies会产生稀疏矩阵路段超过 50 个就不建议用了。频率编码是把路段出现次数占比作为特征适合高基数类别。road_mean_flow是目标编码的简化版注意要在划分训练集之前算否则会泄露验证集信息。更严谨的做法是在训练集上算均值再映射到验证集。注意目标编码是数据泄露的重灾区。如果非要用必须用 K 折方式在训练集内部计算别图省事在全量数据上算。4. 模型训练与验证LSTM 和 XGBoost 两条路怎么选4.1 数据集划分与标准化时序数据不能随机打乱划分必须按时间切分。下面这段代码按 7:2:1 切训练、验证、测试集。from sklearn.preprocessing import StandardScaler # 按时间排序后切分 df_model df_model.sort_values(timestamp).reset_index(dropTrue) n len(df_model) train_end int(n * 0.7) val_end int(n * 0.9) train_df df_model.iloc[:train_end] val_df df_model.iloc[train_end:val_end] test_df df_model.iloc[val_end:] # 特征列和标签列 feature_cols [c for c in df_model.columns if c not in [timestamp, flow, road_id]] target_col flow # 标准化只在训练集上 fit scaler StandardScaler() train_scaled scaler.fit_transform(train_df[feature_cols]) val_scaled scaler.transform(val_df[feature_cols]) test_scaled scaler.transform(test_df[feature_cols]) print(f训练集: {train_scaled.shape}, 验证集: {val_scaled.shape}, 测试集: {test_scaled.shape})fit_transform只在训练集上调用验证集和测试集只用transform。这是铁律但每年还是能看到有人在全量数据上 fit。按时间切分而不是随机切分是因为随机切分会让模型看到未来的数据验证指标虚高上线就崩。4.2 XGBoost 基线快速验证在投入深度学习之前先用 XGBoost 跑一个基线。如果 XGBoost 效果已经够用就没必要上 LSTM。import xgboost as xgb from sklearn.metrics import mean_absolute_error, mean_squared_error # 构造 DMatrix dtrain xgb.DMatrix(train_scaled, labeltrain_df[target_col]) dval xgb.DMatrix(val_scaled, labelval_df[target_col]) dtest xgb.DMatrix(test_scaled, labeltest_df[target_col]) # 参数设置 params { objective: reg:squarederror, max_depth: 6, learning_rate: 0.05, subsample: 0.8, colsample_bytree: 0.8, seed: 42 } # 训练并记录验证集表现 evals [(dtrain, train), (dval, val)] model xgb.train(params, dtrain, num_boost_round500, evalsevals, early_stopping_rounds30, verbose_eval50) # 测试集评估 pred model.predict(dtest) mae mean_absolute_error(test_df[target_col], pred) rmse mean_squared_error(test_df[target_col], pred) ** 0.5 print(fXGBoost MAE: {mae:.2f}, RMSE: {rmse:.2f})early_stopping_rounds30表示验证集 30 轮不提升就停防止过拟合。max_depth6是交通流量数据的常用起点再深容易记住噪声。subsample和colsample_bytree都设 0.8增加随机性提升泛化。如果 MAE 在测试集上比验证集高出一大截说明时间切分点之后的数据分布变了可能是节假日或天气影响。这时候要么加更多特征要么用滚动预测代替单次切分。4.3 LSTM 序列建模与窗口构造LSTM 需要三维输入(样本数, 时间步, 特征数)。下面这段代码把二维表转成三维张量。import numpy as np import tensorflow as tf from tensorflow.keras import layers, models def create_sequences(data, target, window_size): X, y [], [] for i in range(len(data) - window_size): X.append(data[i:iwindow_size]) y.append(target[iwindow_size]) return np.array(X), np.array(y) window_size 12 # 过去 12 个时刻 X_train, y_train create_sequences(train_scaled, train_df[target_col].values, window_size) X_val, y_val create_sequences(val_scaled, val_df[target_col].values, window_size) X_test, y_test create_sequences(test_scaled, test_df[target_col].values, window_size) print(fLSTM 输入形状: {X_train.shape}) # 构建模型 model models.Sequential([ layers.LSTM(64, return_sequencesTrue, input_shape(window_size, X_train.shape[2])), layers.LSTM(32), layers.Dense(16, activationrelu), layers.Dense(1) ]) model.compile(optimizeradam, lossmse, metrics[mae]) history model.fit(X_train, y_train, validation_data(X_val, y_val), epochs50, batch_size64, verbose1)window_size12对应 5 分钟粒度下的过去一小时。两层 LSTM 比单层更容易捕捉长程依赖但参数量翻倍数据量少于 10 万条时建议先用单层。return_sequencesTrue只在第一层需要第二层输出二维向量给全连接层。训练时看val_loss曲线如果训练 loss 持续下降但验证 loss 反弹就是过拟合加 Dropout 或减小 LSTM 单元数。如果两条曲线都下不去可能是 window_size 太小或特征不够。5. 避坑与排查交通流量数据集的五个血泪经验5.1 时间戳时区不统一导致周期特征错乱现象模型在训练集上表现正常但预测早高峰总是偏移一两个小时。原因数据里混了 UTC 和本地时间pd.to_datetime默认不转换时区导致小时特征错位。解决读入后统一加utcTrue再转目标时区或者先检查时间戳范围是否跨了时区切换点。df[timestamp] pd.to_datetime(df[timestamp], utcTrue) df[timestamp] df[timestamp].dt.tz_convert(Asia/Shanghai)5.2 路段编号被自动转成整数丢失前导零现象One-Hot 之后路段数量比实际少因为001和1被合并了。原因read_csv默认推断类型把字符串编号转成了整数。解决读取时指定dtype{road_id: str}或者用converters参数强制字符串。5.3 缺失值填充方式选错引入虚假规律现象模型在夜间时段预测偏高。原因用全局均值填充夜间缺失流量而夜间实际流量接近零。解决按路段和小时分组填充或者用前向填充加标记列。别用全局均值那是给模型喂假数据。5.4 滑动窗口跨路段边界导致特征污染现象验证集误差比训练集高很多且集中在路段切换处。原因构造滞后特征时没按路段分组上一个路段的末尾流量混进了下一个路段的开头。解决所有shift和rolling操作前先groupby(road_id)构造完再dropna。5.5 验证集划分时数据泄露现象验证集 MAE 低得离谱测试集一跑就翻车。原因标准化时在全量数据上fit或者目标编码用了全量均值。解决所有拟合操作只在训练集上做验证集和测试集只做transform。时序数据还要确保切分点之后的数据完全不可见。提示每次跑完模型把验证集和测试集的指标并排看。差距超过 20% 就先查泄露别急着调参。6. 进阶技巧用滚动预测替代单次切分验证单次时间切分有个天然缺陷只验证了一个时间点之后的表现。如果测试期正好赶上节假日指标会失真。更稳的做法是滚动预测——用过去 N 天训练预测下一天然后窗口后移重复多次取平均。def rolling_validation(df, feature_cols, target_col, train_days30, test_days1): results [] dates df[timestamp].dt.date.unique() for i in range(train_days, len(dates) - test_days 1, test_days): train_dates dates[i-train_days:i] test_dates dates[i:itest_days] train_mask df[timestamp].dt.date.isin(train_dates) test_mask df[timestamp].dt.date.isin(test_dates) scaler StandardScaler() X_train scaler.fit_transform(df.loc[train_mask, feature_cols]) X_test scaler.transform(df.loc[test_mask, feature_cols]) model xgb.XGBRegressor(n_estimators300, max_depth6, learning_rate0.05) model.fit(X_train, df.loc[train_mask, target_col]) pred model.predict(X_test) mae mean_absolute_error(df.loc[test_mask, target_col], pred) results.append(mae) return np.mean(results), np.std(results) mean_mae, std_mae rolling_validation(df_model, feature_cols, target_col) print(f滚动验证 MAE: {mean_mae:.2f} ± {std_mae:.2f})这段代码的核心是for循环里的窗口滑动。train_days30表示每次用 30 天数据训练test_days1表示预测下一天。dates去重后按天遍历避免同一天被重复切分。每次循环重新fitscaler 和模型确保没有跨窗口泄露。滚动验证的代价是计算量成倍增加30 天窗口滑 60 次就是 60 次训练。但它的好处是能给出指标的标准差——如果标准差很大说明模型对时间敏感上线后需要频繁更新。我一般会在项目初期跑一次滚动验证确认模型稳定性后再用单次切分做快速迭代。从那以后我每次拿到新的时序数据集都强制先跑一遍滚动验证再决定要不要上深度学习。单次切分的指标太容易骗人滚动验证的均值和方差才是真实水平。希望帮到你。本文还有配套的精品资源点击获取 交通流量数据集实战：从时序预测到拥堵识别的完整流程
简介交通流量数据集.zip 面向交通工程、城市规划与人工智能方向的研究者及算法学习者用于交通流量预测、拥堵识别与智能交通系统建模等场景。压缩包共含12243个文件以6121个txt标注文件和6121个jpg图像文件为主另有1个yaml配置文件整体约162.31MBtxt文件通常承载车辆计数、速度、时间戳等结构化记录jpg图像对应各监测点位的交通场景yaml则用于组织数据路径与类别信息便于直接接入目标检测或时序分析流程。内容覆盖多个路口与路段的流量记录可支撑流量预测、瓶颈识别、信号控制优化及环境影响评估等任务。目前已有1890人学习下载适合需要真实交通数据开展实验、复现模型或完成课程与竞赛项目的读者参考使用。1. 交通流量数据集从时序预测到拥堵识别的落地起点做交通流量预测的工程师大概都有过这种经历模型结构调了一周注意力机制换了三版结果一跑验证集误差比上周还大。回头一查训练集里节假日和雨天的样本几乎没标早晚高峰的周期特征被随机打散模型学到的全是噪声。问题往往不在网络本身而在数据这一层。这份交通流量数据集.zip 就是冲着这个痛点来的——它把某城市路网的多时段流量观测整理成结构化表格包含时间戳、路段编号、车道方向、流量计数和平均车速等字段适合直接喂给 LSTM、Transformer 或 XGBoost 做时序回归与拥堵分类。如果你正在找一份能跑通完整流程、又不用花两周做清洗的交通数据集它值得先拆开看看。2. 拆包与字段解析先搞清楚每一列能不能用2.1 压缩包结构与文件格式拿到一个数据集压缩包我习惯先不急着写代码而是用命令行把目录树和文件大小过一遍。这一步能快速判断数据是单表还是分片、有没有缺失的日期文件、编码是不是 UTF-8。# 查看压缩包内文件列表不实际解压 unzip -l traffic_flow_dataset.zip # 解压到指定目录保留原始结构 unzip traffic_flow_dataset.zip -d ./traffic_data # 查看解压后各文件大小和行数 ls -lh ./traffic_data/ wc -l ./traffic_data/*.csvunzip -l只列目录适合在解压前确认文件数量是否符合预期。-d参数指定输出目录避免文件散落在当前路径。wc -l统计行数时要注意如果 CSV 字段内含有换行符行数会虚高所以它只能作为粗略参考精确行数还是得用 pandas 读一遍。常见做法是解压后先看有没有 README 或字段说明文件。如果压缩包里只有 CSV 没有说明文档那就得靠列名和取值分布反推含义。这一步别偷懒字段理解错了后面特征工程全是白做。2.2 字段含义与数据类型推断用 pandas 读进来之后第一件事是看 dtypes 和 head。交通流量数据最容易翻车的地方是时间列被当成字符串、路段编号被当成数值。下面这段代码把字段类型和缺失情况一次性打出来。import pandas as pd # 读取时先把所有列当字符串避免自动类型推断出错 df pd.read_csv(./traffic_data/traffic_flow.csv, dtypestr) # 查看前五行和字段类型 print(df.head()) print(df.dtypes) # 统计每列缺失率和唯一值数量 for col in df.columns: missing df[col].isna().mean() unique df[col].nunique() print(f{col}: 缺失率{missing:.2%}, 唯一值{unique}) # 尝试解析时间列记录失败的行 df[timestamp] pd.to_datetime(df[timestamp], errorscoerce) bad_time df[df[timestamp].isna()] print(f时间解析失败行数: {len(bad_time)})dtypestr是为了防止 pandas 把路段编号001读成整数1也防止时间列被误判。errorscoerce会把解析失败的值变成 NaT方便你统计有多少脏时间戳。如果失败行数超过总行数的 1%说明时间格式不统一可能需要指定format参数或先做字符串清洗。字段类型确认后重点看三列时间戳、路段编号、流量值。时间戳决定你能不能做周期特征路段编号决定能不能做空间聚合流量值决定回归还是分类。平均车速那一列如果缺失严重可以考虑只用流量做单变量预测别硬塞进模型。2.3 时间粒度与采样频率确认交通流量数据的采样频率直接决定模型输入窗口怎么切。5 分钟粒度和 15 分钟粒度的处理方式完全不同。下面这段代码用来确认实际采样间隔。# 按路段分组后计算时间差的中位数 df_sorted df.sort_values([road_id, timestamp]) df_sorted[time_diff] df_sorted.groupby(road_id)[timestamp].diff() # 统计各时间差的出现次数 diff_counts df_sorted[time_diff].value_counts().head(10) print(diff_counts) # 如果中位数是 5 分钟说明是 5 分钟粒度 median_diff df_sorted[time_diff].median() print(f采样间隔中位数: {median_diff})groupby(road_id)是关键因为不同路段的时间序列是独立的直接对全表做 diff 会跨路段计算出错误间隔。value_counts()看分布比只看中位数更可靠——如果出现多个间隔值说明数据里有缺采或重复采集。常见情况是白天 5 分钟、夜间 15 分钟这种非均匀采样在建模前要么重采样到统一粒度要么在模型里加时间间隔特征。提示如果采样间隔不统一别急着用resample填充。先确认缺失时段是设备离线还是正常无车前者该补零后者该补 NaN。3. 特征工程实战把原始流量表变成模型能吃的张量3.1 时间周期特征构造交通流量最核心的规律就是周期性早高峰、晚高峰、工作日与周末。把时间戳拆成小时、星期、是否节假日是最基础也最有效的特征。import numpy as np df[hour] df[timestamp].dt.hour df[dayofweek] df[timestamp].dt.dayofweek # 0周一, 6周日 df[is_weekend] (df[dayofweek] 5).astype(int) # 用正弦余弦编码小时保留周期性 df[hour_sin] np.sin(2 * np.pi * df[hour] / 24) df[hour_cos] np.cos(2 * np.pi * df[hour] / 24) # 构造高峰标记7-9 点和 17-19 点 df[is_peak] df[hour].isin([7, 8, 9, 17, 18, 19]).astype(int) print(df[[timestamp, hour, hour_sin, hour_cos, is_peak]].head())正弦余弦编码的作用是让 23 点和 0 点在特征空间里距离更近而不是像原始小时数那样相差 23。is_peak这种硬标记适合树模型神经网络里用 sin/cos 更平滑。两种都保留让模型自己选。如果数据跨月还可以加dayofmonth和is_month_start。但要注意别加太多时间特征否则模型容易记住训练集的具体日期而不是学周期规律。我一般控制在 6 个以内。3.2 滑动窗口与滞后特征时序预测的核心是把过去 N 个时刻的流量作为输入预测未来 M 个时刻。下面这段代码构造滞后特征和滑动统计量。# 按路段分组后构造滞后特征 for lag in [1, 2, 3, 6, 12]: df[fflow_lag_{lag}] df.groupby(road_id)[flow].shift(lag) # 滑动窗口均值和标准差 df[flow_roll_mean_6] df.groupby(road_id)[flow].transform( lambda x: x.rolling(window6, min_periods1).mean() ) df[flow_roll_std_6] df.groupby(road_id)[flow].transform( lambda x: x.rolling(window6, min_periods1).std() ) # 去掉因滞后产生的空值行 df_model df.dropna().reset_index(dropTrue) print(f建模可用行数: {len(df_model)})shift(lag)必须放在groupby之后否则会把上一个路段的最后几条记录混进当前路段。rolling的min_periods1保证序列开头不会全变 NaN但代价是前几行的统计量不稳定。如果对精度要求高可以把前 12 行直接丢掉。滞后阶数的选择看采样间隔5 分钟粒度下lag 1 到 12 覆盖过去一小时15 分钟粒度下lag 1 到 4 就够了。别盲目堆滞后特征每多一列就多一份过拟合风险。3.3 路段编码与空间特征路段编号是类别变量直接扔进模型会被当成数值大小。常见做法是 One-Hot 或 Embedding。路段数量少就用 One-Hot多就用 Embedding。# 路段数量少于 50 时用 One-Hot road_count df_model[road_id].nunique() print(f路段数量: {road_count}) if road_count 50: df_model pd.get_dummies(df_model, columns[road_id], prefixroad) else: # 路段多时用频率编码替代 road_freq df_model[road_id].value_counts(normalizeTrue) df_model[road_freq] df_model[road_id].map(road_freq) # 构造路段平均流量作为静态特征 road_mean df_model.groupby(road_id)[flow].mean().to_dict() df_model[road_mean_flow] df_model[road_id].map(road_mean)get_dummies会产生稀疏矩阵路段超过 50 个就不建议用了。频率编码是把路段出现次数占比作为特征适合高基数类别。road_mean_flow是目标编码的简化版注意要在划分训练集之前算否则会泄露验证集信息。更严谨的做法是在训练集上算均值再映射到验证集。注意目标编码是数据泄露的重灾区。如果非要用必须用 K 折方式在训练集内部计算别图省事在全量数据上算。4. 模型训练与验证LSTM 和 XGBoost 两条路怎么选4.1 数据集划分与标准化时序数据不能随机打乱划分必须按时间切分。下面这段代码按 7:2:1 切训练、验证、测试集。from sklearn.preprocessing import StandardScaler # 按时间排序后切分 df_model df_model.sort_values(timestamp).reset_index(dropTrue) n len(df_model) train_end int(n * 0.7) val_end int(n * 0.9) train_df df_model.iloc[:train_end] val_df df_model.iloc[train_end:val_end] test_df df_model.iloc[val_end:] # 特征列和标签列 feature_cols [c for c in df_model.columns if c not in [timestamp, flow, road_id]] target_col flow # 标准化只在训练集上 fit scaler StandardScaler() train_scaled scaler.fit_transform(train_df[feature_cols]) val_scaled scaler.transform(val_df[feature_cols]) test_scaled scaler.transform(test_df[feature_cols]) print(f训练集: {train_scaled.shape}, 验证集: {val_scaled.shape}, 测试集: {test_scaled.shape})fit_transform只在训练集上调用验证集和测试集只用transform。这是铁律但每年还是能看到有人在全量数据上 fit。按时间切分而不是随机切分是因为随机切分会让模型看到未来的数据验证指标虚高上线就崩。4.2 XGBoost 基线快速验证在投入深度学习之前先用 XGBoost 跑一个基线。如果 XGBoost 效果已经够用就没必要上 LSTM。import xgboost as xgb from sklearn.metrics import mean_absolute_error, mean_squared_error # 构造 DMatrix dtrain xgb.DMatrix(train_scaled, labeltrain_df[target_col]) dval xgb.DMatrix(val_scaled, labelval_df[target_col]) dtest xgb.DMatrix(test_scaled, labeltest_df[target_col]) # 参数设置 params { objective: reg:squarederror, max_depth: 6, learning_rate: 0.05, subsample: 0.8, colsample_bytree: 0.8, seed: 42 } # 训练并记录验证集表现 evals [(dtrain, train), (dval, val)] model xgb.train(params, dtrain, num_boost_round500, evalsevals, early_stopping_rounds30, verbose_eval50) # 测试集评估 pred model.predict(dtest) mae mean_absolute_error(test_df[target_col], pred) rmse mean_squared_error(test_df[target_col], pred) ** 0.5 print(fXGBoost MAE: {mae:.2f}, RMSE: {rmse:.2f})early_stopping_rounds30表示验证集 30 轮不提升就停防止过拟合。max_depth6是交通流量数据的常用起点再深容易记住噪声。subsample和colsample_bytree都设 0.8增加随机性提升泛化。如果 MAE 在测试集上比验证集高出一大截说明时间切分点之后的数据分布变了可能是节假日或天气影响。这时候要么加更多特征要么用滚动预测代替单次切分。4.3 LSTM 序列建模与窗口构造LSTM 需要三维输入(样本数, 时间步, 特征数)。下面这段代码把二维表转成三维张量。import numpy as np import tensorflow as tf from tensorflow.keras import layers, models def create_sequences(data, target, window_size): X, y [], [] for i in range(len(data) - window_size): X.append(data[i:iwindow_size]) y.append(target[iwindow_size]) return np.array(X), np.array(y) window_size 12 # 过去 12 个时刻 X_train, y_train create_sequences(train_scaled, train_df[target_col].values, window_size) X_val, y_val create_sequences(val_scaled, val_df[target_col].values, window_size) X_test, y_test create_sequences(test_scaled, test_df[target_col].values, window_size) print(fLSTM 输入形状: {X_train.shape}) # 构建模型 model models.Sequential([ layers.LSTM(64, return_sequencesTrue, input_shape(window_size, X_train.shape[2])), layers.LSTM(32), layers.Dense(16, activationrelu), layers.Dense(1) ]) model.compile(optimizeradam, lossmse, metrics[mae]) history model.fit(X_train, y_train, validation_data(X_val, y_val), epochs50, batch_size64, verbose1)window_size12对应 5 分钟粒度下的过去一小时。两层 LSTM 比单层更容易捕捉长程依赖但参数量翻倍数据量少于 10 万条时建议先用单层。return_sequencesTrue只在第一层需要第二层输出二维向量给全连接层。训练时看val_loss曲线如果训练 loss 持续下降但验证 loss 反弹就是过拟合加 Dropout 或减小 LSTM 单元数。如果两条曲线都下不去可能是 window_size 太小或特征不够。5. 避坑与排查交通流量数据集的五个血泪经验5.1 时间戳时区不统一导致周期特征错乱现象模型在训练集上表现正常但预测早高峰总是偏移一两个小时。原因数据里混了 UTC 和本地时间pd.to_datetime默认不转换时区导致小时特征错位。解决读入后统一加utcTrue再转目标时区或者先检查时间戳范围是否跨了时区切换点。df[timestamp] pd.to_datetime(df[timestamp], utcTrue) df[timestamp] df[timestamp].dt.tz_convert(Asia/Shanghai)5.2 路段编号被自动转成整数丢失前导零现象One-Hot 之后路段数量比实际少因为001和1被合并了。原因read_csv默认推断类型把字符串编号转成了整数。解决读取时指定dtype{road_id: str}或者用converters参数强制字符串。5.3 缺失值填充方式选错引入虚假规律现象模型在夜间时段预测偏高。原因用全局均值填充夜间缺失流量而夜间实际流量接近零。解决按路段和小时分组填充或者用前向填充加标记列。别用全局均值那是给模型喂假数据。5.4 滑动窗口跨路段边界导致特征污染现象验证集误差比训练集高很多且集中在路段切换处。原因构造滞后特征时没按路段分组上一个路段的末尾流量混进了下一个路段的开头。解决所有shift和rolling操作前先groupby(road_id)构造完再dropna。5.5 验证集划分时数据泄露现象验证集 MAE 低得离谱测试集一跑就翻车。原因标准化时在全量数据上fit或者目标编码用了全量均值。解决所有拟合操作只在训练集上做验证集和测试集只做transform。时序数据还要确保切分点之后的数据完全不可见。提示每次跑完模型把验证集和测试集的指标并排看。差距超过 20% 就先查泄露别急着调参。6. 进阶技巧用滚动预测替代单次切分验证单次时间切分有个天然缺陷只验证了一个时间点之后的表现。如果测试期正好赶上节假日指标会失真。更稳的做法是滚动预测——用过去 N 天训练预测下一天然后窗口后移重复多次取平均。def rolling_validation(df, feature_cols, target_col, train_days30, test_days1): results [] dates df[timestamp].dt.date.unique() for i in range(train_days, len(dates) - test_days 1, test_days): train_dates dates[i-train_days:i] test_dates dates[i:itest_days] train_mask df[timestamp].dt.date.isin(train_dates) test_mask df[timestamp].dt.date.isin(test_dates) scaler StandardScaler() X_train scaler.fit_transform(df.loc[train_mask, feature_cols]) X_test scaler.transform(df.loc[test_mask, feature_cols]) model xgb.XGBRegressor(n_estimators300, max_depth6, learning_rate0.05) model.fit(X_train, df.loc[train_mask, target_col]) pred model.predict(X_test) mae mean_absolute_error(df.loc[test_mask, target_col], pred) results.append(mae) return np.mean(results), np.std(results) mean_mae, std_mae rolling_validation(df_model, feature_cols, target_col) print(f滚动验证 MAE: {mean_mae:.2f} ± {std_mae:.2f})这段代码的核心是for循环里的窗口滑动。train_days30表示每次用 30 天数据训练test_days1表示预测下一天。dates去重后按天遍历避免同一天被重复切分。每次循环重新fitscaler 和模型确保没有跨窗口泄露。滚动验证的代价是计算量成倍增加30 天窗口滑 60 次就是 60 次训练。但它的好处是能给出指标的标准差——如果标准差很大说明模型对时间敏感上线后需要频繁更新。我一般会在项目初期跑一次滚动验证确认模型稳定性后再用单次切分做快速迭代。从那以后我每次拿到新的时序数据集都强制先跑一遍滚动验证再决定要不要上深度学习。单次切分的指标太容易骗人滚动验证的均值和方差才是真实水平。希望帮到你。本文还有配套的精品资源点击获取