尧图网站开发 
简介这份资源面向具备一定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(), freq5min) 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_value0).reset_index()这段代码的逻辑是先按路段分组再按时间重采样计数然后用MultiIndex笛卡尔积把“路段 × 完整时间轴”补全缺失填 0。参数上resample(5min)的粒度要和后续特征窗口匹配fill_value0只对流量成立如果后面加avg_speed缺失应填NaN再插值不能填 0否则会把“没车”和“堵死”混为一谈。2.2 把时间戳拆成模型能学的周期特征原始时间戳对模型没有直接意义但“星期几”“几点”“是否节假日”对流量影响极大。常见做法是做周期性编码而不是直接给 0-23 的整数因为 23 点和 0 点在时间上相邻整数编码会让模型以为它们很远。用正弦余弦编码可以保留这种循环关系。import numpy as np def add_cycle_features(df, time_coltimestamp): 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_in12, n_out3): series: 单个路段按时间排序的 flow 序列 n_in: 用过去 12 个时间片5min*121小时 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_in12, n_out3) X_all.append(X); y_all.append(y) X_all np.concatenate(X_all); y_all np.concatenate(y_all)逻辑说明n_in12对应过去 1 小时n_out3对应未来 15 分钟这是交通流量预测里比较稳的配置。参数怎么改如果要做 30 分钟预测n_out调到 6如果数据粒度是 15 分钟n_in4就是过去 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, activationtanh, input_shape(n_in, 1), return_sequencesFalse), Dropout(0.2), Dense(n_out) # 直接输出未来 n_out 个时间片 ]) model.compile(optimizeradam, lossmse) model.fit(X_scaled[..., None], y_scaled, epochs30, batch_size64, validation_split0.2)参数说明LSTM(64)的 64 是隐藏单元数数据量小就降到 32避免过拟合Dropout(0.2)是防过拟合的常规操作Dense(n_out)让模型一次输出多步比递归预测更稳。注意X_scaled[..., None]是把二维特征扩成 LSTM 需要的三维(样本, 时间步, 特征)。如果换成树模型直接把X_all展平成(样本, n_in)喂给XGBRegressorn_out每个步长单独训一个模型或用多输出回归。归一化必须用训练集拟合scaler再变换验证集否则就是典型的数据泄漏离线 RMSE 会虚低。4. 交通拥堵预测把连续流量变成拥堵等级4.1 拥堵标签怎么定义才不拍脑袋交通拥堵预测和流量预测最大的区别是流量是回归拥堵通常是分类。标签定义直接决定模型有没有用。常见做法有三种按速度阈值如平均速度低于 20km/h 判拥堵、按拥堵指数流量/通行能力、按占有率。我一般用“速度 占有率”双条件单看速度会把事故和红灯排队混淆。def label_congestion(row, speed_th20, occ_th0.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, axis1)参数上speed_th20和occ_th0.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_periods1).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_size0.2, shuffleFalse) # 时序不能随机打乱 clf lgb.LGBMClassifier(n_estimators300, learning_rate0.05, num_leaves31) clf.fit(X_tr, y_tr) print(classification_report(y_te, clf.predict(X_te)))逻辑说明rolling(w).mean()造的是历史窗口统计w3/6/12分别对应 15/30/60 分钟。shuffleFalse是时序预测的铁律随机打乱会让未来信息泄漏到训练集。LGBMClassifier的参数里num_leaves31控制复杂度数据小就降到 15。输出概率用clf.predict_proba(X_te)再按业务设阈值比如拥堵概率大于 0.7 触发预警。这一步做完流量预测和拥堵预测就串起来了流量预测给未来流量拥堵预测给未来状态概率两者可以共用一套滑窗特征。5. 避坑与排查流量和拥堵预测里最容易翻车的 5 个点5.1 现象离线 RMSE 很低上线预测全错原因滑窗造样本时用了未来信息比如归一化在全集上拟合、或者rolling默认包含了当前时刻。解决所有统计特征必须shift(1)后再 rolling归一化只在训练集拟合。检查方法是把预测结果和真实值按时间画出来看是不是整体平移了一个时间片。5.2 现象模型把凌晨 3 点的流量预测成早高峰原因周期特征编码错误或者时间片补齐时把不同日期的同一时刻混在一起。解决确认sin_day/cos_day的分母是 1440sin_week/cos_week的分母是 10080补齐时间轴时用pd.date_range而不是简单reindex到字符串。5.3 现象拥堵预测的召回率极低几乎不报拥堵原因拥堵样本太少模型被多数类带偏。解决用class_weightbalanced或对拥堵样本过采样同时检查标签阈值是不是太严speed_th20在快速路上可能永远触发不了。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_in12, n_out3, step12): 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控制评估密度step12就是每小时评估一次。基线用历史均值是为了给模型一个最低参照——如果你的 LSTM 连历史均值都打不过说明特征或训练有问题。进阶方向可以试 Temporal Fusion Transformer 做多步概率预测或者把天气、节假日作为外生变量加进去。我自己的习惯是任何时序模型上线前先跑一遍滚动预测把每个路段的 MAE 和拥堵召回率列成表哪个路段拖后腿一目了然。这个方案值不值得做取决于你有没有稳定的历史数据——有 3 个月以上、粒度 5 分钟的流量和速度就值得投入数据太短或缺失太多先把采集补上再谈模型。希望帮到你。本文还有配套的精品资源点击获取 
简介这份资源面向具备一定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(), freq5min) 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_value0).reset_index()这段代码的逻辑是先按路段分组再按时间重采样计数然后用MultiIndex笛卡尔积把“路段 × 完整时间轴”补全缺失填 0。参数上resample(5min)的粒度要和后续特征窗口匹配fill_value0只对流量成立如果后面加avg_speed缺失应填NaN再插值不能填 0否则会把“没车”和“堵死”混为一谈。2.2 把时间戳拆成模型能学的周期特征原始时间戳对模型没有直接意义但“星期几”“几点”“是否节假日”对流量影响极大。常见做法是做周期性编码而不是直接给 0-23 的整数因为 23 点和 0 点在时间上相邻整数编码会让模型以为它们很远。用正弦余弦编码可以保留这种循环关系。import numpy as np def add_cycle_features(df, time_coltimestamp): 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_in12, n_out3): series: 单个路段按时间排序的 flow 序列 n_in: 用过去 12 个时间片5min*121小时 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_in12, n_out3) X_all.append(X); y_all.append(y) X_all np.concatenate(X_all); y_all np.concatenate(y_all)逻辑说明n_in12对应过去 1 小时n_out3对应未来 15 分钟这是交通流量预测里比较稳的配置。参数怎么改如果要做 30 分钟预测n_out调到 6如果数据粒度是 15 分钟n_in4就是过去 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, activationtanh, input_shape(n_in, 1), return_sequencesFalse), Dropout(0.2), Dense(n_out) # 直接输出未来 n_out 个时间片 ]) model.compile(optimizeradam, lossmse) model.fit(X_scaled[..., None], y_scaled, epochs30, batch_size64, validation_split0.2)参数说明LSTM(64)的 64 是隐藏单元数数据量小就降到 32避免过拟合Dropout(0.2)是防过拟合的常规操作Dense(n_out)让模型一次输出多步比递归预测更稳。注意X_scaled[..., None]是把二维特征扩成 LSTM 需要的三维(样本, 时间步, 特征)。如果换成树模型直接把X_all展平成(样本, n_in)喂给XGBRegressorn_out每个步长单独训一个模型或用多输出回归。归一化必须用训练集拟合scaler再变换验证集否则就是典型的数据泄漏离线 RMSE 会虚低。4. 交通拥堵预测把连续流量变成拥堵等级4.1 拥堵标签怎么定义才不拍脑袋交通拥堵预测和流量预测最大的区别是流量是回归拥堵通常是分类。标签定义直接决定模型有没有用。常见做法有三种按速度阈值如平均速度低于 20km/h 判拥堵、按拥堵指数流量/通行能力、按占有率。我一般用“速度 占有率”双条件单看速度会把事故和红灯排队混淆。def label_congestion(row, speed_th20, occ_th0.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, axis1)参数上speed_th20和occ_th0.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_periods1).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_size0.2, shuffleFalse) # 时序不能随机打乱 clf lgb.LGBMClassifier(n_estimators300, learning_rate0.05, num_leaves31) clf.fit(X_tr, y_tr) print(classification_report(y_te, clf.predict(X_te)))逻辑说明rolling(w).mean()造的是历史窗口统计w3/6/12分别对应 15/30/60 分钟。shuffleFalse是时序预测的铁律随机打乱会让未来信息泄漏到训练集。LGBMClassifier的参数里num_leaves31控制复杂度数据小就降到 15。输出概率用clf.predict_proba(X_te)再按业务设阈值比如拥堵概率大于 0.7 触发预警。这一步做完流量预测和拥堵预测就串起来了流量预测给未来流量拥堵预测给未来状态概率两者可以共用一套滑窗特征。5. 避坑与排查流量和拥堵预测里最容易翻车的 5 个点5.1 现象离线 RMSE 很低上线预测全错原因滑窗造样本时用了未来信息比如归一化在全集上拟合、或者rolling默认包含了当前时刻。解决所有统计特征必须shift(1)后再 rolling归一化只在训练集拟合。检查方法是把预测结果和真实值按时间画出来看是不是整体平移了一个时间片。5.2 现象模型把凌晨 3 点的流量预测成早高峰原因周期特征编码错误或者时间片补齐时把不同日期的同一时刻混在一起。解决确认sin_day/cos_day的分母是 1440sin_week/cos_week的分母是 10080补齐时间轴时用pd.date_range而不是简单reindex到字符串。5.3 现象拥堵预测的召回率极低几乎不报拥堵原因拥堵样本太少模型被多数类带偏。解决用class_weightbalanced或对拥堵样本过采样同时检查标签阈值是不是太严speed_th20在快速路上可能永远触发不了。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_in12, n_out3, step12): 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控制评估密度step12就是每小时评估一次。基线用历史均值是为了给模型一个最低参照——如果你的 LSTM 连历史均值都打不过说明特征或训练有问题。进阶方向可以试 Temporal Fusion Transformer 做多步概率预测或者把天气、节假日作为外生变量加进去。我自己的习惯是任何时序模型上线前先跑一遍滚动预测把每个路段的 MAE 和拥堵召回率列成表哪个路段拖后腿一目了然。这个方案值不值得做取决于你有没有稳定的历史数据——有 3 个月以上、粒度 5 分钟的流量和速度就值得投入数据太短或缺失太多先把采集补上再谈模型。希望帮到你。本文还有配套的精品资源点击获取