简介：这份资源面向具备一定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(name="flow") # 补全缺失的时间窗口，缺失说明该时段无车，流量为0 full_index = pd.date_range( start=flow["window"].min(), end=flow["window"].max(), freq="5min" ) 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_value=0).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_value=0是关键，交通流量在夜间低峰期确实可能为 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[f"lag_{lag}"] = flow.groupby("section_id")["flow"].shift(lag) # 滑动窗口统计 flow["rolling_mean_6"] = flow.groupby("section_id")["flow"].transform( lambda x: x.rolling(6, min_periods=1).mean() ) flow["rolling_std_6"] = flow.groupby("section_id")["flow"].transform( lambda x: x.rolling(6, min_periods=1).std() ) # 丢弃因shift产生的空值行 flow = flow.dropna().reset_index(drop=True)

这里有个容易翻车的地方：groupby("section_id")之后再 shift 和 rolling，顺序不能反。如果先 shift 再 groupby，不同断面的数据会串在一起，lag 特征就废了。另外min_periods=1是为了让序列开头也有值，但如果你对缺失容忍度低，可以设成 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_splits=3) 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_round=1000, 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_round=int(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(f"MAE: {mae:.2f}, RMSE: {rmse:.2f}") # 特征重要性 importance = pd.DataFrame({ "feature": feature_cols, "importance": final_model.feature_importance("gain") }).sort_values("importance", ascending=False) print(importance.head(10))

参数上，learning_rate=0.05配合num_boost_round=1000和早停，是比较稳的组合。num_leaves=63和max_depth=8控制模型复杂度，交通数据噪声大，树太深容易过拟合。min_data_in_leaf=50保证每个叶子节点有足够样本，避免学到偶然波动。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)), axis=1 ) 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_value=0，把无车时段和缺失时段混为一谈。如果检测器夜间偶尔离线，缺失被当成 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_round=200 ) # 最终预测 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, steps=3): 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 验证预测效果：别只看 MAE

MAE 是平均绝对误差，但它不区分高峰和低峰。一个模型可能在低峰期很准，高峰期误差很大，MAE 看起来还行，实际不能用。我一般会分时段看误差：早高峰（7:00-9:00）、晚高峰（17:00-19:00）、平峰、夜间，分别算 MAE。如果高峰 MAE 是平峰的 2 倍以上，说明模型对高峰的拟合不够，需要加高峰样本权重或单独建模。

另一个指标是预测方向准确率：预测流量比上一时段高，实际也高，就算方向正确。交通调度更关心趋势方向，方向对了，幅度差一点可以接受。方向准确率能到 75% 以上，这个模型就有实用价值了。

这张表是我在一个实际项目里的记录，早高峰 MAE 明显高于平峰，但方向准确率还能接受。如果晚高峰 MAE 超过 30，就要考虑加天气特征或事故特征了。

做交通流量预测这几年，我最大的习惯是：每次模型上线前，一定拿最近一周的数据做一次“盲测”，不看任何指标，先把预测曲线和实际曲线画在一起，用眼睛看。眼睛能发现指标发现不了的问题，比如预测曲线比实际曲线平滑太多，说明模型没学到突发波动；或者预测总是滞后实际一个时段，说明 lag 特征没构造对。指标是黑匣子，图是后悔药。希望帮到你。

本文还有配套的精品资源，点击获取