简介这份资源面向具备一定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, errorscoerce) # 按 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, lookback12, horizon3): X, y_vol, y_cong [], [], [] for link, g in data.groupby(linkID): g g.sort_values(datetime).reset_index(dropTrue) 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 分组再滑窗保证窗口不跨传感器。lookback12对应 1 小时历史horizon3对应预测未来 15 分钟。y_vol 取未来 horizon 个时间片的 volumey_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_estimators500, learning_rate0.05, num_leaves63) 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_estimators500, learning_rate0.05, num_leaves63, class_weightbalanced) clf.fit(X_tr, y_cong_train) pred_cong clf.predict(X_te) print(Congestion macro-F1:, f1_score(y_cong_test, pred_cong, averagemacro))逻辑说明展平是为了让树模型能吃时序窗口每个时间片的三维特征拼成一维。标准化在 fit 之后立刻 transform 测试集顺序不能反。回归任务这里只取了未来第一个时间片的 volume 做示例要预测多步就循环或改成多输出。参数说明n_estimators500、learning_rate0.05、num_leaves63是一组稳妥的起点数据量大可以加到 1000 棵树学习率降到 0.02。class_weightbalanced对拥堵分类很关键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_weightbalanced或对 heavy 样本过采样。评估改用 macro-F1别只看 accuracy。4.3 现象volume 预测 MAE 小得离谱原因特征里混入了目标泄漏比如把当前时刻的 volume 同时当特征和标签或者滑窗边界算错窗口末端和预测起点重叠。解决仔细核对滑窗索引确保ilookback之后才是预测目标特征窗口和目标窗口不重叠。4.4 现象某些 linkID 预测完全不可用原因该传感器数据完整度低插值过多或者该路段本身流量模式特殊训练样本少。解决在清洗阶段就按完整度过滤低于阈值的传感器直接排除不要硬训。也可以按 linkID 分组评估找出表现差的传感器单独看。4.5 现象换一批数据后脚本报字段错误原因原始数据的 time 字段格式不统一或者 congestionLevel 出现了映射表里没有的新值。解决pd.to_datetime加errorscoerce兜底映射用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], averagemacro) print(f{h}点 macro-F1: {score:.3f})逻辑说明按小时切片评估能暴露模型在特定时段的短板。参数上averagemacro保证少数类 heavy 的贡献不被淹没。如果某个高峰时段 F1 低于 0.5基本可以判定该时段特征不足考虑加入「是否高峰」「前一小时平均速度」这类派生特征。还有个习惯每次改完特征工程或切分方式我都会固定跑一遍同一组验证集记录指标变化。交通数据周期性太强很容易出现「这次调参涨了」其实是切分随机性带来的玄学波动。固定验证集是唯一的后悔药。从那以后我每次动滑窗参数或特征列都强制走一遍按小时拆分的评估确认不是某个时段被牺牲换来的整体提升。希望帮到你。本文还有配套的精品资源点击获取 Python交通流预测实战：855个传感器数据清洗与拥堵等级建模
简介这份资源面向具备一定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, errorscoerce) # 按 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, lookback12, horizon3): X, y_vol, y_cong [], [], [] for link, g in data.groupby(linkID): g g.sort_values(datetime).reset_index(dropTrue) 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 分组再滑窗保证窗口不跨传感器。lookback12对应 1 小时历史horizon3对应预测未来 15 分钟。y_vol 取未来 horizon 个时间片的 volumey_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_estimators500, learning_rate0.05, num_leaves63) 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_estimators500, learning_rate0.05, num_leaves63, class_weightbalanced) clf.fit(X_tr, y_cong_train) pred_cong clf.predict(X_te) print(Congestion macro-F1:, f1_score(y_cong_test, pred_cong, averagemacro))逻辑说明展平是为了让树模型能吃时序窗口每个时间片的三维特征拼成一维。标准化在 fit 之后立刻 transform 测试集顺序不能反。回归任务这里只取了未来第一个时间片的 volume 做示例要预测多步就循环或改成多输出。参数说明n_estimators500、learning_rate0.05、num_leaves63是一组稳妥的起点数据量大可以加到 1000 棵树学习率降到 0.02。class_weightbalanced对拥堵分类很关键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_weightbalanced或对 heavy 样本过采样。评估改用 macro-F1别只看 accuracy。4.3 现象volume 预测 MAE 小得离谱原因特征里混入了目标泄漏比如把当前时刻的 volume 同时当特征和标签或者滑窗边界算错窗口末端和预测起点重叠。解决仔细核对滑窗索引确保ilookback之后才是预测目标特征窗口和目标窗口不重叠。4.4 现象某些 linkID 预测完全不可用原因该传感器数据完整度低插值过多或者该路段本身流量模式特殊训练样本少。解决在清洗阶段就按完整度过滤低于阈值的传感器直接排除不要硬训。也可以按 linkID 分组评估找出表现差的传感器单独看。4.5 现象换一批数据后脚本报字段错误原因原始数据的 time 字段格式不统一或者 congestionLevel 出现了映射表里没有的新值。解决pd.to_datetime加errorscoerce兜底映射用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], averagemacro) print(f{h}点 macro-F1: {score:.3f})逻辑说明按小时切片评估能暴露模型在特定时段的短板。参数上averagemacro保证少数类 heavy 的贡献不被淹没。如果某个高峰时段 F1 低于 0.5基本可以判定该时段特征不足考虑加入「是否高峰」「前一小时平均速度」这类派生特征。还有个习惯每次改完特征工程或切分方式我都会固定跑一遍同一组验证集记录指标变化。交通数据周期性太强很容易出现「这次调参涨了」其实是切分随机性带来的玄学波动。固定验证集是唯一的后悔药。从那以后我每次动滑窗参数或特征列都强制走一遍按小时拆分的评估确认不是某个时段被牺牲换来的整体提升。希望帮到你。本文还有配套的精品资源点击获取