拓冰网站定制 
做交通流量预测最怕只把它当成普通时间序列问题处理。你在路口装了地磁检测器拿到一组5分钟粒度的流量数据用LSTM去预测未来半小时的流量跑出来的效果通常“单看时间维度还行”可真遇到早高峰突发事故、信号配时调整、周边道路施工纯时间模型会明显慢半拍。原因很简单流量是“传染”的。上游断面一堵下游十几分钟后跟着堵隔壁道路的车流也会在某些时段涌过来。这种同时受空间结构和时间演化双重影响的问题正是ST-CNN时空卷积神经网络要解决的核心场景。这篇文章我会把最近做的一个Python版ST-CNN交通流量预测项目完整拆开讲一遍包括模拟数据怎么造、模型怎么搭、训练和评估怎么落地以及实际调试中踩过的几个坑。代码是基于PyTorch的我给了完整可运行的脚本CPU机器也能跑环境只要装好Python、PyTorch和matplotlib就行。适合刚入门时空预测的读者也适合已经在做时间序列、想把“空间关系”真正用起来的朋友。1. 为什么交通流量预测要用ST-CNN先想清楚再动手1.1 交通数据的双重依赖空间依赖 时间依赖交通流量数据跟普通的时间序列有个本质区别它不是一列数字而是一堆检测器同时观测到的一个“网络状态”。先说时间依赖。任何一个断面的流量都跟它过去十几分钟甚至几个小时的状态强相关。早高峰是从6点半开始爬坡、8点见顶晚高峰又是另一条曲线到了周末峰值形态又不一样。这种规律性当然可以交给LSTM、GRU这类循环神经网络去学但循环网络有个毛病它对“长期空间传导”不敏感某个远方路口的变化传到当前断面需要靠隐状态硬记实际效果经常是“记不住那么远”。再看空间依赖。道路网络天然是连通的上下游和相邻路段的流量具有很强的相关性。最典型的例子是交通事故一个车道被占用之后上游流量会急剧下降车过不去下游流量短期内可能不变但再往下游看流量会在几分钟后出现一个“冲击波”式的波动。这种空间上的传导和时间上的滞后是耦合在一起的你想分开处理都分不开。如果只做时间建模相当于假设每个检测器是孤岛如果只做空间建模又等于假设所有检测器都是同一时刻的静态快照。两个都不对。ST-CNN的出发点非常朴素把时间维和空间维同时放进卷积网络里让模型自己去学习这两个维度上的依赖模式。1.2 ST-CNN是怎么解决这个问题的ST-CNN的核心思想是把一段时间的交通状态组织成一个二维结构一个轴是时间另一个轴是空间位置。如果传感器是按道路顺序排列的那整个输入就是一个 (时间步 × 传感器数) 的矩阵类似一张灰度图。在这个矩阵上卷积核有两种滑法时间卷积核比如3×1沿着时间轴滑动一次覆盖相邻的几个时间步作用是提取流量变化的局部趋势和周期片段。空间卷积核比如1×3沿着传感器轴滑动一次覆盖相邻的几个空间位置作用是提取上下游和邻近路段的联动关系。把这两种卷积结果叠加起来再堆叠几层网络就能同时看到“这个断面在几分钟前经历了什么”和“旁边几个断面现在是什么状态”。相比STGCN那类图卷积方案ST-CNN在规则排列的数据上更简单、更容易调参也更容易跑通等真到了不规则的复杂路网再往图卷积方向升级也不迟。1.3 这篇实战的路线图和最终效果后面几个章节我按一个真实项目落地时的顺序来写先解决数据问题模拟数据生成、滑窗切分、归一化再搭模型和训练流程然后给一份完整可运行的代码最后聊调试经验和坑。项目的设定是20个传感器、5分钟粒度、14天的流量数据用过去1小时12个步长预测未来15分钟3个步长的输出规模最终在测试集上MAE和RMSE都能压到明显低于数据波动幅度的水平可视化预测曲线也能和真实曲线贴合。2. 第一步数据构造与预处理别在源头翻车2.1 真实数据集与训练用模拟数据怎么选做时空预测的人应该都听说过PeMS、METR-LA这类公开数据集它们是由加州高速公路上的真实检测器采集的包含流量、速度、占有率等字段是学术界比较通用的benchmark。但直接用这些数据有一个门槛你要先下载、清洗、对齐时间戳还要处理缺失值对第一次跑模型的同学来说很容易在数据预处理阶段就被劝退。所以这个项目里我选择先构造一份模拟数据。模拟数据的好处是第一空间相关性和时间周期性是“内置”的你知道真值长什么样能判断模型有没有学到东西第二代码拿过去就能直接跑不需要下载任何外部文件第三等你想换真实数据时只需要替换数据加载的函数模型和训练部分完全不用动。2.2 生成带空间相关性的模拟流量数据模拟数据不能拍脑袋随机生成否则模型学到的规律没有意义。我在构造时让这份数据具备三个真实交通流量的特征周期性每天早晚双高峰早高峰8点左右更尖晚高峰18点左右更宽。空间差异不同位置的传感器基础流量不同但相邻传感器的流量水平接近。空间传导对整条线路做一次空间方向的平滑卷积让相邻传感器的流量序列“长相”更像。生成代码如下import numpy as np def generate_traffic_flow(n_sensors20, n_days14, interval5, seed42): rng np.random.default_rng(seed) steps_per_day 24 * 60 // interval # 288 total_steps n_days * steps_per_day # 4032 t np.arange(total_steps) tod (t % steps_per_day) / steps_per_day # 一天内的归一化时刻0~1 # 早高峰8:00、晚高峰18:00用高斯峰叠加近似 morning np.exp(-((tod - 8 / 24) ** 2) / (2 * (0.7 / 24) ** 2)) evening np.exp(-((tod - 18 / 24) ** 2) / (2 * (1.2 / 24) ** 2)) daily 160 * morning 130 * evening # 20个传感器沿线排列基础流量随位置连续变化 pos np.linspace(0, 1, n_sensors) base 120 80 * np.sin(2 * np.pi * pos * 2) 30 * pos flow np.zeros((n_sensors, total_steps)) for i in range(n_sensors): flow[i] base[i] daily rng.normal(0, 10, total_steps) # 空间方向平滑模拟上下游流量传导 kernel np.array([0.05, 0.15, 0.6, 0.15, 0.05]) smoothed np.zeros_like(flow) for step in range(total_steps): smoothed[:, step] np.convolve(flow[:, step], kernel, modesame) return smoothed.astype(np.float32), pos这段代码里最关键的是最后的空间平滑对每一个时刻的20个传感器读数用卷积核[0.05, 0.15, 0.6, 0.15, 0.05]在空间方向上做加权平均。这样相邻传感器之间就有了共享的“局部状态”模型需要做的就是从这种局部依赖里找到预测线索这正是ST-CNN的训练目标。2.3 滑窗切分、归一化与训练/验证/测试集划分数据生成之后要把它变成模型能吃的样本。交通流量预测的标准做法是滑窗用过去T个时刻的流量预测未来H个时刻的流量。这里有一个非常重要的细节归一化的统计量只能用训练集来算。很多新手直接对整个数据求均值和方差然后再切分看起来没什么问题实际上这属于数据泄漏——模型在训练阶段已经“偷偷看到”了测试集的统计信息测试误差会被人为压低。我在代码里先按原始时间序列切出前70%作为训练区间用这段区间的均值和标准差来归一化整条序列再滑窗。这样才是干净的处理方式。def make_loaders(flow, T12, H3, batch_size64, ratio(0.7, 0.15, 0.15)): n_sensors, total flow.shape train_len int(total * ratio[0]) # 只用训练区间统计量做归一化避免数据泄漏 mean flow[:, :train_len].mean(axis1, keepdimsTrue) std flow[:, :train_len].std(axis1, keepdimsTrue) 1e-6 norm_flow (flow - mean) / std # 滑窗构造 (X, Y)X为(T, S)Y为(H, S) X, Y [], [] for i in range(total - T - H 1): X.append(norm_flow[:, i:i T].T) Y.append(norm_flow[:, i T:i T H].T) X np.array(X).astype(np.float32) Y np.array(Y).astype(np.float32) n_total len(X) n1 int(n_total * ratio[0]) n2 n1 int(n_total * ratio[1]) import torch from torch.utils.data import TensorDataset, DataLoader train_ds TensorDataset(torch.from_numpy(X[:n1]), torch.from_numpy(Y[:n1])) val_ds TensorDataset(torch.from_numpy(X[n1:n2]), torch.from_numpy(Y[n1:n2])) test_ds TensorDataset(torch.from_numpy(X[n2:]), torch.from_numpy(Y[n2:])) train_loader DataLoader(train_ds, batch_sizebatch_size, shuffleTrue) val_loader DataLoader(val_ds, batch_sizebatch_size, shuffleFalse) test_loader DataLoader(test_ds, batch_sizebatch_size, shuffleFalse) return train_loader, val_loader, test_loader, mean.reshape(-1), std.reshape(-1)注意所有窗口的X和Y都来自归一化之后的数据。最后返回的mean和std是给评估阶段反归一化用的不然你算出来的MAE是“标准化单位”没法直接说成“每小时多少辆车”。3. 第二步ST-CNN模型搭建与训练3.1 网络结构逐层拆解模型结构我设计成三个时空卷积块再加全连接输出层。每个时空卷积块内部有两条平行的卷积路径一条用 (1, 3) 的卷积核做空间卷积另一条用 (3, 1) 的卷积核做时间卷积两条路径的结果相加再经过ReLU激活。这种设计有个直观的解释空间卷积核每一次只看同一个时刻的相邻几个传感器提取“断面之间的联动”时间卷积核每一次只看同一个传感器的相邻几个时刻提取“流量变化的短时趋势”。二者相加相当于让每个位置的输出同时融合了时空两类信息。import torch import torch.nn as nn import torch.nn.functional as F class STBlock(nn.Module): def __init__(self, in_ch, out_ch, K3): super().__init__() self.spatial nn.Conv2d(in_ch, out_ch, kernel_size(1, K), padding(0, K // 2)) self.temporal nn.Conv2d(in_ch, out_ch, kernel_size(K, 1), padding(K // 2, 0)) def forward(self, x): return F.relu(self.spatial(x) self.temporal(x)) class STCNN(nn.Module): def __init__(self, T, S, H, hidden32, K3): super().__init__() self.T, self.S, self.H T, S, H self.block1 STBlock(1, hidden, K) self.block2 STBlock(hidden, hidden, K) self.block3 STBlock(hidden, hidden, K) self.fc1 nn.Linear(hidden * T * S, 128) self.fc2 nn.Linear(128, H * S) self.drop nn.Dropout(0.3) def forward(self, x): # x: (batch, T, S) x x.unsqueeze(1) # (batch, 1, T, S) x self.block1(x) x self.block2(x) x self.block3(x) x x.flatten(1) # (batch, hidden*T*S) x F.relu(self.fc1(x)) x self.drop(x) x self.fc2(x) # (batch, H*S) return x.view(-1, self.H, self.S)输入在进入卷积之前要做一次unsqueeze(1)把 (batch, T, S) 变成 (batch, 1, T, S)类似灰度图的单通道T对应图像高度S对应图像宽度。三个卷积块都用了padding所以T和S的尺寸在特征传播过程中保持不变最后的全连接层可以直接拿到hidden * T * S的特征数量。对于20个传感器、12个时间步、32个隐藏通道这个特征量是 32×12×207680第一个全连接层压到128维再输出参数量完全是可控的。3.2 训练流程、损失函数与优化器选择流量预测本质是回归问题损失函数我选MSE均方误差。MSE对大误差的惩罚更狠预测值和真实值差得远的时候梯度会明显变大网络会更关注那些“严重错误”的样本。这在实际中比较合适毕竟交通管理场景里漏报一次突发拥堵比正常状态下差几个车流的代价更大。优化器用的Adam学习率1e-3加了weight_decay1e-4做L2正则。训练过程中用ReduceLROnPlateau做学习率衰减验证集loss连续5轮不下降学习率减半。我没有在代码里固定跑几十轮然后结束而是加了“保存验证集最优模型”的逻辑最后加载最优权重做测试这是实际项目里更稳妥的做法。def train_model(model, train_loader, val_loader, epochs60, lr1e-3, devicecpu): criterion nn.MSELoss() opt torch.optim.Adam(model.parameters(), lrlr, weight_decay1e-4) scheduler torch.optim.lr_scheduler.ReduceLROnPlateau( opt, modemin, factor0.5, patience5) best_val float(inf) for epoch in range(1, epochs 1): model.train() train_loss 0.0 for xb, yb in train_loader: xb, yb xb.to(device), yb.to(device) opt.zero_grad() loss criterion(model(xb), yb) loss.backward() opt.step() train_loss loss.item() * xb.size(0) train_loss / len(train_loader.dataset) model.eval() val_loss 0.0 with torch.no_grad(): for xb, yb in val_loader: xb, yb xb.to(device), yb.to(device) val_loss criterion(model(xb), yb).item() * xb.size(0) val_loss / len(val_loader.dataset) scheduler.step(val_loss) if val_loss best_val: best_val val_loss torch.save(model.state_dict(), best_stcnn.pt) if epoch 1 or epoch % 5 0: print(fepoch {epoch:3d} train {train_loss:.5f} val {val_loss:.5f})3.3 评估指标与反归一化预测完不能只看loss工程上还要算MAE、RMSE、MAPE这三个指标。我用一张表总结一下它们各自关注的侧重点指标公式关注点使用场景MAEmean(|pred - true|)平均绝对误差直观反映平均偏差单位好解释RMSEsqrt(mean((pred - true)^2))大误差被放大更关注极端预测错误MAPEmean(|pred - true| / |true|) * 100%相对误差百分比不同断面流量水平差异大时对比重要提醒因为训练前做过归一化必须用之前保存的mean和std把预测结果和真实值都反变换回原始流量单位再计算指标否则结果没有物理意义。MAPE还有一个细节真实流量接近0的时候会算出无穷大所以分母要加一个很小的epsilon或者干脆只统计流量高于某个阈值的样本。def evaluate(model, loader, mean, std, devicecpu): model.eval() preds, trues [], [] with torch.no_grad(): for xb, yb in loader: preds.append(model(xb.to(device)).cpu().numpy()) trues.append(yb.numpy()) pred np.concatenate(preds) true np.concatenate(trues) # 反归一化mean/std是每个传感器一个值形状为(S,) pred pred * std.reshape(1, 1, -1) mean.reshape(1, 1, -1) true true * std.reshape(1, 1, -1) mean.reshape(1, 1, -1) mae np.mean(np.abs(pred - true)) rmse np.sqrt(np.mean((pred - true) ** 2)) mape np.mean(np.abs((pred - true) / (true 1e-6))) * 100 return mae, rmse, mape注意pred和true的形状是 (样本数, H, S)也就是每个样本、每个预测时刻、每个传感器各有一个值。mean.reshape(1, 1, -1)就是把长度为S的统计量广播到 (1, 1, S)和三维数组的最后一维对齐。4. 完整代码从数据到预测一条龙4.1 环境准备与依赖安装跑这份代码之前先把Python环境备好。我用的是Python 3.10PyTorch 2.xmatplotlib做画图numpy做数据处理。如果你是新环境用下面的命令装依赖就行pip install numpy torch matplotlib关于Python安装、pycharm配置Python环境、VSCode里怎么切换解释器这类基础问题网上资料已经很多这里不展开。有一点经验是PyTorch的CPU版本就够跑这个小项目没必要为了这个小例子专门装CUDA等数据量上来了再考虑GPU环境。4.2 完整可运行脚本前面几节是分块讲解这里给一份完整的单文件脚本。把下面的代码保存成stcnn_traffic.py直接运行就能看到训练日志、测试指标和预测对比图。import numpy as np import torch import torch.nn as nn import torch.nn.functional as F from torch.utils.data import TensorDataset, DataLoader import matplotlib.pyplot as plt device torch.device(cuda if torch.cuda.is_available() else cpu) print(using device:, device) def generate_traffic_flow(n_sensors20, n_days14, interval5, seed42): rng np.random.default_rng(seed) steps_per_day 24 * 60 // interval total_steps n_days * steps_per_day t np.arange(total_steps) tod (t % steps_per_day) / steps_per_day morning np.exp(-((tod - 8 / 24) ** 2) / (2 * (0.7 / 24) ** 2)) evening np.exp(-((tod - 18 / 24) ** 2) / (2 * (1.2 / 24) ** 2)) daily 160 * morning 130 * evening pos np.linspace(0, 1, n_sensors) base 120 80 * np.sin(2 * np.pi * pos * 2) 30 * pos flow np.zeros((n_sensors, total_steps)) for i in range(n_sensors): flow[i] base[i] daily rng.normal(0, 10, total_steps) kernel np.array([0.05, 0.15, 0.6, 0.15, 0.05]) smoothed np.zeros_like(flow) for step in range(total_steps): smoothed[:, step] np.convolve(flow[:, step], kernel, modesame) return smoothed.astype(np.float32), pos def make_loaders(flow, T12, H3, batch_size64, ratio(0.7, 0.15, 0.15)): n_sensors, total flow.shape train_len int(total * ratio[0]) mean flow[:, :train_len].mean(axis1, keepdimsTrue) std flow[:, :train_len].std(axis1, keepdimsTrue) 1e-6 norm_flow (flow - mean) / std X, Y [], [] for i in range(total - T - H 1): X.append(norm_flow[:, i:i T].T) Y.append(norm_flow[:, i T:i T H].T) X np.array(X).astype(np.float32) Y np.array(Y).astype(np.float32) n_total len(X) n1 int(n_total * ratio[0]) n2 n1 int(n_total * ratio[1]) train_ds TensorDataset(torch.from_numpy(X[:n1]), torch.from_numpy(Y[:n1])) val_ds TensorDataset(torch.from_numpy(X[n1:n2]), torch.from_numpy(Y[n1:n2])) test_ds TensorDataset(torch.from_numpy(X[n2:]), torch.from_numpy(Y[n2:])) train_loader DataLoader(train_ds, batch_sizebatch_size, shuffleTrue) val_loader DataLoader(val_ds, batch_sizebatch_size, shuffleFalse) test_loader DataLoader(test_ds, batch_sizebatch_size, shuffleFalse) return train_loader, val_loader, test_loader, mean.reshape(-1), std.reshape(-1) class STBlock(nn.Module): def __init__(self, in_ch, out_ch, K3): super().__init__() self.spatial nn.Conv2d(in_ch, out_ch, kernel_size(1, K), padding(0, K // 2)) self.temporal nn.Conv2d(in_ch, out_ch, kernel_size(K, 1), padding(K // 2, 0)) def forward(self, x): return F.relu(self.spatial(x) self.temporal(x)) class STCNN(nn.Module): def __init__(self, T, S, H, hidden32, K3): super().__init__() self.T, self.S, self.H T, S, H self.block1 STBlock(1, hidden, K) self.block2 STBlock(hidden, hidden, K) self.block3 STBlock(hidden, hidden, K) self.fc1 nn.Linear(hidden * T * S, 128) self.fc2 nn.Linear(128, H * S) self.drop nn.Dropout(0.3) def forward(self, x): x x.unsqueeze(1) x self.block1(x) x self.block2(x) x self.block3(x) x x.flatten(1) x F.relu(self.fc1(x)) x self.drop(x) x self.fc2(x) return x.view(-1, self.H, self.S) def train_model(model, train_loader, val_loader, epochs60, lr1e-3): criterion nn.MSELoss() opt torch.optim.Adam(model.parameters(), lrlr, weight_decay1e-4) scheduler torch.optim.lr_scheduler.ReduceLROnPlateau( opt, modemin, factor0.5, patience5) best_val float(inf) for epoch in range(1, epochs 1): model.train() train_loss 0.0 for xb, yb in train_loader: xb, yb xb.to(device), yb.to(device) opt.zero_grad() loss criterion(model(xb), yb) loss.backward() opt.step() train_loss loss.item() * xb.size(0) train_loss / len(train_loader.dataset) model.eval() val_loss 0.0 with torch.no_grad(): for xb, yb in val_loader: xb, yb xb.to(device), yb.to(device) val_loss criterion(model(xb), yb).item() * xb.size(0) val_loss / len(val_loader.dataset) scheduler.step(val_loss) if val_loss best_val: best_val val_loss torch.save(model.state_dict(), best_stcnn.pt) if epoch 1 or epoch % 5 0: print(fepoch {epoch:3d} train {train_loss:.5f} val {val_loss:.5f}) def evaluate(model, loader, mean, std): model.eval() preds, trues [], [] with torch.no_grad(): for xb, yb in loader: preds.append(model(xb.to(device)).cpu().numpy()) trues.append(yb.numpy()) pred np.concatenate(preds) true np.concatenate(trues) pred pred * std.reshape(1, 1, -1) mean.reshape(1, 1, -1) true true * std.reshape(1, 1, -1) mean.reshape(1, 1, -1) mae np.mean(np.abs(pred - true)) rmse np.sqrt(np.mean((pred - true) ** 2)) mape np.mean(np.abs((pred - true) / (true 1e-6))) * 100 return mae, rmse, mape def plot_prediction(pred, true, sensor_id0, horizon0, steps50): plt.figure(figsize(10, 4)) plt.plot(range(steps), true[:steps, horizon, sensor_id], labeltrue) plt.plot(range(steps), pred[:steps, horizon, sensor_id], labelpred) plt.xlabel(sample index) plt.ylabel(traffic flow) plt.title(fSensor {sensor_id}, predict next {horizon 1} steps) plt.legend() plt.tight_layout() plt.savefig(stcnn_pred.png, dpi150) print(prediction figure saved to stcnn_pred.png) if __name__ __main__: flow, pos generate_traffic_flow() T, H, batch_size, epochs 12, 3, 64, 60 train_loader, val_loader, test_loader, mean, std make_loaders( flow, TT, HH, batch_sizebatch_size) model STCNN(T, flow.shape[0], H).to(device) train_model(model, train_loader, val_loader, epochsepochs) model.load_state_dict(torch.load(best_stcnn.pt, map_locationdevice)) mae, rmse, mape evaluate(model, test_loader, mean, std) print(ftest MAE{mae:.3f} RMSE{rmse:.3f} MAPE{mape:.2f}%) preds [] with torch.no_grad(): for xb, _ in test_loader: preds.append(model(xb.to(device)).cpu().numpy()) pred np.concatenate(preds) * std.reshape(1, 1, -1) mean.reshape(1, 1, -1) true np.concatenate([yb.numpy() for _, yb in test_loader]) * std.reshape(1, 1, -1) mean.reshape(1, 1, -1) plot_prediction(pred, true, sensor_id0, horizon0, steps50)4.3 运行结果解读代码跑起来后训练日志大概长这样具体会受随机种子和机器影响epoch 1 train 0.63420 val 0.39012 epoch 5 train 0.15032 val 0.12840 epoch 10 train 0.09041 val 0.08435 ... epoch 60 train 0.02144 val 0.02086测试集上的MAE通常能压到5~8单位是辆/5分钟考虑到真实流量的波动范围在100到400之间这个相对误差已经相当理想了。再看可视化图预测曲线在上升沿和下降沿都能咬住真实曲线说明模型确实学到了早晚高峰的形态特征而不是简单地把上一时刻的流量往后平移。如果你在运行中发现没有任何输出先检查两件事一是matplotlib有没有设置交互后端导致画图窗口卡住二是你的torch能否正常调用CPU算子。这两个都算家常便饭遇到别慌。5. 常见问题与排查技巧实录5.1 张量维度不对十个报错九个因为这初学者跑这个脚本最容易碰到的报错是卷积输入维度不匹配。PyTorch的Conv2d要求输入是四维 (batch, channel, height, width)但滑窗切出来的数据是三维 (batch, T, S)所以在forward里必须x.unsqueeze(1)。如果你把模型单独拿来推理千万记得先传 (B, T, S) 形状的tensor不要在外部先unsqueeze后又跑到模型里再unsqueeze一次否则会多出一个维度。判断维度问题有个笨但有效的方法在forward里临时打印每个阶段x.shape。等代码稳定之后删掉就好。我还遇到过全连接层维度写死导致换传感器数量之后直接报错的情况建议全连接输入层不要手算数字而是写hidden * T * S让代码自己对齐。5.2 归一化与数据泄漏这个坑我在第2章提过一次因为太重要所以再强调一遍。数据泄漏有三种常见形式使用全量数据计算均值和标准差再切分测试集的分布信息提前进了训练过程。模型训练时用了测试集的样本做early stopping或学习率调整。在真实场景中窗口本身跨越了训练和测试的边界导致“未来”信息被训练集看到。代码里的实现只会泄漏最后一种窗口有重叠在13天的短数据上影响不大但如果你要做严谨的实验正确的做法是先按时间把原始序列切成三段再在每个段内分别滑窗。这也是Stack Overflow上关于时间序列预测被问烂了的问题之一我见过不少论文审稿意见专门盯着这一点。5.3 训练不收敛、过拟合怎么判断和处理如果你发现loss一直不降先别急着该网络结构按这个顺序排查输入是否归一化过。没归一化时数值范围大梯度容易震荡。学习率是否合适。1e-3在Adam里是常见起点如果loss剧烈跳动降到1e-4。有没有用对损失函数。分类问题误用了CrossEntropy、回归问题误用了BCELoss这类错误网上天天见。过拟合的表现是训练loss持续下降验证loss先降后升。代码里的Dropout和weight_decay就是干这个的但要注意Dropout只在训练时生效PyTorch的model.eval()会自动关掉它所以评估代码里必须调用model.eval()否则结果会不稳定。还有一个很多教程不会告诉你的事交通流量预测的验证集loss没必要追求降到极低。因为流量本身带有很强的随机噪声模型把噪声都拟合进去反而会牺牲对突发情况的泛化能力。我实际观察下来验证loss降到某个平台后继续训练只是让曲线更光滑对测试集指标帮助不大。5.4 从模拟数据迁移到真实数据最后聊一下怎么把这套代码用到真实数据。你只需要替换数据生成部分让flow变成形状为 (传感器数, 时间步数) 的矩阵模型和训练流程都不用大改。有几个真实场景的注意点缺失值处理检测器经常断传常见做法是对短缺失用线性插值长缺失直接剔除对应的时间段。时间粒度不一致不同数据源可能是5分钟、10分钟或15分钟统一重采样之后再做滑窗。传感器顺序ST-CNN的空间卷积假定传感器是按物理位置有序排列的。如果订单号没有空间意义可以先把传感器按地理位置排序或者做地图网格化把不规则的传感器映射到规则网格上。最后说一个我自己的体会做这类时空预测项目数据预处理和评估设计的工作量往往比搭模型还大。很多新手机器学习拿到数据先急着把模型跑起来结果后面返工最多的恰恰是归一化、切分和指标口径这些“脏活”。如果你照着这篇文章的流程走一遍把这些环节都理顺了再往STGCN、Transformer之类更复杂的模型上走会轻松很多。 ST-CNN交通流量预测实战：时空双维度建模与PyTorch实现
做交通流量预测最怕只把它当成普通时间序列问题处理。你在路口装了地磁检测器拿到一组5分钟粒度的流量数据用LSTM去预测未来半小时的流量跑出来的效果通常“单看时间维度还行”可真遇到早高峰突发事故、信号配时调整、周边道路施工纯时间模型会明显慢半拍。原因很简单流量是“传染”的。上游断面一堵下游十几分钟后跟着堵隔壁道路的车流也会在某些时段涌过来。这种同时受空间结构和时间演化双重影响的问题正是ST-CNN时空卷积神经网络要解决的核心场景。这篇文章我会把最近做的一个Python版ST-CNN交通流量预测项目完整拆开讲一遍包括模拟数据怎么造、模型怎么搭、训练和评估怎么落地以及实际调试中踩过的几个坑。代码是基于PyTorch的我给了完整可运行的脚本CPU机器也能跑环境只要装好Python、PyTorch和matplotlib就行。适合刚入门时空预测的读者也适合已经在做时间序列、想把“空间关系”真正用起来的朋友。1. 为什么交通流量预测要用ST-CNN先想清楚再动手1.1 交通数据的双重依赖空间依赖 时间依赖交通流量数据跟普通的时间序列有个本质区别它不是一列数字而是一堆检测器同时观测到的一个“网络状态”。先说时间依赖。任何一个断面的流量都跟它过去十几分钟甚至几个小时的状态强相关。早高峰是从6点半开始爬坡、8点见顶晚高峰又是另一条曲线到了周末峰值形态又不一样。这种规律性当然可以交给LSTM、GRU这类循环神经网络去学但循环网络有个毛病它对“长期空间传导”不敏感某个远方路口的变化传到当前断面需要靠隐状态硬记实际效果经常是“记不住那么远”。再看空间依赖。道路网络天然是连通的上下游和相邻路段的流量具有很强的相关性。最典型的例子是交通事故一个车道被占用之后上游流量会急剧下降车过不去下游流量短期内可能不变但再往下游看流量会在几分钟后出现一个“冲击波”式的波动。这种空间上的传导和时间上的滞后是耦合在一起的你想分开处理都分不开。如果只做时间建模相当于假设每个检测器是孤岛如果只做空间建模又等于假设所有检测器都是同一时刻的静态快照。两个都不对。ST-CNN的出发点非常朴素把时间维和空间维同时放进卷积网络里让模型自己去学习这两个维度上的依赖模式。1.2 ST-CNN是怎么解决这个问题的ST-CNN的核心思想是把一段时间的交通状态组织成一个二维结构一个轴是时间另一个轴是空间位置。如果传感器是按道路顺序排列的那整个输入就是一个 (时间步 × 传感器数) 的矩阵类似一张灰度图。在这个矩阵上卷积核有两种滑法时间卷积核比如3×1沿着时间轴滑动一次覆盖相邻的几个时间步作用是提取流量变化的局部趋势和周期片段。空间卷积核比如1×3沿着传感器轴滑动一次覆盖相邻的几个空间位置作用是提取上下游和邻近路段的联动关系。把这两种卷积结果叠加起来再堆叠几层网络就能同时看到“这个断面在几分钟前经历了什么”和“旁边几个断面现在是什么状态”。相比STGCN那类图卷积方案ST-CNN在规则排列的数据上更简单、更容易调参也更容易跑通等真到了不规则的复杂路网再往图卷积方向升级也不迟。1.3 这篇实战的路线图和最终效果后面几个章节我按一个真实项目落地时的顺序来写先解决数据问题模拟数据生成、滑窗切分、归一化再搭模型和训练流程然后给一份完整可运行的代码最后聊调试经验和坑。项目的设定是20个传感器、5分钟粒度、14天的流量数据用过去1小时12个步长预测未来15分钟3个步长的输出规模最终在测试集上MAE和RMSE都能压到明显低于数据波动幅度的水平可视化预测曲线也能和真实曲线贴合。2. 第一步数据构造与预处理别在源头翻车2.1 真实数据集与训练用模拟数据怎么选做时空预测的人应该都听说过PeMS、METR-LA这类公开数据集它们是由加州高速公路上的真实检测器采集的包含流量、速度、占有率等字段是学术界比较通用的benchmark。但直接用这些数据有一个门槛你要先下载、清洗、对齐时间戳还要处理缺失值对第一次跑模型的同学来说很容易在数据预处理阶段就被劝退。所以这个项目里我选择先构造一份模拟数据。模拟数据的好处是第一空间相关性和时间周期性是“内置”的你知道真值长什么样能判断模型有没有学到东西第二代码拿过去就能直接跑不需要下载任何外部文件第三等你想换真实数据时只需要替换数据加载的函数模型和训练部分完全不用动。2.2 生成带空间相关性的模拟流量数据模拟数据不能拍脑袋随机生成否则模型学到的规律没有意义。我在构造时让这份数据具备三个真实交通流量的特征周期性每天早晚双高峰早高峰8点左右更尖晚高峰18点左右更宽。空间差异不同位置的传感器基础流量不同但相邻传感器的流量水平接近。空间传导对整条线路做一次空间方向的平滑卷积让相邻传感器的流量序列“长相”更像。生成代码如下import numpy as np def generate_traffic_flow(n_sensors20, n_days14, interval5, seed42): rng np.random.default_rng(seed) steps_per_day 24 * 60 // interval # 288 total_steps n_days * steps_per_day # 4032 t np.arange(total_steps) tod (t % steps_per_day) / steps_per_day # 一天内的归一化时刻0~1 # 早高峰8:00、晚高峰18:00用高斯峰叠加近似 morning np.exp(-((tod - 8 / 24) ** 2) / (2 * (0.7 / 24) ** 2)) evening np.exp(-((tod - 18 / 24) ** 2) / (2 * (1.2 / 24) ** 2)) daily 160 * morning 130 * evening # 20个传感器沿线排列基础流量随位置连续变化 pos np.linspace(0, 1, n_sensors) base 120 80 * np.sin(2 * np.pi * pos * 2) 30 * pos flow np.zeros((n_sensors, total_steps)) for i in range(n_sensors): flow[i] base[i] daily rng.normal(0, 10, total_steps) # 空间方向平滑模拟上下游流量传导 kernel np.array([0.05, 0.15, 0.6, 0.15, 0.05]) smoothed np.zeros_like(flow) for step in range(total_steps): smoothed[:, step] np.convolve(flow[:, step], kernel, modesame) return smoothed.astype(np.float32), pos这段代码里最关键的是最后的空间平滑对每一个时刻的20个传感器读数用卷积核[0.05, 0.15, 0.6, 0.15, 0.05]在空间方向上做加权平均。这样相邻传感器之间就有了共享的“局部状态”模型需要做的就是从这种局部依赖里找到预测线索这正是ST-CNN的训练目标。2.3 滑窗切分、归一化与训练/验证/测试集划分数据生成之后要把它变成模型能吃的样本。交通流量预测的标准做法是滑窗用过去T个时刻的流量预测未来H个时刻的流量。这里有一个非常重要的细节归一化的统计量只能用训练集来算。很多新手直接对整个数据求均值和方差然后再切分看起来没什么问题实际上这属于数据泄漏——模型在训练阶段已经“偷偷看到”了测试集的统计信息测试误差会被人为压低。我在代码里先按原始时间序列切出前70%作为训练区间用这段区间的均值和标准差来归一化整条序列再滑窗。这样才是干净的处理方式。def make_loaders(flow, T12, H3, batch_size64, ratio(0.7, 0.15, 0.15)): n_sensors, total flow.shape train_len int(total * ratio[0]) # 只用训练区间统计量做归一化避免数据泄漏 mean flow[:, :train_len].mean(axis1, keepdimsTrue) std flow[:, :train_len].std(axis1, keepdimsTrue) 1e-6 norm_flow (flow - mean) / std # 滑窗构造 (X, Y)X为(T, S)Y为(H, S) X, Y [], [] for i in range(total - T - H 1): X.append(norm_flow[:, i:i T].T) Y.append(norm_flow[:, i T:i T H].T) X np.array(X).astype(np.float32) Y np.array(Y).astype(np.float32) n_total len(X) n1 int(n_total * ratio[0]) n2 n1 int(n_total * ratio[1]) import torch from torch.utils.data import TensorDataset, DataLoader train_ds TensorDataset(torch.from_numpy(X[:n1]), torch.from_numpy(Y[:n1])) val_ds TensorDataset(torch.from_numpy(X[n1:n2]), torch.from_numpy(Y[n1:n2])) test_ds TensorDataset(torch.from_numpy(X[n2:]), torch.from_numpy(Y[n2:])) train_loader DataLoader(train_ds, batch_sizebatch_size, shuffleTrue) val_loader DataLoader(val_ds, batch_sizebatch_size, shuffleFalse) test_loader DataLoader(test_ds, batch_sizebatch_size, shuffleFalse) return train_loader, val_loader, test_loader, mean.reshape(-1), std.reshape(-1)注意所有窗口的X和Y都来自归一化之后的数据。最后返回的mean和std是给评估阶段反归一化用的不然你算出来的MAE是“标准化单位”没法直接说成“每小时多少辆车”。3. 第二步ST-CNN模型搭建与训练3.1 网络结构逐层拆解模型结构我设计成三个时空卷积块再加全连接输出层。每个时空卷积块内部有两条平行的卷积路径一条用 (1, 3) 的卷积核做空间卷积另一条用 (3, 1) 的卷积核做时间卷积两条路径的结果相加再经过ReLU激活。这种设计有个直观的解释空间卷积核每一次只看同一个时刻的相邻几个传感器提取“断面之间的联动”时间卷积核每一次只看同一个传感器的相邻几个时刻提取“流量变化的短时趋势”。二者相加相当于让每个位置的输出同时融合了时空两类信息。import torch import torch.nn as nn import torch.nn.functional as F class STBlock(nn.Module): def __init__(self, in_ch, out_ch, K3): super().__init__() self.spatial nn.Conv2d(in_ch, out_ch, kernel_size(1, K), padding(0, K // 2)) self.temporal nn.Conv2d(in_ch, out_ch, kernel_size(K, 1), padding(K // 2, 0)) def forward(self, x): return F.relu(self.spatial(x) self.temporal(x)) class STCNN(nn.Module): def __init__(self, T, S, H, hidden32, K3): super().__init__() self.T, self.S, self.H T, S, H self.block1 STBlock(1, hidden, K) self.block2 STBlock(hidden, hidden, K) self.block3 STBlock(hidden, hidden, K) self.fc1 nn.Linear(hidden * T * S, 128) self.fc2 nn.Linear(128, H * S) self.drop nn.Dropout(0.3) def forward(self, x): # x: (batch, T, S) x x.unsqueeze(1) # (batch, 1, T, S) x self.block1(x) x self.block2(x) x self.block3(x) x x.flatten(1) # (batch, hidden*T*S) x F.relu(self.fc1(x)) x self.drop(x) x self.fc2(x) # (batch, H*S) return x.view(-1, self.H, self.S)输入在进入卷积之前要做一次unsqueeze(1)把 (batch, T, S) 变成 (batch, 1, T, S)类似灰度图的单通道T对应图像高度S对应图像宽度。三个卷积块都用了padding所以T和S的尺寸在特征传播过程中保持不变最后的全连接层可以直接拿到hidden * T * S的特征数量。对于20个传感器、12个时间步、32个隐藏通道这个特征量是 32×12×207680第一个全连接层压到128维再输出参数量完全是可控的。3.2 训练流程、损失函数与优化器选择流量预测本质是回归问题损失函数我选MSE均方误差。MSE对大误差的惩罚更狠预测值和真实值差得远的时候梯度会明显变大网络会更关注那些“严重错误”的样本。这在实际中比较合适毕竟交通管理场景里漏报一次突发拥堵比正常状态下差几个车流的代价更大。优化器用的Adam学习率1e-3加了weight_decay1e-4做L2正则。训练过程中用ReduceLROnPlateau做学习率衰减验证集loss连续5轮不下降学习率减半。我没有在代码里固定跑几十轮然后结束而是加了“保存验证集最优模型”的逻辑最后加载最优权重做测试这是实际项目里更稳妥的做法。def train_model(model, train_loader, val_loader, epochs60, lr1e-3, devicecpu): criterion nn.MSELoss() opt torch.optim.Adam(model.parameters(), lrlr, weight_decay1e-4) scheduler torch.optim.lr_scheduler.ReduceLROnPlateau( opt, modemin, factor0.5, patience5) best_val float(inf) for epoch in range(1, epochs 1): model.train() train_loss 0.0 for xb, yb in train_loader: xb, yb xb.to(device), yb.to(device) opt.zero_grad() loss criterion(model(xb), yb) loss.backward() opt.step() train_loss loss.item() * xb.size(0) train_loss / len(train_loader.dataset) model.eval() val_loss 0.0 with torch.no_grad(): for xb, yb in val_loader: xb, yb xb.to(device), yb.to(device) val_loss criterion(model(xb), yb).item() * xb.size(0) val_loss / len(val_loader.dataset) scheduler.step(val_loss) if val_loss best_val: best_val val_loss torch.save(model.state_dict(), best_stcnn.pt) if epoch 1 or epoch % 5 0: print(fepoch {epoch:3d} train {train_loss:.5f} val {val_loss:.5f})3.3 评估指标与反归一化预测完不能只看loss工程上还要算MAE、RMSE、MAPE这三个指标。我用一张表总结一下它们各自关注的侧重点指标公式关注点使用场景MAEmean(|pred - true|)平均绝对误差直观反映平均偏差单位好解释RMSEsqrt(mean((pred - true)^2))大误差被放大更关注极端预测错误MAPEmean(|pred - true| / |true|) * 100%相对误差百分比不同断面流量水平差异大时对比重要提醒因为训练前做过归一化必须用之前保存的mean和std把预测结果和真实值都反变换回原始流量单位再计算指标否则结果没有物理意义。MAPE还有一个细节真实流量接近0的时候会算出无穷大所以分母要加一个很小的epsilon或者干脆只统计流量高于某个阈值的样本。def evaluate(model, loader, mean, std, devicecpu): model.eval() preds, trues [], [] with torch.no_grad(): for xb, yb in loader: preds.append(model(xb.to(device)).cpu().numpy()) trues.append(yb.numpy()) pred np.concatenate(preds) true np.concatenate(trues) # 反归一化mean/std是每个传感器一个值形状为(S,) pred pred * std.reshape(1, 1, -1) mean.reshape(1, 1, -1) true true * std.reshape(1, 1, -1) mean.reshape(1, 1, -1) mae np.mean(np.abs(pred - true)) rmse np.sqrt(np.mean((pred - true) ** 2)) mape np.mean(np.abs((pred - true) / (true 1e-6))) * 100 return mae, rmse, mape注意pred和true的形状是 (样本数, H, S)也就是每个样本、每个预测时刻、每个传感器各有一个值。mean.reshape(1, 1, -1)就是把长度为S的统计量广播到 (1, 1, S)和三维数组的最后一维对齐。4. 完整代码从数据到预测一条龙4.1 环境准备与依赖安装跑这份代码之前先把Python环境备好。我用的是Python 3.10PyTorch 2.xmatplotlib做画图numpy做数据处理。如果你是新环境用下面的命令装依赖就行pip install numpy torch matplotlib关于Python安装、pycharm配置Python环境、VSCode里怎么切换解释器这类基础问题网上资料已经很多这里不展开。有一点经验是PyTorch的CPU版本就够跑这个小项目没必要为了这个小例子专门装CUDA等数据量上来了再考虑GPU环境。4.2 完整可运行脚本前面几节是分块讲解这里给一份完整的单文件脚本。把下面的代码保存成stcnn_traffic.py直接运行就能看到训练日志、测试指标和预测对比图。import numpy as np import torch import torch.nn as nn import torch.nn.functional as F from torch.utils.data import TensorDataset, DataLoader import matplotlib.pyplot as plt device torch.device(cuda if torch.cuda.is_available() else cpu) print(using device:, device) def generate_traffic_flow(n_sensors20, n_days14, interval5, seed42): rng np.random.default_rng(seed) steps_per_day 24 * 60 // interval total_steps n_days * steps_per_day t np.arange(total_steps) tod (t % steps_per_day) / steps_per_day morning np.exp(-((tod - 8 / 24) ** 2) / (2 * (0.7 / 24) ** 2)) evening np.exp(-((tod - 18 / 24) ** 2) / (2 * (1.2 / 24) ** 2)) daily 160 * morning 130 * evening pos np.linspace(0, 1, n_sensors) base 120 80 * np.sin(2 * np.pi * pos * 2) 30 * pos flow np.zeros((n_sensors, total_steps)) for i in range(n_sensors): flow[i] base[i] daily rng.normal(0, 10, total_steps) kernel np.array([0.05, 0.15, 0.6, 0.15, 0.05]) smoothed np.zeros_like(flow) for step in range(total_steps): smoothed[:, step] np.convolve(flow[:, step], kernel, modesame) return smoothed.astype(np.float32), pos def make_loaders(flow, T12, H3, batch_size64, ratio(0.7, 0.15, 0.15)): n_sensors, total flow.shape train_len int(total * ratio[0]) mean flow[:, :train_len].mean(axis1, keepdimsTrue) std flow[:, :train_len].std(axis1, keepdimsTrue) 1e-6 norm_flow (flow - mean) / std X, Y [], [] for i in range(total - T - H 1): X.append(norm_flow[:, i:i T].T) Y.append(norm_flow[:, i T:i T H].T) X np.array(X).astype(np.float32) Y np.array(Y).astype(np.float32) n_total len(X) n1 int(n_total * ratio[0]) n2 n1 int(n_total * ratio[1]) train_ds TensorDataset(torch.from_numpy(X[:n1]), torch.from_numpy(Y[:n1])) val_ds TensorDataset(torch.from_numpy(X[n1:n2]), torch.from_numpy(Y[n1:n2])) test_ds TensorDataset(torch.from_numpy(X[n2:]), torch.from_numpy(Y[n2:])) train_loader DataLoader(train_ds, batch_sizebatch_size, shuffleTrue) val_loader DataLoader(val_ds, batch_sizebatch_size, shuffleFalse) test_loader DataLoader(test_ds, batch_sizebatch_size, shuffleFalse) return train_loader, val_loader, test_loader, mean.reshape(-1), std.reshape(-1) class STBlock(nn.Module): def __init__(self, in_ch, out_ch, K3): super().__init__() self.spatial nn.Conv2d(in_ch, out_ch, kernel_size(1, K), padding(0, K // 2)) self.temporal nn.Conv2d(in_ch, out_ch, kernel_size(K, 1), padding(K // 2, 0)) def forward(self, x): return F.relu(self.spatial(x) self.temporal(x)) class STCNN(nn.Module): def __init__(self, T, S, H, hidden32, K3): super().__init__() self.T, self.S, self.H T, S, H self.block1 STBlock(1, hidden, K) self.block2 STBlock(hidden, hidden, K) self.block3 STBlock(hidden, hidden, K) self.fc1 nn.Linear(hidden * T * S, 128) self.fc2 nn.Linear(128, H * S) self.drop nn.Dropout(0.3) def forward(self, x): x x.unsqueeze(1) x self.block1(x) x self.block2(x) x self.block3(x) x x.flatten(1) x F.relu(self.fc1(x)) x self.drop(x) x self.fc2(x) return x.view(-1, self.H, self.S) def train_model(model, train_loader, val_loader, epochs60, lr1e-3): criterion nn.MSELoss() opt torch.optim.Adam(model.parameters(), lrlr, weight_decay1e-4) scheduler torch.optim.lr_scheduler.ReduceLROnPlateau( opt, modemin, factor0.5, patience5) best_val float(inf) for epoch in range(1, epochs 1): model.train() train_loss 0.0 for xb, yb in train_loader: xb, yb xb.to(device), yb.to(device) opt.zero_grad() loss criterion(model(xb), yb) loss.backward() opt.step() train_loss loss.item() * xb.size(0) train_loss / len(train_loader.dataset) model.eval() val_loss 0.0 with torch.no_grad(): for xb, yb in val_loader: xb, yb xb.to(device), yb.to(device) val_loss criterion(model(xb), yb).item() * xb.size(0) val_loss / len(val_loader.dataset) scheduler.step(val_loss) if val_loss best_val: best_val val_loss torch.save(model.state_dict(), best_stcnn.pt) if epoch 1 or epoch % 5 0: print(fepoch {epoch:3d} train {train_loss:.5f} val {val_loss:.5f}) def evaluate(model, loader, mean, std): model.eval() preds, trues [], [] with torch.no_grad(): for xb, yb in loader: preds.append(model(xb.to(device)).cpu().numpy()) trues.append(yb.numpy()) pred np.concatenate(preds) true np.concatenate(trues) pred pred * std.reshape(1, 1, -1) mean.reshape(1, 1, -1) true true * std.reshape(1, 1, -1) mean.reshape(1, 1, -1) mae np.mean(np.abs(pred - true)) rmse np.sqrt(np.mean((pred - true) ** 2)) mape np.mean(np.abs((pred - true) / (true 1e-6))) * 100 return mae, rmse, mape def plot_prediction(pred, true, sensor_id0, horizon0, steps50): plt.figure(figsize(10, 4)) plt.plot(range(steps), true[:steps, horizon, sensor_id], labeltrue) plt.plot(range(steps), pred[:steps, horizon, sensor_id], labelpred) plt.xlabel(sample index) plt.ylabel(traffic flow) plt.title(fSensor {sensor_id}, predict next {horizon 1} steps) plt.legend() plt.tight_layout() plt.savefig(stcnn_pred.png, dpi150) print(prediction figure saved to stcnn_pred.png) if __name__ __main__: flow, pos generate_traffic_flow() T, H, batch_size, epochs 12, 3, 64, 60 train_loader, val_loader, test_loader, mean, std make_loaders( flow, TT, HH, batch_sizebatch_size) model STCNN(T, flow.shape[0], H).to(device) train_model(model, train_loader, val_loader, epochsepochs) model.load_state_dict(torch.load(best_stcnn.pt, map_locationdevice)) mae, rmse, mape evaluate(model, test_loader, mean, std) print(ftest MAE{mae:.3f} RMSE{rmse:.3f} MAPE{mape:.2f}%) preds [] with torch.no_grad(): for xb, _ in test_loader: preds.append(model(xb.to(device)).cpu().numpy()) pred np.concatenate(preds) * std.reshape(1, 1, -1) mean.reshape(1, 1, -1) true np.concatenate([yb.numpy() for _, yb in test_loader]) * std.reshape(1, 1, -1) mean.reshape(1, 1, -1) plot_prediction(pred, true, sensor_id0, horizon0, steps50)4.3 运行结果解读代码跑起来后训练日志大概长这样具体会受随机种子和机器影响epoch 1 train 0.63420 val 0.39012 epoch 5 train 0.15032 val 0.12840 epoch 10 train 0.09041 val 0.08435 ... epoch 60 train 0.02144 val 0.02086测试集上的MAE通常能压到5~8单位是辆/5分钟考虑到真实流量的波动范围在100到400之间这个相对误差已经相当理想了。再看可视化图预测曲线在上升沿和下降沿都能咬住真实曲线说明模型确实学到了早晚高峰的形态特征而不是简单地把上一时刻的流量往后平移。如果你在运行中发现没有任何输出先检查两件事一是matplotlib有没有设置交互后端导致画图窗口卡住二是你的torch能否正常调用CPU算子。这两个都算家常便饭遇到别慌。5. 常见问题与排查技巧实录5.1 张量维度不对十个报错九个因为这初学者跑这个脚本最容易碰到的报错是卷积输入维度不匹配。PyTorch的Conv2d要求输入是四维 (batch, channel, height, width)但滑窗切出来的数据是三维 (batch, T, S)所以在forward里必须x.unsqueeze(1)。如果你把模型单独拿来推理千万记得先传 (B, T, S) 形状的tensor不要在外部先unsqueeze后又跑到模型里再unsqueeze一次否则会多出一个维度。判断维度问题有个笨但有效的方法在forward里临时打印每个阶段x.shape。等代码稳定之后删掉就好。我还遇到过全连接层维度写死导致换传感器数量之后直接报错的情况建议全连接输入层不要手算数字而是写hidden * T * S让代码自己对齐。5.2 归一化与数据泄漏这个坑我在第2章提过一次因为太重要所以再强调一遍。数据泄漏有三种常见形式使用全量数据计算均值和标准差再切分测试集的分布信息提前进了训练过程。模型训练时用了测试集的样本做early stopping或学习率调整。在真实场景中窗口本身跨越了训练和测试的边界导致“未来”信息被训练集看到。代码里的实现只会泄漏最后一种窗口有重叠在13天的短数据上影响不大但如果你要做严谨的实验正确的做法是先按时间把原始序列切成三段再在每个段内分别滑窗。这也是Stack Overflow上关于时间序列预测被问烂了的问题之一我见过不少论文审稿意见专门盯着这一点。5.3 训练不收敛、过拟合怎么判断和处理如果你发现loss一直不降先别急着该网络结构按这个顺序排查输入是否归一化过。没归一化时数值范围大梯度容易震荡。学习率是否合适。1e-3在Adam里是常见起点如果loss剧烈跳动降到1e-4。有没有用对损失函数。分类问题误用了CrossEntropy、回归问题误用了BCELoss这类错误网上天天见。过拟合的表现是训练loss持续下降验证loss先降后升。代码里的Dropout和weight_decay就是干这个的但要注意Dropout只在训练时生效PyTorch的model.eval()会自动关掉它所以评估代码里必须调用model.eval()否则结果会不稳定。还有一个很多教程不会告诉你的事交通流量预测的验证集loss没必要追求降到极低。因为流量本身带有很强的随机噪声模型把噪声都拟合进去反而会牺牲对突发情况的泛化能力。我实际观察下来验证loss降到某个平台后继续训练只是让曲线更光滑对测试集指标帮助不大。5.4 从模拟数据迁移到真实数据最后聊一下怎么把这套代码用到真实数据。你只需要替换数据生成部分让flow变成形状为 (传感器数, 时间步数) 的矩阵模型和训练流程都不用大改。有几个真实场景的注意点缺失值处理检测器经常断传常见做法是对短缺失用线性插值长缺失直接剔除对应的时间段。时间粒度不一致不同数据源可能是5分钟、10分钟或15分钟统一重采样之后再做滑窗。传感器顺序ST-CNN的空间卷积假定传感器是按物理位置有序排列的。如果订单号没有空间意义可以先把传感器按地理位置排序或者做地图网格化把不规则的传感器映射到规则网格上。最后说一个我自己的体会做这类时空预测项目数据预处理和评估设计的工作量往往比搭模型还大。很多新手机器学习拿到数据先急着把模型跑起来结果后面返工最多的恰恰是归一化、切分和指标口径这些“脏活”。如果你照着这篇文章的流程走一遍把这些环节都理顺了再往STGCN、Transformer之类更复杂的模型上走会轻松很多。