短时交通流量预测实战：从数据预处理到模型选型与避坑指南

简介：这份资源面向交通工程、智能交通与数据分析方向的学习者，聚焦短时交通流量预测这一典型场景，提供一份可直接运行的MATLAB实现脚本，帮助读者理解从历史流量数据到未来时段流量估算的完整建模思路。压缩包内共1个文件，为m脚本类型，体积约1KB，轻量便于快速阅读与二次修改。脚本内容覆盖数据预处理、特征工程、模型构建、训练优化、预测评估与结果可视化等关键环节，涉及时间序列分析、ARIMA、灰色模型、支持向量机与神经网络等常见方法的落地实践，可用于课程设计、毕业课题或交通调度研究中的算法验证。目前已有340人学习，适合希望以最小成本掌握短时交通预测代码框架、并在此基础上替换数据与调参的读者参考。

1. 短时交通流量预测：从 yc.rar 这个文件名说起

看到yc.rar_交通_交通流量_交通流量预测_流量预测_短时交通预测这串标签，做过交通数据项目的人大概会心一笑——这多半是某次课程设计、竞赛或课题结项时打包的压缩包，里面塞着路网检测器数据、几份 baseline 脚本和一份写得不太完整的说明。它真正指向的问题很具体：给定某条路段过去若干个时间片的流量，预测未来 5 到 30 分钟的流量。短时交通预测和宏观的四阶段法不是一回事，它要的是分钟级、可滚动、能塞进信号配时或诱导屏的数值。适合谁？做智慧高速、城市大脑、交通信号优化的算法同学，以及手上有线圈、地磁、卡口或浮动车数据、想先跑通一条 baseline 的工程师。这一篇不讲空泛的“智能交通愿景”，只把从数据到模型到上线评估这条链路拆开，让你拿到一份类似yc.rar的数据后知道先干什么、参数怎么定、哪里最容易翻车。

2. 数据到手先别急着建模：短时交通流的数据结构与预处理

2.1 短时交通流数据的三种常见形态

拿到yc.rar这类压缩包，第一件事是解压后看目录结构，而不是直接pd.read_csv。短时交通流量数据在工程里通常有三种落地形态，识别错了后面全白做。

第一种是宽表：一行是一个时间片，一列是一个检测器或路段 ID，单元格是流量值。这种格式来自 SCATS、UTC/SCOOT 或国产信号系统的导出，最直观，但列数可能上千，缺失值以空或 -1 表示。第二种是长表：字段固定为timestamp, sensor_id, flow, occupancy, speed，一行一条记录，适合直接喂给深度学习框架。第三种是张量切片：已经按(时间步, 节点, 特征)组织好的.npz或.h5，常见于 METR-LA、PEMS-BAY 这类公开数据集的二次分发。

判断方法很简单：看第一行是不是时间戳开头。如果是，大概率宽表；如果字段名里有sensor_id或detector_id，是长表。宽表转长表用melt，长表转张量用pivot加滑窗，这一步别偷懒，后面所有模型都依赖统一格式。

2.2 缺失值、异常值和周期性的处理顺序

预处理顺序错了，后面调参就是玄学。我一般按「时间对齐 → 异常剔除 → 缺失填补 → 归一化」四步走，顺序不能换。

时间对齐：检测器上报周期常见是 1 分钟、2 分钟、5 分钟，混采时统一重采样到 5 分钟。用resample('5T').mean()而不是sum()，因为流量是速率量，求和会放大误差。

异常剔除：流量为负、连续多小时恒定不变、单步跳变超过历史均值 5 倍，这三类先标记为 NaN。别直接删行，删了时间轴就断了。

缺失填补：短时预测里缺失比例低于 5% 时，线性插值加周期均值修正足够；高于 20% 的路段建议整段丢弃，否则模型学到的全是填补噪声。周期均值修正指的是用「上周同一时刻同一星期几」的均值去补，比单纯前向填充稳。

归一化：流量值用 Min-Max 或 Z-Score 都行，但必须按训练集统计量归一化，再应用到验证和测试集。我见过太多人全量归一化，结果测试集信息泄漏，离线指标漂亮，上线就崩。

import pandas as pd import numpy as np # 假设长表字段: timestamp, sensor_id, flow df = pd.read_csv('traffic_long.csv', parse_dates=['timestamp']) # 1. 时间对齐到5分钟 df = df.set_index('timestamp').groupby('sensor_id').resample('5T')['flow'].mean().reset_index() # 2. 异常标记: 负值、单步跳变>5倍历史均值 df['flow'] = df['flow'].where(df['flow'] >= 0, np.nan) df['roll_mean'] = df.groupby('sensor_id')['flow'].transform(lambda x: x.rolling(12, min_periods=1).mean()) df.loc[df['flow'] > 5 * df['roll_mean'], 'flow'] = np.nan # 3. 缺失填补: 线性插值 + 周期修正 df['flow'] = df.groupby('sensor_id')['flow'].transform(lambda x: x.interpolate(method='linear', limit=6)) # 4. 按训练集统计量归一化 train_mask = df['timestamp'] < '2024-01-01' mu, sigma = df.loc[train_mask, 'flow'].mean(), df.loc[train_mask, 'flow'].std() df['flow_norm'] = (df['flow'] - mu) / sigma

这段代码的关键参数是limit=6，对应 30 分钟，超过就说明该路段数据质量太差，插值不可信。rolling(12)是 1 小时窗口，用来判断跳变。归一化用训练集统计量，这是铁律。

2.3 滑窗样本构造与训练集划分的坑

短时预测的样本构造是(历史 T 步, 预测 H 步)。T 常见取 12（1 小时），H 取 1、3、6（5、15、30 分钟）。滑窗步长一般取 1，但训练集划分不能随机打乱，必须按时间切分，否则未来信息泄漏，验证集指标虚高。

def make_samples(series, T=12, H=6): X, y = [], [] for i in range(len(series) - T - H + 1): X.append(series[i:i+T]) y.append(series[i+T:i+T+H]) return np.array(X), np.array(y) # 按时间7:1:2切分，不shuffle n = len(df) train_end, val_end = int(n*0.7), int(n*0.8) X, y = make_samples(df['flow_norm'].values) X_train, y_train = X[:train_end], y[:train_end] X_val, y_val = X[train_end:val_end], y[train_end:val_end] X_test, y_test = X[val_end:], y[val_end:]

注意make_samples是在整个序列上滑窗后再切分，这样边界处会有少量重叠，工程上可接受。如果严格要求无重叠，就分段滑窗。验证集的作用是早停和选超参，测试集只在最后跑一次。

3. 模型选型：从历史均值到时空图网络，短时预测该用哪个

3.1 基线模型不是摆设：HA、ARIMA 和 LightGBM 的适用边界

很多人一上来就上 STGCN、Graph WaveNet，结果发现连历史均值都跑不过。短时交通流有强周期性，历史均值（HA）在平稳路段上 MAPE 能到 8% 到 12%，这是任何复杂模型的底线。先跑 HA，再跑 ARIMA，再跑 LightGBM，最后才考虑深度学习。

HA 的做法：预测值等于上周同一时刻同一星期几的流量均值。实现三行代码，但它是判断数据是否有周期性的试金石。如果 HA 都很差，说明数据本身噪声大或周期被破坏，先回去查预处理。

ARIMA 适合单路段、平稳序列，参数(p,d,q)用 AIC 定阶。缺点是只能单变量、不能加外部特征。LightGBM 把历史 T 步展平成特征，加上小时、星期几、是否节假日，训练快、可解释，在中小规模路网上经常是性价比最高的选择。

这张表是经验值，具体看数据质量。选型原则：数据少于 1 万条样本，别碰深度学习；路网拓扑明确且数据充足，再上时空图模型。

3.2 时空图网络的输入组织：节点、边和特征矩阵

如果决定上 STGCN 或 Graph WaveNet，核心是把路网建成图。节点是检测器或路段，边是邻接关系。邻接矩阵常见三种构建方式：基于距离的高斯核、基于路网拓扑的连通性、基于历史流量相关性的动态图。

输入张量形状是(batch, T, N, F)，T 是历史步数，N 是节点数，F 是特征数（流量、占有率、速度）。邻接矩阵 A 形状(N, N)，归一化后用D^-1/2 A D^-1/2。

import numpy as np # 基于距离的高斯核邻接矩阵 def gaussian_adj(coords, sigma=0.5, threshold=0.1): N = len(coords) A = np.zeros((N, N)) for i in range(N): for j in range(N): if i != j: d = np.linalg.norm(coords[i] - coords[j]) A[i, j] = np.exp(-d**2 / sigma**2) A[A < threshold] = 0 # 对称归一化 D = np.diag(A.sum(1)) D_inv_sqrt = np.linalg.inv(np.sqrt(D + 1e-6)) return D_inv_sqrt @ A @ D_inv_sqrt

sigma控制邻接衰减速度，一般取节点间平均距离的 0.5 到 1 倍。threshold过滤弱连接，避免图太稠密导致过平滑。如果拿不到坐标，就用流量相关系数矩阵，取 Top-K 邻居。

3.3 训练配置：损失函数、学习率和早停

短时预测的损失函数，MSE 是默认，但流量有零值和低值，MAPE 在低流量段会爆炸。我一般用Masked MAE，把零值样本 mask 掉，或者用 Huber Loss 兼顾异常值。

学习率用 1e-3 配 Adam，batch size 64，训练 100 epoch，早停 patience 设 10。如果 loss 震荡，先降学习率到 5e-4，再检查归一化是否按训练集做的。

import torch import torch.nn as nn class MaskedMAELoss(nn.Module): def __init__(self, null_val=0.0): super().__init__() self.null_val = null_val def forward(self, pred, target): mask = (target != self.null_val).float() loss = torch.abs(pred - target) * mask return loss.sum() / (mask.sum() + 1e-6) # 训练循环关键片段 optimizer = torch.optim.Adam(model.parameters(), lr=1e-3) scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(optimizer, patience=5, factor=0.5) criterion = MaskedMAELoss() best_val = float('inf') patience_cnt = 0 for epoch in range(100): model.train() # ... 训练 ... model.eval() with torch.no_grad(): val_loss = criterion(model(X_val), y_val) scheduler.step(val_loss) if val_loss < best_val: best_val = val_loss torch.save(model.state_dict(), 'best.pt') patience_cnt = 0 else: patience_cnt += 1 if patience_cnt >= 10: break

ReduceLROnPlateau的patience=5和早停patience=10是两回事，前者调学习率，后者停训练。别搞混。

4. 短时交通预测避坑指南：五个让指标崩掉的真实原因

4.1 现象：离线 MAPE 5%，上线 MAPE 20%

原因：归一化用了全量数据统计量，测试集信息泄漏；或者滑窗样本构造时，训练集和验证集在时间上有重叠。解决：严格按时间切分，归一化统计量只从训练集算，滑窗边界加 gap，比如训练集最后 T 步不参与验证集构造。

4.2 现象：模型预测值几乎是一条直线

原因：损失函数被大量零流量或低流量样本主导，模型学到的是均值。解决：换 Masked MAE，或者对流量做对数变换log1p再归一化。另外检查输入特征里是不是只有流量，没有时间编码，模型分不清早晚高峰。

4.3 现象：多路段联合训练后，单路段精度反而下降

原因：不同路段的流量量级差异大，共享归一化参数导致小流量路段被淹没。解决：按路段分别归一化，或者用路段 ID 做 embedding 输入模型。图网络里，邻接矩阵如果太稠密，过平滑会让所有节点输出趋同，调大threshold或减少邻居数。

4.4 现象：预测滞后一个时间片

原因：滑窗构造时，预测目标y和输入X的最后一个时间步重叠，模型直接复制了最后一个值。解决：确保y从i+T开始，和X的i+T-1不重叠。这个坑在自回归预测里特别常见，指标看着好，实际是作弊。

4.5 现象：训练 loss 下降但验证 loss 上升

原因：过拟合，或者验证集分布和训练集不一致（比如验证集包含节假日，训练集没有）。解决：加 Dropout、权重衰减，或者把节假日样本单独分层。如果验证集是未来一段时间，检查是否有政策、天气等外部事件导致分布漂移，这类样本考虑剔除或单独建模。

5. 上线前的验证与滚动预测技巧

5.1 多步预测的滚动策略与误差累积

短时预测通常要输出未来 6 个时间片（30 分钟）。有两种策略：直接多步，模型一次输出 H 步；滚动单步，每次预测一步，把预测值喂回输入。直接多步误差不累积，但 H 大时精度下降；滚动单步短期准，但误差会累积，30 分钟后可能偏得离谱。

我一般用直接多步，H=6，损失函数对所有步一视同仁。如果业务只关心 15 分钟，可以对前 3 步加权，权重 0.6/0.3/0.1。

5.2 评估指标的选择：MAPE、RMSE 和 WAPE 怎么报

MAPE 在低流量段会爆炸，报指标时一定要说明是否剔除零值。RMSE 对大误差敏感，适合关注峰值预测。WAPE（加权绝对百分比误差）是sum(|pred-true|)/sum(true)，对低流量更稳，推荐作为主指标。

5.3 一个具体技巧：用残差修正提升高峰段精度

模型在早晚高峰往往低估，因为高峰流量变化剧烈。我的习惯是训练完主模型后，对残差true - pred再训一个轻量级 LightGBM，输入是小时、星期几、前一步流量，输出残差修正量。高峰段 MAPE 通常能再降 1 到 2 个百分点。

# 残差修正 residual = y_val - model(X_val).detach().numpy() residual = residual.reshape(-1) # 展平 feat = np.column_stack([hour_val, weekday_val, X_val[:, -1, :].reshape(len(X_val), -1)]) corrector = lgb.LGBMRegressor(n_estimators=100, learning_rate=0.05) corrector.fit(feat, residual) # 预测时: final_pred = model_pred + corrector.predict(feat)

这个技巧的边界是：残差本身要有规律，如果残差是白噪声，修正无效。先画残差的自相关图，ACF 截尾明显再上。

做短时交通预测这些年，我最大的教训是：别在预处理没做干净的时候调模型。数据里的缺失和异常不处理，再复杂的网络也是学噪声。另一个习惯是每次上线前，一定用最近一周的真实数据跑一遍滚动预测，看误差分布，而不是只看平均指标。希望帮到你。

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