基于Python的道路流量预测与交通拥堵预测实战指南

简介：这份资源面向具备一定Python基础、希望入门交通流预测与拥堵识别的学习者与开发者，围绕GCM走廊真实交通数据展开，解决如何利用历史传感器数据提前判断道路拥堵程度的问题。压缩包共7个文件，包含3个py脚本、2个txt说明、1个md项目说明和1个csv数据文件，整体约9KB，其中py脚本承担数据过滤、训练与测试流程，md与txt用于交代项目背景和运行方式，csv则提供样例数据支撑。数据覆盖855个传感器每5分钟记录一次的交通状态，含流量、速度、占有率及non、light、medium、heavy四级拥堵标签，可支撑从特征处理到模型训练再到结果验证的完整链路。目前已有327人学习下载，适合作为课程设计、小型实验或交通预测入门的参考方案，帮助读者理解数据清洗、特征构造与拥堵分级预测的基本思路。

1. 从一份 zip 说起：道路流量预测到底在预测什么

拿到「基于python实现道路一段时间内的车辆流量预测交通拥堵预测+项目说明.zip」这个标题，多数人第一反应是去找代码跑起来，但真正决定这个项目能不能落地的，是搞清楚它预测的到底是什么量。道路流量预测不是算命，它输入的是历史时间序列——某条路段每 5 分钟、15 分钟或 1 小时通过的车流量，输出的是未来一段时间同一路段的流量数值，再由流量数值反推拥堵等级。这个「一段时间」是核心参数：预测未来 15 分钟叫短时预测，用于信号灯配时；预测未来 1 小时叫短中期预测，用于路径诱导；预测未来一天叫长期预测，用于路网规划。三者用的模型、特征、评价指标完全不同，一份 zip 里通常只覆盖其中一种，跑之前先确认它站在哪个时间尺度上。

这个方向适合三类人：做智慧交通的算法工程师、拿交通数据做课程设计的学生、以及想用 python 把结构化数据预测跑通的数据从业者。它不需要 GPU 集群，一台普通笔记本就能复现，但坑集中在数据清洗和特征构造上，而不是模型本身。下面按「数据怎么来 → 特征怎么造 → 模型怎么选 → 怎么评估 → 怎么避坑」的顺序拆开讲，每一步都给可抄的命令和参数。

2. 数据从哪来：流量预测的输入层怎么搭

2.1 三类常见数据源与字段结构

车辆流量预测的数据源决定了整个项目的上限。常见做法有三类：一是公开交通数据集，字段通常是「路段 ID + 时间戳 + 流量 + 占有率 + 平均速度」；二是地磁或线圈检测器导出数据，字段是「检测器编号 + 采集时间 + 车流量 + 车道号」；三是导航平台开放的路况接口，字段是「路段 + 时间片 + 拥堵指数 + 旅行时间」。这三类的共同点是都带时间戳和路段标识，区别在于采样粒度和是否有速度字段。

选数据源时优先看两点：时间戳是否连续、路段是否固定。如果时间戳有大量缺失，后面做滑动窗口会直接崩掉；如果路段 ID 每天变，就没法构造稳定的序列样本。我一般会先统计每个路段的记录条数和时间跨度，只保留覆盖率超过 90% 的路段，这一步能砍掉一半脏数据。

2.2 用 pandas 做时间序列对齐与缺失填充

拿到原始表后，第一件事是把时间戳转成统一格式并按路段分组重采样。下面这段代码做三件事：解析时间、按「路段 + 时间」聚合、对缺失时间片做前向填充加线性插值。

import pandas as pd import numpy as np # 读取原始数据，假设字段为 road_id, ts, flow df = pd.read_csv("raw_flow.csv") df["ts"] = pd.to_datetime(df["ts"], format="%Y-%m-%d %H:%M:%S") # 按路段和时间聚合，同一时间片多条记录取均值 df = df.groupby(["road_id", "ts"], as_index=False)["flow"].mean() # 对每个路段重采样到 15 分钟粒度，缺失值先标记 df = df.set_index("ts").groupby("road_id").resample("15min").mean().reset_index() # 前向填充 + 线性插值，限制连续填充不超过 4 个点 df["flow"] = df.groupby("road_id")["flow"].transform( lambda s: s.ffill(limit=4).interpolate(method="linear") ) # 丢弃仍然缺失的记录 df = df.dropna(subset=["flow"]) print(df.groupby("road_id")["flow"].describe())

逻辑说明：groupby + resample是时间序列对齐的标准组合，resample("15min")里的粒度要和你的预测目标一致，预测 15 分钟就重采样到 15 分钟。ffill(limit=4)表示最多连续填 4 个缺失点，超过就放弃，避免用太久之前的值污染序列。interpolate(method="linear")在两端缺失时不起作用，所以最后还要dropna。参数上，limit 设太大（比如 12）会让模型学到虚假的平稳段，设太小又浪费数据，4 是我在 15 分钟粒度下比较稳的经验值。

2.3 构造监督学习样本：滑动窗口的窗口长度怎么定

时间序列预测要把序列切成「输入窗口 → 输出窗口」的样本对。窗口长度不是拍脑袋定的，要看数据的自相关函数。下面代码先算自相关，再按结果切样本。

from statsmodels.tsa.stattools import acf # 取一条路段的流量序列算自相关 one_road = df[df["road_id"] == df["road_id"].iloc[0]]["flow"].values acf_vals = acf(one_road, nlags=48) # 找到自相关首次跌破 0.5 的滞后阶数 lag_cut = np.argmax(acf_vals < 0.5) print("建议输入窗口长度:", lag_cut) def make_samples(series, in_len, out_len): X, y = [], [] for i in range(len(series) - in_len - out_len + 1): X.append(series[i:i + in_len]) y.append(series[i + in_len:i + in_len + out_len]) return np.array(X), np.array(y) X, y = make_samples(one_road, in_len=lag_cut, out_len=4) print(X.shape, y.shape)

逻辑说明：acf算的是序列和自身滞后版本的相关系数，跌破 0.5 说明再往前的值和当前值关系已经不强，拿它当输入窗口上限比较合理。out_len=4表示预测未来 4 个时间片，15 分钟粒度下就是未来 1 小时。参数上，输入窗口太短模型看不到周期，太长会引入噪声且增加计算量；输出窗口太长会让误差累积，短时预测一般不超过 8 个片。这一步做完，X和y就是可以直接喂给模型的数组，注意要按时间顺序划分训练集和测试集，不能随机打乱。

3. 模型怎么选：从统计方法到深度模型的取舍

3.1 基线模型：历史均值与 ARIMA 先跑通

任何预测项目都应该先有一个基线，否则你无法判断复杂模型到底有没有用。最简单的基线是「用上周同一时刻的流量作为预测值」，代码只有几行：

# 历史同期均值基线：用前 7 天同一时间片的均值预测 df["weekday"] = df["ts"].dt.weekday df["slot"] = df["ts"].dt.hour * 4 + df["ts"].dt.minute // 15 baseline = df.groupby(["road_id", "weekday", "slot"])["flow"].transform("mean") df["baseline_pred"] = baseline.shift(7 * 96) # 往前推 7 天

逻辑说明：slot把一天切成 96 个 15 分钟片，shift(7 * 96)表示用 7 天前同一时间片的值。这个基线在流量周期性强的路段上能拿到不错的 MAE，如果复杂模型连它都打不过，说明特征或训练有问题。ARIMA 作为统计方法代表，适合单路段、数据量小的场景，order=(p,d,q)里 p 看 ACF 截尾、q 看 PACF 截尾，d 用差分次数，一般流量数据做一阶差分就平稳。

3.2 树模型与深度模型的适用边界

树模型（XGBoost、LightGBM）在流量预测里被低估了。把时间特征（小时、星期、是否节假日）、滞后特征（前 1、2、4、96 个时间片的流量）、滑动统计量（过去 1 小时均值、方差）拼成一张宽表，树模型往往能逼近深度模型的效果，而且训练快、可解释。深度模型里，LSTM 和 GRU 适合捕捉长依赖，TCN 和 Transformer 适合并行训练和更长序列，但数据量少于几万条时深度模型容易过拟合。

选型建议：先用基线定标，再用 LightGBM 做主力，如果数据量足够且追求极致精度再上深度模型。不要一上来就 LSTM，调参成本高且容易在验证集上过拟合。

3.3 用 LightGBM 跑通完整训练与预测

下面代码把特征工程和训练串起来，重点是滞后特征的构造方式。

import lightgbm as lgb from sklearn.metrics import mean_absolute_error def build_features(data, lags=(1, 2, 4, 96)): data = data.sort_values("ts").copy() for lag in lags: data[f"lag_{lag}"] = data["flow"].shift(lag) data["hour"] = data["ts"].dt.hour data["weekday"] = data["ts"].dt.weekday data["is_weekend"] = (data["weekday"] >= 5).astype(int) data["roll_mean_4"] = data["flow"].shift(1).rolling(4).mean() data["roll_std_4"] = data["flow"].shift(1).rolling(4).std() return data.dropna() feat_df = df.groupby("road_id", group_keys=False).apply(build_features) feature_cols = [c for c in feat_df.columns if c.startswith(("lag_", "roll_", "hour", "weekday", "is_weekend"))] split = int(len(feat_df) * 0.8) train, test = feat_df.iloc[:split], feat_df.iloc[split:] model = lgb.LGBMRegressor( n_estimators=800, learning_rate=0.05, num_leaves=63, min_child_samples=20, subsample=0.8, colsample_bytree=0.8 ) model.fit(train[feature_cols], train["flow"]) pred = model.predict(test[feature_cols]) print("MAE:", mean_absolute_error(test["flow"], pred))

逻辑说明：shift(lag)构造滞后特征时必须在排序后做，否则会跨路段串数据，所以用groupby("road_id").apply保证每个路段独立。roll_mean_4和roll_std_4先shift(1)再滚动，避免用到当前时刻的值造成标签泄漏。参数上，num_leaves=63控制树复杂度，流量数据噪声大，叶子数不宜超过 127；min_child_samples=20防止过拟合到个别异常时间片；learning_rate=0.05配n_estimators=800是精度和训练时间的平衡点。跑完看 MAE，如果比基线还差，先检查滞后特征有没有泄漏。

4. 从流量到拥堵：预测结果怎么映射成拥堵等级

4.1 流量-速度-拥堵指数的换算逻辑

流量预测本身不直接等于拥堵预测。拥堵的本质是速度下降，而速度和流量在交通流理论里是「基本图」关系：流量低时速度高，流量升到某个临界点后速度骤降，再往上流量反而下降（拥堵态）。所以拿到预测流量后，要结合路段通行能力做映射。常见做法是设定阈值：预测流量 / 通行能力 < 0.6 为畅通，0.6～0.8 为缓行，0.8～1.0 为拥堵，超过 1.0 为严重拥堵。通行能力可以从路段车道数和限速估算，也可以从历史数据里取流量 95 分位数作为近似。

# 用历史 95 分位数近似通行能力，映射拥堵等级 cap = df.groupby("road_id")["flow"].quantile(0.95).rename("capacity") df = df.merge(cap, on="road_id") df["ratio"] = df["flow"] / df["capacity"] df["congestion"] = pd.cut(df["ratio"], bins=[0, 0.6, 0.8, 1.0, np.inf], labels=["畅通", "缓行", "拥堵", "严重拥堵"])

逻辑说明：quantile(0.95)取每个路段流量的 95 分位作为通行能力代理值，比用固定值更适应不同路段。pd.cut的 bins 边界可以根据实际路网调整，城市快速路和支路的标准不一样。这一步做完，流量预测的连续值就变成了拥堵等级，方便做分类评估和可视化。

4.2 拥堵预测的评价指标怎么选

流量预测看 MAE、RMSE、MAPE，拥堵预测看准确率、召回率和 F1。注意拥堵样本通常不均衡，畅通占大多数，如果只看准确率，全预测畅通也能拿高分。所以要用加权 F1 或对拥堵类单独看召回。下面代码输出分类报告：

from sklearn.metrics import classification_report # test 集上的拥堵等级对比 print(classification_report(test["congestion"], test["congestion_pred"], digits=3))

逻辑说明：classification_report会给出每类的 precision、recall、f1-score，重点看「拥堵」和「严重拥堵」两行的 recall，如果低于 0.5 说明模型漏报严重，需要调整阈值或增加拥堵样本权重。参数上，digits=3只是显示精度，不影响计算。如果要做业务报警，宁可召回高一点、精确率低一点，漏报拥堵比误报代价大。

5. 避坑与排查：跑这个项目最容易翻车的 5 个地方

5.1 时间戳时区不统一导致序列错位

现象：重采样后每个路段的时间片数量不一致，有的多出几行。原因：原始数据里混了本地时间和 UTC 时间，或者夏令时切换。解决：统一用pd.to_datetime(..., utc=True)转成 UTC 再转本地时区，或者直接tz_localize(None)去掉时区信息，保证全表一致。

5.2 滞后特征用了未来值造成标签泄漏

现象：训练集 MAE 极低，测试集 MAE 暴涨。原因：构造lag或滚动统计时没有shift，把当前时刻的流量也放进了特征。解决：所有基于流量的特征必须先shift(1)再计算，滚动窗口的rolling也要在shift之后做。检查方法很简单，看特征和标签的相关系数，如果超过 0.99 基本就是泄漏。

5.3 按随机方式划分训练测试集

现象：模型在测试集上表现好得离谱。原因：时间序列随机打乱后，测试集的相邻时间片出现在训练集里，模型等于抄答案。解决：严格按时间顺序切分，前 80% 训练、后 20% 测试，或者用滚动窗口做时间序列交叉验证。

5.4 缺失值填充过度导致模型学到假平稳

现象：预测曲线过于平滑，跟不上真实流量的突变。原因：ffill或interpolate填了太多连续缺失点。解决：限制填充长度，连续缺失超过 4 个时间片就丢弃该段，或者用同期均值填充而不是前向填充。填充后画一下流量曲线，如果某段明显是直线，就是填过头了。

5.5 路段 ID 变化导致样本错位

现象：groupby后样本数量对不上，或者某些路段预测全是同一个值。原因：原始数据里同一路段用了不同编码，或者路段 ID 含空格、大小写不一致。解决：读数据后先df["road_id"] = df["road_id"].str.strip().str.upper()做标准化，再统计每个 ID 的记录数，把记录数过少的 ID 合并或剔除。

6. 进阶技巧：用滚动预测和误差反馈把精度再压一档

单步预测跑通后，真正影响落地效果的是多步滚动预测。做法是：先用模型预测未来 1 个时间片，把这个预测值当作已知值拼回输入序列，再预测下一个时间片，如此滚动。这样误差会累积，但比一次性输出多步更贴近真实业务场景。下面是一个滚动预测的骨架：

def rolling_forecast(model, history, feature_cols, steps=4): preds = [] seq = list(history) for _ in range(steps): feat = build_single_feature(seq) # 用当前序列构造一行特征 p = model.predict(feat[feature_cols])[0] preds.append(p) seq.append(p) # 把预测值拼回序列 return preds

逻辑说明：build_single_feature要复用训练时的特征构造逻辑，保证滞后和滚动统计的口径一致。seq.append(p)是滚动预测的关键，用预测值替代真实值继续往下推。参数上，steps不要超过 8，再多误差会失控。为了抑制误差累积，可以每隔几步用真实值校正一次，这叫「滚动更新」，在实时系统里很常见。

另一个技巧是误差反馈：把上一轮的预测误差作为特征加进下一轮训练。具体做法是训练一个残差模型，输入是原始特征加上上一轮预测值，输出是真实值和预测值的差，最后把两个模型的结果相加。这个方法在流量预测比赛里经常能再降 5%～10% 的 MAE，代价是训练和推理都多一步。

验证方法上，不要只看一个路段的 MAE，要按路段分组统计，看哪些路段误差大。通常误差大的路段是流量突变频繁、事件多的路段，这类路段可以考虑单独建模或者加入事件特征。我自己的习惯是每次调完参，先把每个路段的 MAE 排序，取最差的 5 个路段画预测对比图，肉眼看一下是滞后、幅度不对还是完全跑偏，比盯着一个总指标有用得多。这套流程跑下来，一份 zip 里的代码能不能用、该怎么改，心里就有数了。希望帮到你。

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