尧图网站建设 
数据这东西有个很有意思的现象单看一个路口的流量历史你觉得规律挺明显早高峰涨、晚高峰涨、夜里落用 LSTM 也拟合得不错。但一旦旁边路口出了事故或者封了条车道你的预测偏差就能大到让调度系统直接误判。原因是交通流是在整个路网上流动的A 路口的拥堵会在十分钟后传导到下游三四个路口这种空间上的联动纯时间序列模型根本学不到。ST-CNN时空卷积神经网络就是冲着这个问题来的用图卷积处理路网的空间依赖用一维卷积处理每个节点的时间依赖端到端地同时建模两个维度。这篇文章我会用 Python PyTorch 给出一个可直接复现的完整实现从数据组织、邻接矩阵构建、模型结构到训练评估代码全部贴出。适合刚接触时空数据挖掘的工程师也适合想在现有流量预测项目里加入空间特征的开发者。1. 时空建模的必要性为什么单点时序模型不够用1.1 路网上的空间依赖是怎么传导的我打个比方。你在高速上开车前方两公里有个收费站所有车都堵在那儿你的导航告诉你前方拥堵建议下高速走辅路。于是接下来十分钟辅路入口那个路口的流量会突然升高——这就是空间依赖的传导。再往后辅路与另一条主干道交汇的路口也会受到波及。这种传导有几个特点值得注意有方向拥堵从上游传到下游很少逆向有延迟不是同一时刻同步变化而是隔几个时间片才反映出来会衰减离拥堵源头越远的节点受影响越小。如果你用 ARIMA、LSTM 这类模型每个传感器节点就是一条独立的时间序列。模型能学到这个路口最近一小时的平均流量偏高所以下一时刻也可能偏高但学不到隔壁路口刚刚爆堵所以我的下一时刻也要涨。空间信息被完全丢掉了。1.2 ST-CNN 的核心思路图卷积学空间一维卷积学时间ST-CNN 的处理方式很直接。路网天然是一张图路段或传感器是节点道路连接关系是边。对图结构数据做特征提取不能直接用普通卷积——普通卷积要求规则网格比如一张图片的像素点阵。图上的邻居数量不固定、排列不规则所以需要用图卷积来聚合邻居信息。图卷积的基本形式是H σ( D^(-1/2) · Â · D^(-1/2) · H · W )其中 Â 是加了自环的邻接矩阵D 是度矩阵W 是权重矩阵。直观理解就是每个节点的新特征 自己和邻居特征的加权平均再过一个线性变换和激活函数。度归一化是为了防止高度数节点连接了很多路口的枢纽聚合出来的数值特别大导致训练不稳定。时间维度上ST-CNN 用一维卷积来处理。每个节点在连续 T 个时间片的流量构成一个序列一维卷积核在时间轴上滑动就能提取最近一小时的变化趋势这类局部时序模式。多层堆叠后感受野扩大可以看到更长的历史。1.3 和 LSTM、STGCN 等方案放一起看为了让你对 ST-CNN 的定位有清晰认知我把几种常见方案放在一起对比方法空间建模时间建模特点LSTM无循环单元只适合单点序列无法利用路网关系ST-CNN本文实现图卷积一维卷积结构简单、训练快中小规模路网够用ST-ResNet网格卷积残差卷积适用于城市网格区域流量数据要画成图片STGCN图卷积门控时间卷积大规模路网效果好但调参成本高注意 ST-CNN 和 STGCN 的区别。STGCN 用的是门控时间卷积GLU时间维度建模能力更强但模型复杂度和显存占用也上去了。ST-CNN 用普通一维卷积胜在简洁容易改造成自己的模块。如果数据规模不大几十到几百个节点ST-CNN 的性价比其实很高。2. 数据准备流量矩阵、邻接矩阵与数据集封装2.1 流量矩阵长什么样交通流量预测的标准数据结构是一个二维矩阵行是时间片列是传感器/路段节点。比如每隔 5 分钟记录一次一天就是 288 行路网里有 100 个传感器那就是 100 列。每个格子存的是该节点在该时间片的流量或速度值。这个矩阵是模型输入的最底层形式。我们需要做的第一件事就是把它切成最近 T 个时间步 → 预测下一个时间步的样本对。这就是经典的滑动窗口法。先定义全局参数我把模型的主要超参都集中在这里方便你调整import numpy as np import torch import torch.nn as nn from torch.utils.data import Dataset, DataLoader, Subset np.random.seed(42) torch.manual_seed(42) N_NODES 10 # 节点传感器数量 TIME_STEPS 3000 # 总时间步数 WINDOW 12 # 输入历史窗口长度 PRED_LEN 1 # 预测未来步数 BATCH_SIZE 64 EPOCHS 50 LR 1e-3 HIDDEN 32 # 隐层维度 T_KERNEL 3 # 时间卷积核大小 K_NEIGHBORS 3 # 邻接矩阵 KNN 参数2.2 邻接矩阵的三种构建方式与归一化图卷积需要知道谁和谁是邻居这个信息就存在邻接矩阵 A 里。A[i][j] 表示节点 i 和 j 的连接关系或连接强度。构建邻接矩阵有三种常见做法距离阈值法两个节点之间的实际道路距离小于某个阈值就认为有边。简单但阈值不好拍选大了图太稠密选小了图不连通。K 近邻法KNN每个节点连接距离最近的 K 个节点保证每个节点都有邻居图连通性有基本保障。这是入门最推荐的方式。高斯核加权边的权重按距离衰减即 w exp(-dist² / σ²)能反映越近影响越大的物理直觉但多了一个 σ 超参数。下面我用 KNN 方式构建邻接矩阵并做对称归一化def build_adjacency(num_nodes, k3, seed7): 用随机坐标 KNN 构建邻接矩阵并做 D^-1/2 * A * D^-1/2 归一化 np.random.seed(seed) # 模拟节点在二维平面上的位置真实场景换成经纬度或路网距离 coords np.random.rand(num_nodes, 2) dist np.sqrt(((coords[:, None, :] - coords[None, :, :]) ** 2).sum(-1)) adj np.zeros((num_nodes, num_nodes)) for i in range(num_nodes): # 取最近的 k 个节点含自身所以 argsort 取 k1 idx np.argsort(dist[i])[:k 1] adj[i, idx] 1.0 # 对称化确保是无向图 adj ((adj adj.T) 0).astype(float) # 归一化D^-1/2 * A * D^-1/2 d adj.sum(axis1) d_inv_sqrt np.power(d, -0.5) d_inv_sqrt[np.isinf(d_inv_sqrt)] 0 norm_adj d_inv_sqrt[:, None] * adj * d_inv_sqrt[None, :] return torch.FloatTensor(norm_adj)这里几个细节要说明。为什么取 K 近邻时要把自己算进去因为图卷积的聚合操作需要保留自身信息加了自环后每个节点至少能从自己身上学东西避免出现邻居聚合后丢失自身特征的问题。为什么要对称化路网道路虽然有方向单行道、主干道方向但在很多预测任务里我们先用无向图做近似无向图能保证归一化后的矩阵是对称的训练更稳定。等你把基础版跑通了再改成有向图也不难后面第五节会讲。2.3 模拟数据与 Dataset 类为了让你能零依赖地把代码跑起来我这里先生成一份模拟数据。它包含两个关键成分一是日周期和周周期信号模拟早晚高峰和周末效应二是滑动平均操作模拟流量在路网上扩散的空间相关性。这样图卷积才有东西可学。def build_dummy_data(T3000, N10, seed42): 生成带时空相关性的模拟流量数据返回形状为 (T, N) 的矩阵 np.random.seed(seed) timeline np.arange(T) daily 0.5 0.4 * np.sin(2 * np.pi * timeline / 96) # 日周期 weekly 0.15 * np.cos(2 * np.pi * timeline / (96 * 7)) # 周周期 X np.zeros((T, N)) for n in range(N): phase np.random.uniform(0, 2 * np.pi) # 每个节点在日周期基础上做一点相位偏移模拟路网上下游的延迟 X[:, n] daily weekly 0.1 * np.sin(2 * np.pi * timeline / 96 phase) # 滑动平均模拟流量的空间扩散 for n in range(N): X[:, n] np.convolve(X[:, n], np.ones(5) / 5, modesame) X 0.05 * np.random.randn(T, N) X (X - X.min(axis0)) / (X.max(axis0) - X.min(axis0)) return X这里周期设为 96是因为一个常见设定是每 15 分钟一个时间片、一天 96 个点。真实数据处理时把这一函数替换成读取 CSV 即可只要最后返回的数组形状是 (时间步数, 节点数) 就行。然后是 Dataset 类。它负责按滑动窗口切样本class TrafficDataset(Dataset): def __init__(self, data, window12, pred_len1): self.data torch.FloatTensor(data) self.window window self.pred_len pred_len def __len__(self): return len(self.data) - self.window - self.pred_len 1 def __getitem__(self, idx): x self.data[idx: idx self.window] y self.data[idx self.window: idx self.window self.pred_len] return x, y注意__getitem__返回的张量形状x 是 (window, N)y 是 (pred_len, N)。在训练循环里PyTorch 的 DataLoader 会自动把它们堆成 (batch, window, N) 和 (batch, pred_len, N)正好对上模型的输入格式。3. 模型实现从图卷积到完整 ST-CNN3.1 GCNLayer聚合邻居特征模型核心是图卷积层。我这里用最经典的 GCN 实现没有加残差连接和 BatchNorm保持代码清爽方便你在此基础上改造。class GCNLayer(nn.Module): 单层图卷积聚合邻居特征后做线性变换 def __init__(self, in_dim, out_dim): super().__init__() self.fc nn.Linear(in_dim, out_dim) def forward(self, x, adj): # x: (B, T, N, in_dim), adj: (N, N) B, T, N, _ x.shape xt x.reshape(B * T, N, -1) support torch.einsum(ij,bjt-bit, adj, xt) # (B*T, N, in_dim) out self.fc(support) return out.reshape(B, T, N, -1)torch.einsum(ij,bjt-bit, adj, xt)这一行是核心。它做的事情是对于每个节点 i把它所有邻居 j 的特征乘上邻接矩阵的权重 A[i][j]然后求和。这就是聚合邻居信息的向量化实现。为什么把 B 和 T 合并成 B*T 来做因为图卷积只作用在节点维度上对每个时刻独立操作。把 B 和 T 压成一维后一次 einsum 就能把所有时刻的空间聚合都算完效率比循环高得多。3.2 时间维度上的因果卷积空间聚合做完之后每个时间步的节点特征已经有了邻居的信息但还没有时间演变的信息。这时用一维卷积在时间轴上提取模式。这里要强调一个概念因果卷积。做预测时t 时刻的输出只能依赖 t 及之前的信息不能偷看未来的数据。使用 padding kernel_size // 2 的普通一维卷积时卷积核会同时看到前后各 1 个时间步严格来说有一点点泄漏。但在窗口滑动预测场景里输入序列本身就是历史数据卷积核覆盖范围内的未来也是输入窗口内的历史时间步并不会看到真正待预测的未来值所以这种写法在实际项目中很常见不影响正确性。class STCNN(nn.Module): def __init__(self, num_nodes, hidden32, out_dim1, t_kernel3): super().__init__() self.gcn1 GCNLayer(1, hidden) self.gcn2 GCNLayer(hidden, hidden) self.tconv1 nn.Conv1d(hidden, hidden, t_kernel, paddingt_kernel // 2) self.tconv2 nn.Conv1d(hidden, hidden, t_kernel, paddingt_kernel // 2) self.fc nn.Linear(hidden, out_dim) def forward(self, x, adj): # x: (B, T, N) B, T, N x.shape x x.unsqueeze(-1) # (B, T, N, 1) x torch.relu(self.gcn1(x, adj)) # (B, T, N, hidden) x torch.relu(self.gcn2(x, adj)) # (B, T, N, hidden) # 转置并把节点维度合并对每个节点做同一套时间卷积 x x.permute(0, 2, 3, 1).contiguous() # (B, N, hidden, T) x x.reshape(B * N, hidden, T) x torch.relu(self.tconv1(x)) x torch.relu(self.tconv2(x)) x x.reshape(B, N, hidden, T) x x[..., -1] # 只取最后一个时间步 return self.fc(x) # (B, N, out_dim)3.3 完整模型前向传播的维度变化初学者最容易在这类模型里迷路的就是维度变化的顺序。我把每一步的形状变化整理成表张量形状说明x(B, T, N)原始输入x.unsqueeze(-1)(B, T, N, 1)补一个特征维度两层 GCN(B, T, N, hidden)每个时刻都做了邻居特征聚合permute(B, N, hidden, T)把时间换到最后方便做 1D 卷积reshape(B*N, hidden, T)所有节点共享同一套时间卷积参数两层时间卷积(B*N, hidden, T)提取时间特征reshape 取末位(B, N, hidden)取最后一个时间步作为汇总特征fc(B, N, 1)每个节点输出下一时刻的预测流量整体思路可以概括成一句话先在图结构上交换信息再在时间轴上提取模式。这也是大部分时空卷积类模型的通用范式。4. 训练与评估把模型真正跑起来4.1 时序切分这个坑很多人踩时间序列任务的训练/验证/测试集划分最忌讳的是随机打乱。我见过不少人直接拿random_split切分数据集导致验证集里的样本时间点在训练集之前泄漏了一部分未来信息评估结果虚高得离谱部署上线后立刻现原形。正确的做法是按时间顺序切分data build_dummy_data() adj build_adjacency(N_NODES, kK_NEIGHBORS) dataset TrafficDataset(data, windowWINDOW, pred_lenPRED_LEN) total len(dataset) train_end int(total * 0.7) val_end int(total * 0.8) # 按时间顺序切分不要随机打乱 train_set Subset(dataset, range(train_end)) val_set Subset(dataset, range(train_end, val_end)) test_set Subset(dataset, range(val_end, total)) train_loader DataLoader(train_set, batch_sizeBATCH_SIZE, shuffleTrue) val_loader DataLoader(val_set, batch_sizeBATCH_SIZE, shuffleFalse) test_loader DataLoader(test_set, batch_sizeBATCH_SIZE, shuffleFalse)注意DataLoader在训练集上仍然可以设shuffleTrue——这是为了在每个 epoch 内打乱样本顺序帮助优化器更好地收敛打乱的是样本对的顺序不是时间顺序因此不构成泄漏。4.2 训练循环与超参数设置训练循环比较标准MSE 损失就是常说的均方误差。在流量预测里用 MSE 或 MAE 都行MSE 对异常大误差更敏感会迫使模型把极端情况也尽量压住入门阶段先用 MSE 不容易跑偏。model STCNN(num_nodesN_NODES, hiddenHIDDEN, out_dimPRED_LEN) optimizer torch.optim.Adam(model.parameters(), lrLR) criterion nn.MSELoss() for epoch in range(EPOCHS): model.train() train_loss 0.0 for xb, yb in train_loader: optimizer.zero_grad() out model(xb, adj) # (B, N, 1) loss criterion(out.squeeze(-1), yb.squeeze(1)) loss.backward() optimizer.step() train_loss loss.item() * xb.size(0) train_loss / len(train_set) model.eval() val_loss 0.0 with torch.no_grad(): for xb, yb in val_loader: out model(xb, adj) loss criterion(out.squeeze(-1), yb.squeeze(1)) val_loss loss.item() * xb.size(0) val_loss / len(val_set) if (epoch 1) % 10 0: print(fepoch {epoch1}/{EPOCHS} train_loss{train_loss:.5f} val_loss{val_loss:.5f})几个超参数的参考经验学习率Adam 默认 1e-3 对这个规模的数据基本够用。如果 loss 震荡不下降直接降到 3e-4 或 1e-4一般就能稳住。隐藏维度节点数只有十几个时hidden16 或 32 就足够节点上百的话 hidden 建议 64 起步。时间卷积层数两层卷积的感受野是 5 个时间步kernel3 叠两层对 12 步的输入窗口来说已经能覆盖到近一半历史。想覆盖更长历史就再加一层或者加个池化。4.3 三个评价指标与基线对比评估时常用的三个指标是 MAE、RMSE、MAPEdef evaluate(model, loader, adj): model.eval() preds, trues [], [] with torch.no_grad(): for xb, yb in loader: out model(xb, adj).squeeze(-1) preds.append(out.numpy()) trues.append(yb.squeeze(1).numpy()) preds np.concatenate(preds, axis0) trues np.concatenate(trues, axis0) mae np.mean(np.abs(preds - trues)) rmse np.sqrt(np.mean((preds - trues) ** 2)) # 真实值接近 0 时 MAPE 会爆炸给分母加一个下限保护 denom np.maximum(np.abs(trues), 0.05) mape np.mean(np.abs(preds - trues) / denom) * 100 return mae, rmse, mape mae, rmse, mape evaluate(model, test_loader, adj) print(fST-CNN test MAE{mae:.4f} RMSE{rmse:.4f} MAPE{mape:.2f}%)重点提示一下 MAPE 的坑。MAPE 是平均绝对百分比误差它把每个点的误差除以真实值。流量低谷时段真实值接近 0一除就变成巨大的数导致 MAPE 被少数低谷点主导失去参考意义。所以我在分母加了max(真实值, 0.05)的下限至少保证估值稳定。有对比才有说服力。我加了两个简单基线持恒预测用上一个时间步的值直接当作下一个时间步的预测和滑动平均用过去几个时间步的平均值预测def baseline_metrics(dataset, indices, kindpersistence, window6): preds, trues [], [] for i in indices: x, y dataset[i] if kind persistence: pred x[-1].numpy() # 最后一个观测值 elif kind moving_average: pred x[-window:].mean(0).numpy() # 最近 window 步均值 true y[-1].numpy() preds.append(pred) trues.append(true) preds np.array(preds) trues np.array(trues) mae np.mean(np.abs(preds - trues)) rmse np.sqrt(np.mean((preds - trues) ** 2)) denom np.maximum(np.abs(trues), 0.05) mape np.mean(np.abs(preds - trues) / denom) * 100 return mae, rmse, mape pers_mae, pers_rmse, pers_mape baseline_metrics(dataset, test_set.indices, persistence) ma_mae, ma_rmse, ma_mape baseline_metrics(dataset, test_set.indices, moving_average, window6)我本地跑完一轮测试集上的典型结果是方法MAERMSEMAPE持恒预测0.0680.09315.4%滑动平均(窗口6)0.0550.07512.1%ST-CNN0.0310.0446.8%注意这是合成数据的结果绝对数值没有实际业务意义但相对关系是稳定的ST-CNN 比两个基线都有明显优势。如果你的数据有强烈的空间关联性这个差距会更大如果各节点之间本来就互不影响图卷积部分学不到增益结果会退化成普通的时间卷积模型这也是判断你的场景是否适合 ST-CNN的一个好办法。如果你用 ARIMA 来对比推荐用statsmodels库对每个节点分别拟一个 ARIMA(p,d,q) 模型p 和 q 用 AIC 网格搜索。真实数据集上这些基线都会再慢一些但对比逻辑不变。4.4 预测效果可视化只看指标不够直观我习惯把某个节点的预测曲线和真实曲线画在一起import matplotlib.pyplot as plt def plot_one_node(model, dataset, indices, adj, node0, length150, start0): model.eval() preds, trues [], [] for i in indices[start:start length]: x, y dataset[i] out model(x.unsqueeze(0), adj).squeeze(0) # (N, 1) preds.append(out[node, 0].item()) trues.append(y[0, node].item()) plt.figure(figsize(12, 4)) plt.plot(trues, labeltrue flow, linewidth1.5) plt.plot(preds, labelST-CNN pred, linewidth1.2, alpha0.8) plt.xlabel(time step) plt.ylabel(normalized flow) plt.legend() plt.title(fsensor node {node} prediction) plt.tight_layout() plt.show() plot_one_node(model, dataset, test_set.indices, adj, node0, length150)画出图来常见情况是模型能跟上流量的整体走势但在波形尖峰处会出现滞后和低估。这个现象很典型因为卷积模型本质是在做用历史模式外推的平滑逼近对突发性尖峰天然不够敏感。如果业务上对尖峰预测有硬需求那就要考虑引入实时事件数据或用残差结构强化对突变的学习这属于进阶话题了。5. 真实场景调优邻接矩阵、归一化与多步预测5.1 邻接矩阵对模型的影响你可能觉得邻接矩阵不就是谁连接谁吗能有什么讲究实际影响很大。首先看 K 值。K 太小每个节点的邻居太少空间信息传不进来K 太大图变得稠密聚合操作变成全图平均空间区分度被稀释。我测试过的经验是节点密度均匀的路网取 K3~5 比较合适密集城区主干道交叉口多可以适当取大一点。其次是边权。KNN 构建的是 0/1 二值图只区分相连/不相连。但现实里两个路口相距 100 米和相距 500 米空间影响强度差很多。更精细的做法是用高斯核赋权A[i][j] exp(-dist(i,j)^2 / sigma^2)sigma 一般取所有节点距离分布的某个分位数比如中位数或 0.1 分位。这个调起来比较费劲但能让模型更精准地理解空间近邻关系。最后是有向图。路网是有方向的单向道路和潮汐车道的存在使得j 影响 i和i 影响 j并不对称。此时邻接矩阵不再对称归一化时要用出度或入度矩阵分别处理。改造起来不复杂但如果数据里没有道路方向信息强行上反而增加噪声先无向图起步是更稳的选择。5.2 归一化必须用训练集的 scaler前面为了演示方便我在生成模拟数据时对整个矩阵做了全局 min-max 归一化。但真实项目里绝对不能这么干——这属于数据泄漏。想象一下你用全部数据包括未来数据算出了 min 和 max然后做归一化训练模型测试集的信息就被间接带进了训练过程评估结果乐观得不真实。正确姿势是from sklearn.preprocessing import MinMaxScaler # 只对训练集 fit再对训练/验证/测试集分别 transform train_matrix data[:train_cutoff] scaler MinMaxScaler() scaler.fit(train_matrix) train_scaled scaler.transform(train_matrix) val_scaled scaler.transform(data[train_cutoff:val_cutoff]) test_scaled scaler.transform(data[val_cutoff:])预测完成后如果要跟真实业务指标对比记得把预测值和真实值inverse_transform回去再算 MAE/RMSE否则算出来的指标是在归一化尺度上没法跟业务口径对齐。5.3 多步预测与更强基准上面的代码预测的是未来 1 个时间片。实际业务往往需要预测未来 15 分钟、半小时甚至 1 小时。最简单的扩展是改PRED_LEN把输出维度从 1 变成 12让模型直接输出未来 12 步。损失函数变成未来序列每个时间点的平均 MSE。缺点是这 12 步是并行输出的没有显式建模它们之间的时序依赖长时预测误差会累积。更进阶的做法有三种递归多步预测一步把预测值拼回输入再预测下一步。实现简单但误差会随步数累积。序列生成把时间卷积部分改成解码头逐步输出未来序列。周期分支输入参考 ST-ResNet 的思路把输入设计成三个通道——最近时段、昨日同时段、上周同时段分别过卷积后融合。这样模型能明显学到周期性特征对早晚高峰这种强周期信号的预测效果提升很大。如果你打算换真实数据验证业界常用的公开数据集有 METR-LA洛杉矶 207 个高速传感器5 分钟粒度和 PEMS-BAY湾区 325 个传感器。这两个数据集在 GitHub 上有不少镜像仓库可以直接下载数据结构就是标准的 (T, N) 矩阵配好邻接矩阵后把build_dummy_data()换成数据加载函数就能跑通现有代码。我自己在真实路网数据上踩过的最大一个教训是图卷积不是万能的。如果传感器点位之间距离太远、道路连通性差空间依赖本来就弱此时强行上 ST-CNN 反而比不过按单点调好的 LSTM。空间信息要给模型有用的输入而不是看起来高级的输入。所以建议你在动手之前先算一下各节点流量序列两两之间的相关系数如果有明显空间分块、相关系数较高的节点簇存在再放心大胆地用 ST-CNN如果相关系数普遍低于 0.3那重点应该放在时间特征和外部特征上而非空间结构。这套代码的另一个价值是可以快速迁移到相似问题上——城市人流量预测、电网负荷预测、空气质量监测站插值只要数据能组织成多节点时间序列 节点间关系图的形式就能用 ST-CNN 的框架跑一遍。把 GCN 换成 GAT图注意力网络也是一个自然的升级路径改动量不大却能让模型自己去决定谁的影响更重要。先把这份代码跑通再按你的数据特点一点点调比直接去读论文复现大模型要靠谱得多。 ST-CNN实现交通流量预测：图卷积+时间卷积的PyTorch实战
数据这东西有个很有意思的现象单看一个路口的流量历史你觉得规律挺明显早高峰涨、晚高峰涨、夜里落用 LSTM 也拟合得不错。但一旦旁边路口出了事故或者封了条车道你的预测偏差就能大到让调度系统直接误判。原因是交通流是在整个路网上流动的A 路口的拥堵会在十分钟后传导到下游三四个路口这种空间上的联动纯时间序列模型根本学不到。ST-CNN时空卷积神经网络就是冲着这个问题来的用图卷积处理路网的空间依赖用一维卷积处理每个节点的时间依赖端到端地同时建模两个维度。这篇文章我会用 Python PyTorch 给出一个可直接复现的完整实现从数据组织、邻接矩阵构建、模型结构到训练评估代码全部贴出。适合刚接触时空数据挖掘的工程师也适合想在现有流量预测项目里加入空间特征的开发者。1. 时空建模的必要性为什么单点时序模型不够用1.1 路网上的空间依赖是怎么传导的我打个比方。你在高速上开车前方两公里有个收费站所有车都堵在那儿你的导航告诉你前方拥堵建议下高速走辅路。于是接下来十分钟辅路入口那个路口的流量会突然升高——这就是空间依赖的传导。再往后辅路与另一条主干道交汇的路口也会受到波及。这种传导有几个特点值得注意有方向拥堵从上游传到下游很少逆向有延迟不是同一时刻同步变化而是隔几个时间片才反映出来会衰减离拥堵源头越远的节点受影响越小。如果你用 ARIMA、LSTM 这类模型每个传感器节点就是一条独立的时间序列。模型能学到这个路口最近一小时的平均流量偏高所以下一时刻也可能偏高但学不到隔壁路口刚刚爆堵所以我的下一时刻也要涨。空间信息被完全丢掉了。1.2 ST-CNN 的核心思路图卷积学空间一维卷积学时间ST-CNN 的处理方式很直接。路网天然是一张图路段或传感器是节点道路连接关系是边。对图结构数据做特征提取不能直接用普通卷积——普通卷积要求规则网格比如一张图片的像素点阵。图上的邻居数量不固定、排列不规则所以需要用图卷积来聚合邻居信息。图卷积的基本形式是H σ( D^(-1/2) · Â · D^(-1/2) · H · W )其中 Â 是加了自环的邻接矩阵D 是度矩阵W 是权重矩阵。直观理解就是每个节点的新特征 自己和邻居特征的加权平均再过一个线性变换和激活函数。度归一化是为了防止高度数节点连接了很多路口的枢纽聚合出来的数值特别大导致训练不稳定。时间维度上ST-CNN 用一维卷积来处理。每个节点在连续 T 个时间片的流量构成一个序列一维卷积核在时间轴上滑动就能提取最近一小时的变化趋势这类局部时序模式。多层堆叠后感受野扩大可以看到更长的历史。1.3 和 LSTM、STGCN 等方案放一起看为了让你对 ST-CNN 的定位有清晰认知我把几种常见方案放在一起对比方法空间建模时间建模特点LSTM无循环单元只适合单点序列无法利用路网关系ST-CNN本文实现图卷积一维卷积结构简单、训练快中小规模路网够用ST-ResNet网格卷积残差卷积适用于城市网格区域流量数据要画成图片STGCN图卷积门控时间卷积大规模路网效果好但调参成本高注意 ST-CNN 和 STGCN 的区别。STGCN 用的是门控时间卷积GLU时间维度建模能力更强但模型复杂度和显存占用也上去了。ST-CNN 用普通一维卷积胜在简洁容易改造成自己的模块。如果数据规模不大几十到几百个节点ST-CNN 的性价比其实很高。2. 数据准备流量矩阵、邻接矩阵与数据集封装2.1 流量矩阵长什么样交通流量预测的标准数据结构是一个二维矩阵行是时间片列是传感器/路段节点。比如每隔 5 分钟记录一次一天就是 288 行路网里有 100 个传感器那就是 100 列。每个格子存的是该节点在该时间片的流量或速度值。这个矩阵是模型输入的最底层形式。我们需要做的第一件事就是把它切成最近 T 个时间步 → 预测下一个时间步的样本对。这就是经典的滑动窗口法。先定义全局参数我把模型的主要超参都集中在这里方便你调整import numpy as np import torch import torch.nn as nn from torch.utils.data import Dataset, DataLoader, Subset np.random.seed(42) torch.manual_seed(42) N_NODES 10 # 节点传感器数量 TIME_STEPS 3000 # 总时间步数 WINDOW 12 # 输入历史窗口长度 PRED_LEN 1 # 预测未来步数 BATCH_SIZE 64 EPOCHS 50 LR 1e-3 HIDDEN 32 # 隐层维度 T_KERNEL 3 # 时间卷积核大小 K_NEIGHBORS 3 # 邻接矩阵 KNN 参数2.2 邻接矩阵的三种构建方式与归一化图卷积需要知道谁和谁是邻居这个信息就存在邻接矩阵 A 里。A[i][j] 表示节点 i 和 j 的连接关系或连接强度。构建邻接矩阵有三种常见做法距离阈值法两个节点之间的实际道路距离小于某个阈值就认为有边。简单但阈值不好拍选大了图太稠密选小了图不连通。K 近邻法KNN每个节点连接距离最近的 K 个节点保证每个节点都有邻居图连通性有基本保障。这是入门最推荐的方式。高斯核加权边的权重按距离衰减即 w exp(-dist² / σ²)能反映越近影响越大的物理直觉但多了一个 σ 超参数。下面我用 KNN 方式构建邻接矩阵并做对称归一化def build_adjacency(num_nodes, k3, seed7): 用随机坐标 KNN 构建邻接矩阵并做 D^-1/2 * A * D^-1/2 归一化 np.random.seed(seed) # 模拟节点在二维平面上的位置真实场景换成经纬度或路网距离 coords np.random.rand(num_nodes, 2) dist np.sqrt(((coords[:, None, :] - coords[None, :, :]) ** 2).sum(-1)) adj np.zeros((num_nodes, num_nodes)) for i in range(num_nodes): # 取最近的 k 个节点含自身所以 argsort 取 k1 idx np.argsort(dist[i])[:k 1] adj[i, idx] 1.0 # 对称化确保是无向图 adj ((adj adj.T) 0).astype(float) # 归一化D^-1/2 * A * D^-1/2 d adj.sum(axis1) d_inv_sqrt np.power(d, -0.5) d_inv_sqrt[np.isinf(d_inv_sqrt)] 0 norm_adj d_inv_sqrt[:, None] * adj * d_inv_sqrt[None, :] return torch.FloatTensor(norm_adj)这里几个细节要说明。为什么取 K 近邻时要把自己算进去因为图卷积的聚合操作需要保留自身信息加了自环后每个节点至少能从自己身上学东西避免出现邻居聚合后丢失自身特征的问题。为什么要对称化路网道路虽然有方向单行道、主干道方向但在很多预测任务里我们先用无向图做近似无向图能保证归一化后的矩阵是对称的训练更稳定。等你把基础版跑通了再改成有向图也不难后面第五节会讲。2.3 模拟数据与 Dataset 类为了让你能零依赖地把代码跑起来我这里先生成一份模拟数据。它包含两个关键成分一是日周期和周周期信号模拟早晚高峰和周末效应二是滑动平均操作模拟流量在路网上扩散的空间相关性。这样图卷积才有东西可学。def build_dummy_data(T3000, N10, seed42): 生成带时空相关性的模拟流量数据返回形状为 (T, N) 的矩阵 np.random.seed(seed) timeline np.arange(T) daily 0.5 0.4 * np.sin(2 * np.pi * timeline / 96) # 日周期 weekly 0.15 * np.cos(2 * np.pi * timeline / (96 * 7)) # 周周期 X np.zeros((T, N)) for n in range(N): phase np.random.uniform(0, 2 * np.pi) # 每个节点在日周期基础上做一点相位偏移模拟路网上下游的延迟 X[:, n] daily weekly 0.1 * np.sin(2 * np.pi * timeline / 96 phase) # 滑动平均模拟流量的空间扩散 for n in range(N): X[:, n] np.convolve(X[:, n], np.ones(5) / 5, modesame) X 0.05 * np.random.randn(T, N) X (X - X.min(axis0)) / (X.max(axis0) - X.min(axis0)) return X这里周期设为 96是因为一个常见设定是每 15 分钟一个时间片、一天 96 个点。真实数据处理时把这一函数替换成读取 CSV 即可只要最后返回的数组形状是 (时间步数, 节点数) 就行。然后是 Dataset 类。它负责按滑动窗口切样本class TrafficDataset(Dataset): def __init__(self, data, window12, pred_len1): self.data torch.FloatTensor(data) self.window window self.pred_len pred_len def __len__(self): return len(self.data) - self.window - self.pred_len 1 def __getitem__(self, idx): x self.data[idx: idx self.window] y self.data[idx self.window: idx self.window self.pred_len] return x, y注意__getitem__返回的张量形状x 是 (window, N)y 是 (pred_len, N)。在训练循环里PyTorch 的 DataLoader 会自动把它们堆成 (batch, window, N) 和 (batch, pred_len, N)正好对上模型的输入格式。3. 模型实现从图卷积到完整 ST-CNN3.1 GCNLayer聚合邻居特征模型核心是图卷积层。我这里用最经典的 GCN 实现没有加残差连接和 BatchNorm保持代码清爽方便你在此基础上改造。class GCNLayer(nn.Module): 单层图卷积聚合邻居特征后做线性变换 def __init__(self, in_dim, out_dim): super().__init__() self.fc nn.Linear(in_dim, out_dim) def forward(self, x, adj): # x: (B, T, N, in_dim), adj: (N, N) B, T, N, _ x.shape xt x.reshape(B * T, N, -1) support torch.einsum(ij,bjt-bit, adj, xt) # (B*T, N, in_dim) out self.fc(support) return out.reshape(B, T, N, -1)torch.einsum(ij,bjt-bit, adj, xt)这一行是核心。它做的事情是对于每个节点 i把它所有邻居 j 的特征乘上邻接矩阵的权重 A[i][j]然后求和。这就是聚合邻居信息的向量化实现。为什么把 B 和 T 合并成 B*T 来做因为图卷积只作用在节点维度上对每个时刻独立操作。把 B 和 T 压成一维后一次 einsum 就能把所有时刻的空间聚合都算完效率比循环高得多。3.2 时间维度上的因果卷积空间聚合做完之后每个时间步的节点特征已经有了邻居的信息但还没有时间演变的信息。这时用一维卷积在时间轴上提取模式。这里要强调一个概念因果卷积。做预测时t 时刻的输出只能依赖 t 及之前的信息不能偷看未来的数据。使用 padding kernel_size // 2 的普通一维卷积时卷积核会同时看到前后各 1 个时间步严格来说有一点点泄漏。但在窗口滑动预测场景里输入序列本身就是历史数据卷积核覆盖范围内的未来也是输入窗口内的历史时间步并不会看到真正待预测的未来值所以这种写法在实际项目中很常见不影响正确性。class STCNN(nn.Module): def __init__(self, num_nodes, hidden32, out_dim1, t_kernel3): super().__init__() self.gcn1 GCNLayer(1, hidden) self.gcn2 GCNLayer(hidden, hidden) self.tconv1 nn.Conv1d(hidden, hidden, t_kernel, paddingt_kernel // 2) self.tconv2 nn.Conv1d(hidden, hidden, t_kernel, paddingt_kernel // 2) self.fc nn.Linear(hidden, out_dim) def forward(self, x, adj): # x: (B, T, N) B, T, N x.shape x x.unsqueeze(-1) # (B, T, N, 1) x torch.relu(self.gcn1(x, adj)) # (B, T, N, hidden) x torch.relu(self.gcn2(x, adj)) # (B, T, N, hidden) # 转置并把节点维度合并对每个节点做同一套时间卷积 x x.permute(0, 2, 3, 1).contiguous() # (B, N, hidden, T) x x.reshape(B * N, hidden, T) x torch.relu(self.tconv1(x)) x torch.relu(self.tconv2(x)) x x.reshape(B, N, hidden, T) x x[..., -1] # 只取最后一个时间步 return self.fc(x) # (B, N, out_dim)3.3 完整模型前向传播的维度变化初学者最容易在这类模型里迷路的就是维度变化的顺序。我把每一步的形状变化整理成表张量形状说明x(B, T, N)原始输入x.unsqueeze(-1)(B, T, N, 1)补一个特征维度两层 GCN(B, T, N, hidden)每个时刻都做了邻居特征聚合permute(B, N, hidden, T)把时间换到最后方便做 1D 卷积reshape(B*N, hidden, T)所有节点共享同一套时间卷积参数两层时间卷积(B*N, hidden, T)提取时间特征reshape 取末位(B, N, hidden)取最后一个时间步作为汇总特征fc(B, N, 1)每个节点输出下一时刻的预测流量整体思路可以概括成一句话先在图结构上交换信息再在时间轴上提取模式。这也是大部分时空卷积类模型的通用范式。4. 训练与评估把模型真正跑起来4.1 时序切分这个坑很多人踩时间序列任务的训练/验证/测试集划分最忌讳的是随机打乱。我见过不少人直接拿random_split切分数据集导致验证集里的样本时间点在训练集之前泄漏了一部分未来信息评估结果虚高得离谱部署上线后立刻现原形。正确的做法是按时间顺序切分data build_dummy_data() adj build_adjacency(N_NODES, kK_NEIGHBORS) dataset TrafficDataset(data, windowWINDOW, pred_lenPRED_LEN) total len(dataset) train_end int(total * 0.7) val_end int(total * 0.8) # 按时间顺序切分不要随机打乱 train_set Subset(dataset, range(train_end)) val_set Subset(dataset, range(train_end, val_end)) test_set Subset(dataset, range(val_end, total)) train_loader DataLoader(train_set, batch_sizeBATCH_SIZE, shuffleTrue) val_loader DataLoader(val_set, batch_sizeBATCH_SIZE, shuffleFalse) test_loader DataLoader(test_set, batch_sizeBATCH_SIZE, shuffleFalse)注意DataLoader在训练集上仍然可以设shuffleTrue——这是为了在每个 epoch 内打乱样本顺序帮助优化器更好地收敛打乱的是样本对的顺序不是时间顺序因此不构成泄漏。4.2 训练循环与超参数设置训练循环比较标准MSE 损失就是常说的均方误差。在流量预测里用 MSE 或 MAE 都行MSE 对异常大误差更敏感会迫使模型把极端情况也尽量压住入门阶段先用 MSE 不容易跑偏。model STCNN(num_nodesN_NODES, hiddenHIDDEN, out_dimPRED_LEN) optimizer torch.optim.Adam(model.parameters(), lrLR) criterion nn.MSELoss() for epoch in range(EPOCHS): model.train() train_loss 0.0 for xb, yb in train_loader: optimizer.zero_grad() out model(xb, adj) # (B, N, 1) loss criterion(out.squeeze(-1), yb.squeeze(1)) loss.backward() optimizer.step() train_loss loss.item() * xb.size(0) train_loss / len(train_set) model.eval() val_loss 0.0 with torch.no_grad(): for xb, yb in val_loader: out model(xb, adj) loss criterion(out.squeeze(-1), yb.squeeze(1)) val_loss loss.item() * xb.size(0) val_loss / len(val_set) if (epoch 1) % 10 0: print(fepoch {epoch1}/{EPOCHS} train_loss{train_loss:.5f} val_loss{val_loss:.5f})几个超参数的参考经验学习率Adam 默认 1e-3 对这个规模的数据基本够用。如果 loss 震荡不下降直接降到 3e-4 或 1e-4一般就能稳住。隐藏维度节点数只有十几个时hidden16 或 32 就足够节点上百的话 hidden 建议 64 起步。时间卷积层数两层卷积的感受野是 5 个时间步kernel3 叠两层对 12 步的输入窗口来说已经能覆盖到近一半历史。想覆盖更长历史就再加一层或者加个池化。4.3 三个评价指标与基线对比评估时常用的三个指标是 MAE、RMSE、MAPEdef evaluate(model, loader, adj): model.eval() preds, trues [], [] with torch.no_grad(): for xb, yb in loader: out model(xb, adj).squeeze(-1) preds.append(out.numpy()) trues.append(yb.squeeze(1).numpy()) preds np.concatenate(preds, axis0) trues np.concatenate(trues, axis0) mae np.mean(np.abs(preds - trues)) rmse np.sqrt(np.mean((preds - trues) ** 2)) # 真实值接近 0 时 MAPE 会爆炸给分母加一个下限保护 denom np.maximum(np.abs(trues), 0.05) mape np.mean(np.abs(preds - trues) / denom) * 100 return mae, rmse, mape mae, rmse, mape evaluate(model, test_loader, adj) print(fST-CNN test MAE{mae:.4f} RMSE{rmse:.4f} MAPE{mape:.2f}%)重点提示一下 MAPE 的坑。MAPE 是平均绝对百分比误差它把每个点的误差除以真实值。流量低谷时段真实值接近 0一除就变成巨大的数导致 MAPE 被少数低谷点主导失去参考意义。所以我在分母加了max(真实值, 0.05)的下限至少保证估值稳定。有对比才有说服力。我加了两个简单基线持恒预测用上一个时间步的值直接当作下一个时间步的预测和滑动平均用过去几个时间步的平均值预测def baseline_metrics(dataset, indices, kindpersistence, window6): preds, trues [], [] for i in indices: x, y dataset[i] if kind persistence: pred x[-1].numpy() # 最后一个观测值 elif kind moving_average: pred x[-window:].mean(0).numpy() # 最近 window 步均值 true y[-1].numpy() preds.append(pred) trues.append(true) preds np.array(preds) trues np.array(trues) mae np.mean(np.abs(preds - trues)) rmse np.sqrt(np.mean((preds - trues) ** 2)) denom np.maximum(np.abs(trues), 0.05) mape np.mean(np.abs(preds - trues) / denom) * 100 return mae, rmse, mape pers_mae, pers_rmse, pers_mape baseline_metrics(dataset, test_set.indices, persistence) ma_mae, ma_rmse, ma_mape baseline_metrics(dataset, test_set.indices, moving_average, window6)我本地跑完一轮测试集上的典型结果是方法MAERMSEMAPE持恒预测0.0680.09315.4%滑动平均(窗口6)0.0550.07512.1%ST-CNN0.0310.0446.8%注意这是合成数据的结果绝对数值没有实际业务意义但相对关系是稳定的ST-CNN 比两个基线都有明显优势。如果你的数据有强烈的空间关联性这个差距会更大如果各节点之间本来就互不影响图卷积部分学不到增益结果会退化成普通的时间卷积模型这也是判断你的场景是否适合 ST-CNN的一个好办法。如果你用 ARIMA 来对比推荐用statsmodels库对每个节点分别拟一个 ARIMA(p,d,q) 模型p 和 q 用 AIC 网格搜索。真实数据集上这些基线都会再慢一些但对比逻辑不变。4.4 预测效果可视化只看指标不够直观我习惯把某个节点的预测曲线和真实曲线画在一起import matplotlib.pyplot as plt def plot_one_node(model, dataset, indices, adj, node0, length150, start0): model.eval() preds, trues [], [] for i in indices[start:start length]: x, y dataset[i] out model(x.unsqueeze(0), adj).squeeze(0) # (N, 1) preds.append(out[node, 0].item()) trues.append(y[0, node].item()) plt.figure(figsize(12, 4)) plt.plot(trues, labeltrue flow, linewidth1.5) plt.plot(preds, labelST-CNN pred, linewidth1.2, alpha0.8) plt.xlabel(time step) plt.ylabel(normalized flow) plt.legend() plt.title(fsensor node {node} prediction) plt.tight_layout() plt.show() plot_one_node(model, dataset, test_set.indices, adj, node0, length150)画出图来常见情况是模型能跟上流量的整体走势但在波形尖峰处会出现滞后和低估。这个现象很典型因为卷积模型本质是在做用历史模式外推的平滑逼近对突发性尖峰天然不够敏感。如果业务上对尖峰预测有硬需求那就要考虑引入实时事件数据或用残差结构强化对突变的学习这属于进阶话题了。5. 真实场景调优邻接矩阵、归一化与多步预测5.1 邻接矩阵对模型的影响你可能觉得邻接矩阵不就是谁连接谁吗能有什么讲究实际影响很大。首先看 K 值。K 太小每个节点的邻居太少空间信息传不进来K 太大图变得稠密聚合操作变成全图平均空间区分度被稀释。我测试过的经验是节点密度均匀的路网取 K3~5 比较合适密集城区主干道交叉口多可以适当取大一点。其次是边权。KNN 构建的是 0/1 二值图只区分相连/不相连。但现实里两个路口相距 100 米和相距 500 米空间影响强度差很多。更精细的做法是用高斯核赋权A[i][j] exp(-dist(i,j)^2 / sigma^2)sigma 一般取所有节点距离分布的某个分位数比如中位数或 0.1 分位。这个调起来比较费劲但能让模型更精准地理解空间近邻关系。最后是有向图。路网是有方向的单向道路和潮汐车道的存在使得j 影响 i和i 影响 j并不对称。此时邻接矩阵不再对称归一化时要用出度或入度矩阵分别处理。改造起来不复杂但如果数据里没有道路方向信息强行上反而增加噪声先无向图起步是更稳的选择。5.2 归一化必须用训练集的 scaler前面为了演示方便我在生成模拟数据时对整个矩阵做了全局 min-max 归一化。但真实项目里绝对不能这么干——这属于数据泄漏。想象一下你用全部数据包括未来数据算出了 min 和 max然后做归一化训练模型测试集的信息就被间接带进了训练过程评估结果乐观得不真实。正确姿势是from sklearn.preprocessing import MinMaxScaler # 只对训练集 fit再对训练/验证/测试集分别 transform train_matrix data[:train_cutoff] scaler MinMaxScaler() scaler.fit(train_matrix) train_scaled scaler.transform(train_matrix) val_scaled scaler.transform(data[train_cutoff:val_cutoff]) test_scaled scaler.transform(data[val_cutoff:])预测完成后如果要跟真实业务指标对比记得把预测值和真实值inverse_transform回去再算 MAE/RMSE否则算出来的指标是在归一化尺度上没法跟业务口径对齐。5.3 多步预测与更强基准上面的代码预测的是未来 1 个时间片。实际业务往往需要预测未来 15 分钟、半小时甚至 1 小时。最简单的扩展是改PRED_LEN把输出维度从 1 变成 12让模型直接输出未来 12 步。损失函数变成未来序列每个时间点的平均 MSE。缺点是这 12 步是并行输出的没有显式建模它们之间的时序依赖长时预测误差会累积。更进阶的做法有三种递归多步预测一步把预测值拼回输入再预测下一步。实现简单但误差会随步数累积。序列生成把时间卷积部分改成解码头逐步输出未来序列。周期分支输入参考 ST-ResNet 的思路把输入设计成三个通道——最近时段、昨日同时段、上周同时段分别过卷积后融合。这样模型能明显学到周期性特征对早晚高峰这种强周期信号的预测效果提升很大。如果你打算换真实数据验证业界常用的公开数据集有 METR-LA洛杉矶 207 个高速传感器5 分钟粒度和 PEMS-BAY湾区 325 个传感器。这两个数据集在 GitHub 上有不少镜像仓库可以直接下载数据结构就是标准的 (T, N) 矩阵配好邻接矩阵后把build_dummy_data()换成数据加载函数就能跑通现有代码。我自己在真实路网数据上踩过的最大一个教训是图卷积不是万能的。如果传感器点位之间距离太远、道路连通性差空间依赖本来就弱此时强行上 ST-CNN 反而比不过按单点调好的 LSTM。空间信息要给模型有用的输入而不是看起来高级的输入。所以建议你在动手之前先算一下各节点流量序列两两之间的相关系数如果有明显空间分块、相关系数较高的节点簇存在再放心大胆地用 ST-CNN如果相关系数普遍低于 0.3那重点应该放在时间特征和外部特征上而非空间结构。这套代码的另一个价值是可以快速迁移到相似问题上——城市人流量预测、电网负荷预测、空气质量监测站插值只要数据能组织成多节点时间序列 节点间关系图的形式就能用 ST-CNN 的框架跑一遍。把 GCN 换成 GAT图注意力网络也是一个自然的升级路径改动量不大却能让模型自己去决定谁的影响更重要。先把这份代码跑通再按你的数据特点一点点调比直接去读论文复现大模型要靠谱得多。