简介：这份资源面向具备一定Python基础、希望入门交通流预测与拥堵识别的学习者与开发者，围绕GCM走廊真实交通数据，构建从数据清洗到模型训练、测试的完整流程，解决如何利用历史传感器数据提前判断道路拥堵程度的问题。压缩包共7个文件，约9KB，包含3个Python脚本分别承担数据过滤、训练与测试任务，另有2个txt与1个csv提供数据与说明，1个md记录项目说明，结构紧凑、便于快速上手。资源基于855个传感器每5分钟采集的流量、速度、占有率等属性，将拥堵划分为通畅、轻微、中度、重度四类，可帮助读者理解特征工程、时序建模与分类评估的落地方式。目前已有326人学习下载，适合作为课程设计、小型实验或交通预测练手项目的参考模板，也能为后续扩展更复杂模型提供清晰起点。
1. 从一段路的车流说起：Python 做流量与拥堵预测到底在算什么
早高峰你盯着某个路口，车流不是均匀来的，它像潮水，有波峰、有波谷，还有偶发的追尾把整条路堵成停车场。基于 Python 实现道路一段时间内的车辆流量预测与交通拥堵预测，要解决的就是把这种“潮汐感”变成可计算的数字：给定过去若干时间片的过车数、平均车速、占有率，预测未来 15 分钟、30 分钟或 1 小时的车流量，并判断拥堵等级会不会跳变。它适合做交通数据分析的工程师、做智慧路口/信号优化的算法同学，以及想用 Python 把时序预测跑通的新手。整套方案不依赖昂贵设备，常见做法是用地磁、线圈、卡口或公开交通数据集，先做时间片聚合，再上时序模型。下面按“数据怎么来 → 特征怎么造 → 模型怎么选 → 怎么避坑 → 怎么验证”推一遍，能直接抄作业。
2. 数据从哪来、怎么切成模型能吃的形状
2.1 车辆流量预测的输入到底长什么样
做车辆流量预测，第一步不是选模型，而是把原始过车记录变成“时间片 + 流量”的规整表。常见原始数据有两种：一种是卡口过车流水，每辆车一条记录，带时间戳和方向；另一种是线圈/地磁的周期统计，直接给 5 分钟或 15 分钟的计数。无论哪种，最终都要落到同一张宽表：timestamp、road_id、flow、avg_speed、occupancy。其中flow是核心预测目标，avg_speed和occupancy是拥堵预测的关键辅助特征——速度掉、占有率升，往往先于流量饱和出现。
时间片粒度选择有讲究。太细（1 分钟）噪声大、缺失多；太粗（1 小时）丢掉了拥堵的短时突变。我一般用 5 分钟或 15 分钟，和信号配时周期对齐。聚合时用resample而不是简单groupby，因为要保证没有车经过的时间片也保留为 0，否则模型会误以为时间不连续。
import pandas as pd # 原始卡口流水：timestamp 为过车时间，road_id 为路段编号 raw = pd.read_csv("kakou_flow.csv", parse_dates=["timestamp"]) # 按 5 分钟切片，统计每个路段每个时间片的过车数 df = ( raw.set_index("timestamp") .groupby("road_id")["vehicle_id"] .resample("5min") .count() .reset_index() .rename(columns={"vehicle_id": "flow"}) ) # 补齐没有过车的时间片为 0，避免时间轴断裂 full_idx = pd.date_range(df["timestamp"].min(), df["timestamp"].max(), freq="5min") roads = df["road_id"].unique() grid = pd.MultiIndex.from_product([roads, full_idx], names=["road_id", "timestamp"]) df = df.set_index(["road_id", "timestamp"]).reindex(grid, fill_value=0).reset_index()这段代码的逻辑是：先按路段分组再按时间重采样计数，然后用MultiIndex笛卡尔积把“路段 × 完整时间轴”补全，缺失填 0。参数上，resample("5min")的粒度要和后续特征窗口匹配；fill_value=0只对流量成立，如果后面加avg_speed，缺失应填NaN再插值，不能填 0，否则会把“没车”和“堵死”混为一谈。
2.2 把时间戳拆成模型能学的周期特征
原始时间戳对模型没有直接意义，但“星期几”“几点”“是否节假日”对流量影响极大。常见做法是做周期性编码，而不是直接给 0-23 的整数，因为 23 点和 0 点在时间上相邻，整数编码会让模型以为它们很远。用正弦余弦编码可以保留这种循环关系。
import numpy as np def add_cycle_features(df, time_col="timestamp"): df = df.copy() df["hour"] = df[time_col].dt.hour df["minute"] = df[time_col].dt.minute df["dayofweek"] = df[time_col].dt.dayofweek # 一天内的位置，映射到 0~2π pos = (df["hour"] * 60 + df["minute"]) / (24 * 60) df["sin_day"] = np.sin(2 * np.pi * pos) df["cos_day"] = np.cos(2 * np.pi * pos) # 一周内的位置 pos_w = (df["dayofweek"] * 24 * 60 + df["hour"] * 60 + df["minute"]) / (7 * 24 * 60) df["sin_week"] = np.sin(2 * np.pi * pos_w) df["cos_week"] = np.cos(2 * np.pi * pos_w) return df df = add_cycle_features(df)这里sin_day/cos_day刻画一天内的周期，sin_week/cos_week刻画一周内的周期。参数上，分母必须是完整周期长度（一天 1440 分钟、一周 10080 分钟），否则编码会错位。做完这一步，流量预测的输入特征就基本齐了：历史流量滑窗、周期编码、速度/占有率。拥堵预测则在此基础上加一个标签列，比如用速度阈值或拥堵指数把每个时间片标成畅通/缓行/拥堵。
3. 车辆流量预测：从滑窗特征到可复现的时序模型
3.1 用滑动窗口造监督学习样本
时序预测不能直接把整条序列丢给普通回归，要把“过去 N 个时间片”映射到“未来 M 个时间片”。这一步叫滑窗造样本，是车辆流量预测里最容易翻车的地方——窗口错位、未来信息泄漏，都会让离线指标好看、上线就崩。
def make_supervised(series, n_in=12, n_out=3): """series: 单个路段按时间排序的 flow 序列 n_in: 用过去 12 个时间片（5min*12=1小时） n_out: 预测未来 3 个时间片（15分钟） """ X, y = [], [] for i in range(len(series) - n_in - n_out + 1): X.append(series[i : i + n_in]) y.append(series[i + n_in : i + n_in + n_out]) return np.array(X), np.array(y) # 按路段分别造样本，避免不同路段互相污染 X_all, y_all = [], [] for road, g in df.groupby("road_id"): g = g.sort_values("timestamp") X, y = make_supervised(g["flow"].values, n_in=12, n_out=3) X_all.append(X); y_all.append(y) X_all = np.concatenate(X_all); y_all = np.concatenate(y_all)逻辑说明：n_in=12对应过去 1 小时，n_out=3对应未来 15 分钟，这是交通流量预测里比较稳的配置。参数怎么改：如果要做 30 分钟预测，n_out调到 6；如果数据粒度是 15 分钟，n_in=4就是过去 1 小时。关键点是range的边界必须减掉n_in + n_out，否则最后几个样本的y会越界或混入不完整窗口。另外，滑窗必须在每个路段内部独立做，不能跨路段拼接，否则会把 A 路的流量当成 B 路的历史。
3.2 选 LSTM 还是树模型：先看数据量再谈玄学
车辆流量预测的模型选型，常见分两派：树模型（XGBoost/LightGBM）和深度时序模型（LSTM/GRU/Temporal Fusion Transformer）。我的经验是：数据量小于几个月、路段数少，优先树模型，特征工程到位就能打；数据量大、周期复杂、要多步预测，再上 LSTM。树模型对缺失和异常更鲁棒，训练快，调参直观；LSTM 能自动学时序依赖，但需要更多数据和算力，且对归一化敏感。
from sklearn.preprocessing import StandardScaler from tensorflow.keras.models import Sequential from tensorflow.keras.layers import LSTM, Dense, Dropout # 归一化：LSTM 对量纲敏感，必须做 scaler = StandardScaler() X_scaled = scaler.fit_transform(X_all.reshape(-1, n_in)).reshape(X_all.shape) y_scaled = scaler.fit_transform(y_all.reshape(-1, n_out)).reshape(y_all.shape) model = Sequential([ LSTM(64, activation="tanh", input_shape=(n_in, 1), return_sequences=False), Dropout(0.2), Dense(n_out) # 直接输出未来 n_out 个时间片 ]) model.compile(optimizer="adam", loss="mse") model.fit(X_scaled[..., None], y_scaled, epochs=30, batch_size=64, validation_split=0.2)参数说明：LSTM(64)的 64 是隐藏单元数，数据量小就降到 32，避免过拟合；Dropout(0.2)是防过拟合的常规操作；Dense(n_out)让模型一次输出多步，比递归预测更稳。注意X_scaled[..., None]是把二维特征扩成 LSTM 需要的三维(样本, 时间步, 特征)。如果换成树模型，直接把X_all展平成(样本, n_in)喂给XGBRegressor，n_out每个步长单独训一个模型或用多输出回归。归一化必须用训练集拟合scaler，再变换验证集，否则就是典型的数据泄漏，离线 RMSE 会虚低。
4. 交通拥堵预测：把连续流量变成拥堵等级
4.1 拥堵标签怎么定义才不拍脑袋
交通拥堵预测和流量预测最大的区别是：流量是回归，拥堵通常是分类。标签定义直接决定模型有没有用。常见做法有三种：按速度阈值（如平均速度低于 20km/h 判拥堵）、按拥堵指数（流量/通行能力）、按占有率。我一般用“速度 + 占有率”双条件，单看速度会把事故和红灯排队混淆。
def label_congestion(row, speed_th=20, occ_th=0.6): # 速度低且占有率高，才判为拥堵，减少误标 if row["avg_speed"] < speed_th and row["occupancy"] > occ_th: return 2 # 拥堵 elif row["avg_speed"] < 35 or row["occupancy"] > 0.4: return 1 # 缓行 else: return 0 # 畅通 df["congestion"] = df.apply(label_congestion, axis=1)参数上，speed_th=20和occ_th=0.6要按道路等级调：快速路阈值高，主干路低。标签做完要看分布，如果拥堵样本不到 5%，模型会偏向多数类，这时要么调阈值，要么用类别权重。拥堵预测的输入除了历史流量，还要把avg_speed、occupancy的滑窗统计加进去，因为拥堵是状态延续，前一时刻的速度比流量更有指示性。
4.2 用分类模型输出拥堵概率而不是硬标签
拥堵预测更实用的输出是概率，而不是 0/1/2。因为调度人员关心的是“有多大可能堵”，概率可以配合阈值做分级预警。用 LightGBM 或逻辑回归都能做，关键是把时间序列的滑窗特征和周期特征拼在一起。
import lightgbm as lgb from sklearn.model_selection import train_test_split from sklearn.metrics import classification_report # 构造拥堵预测特征：历史流量、速度、占有率的滑窗均值 feat_cols = ["flow", "avg_speed", "occupancy"] for c in feat_cols: for w in [3, 6, 12]: df[f"{c}_mean_{w}"] = df.groupby("road_id")[c].transform( lambda s: s.rolling(w, min_periods=1).mean() ) X = df[[c for c in df.columns if "_mean_" in c] + ["sin_day", "cos_day", "sin_week", "cos_week"]] y = df["congestion"] X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.2, shuffle=False) # 时序不能随机打乱 clf = lgb.LGBMClassifier(n_estimators=300, learning_rate=0.05, num_leaves=31) clf.fit(X_tr, y_tr) print(classification_report(y_te, clf.predict(X_te)))逻辑说明：rolling(w).mean()造的是历史窗口统计，w=3/6/12分别对应 15/30/60 分钟。shuffle=False是时序预测的铁律，随机打乱会让未来信息泄漏到训练集。LGBMClassifier的参数里，num_leaves=31控制复杂度，数据小就降到 15。输出概率用clf.predict_proba(X_te)，再按业务设阈值，比如拥堵概率大于 0.7 触发预警。这一步做完，流量预测和拥堵预测就串起来了：流量预测给未来流量，拥堵预测给未来状态概率，两者可以共用一套滑窗特征。
5. 避坑与排查：流量和拥堵预测里最容易翻车的 5 个点
5.1 现象：离线 RMSE 很低，上线预测全错
原因：滑窗造样本时用了未来信息，比如归一化在全集上拟合、或者rolling默认包含了当前时刻。解决：所有统计特征必须shift(1)后再 rolling，归一化只在训练集拟合。检查方法是把预测结果和真实值按时间画出来，看是不是整体平移了一个时间片。
5.2 现象：模型把凌晨 3 点的流量预测成早高峰
原因：周期特征编码错误，或者时间片补齐时把不同日期的同一时刻混在一起。解决：确认sin_day/cos_day的分母是 1440，sin_week/cos_week的分母是 10080；补齐时间轴时用pd.date_range而不是简单reindex到字符串。
5.3 现象：拥堵预测的召回率极低，几乎不报拥堵
原因：拥堵样本太少，模型被多数类带偏。解决：用class_weight="balanced"或对拥堵样本过采样；同时检查标签阈值是不是太严，speed_th=20在快速路上可能永远触发不了。
5.4 现象：LSTM 训练 loss 不降，或者降了但验证集爆炸
原因：没做归一化，或者batch_size太大导致梯度震荡。解决：先StandardScaler归一化，batch_size从 32 试起，epochs配合EarlyStopping。如果数据量只有几千条，直接换树模型，别硬上 LSTM。
5.5 现象：多路段一起训练，模型对某个路段预测特别差
原因：不同路段的流量量级差异大，模型被大流量路段主导。解决：按路段分别归一化，或者把road_id做 embedding/one-hot 加入特征；更稳的做法是每个路段单独训一个模型，再统一评估。
6. 验证与进阶：用滚动预测和时间序列交叉验证兜住底线
流量和拥堵预测的验证，不能用普通train_test_split。我一般用两种：滚动预测（walk-forward）和时间序列交叉验证。滚动预测是每次用过去数据训，预测下一段，再把真实值并入训练集往前滚。它能模拟上线后的真实更新节奏，指标更可信。
from sklearn.metrics import mean_absolute_error import numpy as np def walk_forward_eval(series, n_in=12, n_out=3, step=12): errors = [] for start in range(0, len(series) - n_in - n_out, step): train = series[: start + n_in] test = series[start + n_in : start + n_in + n_out] # 这里用最简单的历史均值做基线，替换成你的模型即可 pred = np.repeat(train[-n_in:].mean(), n_out) errors.append(mean_absolute_error(test, pred)) return np.mean(errors) # 按路段评估，避免大流量路段掩盖小路段问题 for road, g in df.groupby("road_id"): mae = walk_forward_eval(g.sort_values("timestamp")["flow"].values) print(road, round(mae, 2))这段代码的逻辑是：每次只用当前时刻之前的数据训练，预测未来n_out步，然后窗口向前滑动step。参数上，step控制评估密度，step=12就是每小时评估一次。基线用历史均值，是为了给模型一个最低参照——如果你的 LSTM 连历史均值都打不过，说明特征或训练有问题。进阶方向可以试 Temporal Fusion Transformer 做多步概率预测，或者把天气、节假日作为外生变量加进去。我自己的习惯是：任何时序模型上线前，先跑一遍滚动预测，把每个路段的 MAE 和拥堵召回率列成表，哪个路段拖后腿一目了然。这个方案值不值得做，取决于你有没有稳定的历史数据——有 3 个月以上、粒度 5 分钟的流量和速度，就值得投入；数据太短或缺失太多，先把采集补上再谈模型。希望帮到你。
本文还有配套的精品资源，点击获取
