简介：这份资源面向具备一定Python基础、希望入门智能交通与数据挖掘的学习者，围绕道路短时车流量与拥堵状态预测展开。项目基于GCM Corridor真实交通数据，覆盖16座城镇主干道、855个传感器每5分钟采集的拥堵记录，包含日期、方向、路段、流量、速度、占有率及non/light/medium/heavy四级拥堵标签，可用于构建时序预测与分类模型。压缩包共7个文件，含3个py脚本、2个txt说明、1个md文档和1个csv数据，整体约9KB，脚本大致对应数据过滤、预处理、训练与测试流程，文档用于说明项目结构与运行方式。目前已有326人学习下载。读者可借此掌握交通流特征工程、拥堵等级建模与预测评估的完整思路，并直接复用脚本与数据快速复现实验，适合课程设计、竞赛练手或交通数据分析入门参考。
1. 从 855 个传感器到拥堵等级：这套 Python 交通预测项目能跑出什么
早高峰出门前刷一眼导航，红色路段已经堵成一条线——这种场景下，如果提前 30 分钟知道某条主干道会从中度拥堵滑向重度拥堵，绕行决策的价值就出来了。这个项目干的就是这件事：用 Python 把 GCM Corridor 上 855 个传感器每 5 分钟采集一次的交通流数据，整理成可训练的时序样本，再预测未来一段时间内路段的车辆流量和拥堵等级。数据里每条记录包含 date、time、direction、type、linkID、length、travelTime、volume、speed、occupancy、congestionLevel 十一个字段，拥堵状态分 non、light、medium、heavy 四档，一天 288 条。项目包里有filtrate_sensor.py、process.py、train.py、test.csv、train、test和readme.md，结构不复杂，但把「原始流数据 → 清洗 → 特征 → 模型 → 预测」这条链路走通了。适合正在做交通时序预测、想找一个能直接跑通 baseline 的从业者，也适合刚学完 Python 数据分析、想拿真实数据练手的人。
2. 数据管道拆解：filtrate_sensor.py 与 process.py 到底在做什么
2.1 原始数据的三个脏点
GCM 数据看着规整，实际用起来有三个绕不开的问题。第一，855 个传感器不是每个都全天在线，缺失段直接断在时间序列里；第二，congestionLevel 是字符串标签，non/light/medium/heavy 不能直接喂给模型；第三，volume、speed、occupancy 量纲差异大，volume 动辄几百，occupancy 可能只有零点几，不归一化会让基于距离的模型直接偏掉。
filtrate_sensor.py处理的是第一类问题。它按 linkID 分组，检查每个传感器在一天 288 个时间片上的记录数，低于阈值的整段丢弃或插值。常见做法是设一个 80% 完整度阈值，低于就弃用该传感器当天的数据，而不是硬插——交通流有强周期性，乱插值比缺数据更危险。
process.py处理第二、三类问题。它把拥堵等级映射成 0/1/2/3，同时对数值列做标准化。这里有个细节：标准化参数必须只在训练集上 fit，再 transform 到测试集，否则就是典型的数据泄漏。很多新手在这一步翻车，线下指标漂亮，上线一塌糊涂。
2.2 清洗脚本的关键代码与参数
下面这段是我按项目结构补全的清洗逻辑，核心思路和filtrate_sensor.py一致：
import pandas as pd import numpy as np # 读取原始传感器流数据 df = pd.read_csv("train.csv") # 统一时间字段，构造 5 分钟粒度的时间索引 df["datetime"] = pd.to_datetime(df["date"].astype(str) + df["time"].astype(str).str.zfill(4), format="%Y%m%d%H%M", errors="coerce") # 按 linkID 分组，统计每个传感器当天的有效记录数 completeness = df.groupby(["linkID", df["datetime"].dt.date]).size() valid_links = completeness[completeness >= 288 * 0.8].index.get_level_values(0).unique() # 只保留完整度达标的传感器 df = df[df["linkID"].isin(valid_links)].copy() # 拥堵等级映射为有序整数 level_map = {"non": 0, "light": 1, "medium": 2, "heavy": 3} df["congestion_label"] = df["congestionLevel"].map(level_map) # 数值列缺失用前向填充，再兜底为 0 num_cols = ["volume", "speed", "occupancy", "travelTime"] df[num_cols] = df.groupby("linkID")[num_cols].ffill().fillna(0) print(f"有效传感器数: {len(valid_links)}, 剩余记录: {len(df)}")逻辑说明：先构造 datetime 是为了按天统计完整度，str.zfill(4)是因为原始 time 字段可能是0000这种四位格式，直接转字符串会丢前导零。完整度阈值 0.8 是我一般会用的起点，数据质量差可以降到 0.6，但别低于 0.5，否则一天里一半是插出来的，模型学不到真实模式。ffill按 linkID 分组做，避免跨传感器串数据。
参数说明：288 * 0.8里的 288 是一天 5 分钟粒度的理论条数，这个数来自数据描述，不要改。level_map的顺序不能乱，non 到 heavy 是有序的，如果当成无序类别用 one-hot，会丢掉「中度比轻微更堵」这个信息，树模型还好，线性模型会吃亏。
2.3 特征工程：从单条流到时间窗样本
预测「一段时间内的流量」，不能只拿当前时刻的特征。process.py里通常会构造滑动窗口，比如用过去 12 个时间片（1 小时）的 volume、speed、occupancy 预测未来 3 个时间片（15 分钟）的流量和拥堵等级。这一步决定了模型能不能学到趋势。
def make_windows(data, lookback=12, horizon=3): X, y_vol, y_cong = [], [], [] for link, g in data.groupby("linkID"): g = g.sort_values("datetime").reset_index(drop=True) vals = g[["volume", "speed", "occupancy"]].values labels = g["congestion_label"].values for i in range(len(g) - lookback - horizon + 1): X.append(vals[i:i + lookback]) y_vol.append(vals[i + lookback:i + lookback + horizon, 0]) # volume 在第 0 列 y_cong.append(labels[i + lookback + horizon - 1]) return np.array(X), np.array(y_vol), np.array(y_cong)逻辑说明：按 linkID 分组再滑窗，保证窗口不跨传感器。lookback=12对应 1 小时历史，horizon=3对应预测未来 15 分钟。y_vol 取未来 horizon 个时间片的 volume，y_cong 取窗口末端那一档拥堵等级，做单标签分类。
参数说明：lookback 和 horizon 是最需要调的两个参数。lookback 太短学不到周期，太长引入噪声且显存吃紧；horizon 太长预测精度会掉，交通流 15 分钟内的可预测性明显好于 1 小时。我一般从 12 和 3 起步，再按验证集表现微调。
3. train.py 建模实战：流量回归与拥堵分类怎么选模型
3.1 两个任务分开建模的理由
流量预测是回归，拥堵等级预测是分类，硬塞进一个多任务模型不是不行，但调试成本高。项目里train.py更可能是分开训两个模型，或者用同一个特征集分别喂给回归头和分类头。分开的好处是评估指标清晰：回归看 MAE/RMSE，分类看 accuracy 和 macro-F1。拥堵四类不平衡是常态，non 和 light 占大头，heavy 样本少，只看 accuracy 会被 majority class 带偏，macro-F1 更能反映对少数类的识别能力。
模型选型上，baseline 用 LightGBM 或 XGBoost 就够，把滑窗展平成二维特征即可。想上时序结构，LSTM 或 GRU 是常见选择，输入形状(batch, lookback, features)。如果数据量不大，别一上来就 Transformer，调参成本高，收益未必明显。
3.2 训练脚本骨架与评估
import lightgbm as lgb from sklearn.metrics import mean_absolute_error, f1_score from sklearn.preprocessing import StandardScaler # X_train: (N, lookback, 3) -> 展平为 (N, lookback*3) X_tr = X_train.reshape(len(X_train), -1) X_te = X_test.reshape(len(X_test), -1) # 标准化只在训练集 fit scaler = StandardScaler().fit(X_tr) X_tr, X_te = scaler.transform(X_tr), scaler.transform(X_te) # 流量回归 reg = lgb.LGBMRegressor(n_estimators=500, learning_rate=0.05, num_leaves=63) reg.fit(X_tr, y_vol_train.reshape(len(y_vol_train), -1)[:, 0]) pred_vol = reg.predict(X_te) print("Volume MAE:", mean_absolute_error(y_vol_test[:, 0], pred_vol)) # 拥堵分类 clf = lgb.LGBMClassifier(n_estimators=500, learning_rate=0.05, num_leaves=63, class_weight="balanced") clf.fit(X_tr, y_cong_train) pred_cong = clf.predict(X_te) print("Congestion macro-F1:", f1_score(y_cong_test, pred_cong, average="macro"))逻辑说明：展平是为了让树模型能吃时序窗口，每个时间片的三维特征拼成一维。标准化在 fit 之后立刻 transform 测试集，顺序不能反。回归任务这里只取了未来第一个时间片的 volume 做示例，要预测多步就循环或改成多输出。
参数说明：n_estimators=500、learning_rate=0.05、num_leaves=63是一组稳妥的起点，数据量大可以加到 1000 棵树，学习率降到 0.02。class_weight="balanced"对拥堵分类很关键，heavy 样本少，不加权模型会直接忽略它。评估时 volume 的 MAE 要结合量纲看，如果 volume 在几百量级，MAE 二三十算正常，个位数就说明可能泄漏了。
3.3 时间序列切分不能随机
这是最容易踩的坑。交通数据有时间顺序，随机切训练测试集会让未来信息泄漏到训练里，指标虚高。正确做法是按时间切：前 70% 训练，中间 15% 验证，后 15% 测试。项目里给了train和test两个目录，大概率就是按时间分好的，直接用，别自己再 shuffle。
4. 避坑与排查：跑这套代码最容易翻车的五个地方
4.1 现象：训练 loss 正常但验证指标极差
原因：随机切分导致时间泄漏，或者标准化在切分前对全量数据做了 fit。解决：按时间切分，scaler 只在训练集 fit。检查方法是看训练集和验证集的时间范围有没有重叠。
4.2 现象：拥堵分类全预测成 non 或 light
原因：类别不平衡，且没加 class_weight。解决：加class_weight="balanced"，或对 heavy 样本过采样。评估改用 macro-F1，别只看 accuracy。
4.3 现象：volume 预测 MAE 小得离谱
原因：特征里混入了目标泄漏，比如把当前时刻的 volume 同时当特征和标签，或者滑窗边界算错，窗口末端和预测起点重叠。解决：仔细核对滑窗索引，确保i+lookback之后才是预测目标，特征窗口和目标窗口不重叠。
4.4 现象：某些 linkID 预测完全不可用
原因：该传感器数据完整度低，插值过多，或者该路段本身流量模式特殊，训练样本少。解决：在清洗阶段就按完整度过滤，低于阈值的传感器直接排除，不要硬训。也可以按 linkID 分组评估，找出表现差的传感器单独看。
4.5 现象：换一批数据后脚本报字段错误
原因：原始数据的 time 字段格式不统一，或者 congestionLevel 出现了映射表里没有的新值。解决：pd.to_datetime加errors="coerce"兜底，映射用map后检查 NaN 数量，有未知标签就打印出来人工确认，别默默变成 NaN 喂进模型。
5. 进阶技巧：把单点预测变成可解释的拥堵预警
跑通 baseline 之后，真正有价值的是让预测可解释、可预警。我一般会做两件事。第一，用 LightGBM 的 feature importance 看哪些时间片的哪些特征贡献大，如果模型主要靠最近一两个时间片的 volume，说明它学的是惯性，不是趋势，这种模型在突变场景下会失效。第二，把拥堵分类的概率输出拿出来，不只看预测类别，而是设阈值预警：当 heavy 概率超过 0.6 就触发预警，比硬分类更灵活。
验证方法上，别只看整体指标，按时间段拆开看。早高峰 7:00-9:00 和晚高峰 17:00-19:00 的 F1 往往差很多，如果晚高峰明显低，说明模型没学好晚高峰的模式，可能要加时间段特征或分时段建模。下面这个按小时评估的片段我每次都会跑一遍：
import pandas as pd from sklearn.metrics import f1_score # 假设 test_df 含 datetime 和真实/预测标签 test_df["hour"] = pd.to_datetime(test_df["datetime"]).dt.hour for h in [7, 8, 9, 17, 18, 19]: sub = test_df[test_df["hour"] == h] if len(sub) > 0: score = f1_score(sub["true_cong"], sub["pred_cong"], average="macro") print(f"{h}点 macro-F1: {score:.3f}")逻辑说明：按小时切片评估，能暴露模型在特定时段的短板。参数上，average="macro"保证少数类 heavy 的贡献不被淹没。如果某个高峰时段 F1 低于 0.5，基本可以判定该时段特征不足，考虑加入「是否高峰」「前一小时平均速度」这类派生特征。
还有个习惯：每次改完特征工程或切分方式，我都会固定跑一遍同一组验证集，记录指标变化。交通数据周期性太强，很容易出现「这次调参涨了」其实是切分随机性带来的玄学波动。固定验证集是唯一的后悔药。从那以后我每次动滑窗参数或特征列，都强制走一遍按小时拆分的评估，确认不是某个时段被牺牲换来的整体提升。希望帮到你。
本文还有配套的精品资源，点击获取
