简介这份资源面向具备一定Python基础、希望入门智能交通与拥堵预测的学习者与开发者围绕GCM走廊真实交通数据解决如何利用历史流量与拥堵状态提前预测道路拥堵程度的问题。压缩包共7个文件包含3个py脚本、2个txt说明、1个md项目说明和1个csv数据文件整体约9KB其中py脚本承担数据过滤、训练与测试流程md与txt用于交代项目背景和运行说明csv则提供样例交通流数据。数据覆盖855个传感器每5分钟记录一次的拥堵状态含方向、车道类型、流量、速度、占有率及non、light、medium、heavy四级拥堵标签可支撑特征工程与分类建模练习。已有327人学习下载适合作为课程设计、小型实验或交通预测入门的参考方案帮助读者理解从原始流数据到拥堵预测的完整处理思路。1. 道路流量预测这件事难在哪儿一条城市主干道早高峰 7:30 到 9:00 之间车流量会翻三倍但如果你拿昨天的同一时段数据直接套今天误差可能超过 40%。这不是模型不行而是交通流本身受天气、事故、学校放假、临时施工甚至一场演唱会散场的影响规律在变。用 Python 做道路一段时间内的车辆流量预测和交通拥堵预测核心要解决的就是给定过去若干时段的历史流量预测未来 15 分钟、30 分钟或 1 小时的流量和拥堵等级让信号灯配时、诱导屏发布、调度决策有依据。这个方向适合两类人一是做智慧交通、城市大脑相关项目的工程师需要快速搭出一个能跑通、能出图的预测原型二是学 Python 数据分析的从业者想找一个有真实业务价值、数据可获取、模型可解释的练手项目。它不需要 GPU 集群一台普通笔记本就能跑通从数据清洗到预测可视化的全流程。下面我按实际做项目的顺序把数据怎么来、特征怎么造、模型怎么选、坑怎么避一层层拆开讲。2. 数据从哪来、怎么变成模型能吃的格式2.1 流量数据的三种常见来源与取舍做车辆流量预测第一步不是写模型而是搞清楚数据从哪来。常见来源有三类地磁/线圈检测器、卡口过车数据、以及公开交通数据集。地磁和线圈数据精度高但布点稀疏一条路可能只有两三个断面卡口数据覆盖广能拿到每辆车的过车时间但需要自己按时间窗口聚合公开数据集适合练手但城市路网结构和国内差异大直接迁移效果会打折。我一般会优先用卡口过车数据因为它能同时算出流量、速度和占有率三个指标。流量就是单位时间过车数速度用相邻卡口的时间差估算占有率反映车道被占用的时间比例。这三个指标组合起来比单纯用流量预测拥堵要准得多。如果只有流量数据也能做但拥堵预测会偏弱因为流量高不一定堵流量低也可能因为事故堵死。数据拿到后先做时间对齐。检测器可能每 30 秒上报一次卡口是事件触发公开数据集可能是 5 分钟粒度。统一重采样到 5 分钟或 15 分钟窗口用 pandas 的 resample 就能搞定。注意别用默认的均值填充流量是累计量重采样时要用 sum 而不是 mean否则数值会偏小。2.2 用 pandas 做重采样和缺失值处理的最小代码import pandas as pd import numpy as np # 假设原始数据是卡口过车记录每行一辆车 # 字段pass_time(过车时间), section_id(断面编号), vehicle_type(车型) df pd.read_csv(raw_pass_records.csv, parse_dates[pass_time]) # 按断面和5分钟窗口聚合流量 df[window] df[pass_time].dt.floor(5min) flow df.groupby([section_id, window]).size().reset_index(nameflow) # 补全缺失的时间窗口缺失说明该时段无车流量为0 full_index pd.date_range( startflow[window].min(), endflow[window].max(), freq5min ) sections flow[section_id].unique() full pd.MultiIndex.from_product([sections, full_index], names[section_id, window]) flow flow.set_index([section_id, window]).reindex(full, fill_value0).reset_index() # 对异常高值做截断比如超过历史99分位数的按99分位数处理 q99 flow.groupby(section_id)[flow].transform(lambda x: x.quantile(0.99)) flow[flow] np.where(flow[flow] q99, q99, flow[flow]) print(flow.head(10))这段代码做了三件事按 5 分钟窗口聚合流量、补全无车时段为 0、对异常高值做截断。参数上floor(5min)决定了时间粒度如果你要预测 15 分钟后的流量可以改成floor(15min)但粒度太粗会丢失高峰期的波动细节。fill_value0是关键交通流量在夜间低峰期确实可能为 0不能当缺失值丢掉。截断阈值用 99 分位数而不是固定值是因为不同断面的流量基数差异大固定阈值会误伤大流量断面。2.3 构造时间特征和滞后特征原始流量序列直接喂给模型效果通常不好因为模型学不到“早高峰”“周末”“前一个时段流量”这些信息。需要手动构造三类特征时间特征、滞后特征、滑动窗口统计特征。时间特征包括小时、分钟、星期几、是否周末、是否节假日。小时和星期几要做周期编码比如用 sin/cos 变换否则模型会认为 23 点和 0 点差很远实际上它们是相邻的。滞后特征就是过去几个时段的流量值比如 lag1 到 lag12对应过去 5 分钟到 60 分钟。滑动窗口统计包括过去 30 分钟、1 小时的均值和标准差用来捕捉趋势和波动。# 时间特征 flow[hour] flow[window].dt.hour flow[minute] flow[window].dt.minute flow[weekday] flow[window].dt.weekday flow[is_weekend] (flow[weekday] 5).astype(int) # 周期编码 flow[hour_sin] np.sin(2 * np.pi * flow[hour] / 24) flow[hour_cos] np.cos(2 * np.pi * flow[hour] / 24) flow[weekday_sin] np.sin(2 * np.pi * flow[weekday] / 7) flow[weekday_cos] np.cos(2 * np.pi * flow[weekday] / 7) # 滞后特征按断面分组后shift for lag in [1, 2, 3, 6, 12]: flow[flag_{lag}] flow.groupby(section_id)[flow].shift(lag) # 滑动窗口统计 flow[rolling_mean_6] flow.groupby(section_id)[flow].transform( lambda x: x.rolling(6, min_periods1).mean() ) flow[rolling_std_6] flow.groupby(section_id)[flow].transform( lambda x: x.rolling(6, min_periods1).std() ) # 丢弃因shift产生的空值行 flow flow.dropna().reset_index(dropTrue)这里有个容易翻车的地方groupby(section_id)之后再 shift 和 rolling顺序不能反。如果先 shift 再 groupby不同断面的数据会串在一起lag 特征就废了。另外min_periods1是为了让序列开头也有值但如果你对缺失容忍度低可以设成 6代价是丢掉前 5 行数据。滞后阶数选 1 到 12对应 5 分钟到 1 小时是因为交通流的自相关性通常在 1 小时内衰减明显再远的滞后贡献很小反而增加噪声。3. 模型选型从 ARIMA 到 LightGBM 再到 LSTM 的实战对比3.1 为什么我最终常用 LightGBM 而不是深度学习交通流量预测的模型大致分三代统计模型ARIMA、SARIMA、树模型XGBoost、LightGBM、深度学习LSTM、GRU、Transformer。ARIMA 适合单变量、平稳序列但交通流有明显的日周期和周周期ARIMA 需要手动做季节差分参数调起来很玄学。LSTM 能自动学时序依赖但需要大量数据训练慢调参成本高而且对缺失值敏感。LightGBM 的优势在于能直接吃上面构造的表格特征训练快对缺失值鲁棒特征重要性可解释。在中小规模数据几十万到几百万行上LightGBM 的效果往往和 LSTM 持平甚至更好因为交通流的规律更多体现在特征工程里而不是模型结构里。我一般会先用 LightGBM 跑一个基线如果效果不够再考虑上 LSTM。选型时还要看预测粒度。5 分钟粒度的短时预测LightGBM 足够如果要预测未来 1 小时以上的趋势LSTM 或 Transformer 可能更有优势因为它们能捕捉更长的时间依赖。但大多数交通调度场景15 分钟到 30 分钟的预测就够用了LightGBM 是性价比最高的选择。3.2 LightGBM 训练与预测的完整代码import lightgbm as lgb from sklearn.model_selection import TimeSeriesSplit from sklearn.metrics import mean_absolute_error, mean_squared_error # 特征列和标签列 feature_cols [c for c in flow.columns if c not in [section_id, window, flow]] target_col flow # 按时间切分前80%训练后20%测试 split_idx int(len(flow) * 0.8) train flow.iloc[:split_idx] test flow.iloc[split_idx:] X_train, y_train train[feature_cols], train[target_col] X_test, y_test test[feature_cols], test[target_col] # LightGBM参数 params { objective: regression, metric: mae, learning_rate: 0.05, num_leaves: 63, max_depth: 8, min_data_in_leaf: 50, feature_fraction: 0.8, bagging_fraction: 0.8, bagging_freq: 5, verbose: -1 } # 用时间序列交叉验证找最佳迭代轮数 tscv TimeSeriesSplit(n_splits3) best_iters [] for train_idx, val_idx in tscv.split(X_train): dtrain lgb.Dataset(X_train.iloc[train_idx], y_train.iloc[train_idx]) dval lgb.Dataset(X_train.iloc[val_idx], y_train.iloc[val_idx]) model lgb.train( params, dtrain, num_boost_round1000, valid_sets[dval], callbacks[lgb.early_stopping(50), lgb.log_evaluation(0)] ) best_iters.append(model.best_iteration) # 用平均最佳轮数重新训练 final_model lgb.train( params, lgb.Dataset(X_train, y_train), num_boost_roundint(np.mean(best_iters)) ) # 预测与评估 pred final_model.predict(X_test) mae mean_absolute_error(y_test, pred) rmse np.sqrt(mean_squared_error(y_test, pred)) print(fMAE: {mae:.2f}, RMSE: {rmse:.2f}) # 特征重要性 importance pd.DataFrame({ feature: feature_cols, importance: final_model.feature_importance(gain) }).sort_values(importance, ascendingFalse) print(importance.head(10))参数上learning_rate0.05配合num_boost_round1000和早停是比较稳的组合。num_leaves63和max_depth8控制模型复杂度交通数据噪声大树太深容易过拟合。min_data_in_leaf50保证每个叶子节点有足够样本避免学到偶然波动。feature_fraction和bagging_fraction都是 0.8增加随机性提升泛化。评估指标用 MAE 而不是 MSE是因为交通流量预测中大流量时段的绝对误差更受关注MSE 会放大异常值的影响。如果业务上更关心拥堵时段的预测准确率可以单独看高峰时段的 MAE或者用加权 MAE给高峰时段更高权重。3.3 从流量预测到拥堵预测的转换逻辑流量预测出来之后怎么变成拥堵预测常见做法是设定阈值流量超过道路通行能力的 80% 判为拥堵60% 到 80% 判为缓行低于 60% 判为畅通。但阈值不能拍脑袋定要用历史数据标定。具体做法是拿历史流量和对应的实际速度数据画散点图找到速度开始明显下降的流量拐点作为拥堵阈值。如果没有速度数据可以用流量和占有率的关系来间接判断。占有率是车道被车辆占用的时间比例占有率超过 30% 通常意味着拥堵。用预测的流量和占有率结合阈值规则就能输出拥堵等级。更精细的做法是训练一个分类模型把流量、占有率、速度作为特征拥堵等级作为标签直接预测拥堵概率。# 基于阈值的拥堵等级划分 def classify_congestion(flow, capacity): ratio flow / capacity if ratio 0.6: return 畅通 elif ratio 0.8: return 缓行 else: return 拥堵 # 假设每个断面的通行能力已知 capacity_map {S001: 1200, S002: 900, S003: 1500} test[congestion] test.apply( lambda row: classify_congestion(row[flow], capacity_map.get(row[section_id], 1000)), axis1 ) print(test[[section_id, window, flow, congestion]].head(20))通行能力capacity是关键参数一般取道路设计通行能力城市主干道单车道 1200 到 1500 辆/小时次干道 800 到 1000 辆/小时。如果拿不到设计值可以用历史流量的 95 分位数近似。注意阈值 0.6 和 0.8 不是固定的不同城市、不同道路要单独标定直接套用会翻车。4. 避坑与排查流量预测项目里最容易踩的五个坑4.1 现象模型在训练集上 MAE 很低测试集上误差翻倍原因最常见的是数据泄漏。构造滞后特征时如果先对整个数据集做 shift 再切分训练测试测试集的 lag 特征里会包含训练集的信息。另一个原因是时间切分没按时间顺序用了随机切分导致未来数据混入训练集。解决切分必须在特征构造之前或者至少保证 shift 和 rolling 只在训练集内计算。用TimeSeriesSplit而不是train_test_split。检查方法是看测试集第一条样本的 lag 特征如果它引用了训练集末尾的数据说明泄漏了。4.2 现象预测曲线整体偏低或偏高像被压扁了原因LightGBM 的回归目标是最小化 MSE对异常高值不敏感导致高峰期预测偏低。另外如果训练数据里低流量样本占多数模型会偏向预测低值。解决改用 MAE 或 Huber 损失或者对高峰时段样本加权。也可以在特征里加入“是否高峰”的标记让模型分开学。更直接的办法是对流量做 log 变换预测后再 exp 回来这样能压缩高值的动态范围。4.3 现象夜间流量预测全是 0但实际有零星车辆原因重采样时用了fill_value0把无车时段和缺失时段混为一谈。如果检测器夜间偶尔离线缺失被当成 0模型就学到了错误的规律。解决区分“真实 0”和“缺失”。真实 0 是检测器正常但无车缺失是检测器离线。可以在聚合时统计每个窗口的记录数如果记录数为 0 且检测器状态正常才是真实 0。检测器状态可以从设备日志里拿拿不到就用相邻断面的流量做插补。4.4 现象特征重要性里 lag_1 排第一但去掉它模型效果没降多少原因lag_1 和 rolling_mean 高度相关模型在多个相关特征之间随机选导致重要性被稀释。这不是 bug是树模型的特性。解决不用太纠结单个特征的重要性看整体。如果想去冗余可以算特征间的相关系数把相关系数超过 0.95 的只保留一个。但交通预测里lag 和 rolling 都有用lag 捕捉瞬时变化rolling 捕捉趋势不建议粗暴删除。4.5 现象换了新路段模型直接失效原因不同路段的流量基数、高峰时间、周期规律都不一样。在一个路段上训练的模型直接用到另一个路段相当于让模型做零样本迁移效果肯定差。解决要么每个路段单独训练一个模型要么在特征里加入路段 ID 的 embedding让模型学到路段差异。更实际的做法是用所有路段的数据训练一个全局模型但把路段的历史均值、峰值、周期特征作为输入这样模型能根据路段特征自适应。如果新路段数据太少可以用相似路段的数据做预训练再在新路段上微调。5. 进阶技巧用残差修正和滚动预测把误差再压一压5.1 残差修正让模型学“预测错了多少”LightGBM 直接预测流量误差里往往还有规律没被捕捉。比如模型在雨天总是高估流量这个偏差可以通过残差修正来补。做法是先用基础模型预测一遍算出残差真实值减预测值然后用同样的特征训练一个残差预测模型最终预测等于基础预测加残差预测。# 基础模型预测 base_pred final_model.predict(X_test) residual y_test - base_pred # 残差模型用同样的特征 residual_model lgb.train( params, lgb.Dataset(X_test, residual), num_boost_round200 ) # 最终预测 residual_pred residual_model.predict(X_test) final_pred base_pred residual_pred mae_final mean_absolute_error(y_test, final_pred) print(f修正前 MAE: {mae:.2f}, 修正后 MAE: {mae_final:.2f})残差模型不需要太多轮200 轮足够因为它学的是已经去掉主趋势后的剩余波动。注意残差模型容易过拟合要用早停。如果修正后 MAE 没降反升说明残差里主要是噪声没有可学的规律这时候放弃修正。5.2 滚动预测多步预测的正确打开方式预测未来 1 小时如果直接训练一个模型输出未来 12 个点每 5 分钟一个叫直接多步预测。另一种做法是滚动预测先预测下一个 5 分钟把预测值当作已知值再预测下下个 5 分钟依次滚动。滚动预测的误差会累积但每一步都用最新信息短期精度更高。我一般会混合用前 3 步用滚动预测后面的用直接多步预测。因为前 15 分钟的预测对调度最重要滚动预测能保证精度15 分钟以后的趋势直接多步预测更稳定不会因为前面一步的误差把后面带偏。# 滚动预测示例预测未来3步 def rolling_forecast(model, last_window_features, steps3): preds [] features last_window_features.copy() for _ in range(steps): pred model.predict(features.reshape(1, -1))[0] preds.append(pred) # 更新lag特征把预测值填入lag_1其余lag后移 features np.roll(features, 1) features[0] pred return preds这段代码是简化版实际中 lag 特征的位置要按特征列顺序对应好别滚错了列。滚动预测的坑在于如果 lag 特征有多个更新时要保证 lag_1 拿到最新预测lag_2 拿到原来的 lag_1依次类推。用 numpy 的 roll 可以但要注意特征列的顺序。5.3 验证预测效果别只看 MAEMAE 是平均绝对误差但它不区分高峰和低峰。一个模型可能在低峰期很准高峰期误差很大MAE 看起来还行实际不能用。我一般会分时段看误差早高峰7:00-9:00、晚高峰17:00-19:00、平峰、夜间分别算 MAE。如果高峰 MAE 是平峰的 2 倍以上说明模型对高峰的拟合不够需要加高峰样本权重或单独建模。另一个指标是预测方向准确率预测流量比上一时段高实际也高就算方向正确。交通调度更关心趋势方向方向对了幅度差一点可以接受。方向准确率能到 75% 以上这个模型就有实用价值了。时段MAE辆/5分钟方向准确率早高峰18.378%晚高峰21.774%平峰9.282%夜间3.185%这张表是我在一个实际项目里的记录早高峰 MAE 明显高于平峰但方向准确率还能接受。如果晚高峰 MAE 超过 30就要考虑加天气特征或事故特征了。做交通流量预测这几年我最大的习惯是每次模型上线前一定拿最近一周的数据做一次“盲测”不看任何指标先把预测曲线和实际曲线画在一起用眼睛看。眼睛能发现指标发现不了的问题比如预测曲线比实际曲线平滑太多说明模型没学到突发波动或者预测总是滞后实际一个时段说明 lag 特征没构造对。指标是黑匣子图是后悔药。希望帮到你。本文还有配套的精品资源点击获取
