简介：一套面向高速公路交通流预测的Python完整实现，基于PeMS真实数据集，涵盖LSTM长短时神经网络、SAE堆栈自编码、CNN卷积神经网络以及CNN-LSTM融合模型四种建模方案。代码链路从数据预处理、特征构造到模型训练与效果对比均有覆盖，适合交通领域研究人员、研究生以及想系统学习时序预测建模的开发者参考。压缩包共20个文件、31.48MB，包含7个Python脚本、6个训练好的h5模型权重、4个模型结构json文件、2张模型对比效果图以及PeMS原始数据集压缩包；既有可直接运行的训练脚本，也有现成权重可加载用于推理和结果复现。目前已有2441人学习下载。通过对比CNN-LSTM、LSTM、SAE、CNN在高速公路交通流数据上的预测表现，读者能清晰理解不同网络结构的建模差异与融合设计，省去从数据获取到模型调优的大量重复工作。
1. TrafficFlowPrediction在解决什么：五分钟后的高速车流你怎么给
如果你也接过“高速公路交通流预测”这类需求，应该能体会那种两难：领导要的是未来15分钟断面流量能不能扛住，而你手里往往只有一个 csv、一坨连续流量，和一堆“玄学”模型参数。TrafficFlowPrediction 这个名字包着三件套：python 代码、数据集、LSTM/SAE/CNN 三个模型。它要解决的是短时交通流预测的标准问题——用过去半小时到一小时的历史流量、速度、占有率，预测未来 5 到 30 分钟的车流强度。这套东西适合三类人：要做毕设或小论文的研究生、给交通管控平台写短临预报模块的数据工程师、以及给收费站做排队预警的算法岗。别急着把数据往模型里一扔就等结果，真正值钱的部分在于数据怎么洗、滑窗怎么切、三个模型怎么对同一个场景说人话。
2. LSTM、SAE、CNN三选一还是三合一：交通流预测的模型分工
短时交通流预测本质上是一个有监督的时序回归问题：给定过去 n 个时间步的检测器数据，预测未来 m 个时间步的流量值。LSTM、SAE、CNN 在数学形式上都可用来做这件事，但它们各自依赖的归纳偏置完全不同：LSTM 认时间顺序，SAE 认特征结构，CNN 认局部模式。我见过很多人拿到标题里的三件套后，把三个模型在一份代码里顺序跑完就对比 RMSE，这样其实只验证了一个指标，没有验证“为什么选它”。先讲清楚分工，再上骨架代码，后面第四章才好展开完整训练流程。
2.1 LSTM为什么是默认基线：从记住“昨天早高峰”说起
LSTM 在交通流预测里当默认基线，不是因为它最复杂，而是因为它保留了“时序因果”。输入门、遗忘门、输出门三个门结构让它在处理高速公路流量这种强周期信号时，能记住 7 天前同一天是周五、2 小时前刚进入晚高峰这类长距离依赖。很多公开数据集上的经验结论也是：LSTM 的预测曲线在不做任何花活时，已经比线性回归和 SVR 平滑很多。
import torch import torch.nn as nn class LSTMFlow(nn.Module): def __init__(self, n_features, hidden_size=64, num_layers=2, pred_len=1): super().__init__() self.lstm = nn.LSTM( input_size=n_features, hidden_size=hidden_size, num_layers=num_layers, batch_first=True, dropout=0.2 if num_layers > 1 else 0.0, ) self.fc = nn.Linear(hidden_size, pred_len) def forward(self, x): # x: (batch, input_len, n_features) out, _ = self.lstm(x) # out: (batch, input_len, hidden) return self.fc(out[:, -1, :]) # 取最后一个时间步的隐状态这里只取序列最后一个时间步的隐状态做预测，因为高速公路流量预测是“用过去推未来”，输入末尾携带的最新状态信息量最大。hidden_size 我一般从 64 起调，断面流量数据的特征维度不高，128 通常就够；num_layers 超过 3 层之后收益很小，反而训练曲线容易震荡。要提醒的是，不要在实时预测任务里用双向 LSTM，它会在时间轴上引入未来信息，离线验证很好看，上线后结果是泄密的。
2.2 SAE堆叠自编码器：把断面流量压缩成特征再回归
SAE（Stacked AutoEncoder）在交通流预测里常被当作特征提取器用。它不建模时间顺序，而是把一整段滑窗展平成向量，通过编码器压缩成低维特征，再用解码器重建原始输入。好处是：在流量、速度、占有率三个通道相关性很强的数据里，SAE 能自动提取“今天这个断面处于什么状态”的紧凑表示，去掉冗余噪声。坏处也很明显：展平之后，时间步之间的顺序被全连接层硬记，很难显式表达“先升后降”这类时序关系。
class SAEFlow(nn.Module): def __init__(self, input_len, n_features, compress_dim=32, pred_len=3): super().__init__() self.in_dim = input_len * n_features self.compress_dim = compress_dim self.encoder = nn.Sequential( nn.Linear(self.in_dim, 128), nn.ReLU(), nn.Linear(128, compress_dim), nn.ReLU(), ) self.decoder = nn.Sequential( nn.Linear(compress_dim, 128), nn.ReLU(), nn.Linear(128, self.in_dim), ) # 预训练阶段使用 decoder 做重构，微调阶段再换成回归头 self.regressor = nn.Linear(compress_dim, pred_len)SAE 的标准做法分两阶段：先忽略回归头，单独用重构损失预训练 encoder 和 decoder；预训练收敛后把 decoder 丢掉，用 encoder 输出的压缩向量接一个回归头继续微调。很多论文把这个过程叫“逐层预训练”。实操中有一个血泪经验：预训练轮数不要贪多，重构 loss 降到平台期就停，否则编码器会过度学习重建细节，反而把天气、节假日这类低频信号压掉了。
2.3 一维CNN：用卷积核去扫时间轴
CNN 做交通流预测，靠的是一维卷积在时间轴上的局部感受野。它和 LSTM 的建模角度不同：LSTM 是“记住并更新状态”，CNN 是“用固定大小的卷积核看一段局部曲线”。对高速公路流量来说，连续 3 到 5 个时间步的爬坡、掉头、平稳，就是最典型的局部特征。卷积核扫过整个滑窗后，再接池化层提取全局趋势。
class CNNFlow(nn.Module): def __init__(self, n_features, pred_len=3): super().__init__() self.conv = nn.Sequential( nn.Conv1d(n_features, 64, kernel_size=5, padding=2), # 不缩时间步 nn.BatchNorm1d(64), nn.ReLU(), nn.MaxPool1d(2), nn.Conv1d(64, 64, kernel_size=3, dilation=2, padding=2), # 感受野扩大 nn.ReLU(), nn.AdaptiveAvgPool1d(1), ) self.fc = nn.Linear(64, pred_len) def forward(self, x): # x: (batch, input_len, n_features) x = x.transpose(1, 2) # 转成 (batch, channels, length) h = self.conv(x).squeeze(-1) # 池化后得到 (batch, 64) return self.fc(h)这里把输入从 (batch, input_len, n_features) 转成 (batch, n_features, input_len)，是为了让 Conv1d 把时间轴当作扫描维度。第一层卷积核大小为 5，意思是每看 5 个连续时间步，就提取一个趋势特征；第二层用了 dilation=2，每个卷积点跨 2 个步长取数，感受野更大。CNN 优点是训练快、占内存小，缺点是如果滑窗短但预测步长很长，容易“看到局部、忽略全天周期”，后面第四章会讲怎么配参数避开。
3. 从原始检测器数据到训练样本：滑窗构造与归一化边界
标题里带着“数据集”，但我见过太多人拿到别人整理好的 CSV 后，跳过了数据处理直接跑模型。高速公路检测器数据通常包含时间戳、断面编号、流量、平均速度、占用率，偶尔还有收费数据和天气数据。乱的地方在于缺失值、设备漂移、异常尖峰、时间粒度不一致。这段是我认为整个预测项目里性价比最高的部分，数据洗对了，后面模型随便跑都不会太差。
3.1 缺失值、异常值和“大车压队”造成的锯齿：三条清洗规则
高速公路断面流量数据最常见的三种脏样子：一是线圈或视频检测器在某几分钟掉线，流量变成 0 或者 NaN；二是流量和速度的采样时间没有对齐，断面间错位；三是受大车压队、事故缓行影响，流量出现连续几个点的尖刺。不能直接删，因为交通流数据的意义就是时间连续性，删掉一段等于人为切断序列。
import pandas as pd import numpy as np def clean_flow_data(df, time_col='time', flow_col='flow', freq='5min'): df = df.copy() df[time_col] = pd.to_datetime(df[time_col]) df = df.sort_values(time_col) df = df.set_index(time_col).resample(freq).mean() # 规则1：速度或占用率为0但流量很大，视为检测器异常 df.loc[(df['flow'] > 0) & (df['speed'] == 0), 'flow'] = np.nan # 规则2：缺失值不做均值填充，用时间插值保留曲线形状 df['flow'] = df['flow'].interpolate(method='time', limit=6) df['speed'] = df['speed'].interpolate(method='time', limit=6) # 规则3：用滚动中位数剔除单点尖峰 med = df['flow'].rolling(6, center=True, min_periods=1).median() mad = (df['flow'] - med).abs().rolling(6, center=True, min_periods=1).median() df.loc[(df['flow'] - med).abs() > 5 * mad + 10, 'flow'] = med return df.dropna(subset=['flow'])规则 3 用的是中位数加绝对中位差做阈值，而不是均值加标准差。原因是流量数据本身就是偏态分布，均值会被高峰时段拉高，用均值滤波会把正常早高峰削成缓坡。limit=6 的意思是连续缺失超过 6 个点就不继续插值，因为长时间缺失说明设备故障，插值等于编数据，不如在构造样本时直接丢弃对应窗口。
3.2 用滑窗构造样本：输入长度、预测步长与归一化参数
清洗完的数据还是二维表，要变成模型能吃的样本，就得滑窗。我对 5 分钟粒度的高速公路数据，一般用 input_len=12 代表过去 1 小时，pred_len=3 或 6，代表预测未来 15 或 30 分钟。为什么不用 1 分钟粒度？因为检测器噪声太大，且在线推理压力也大；30 分钟以上的长预测，单靠流量序列已经不够，得加天气和事件特征。
def create_sequences(df, input_len=12, pred_len=6, feat_cols=['flow', 'speed', 'occupancy']): values = df[feat_cols].values hour = df.index.hour.values / 24.0 dow = df.index.dayofweek.values / 7.0 # 特征列最后拼上小时和星期，帮助模型识别周期性 X_data = np.concatenate([ values, np.stack([hour, dow], axis=1) ], axis=1) X, y = [], [] for i in range(len(X_data) - input_len - pred_len + 1): X.append(X_data[i: i + input_len]) y.append(values[i + input_len: i + input_len + pred_len, 0]) return np.array(X), np.array(y)这个函数把过去 12 个时间步的流量、速度、占有率、小时、星期拼成一个 (12, 5) 的窗口，目标是后面 6 个时间步的流量。y 取的是流量列，因为多数场景只关心流量。小时和星期做成了连续特征而不是 one-hot，因为模型是回归任务，连续编码更省参数，也不容易过拟合。注意这里返回的是多个预测步的数组，而不是单步值，意味着我们做的是直接多步预测。
3.3 时序数据切分：不能随机的三个原因
老手第一眼就会问：数据切分怎么切的？如果你用 sklearn 的 train_test_split 随机切，那基本等于白做。原因有三条。第一，随机切分会让测试集里的样本散布在训练集中间，模型在训练时已经见过前一天、后一天的同个时段，测试指标会虚高。第二，滑窗之间是重叠的，相邻样本高度相似，随机划分会造成信息泄漏。第三，业务上线时，你要预测的是“下一个连续时段”，不是“过去某个随机时段”，所以验证方式也必须是时间上的前推。
split_date = '2023-11-01' train_mask = df.index < split_date valid_mask = (df.index >= split_date) & (df.index < '2023-12-01') X_train, y_train = create_sequences(df[train_mask]) X_val, y_val = create_sequences(df[valid_mask])切分顺序按时间排，训练集、验证集、测试集三个区间互不重叠。还有一个细节：缩放器要在训练集上 fit，再分别 transform 训练集、验证集和测试集。这个问题我在第五章会重点讲，因为很多人把整个数据集丢进 MinMaxScaler 再切分，结果验证集指标好看到不真实。
4. 一份代码跑通三种模型：LSTM、SAE、CNN的训练骨架
标题里的 python 代码，核心价值在训练流程的统一。三个模型结构不同，但训练步骤完全一致：造 DataLoader、定义损失函数、写早停循环、存 best model。把训练循环抽成一个函数，后面换模型只是换一行参数，这种做法可以省掉一半的重复代码。我先给统一训练循环，再分别讲三个模型的调参重点。
from torch.utils.data import DataLoader, TensorDataset def train_flow_model(model, X_train, y_train, X_val, y_val, epochs=60, lr=1e-3, batch_size=64, patience=10): train_loader = DataLoader( TensorDataset(torch.tensor(X_train, dtype=torch.float32), torch.tensor(y_train, dtype=torch.float32)), batch_size=batch_size, shuffle=True) optimizer = torch.optim.Adam(model.parameters(), lr=lr) loss_fn = nn.HuberLoss(delta=1.0) best_val = float('inf') bad_epochs = 0 for epoch in range(epochs): model.train() for xb, yb in train_loader: optimizer.zero_grad() y_pred = model(xb) loss = loss_fn(y_pred, yb) loss.backward() nn.utils.clip_grad_norm_(model.parameters(), 5.0) optimizer.step() model.eval() val_loss = 0.0 with torch.no_grad(): y_pred = model(torch.tensor(X_val, dtype=torch.float32)) val_loss = loss_fn(y_pred, torch.tensor(y_val, dtype=torch.float32)).item() # 早停逻辑：验证loss超过patience轮不降就停 if val_loss < best_val - 1e-4: best_val = val_loss bad_epochs = 0 torch.save(model.state_dict(), 'best_model.pt') else: bad_epochs += 1 if bad_epochs >= patience: break model.load_state_dict(torch.load('best_model.pt')) return model这个函数里的关键参数都很常规：HuberLoss 的 delta=1.0 让损失在误差较小时按平方走、误差较大时按绝对值走，适合流量尖峰；clip_grad_norm 设为 5.0，防止 LSTM 在长序列上梯度爆炸；patience=10 是保守值，如果数据集只有两周，建议改成 6，不然容易在最后几个 epoch 记住噪声。验证集直接用全量张量算 loss，因为在短时预测场景里验证集通常只有几千个窗口，不需要再分 batch。
4.1 LSTM的训练配置与收敛要点
LSTM 我一般从 hidden_size=64、num_layers=2、dropout=0.2、lr=1e-3 开始。断面流量的特征维度不多，hidden_size 设到 128 以上对精度帮助很小，只增加训练时间。收敛过程要盯训练 loss 和验证 loss 的间距：如果训练 loss 一直降、验证 loss 从某个 epoch 开始反弹，就是过拟合，此时优先加大 dropout 或减少 num_layers，而不是加数据。LSTM 对学习率比 CNN 敏感，lr=1e-2 很容易直接炸成 NaN，lr=1e-4 又收敛太慢，1e-3 是个折中点。
4.2 SAE的分步预训练和微调
SAE 不能像 LSTM 那样端到端一把梭，否则它的结构优势就体现不出来。正确顺序是：先冻结回归头，单独优化 encoder 和 decoder 的重构损失；预训练收敛后，把 decoder 摘掉，用 encoder 输出的 compress_dim 维向量接回归头，继续端到端微调。
def pretrain_sae(sae, X_train, epochs=20, lr=1e-3): loader = DataLoader(TensorDataset(torch.tensor(X_train, dtype=torch.float32).flatten(1, 2)), batch_size=64, shuffle=True) optimizer = torch.optim.Adam(sae.parameters(), lr=lr) loss_fn = nn.MSELoss() for epoch in range(epochs): for xb in loader: recon = sae.decoder(sae.encoder(xb)) loss = loss_fn(recon, xb) optimizer.zero_grad() loss.backward() optimizer.step()预训练时要先flatten(1, 2)把滑窗展平成向量，让 SAE 处理的是“时间步摊平后的特征”，而不是原始三维张量。预训练完成后再把 model 换成回归模式训练。如果你发现预训练后的微调比直接从随机初始化训练更差，说明 20 个 epoch 里重构过拟合了，把 pre-train epochs 降到 8 到 10 再试。
4.3 CNN对流量尖峰的响应与参数调整
CNNFlow 训练起来最省心，loss 曲线稳定，收敛快。真正需要调的是感受野：第一层 kernel_size=5、第二层 dilation=2，合起来能覆盖大约 9 个时间步，也就是 45 分钟内的趋势。如果 pred_len 是 6（30 分钟），这个感受野够用；如果 pred_len 是 12（60 分钟），建议把第一层 kernel_size 加到 7，或者加一层 dilation=4 的卷积，否则模型会“看不到”足够长的前文。BatchNorm 在这一类小模型里很重要，不加的话收敛会明显变慢。
5. TrafficFlowPrediction避坑清单：从滞后预测到数据泄漏的五次翻车
这一章写成清单，是因为交通流预测的坑非常集中，绝大多数项目翻车都翻在这几个地方。每一条我都按“现象、原因、解决”三步写，方便你排查时直接对照。
5.1 现象：预测曲线贴着真实值但永远慢半拍
你画图时会发现，预测曲线和真实曲线形状几乎一样，但整体向右偏移几个时间步。这是短时预测最容易出现的假性成功。原因在于流量序列自相关性太强，模型只要输出“最近一个时刻的值再顺延一点”，就能在 RMSE 上拿到不错成绩，尤其是平峰时段。解决这个问题的第一件事是换评估视角：把预测曲线滞后 N 个时间步后再和真实值比 RMSE，如果滞后 3 步的误差反而更低，说明模型根本没学到趋势，只是在抄近路。第二件事是别用递归多步预测，改用第三章的直接多步方式，让输出直接对应真实未来值，模型就没有机会靠“多步传播误差”来平滑输出。
5.2 现象：验证集漂亮，上线就崩
这个坑的根源几乎都是归一化泄漏。很多教程会把整个数据集丢进 MinMaxScaler 再切分，这么做的问题是：缩放器用了未来数据的最大值和最小值，验证集和测试集的分布信息提前混进了训练过程。在高速公路场景里，节假日高峰和夜里低峰很容易让 Max/Min 偏移，训练时的输入分布和上线时完全不一样。解决方法是先把数据集按时间切好，再用训练集的统计量做归一化。
from sklearn.preprocessing import MinMaxScaler scaler = MinMaxScaler() scaler.fit(X_train.reshape(-1, X_train.shape[-1])) X_train = scaler.transform(X_train.reshape(-1, X_train.shape[-1])).reshape(X_train.shape)这条规则要严格执行到每个特征列，尤其是流量和占有率，不能只在模型输入上做一次。上线推理时也得用同一个 scaler 实例转换新数据，不能每次重新 fit，否则相当于每天都在换量纲。我在项目里吃过这个亏，换了模型结构都没用，最后发现是归一化的问题，属于典型的方向错了。
5.3 现象：节假日、事故、恶劣天气出现时，模型输出一个“平均流量”
日常验证误差看着还行，但只要碰上国庆前一天、上午十点的事故管制、暴雨天气，预测值就变成一条平缓曲线，完全不响应。原因是训练数据里事件样本占比太低，模型学不到“突发事件导致流量断崖式下降”的映射。解决办法分三步：第一，特征里加入是否节假日、是否周末、小时、星期，至少让模型知道周期性；第二，事故和管制时间如果能有日志，做成一个 0/1 特征拼进滑窗；第三，评估时要单独统计事件日误差，不要和普通日混在一起报，否则你永远不知道模型在关键时刻有多差。HuberLoss 对这类尖峰也有一点帮助，但单靠损失函数救不了缺特征的问题。
5.4 现象：RMSE一直降，但MAPE在高峰期就是下不去
如果你同时看 RMSE 和 MAPE，会发现两个指标经常打架。流量值正常在几百到几千之间，夜间低峰时一个很小的绝对误差就能把 MAPE 拉爆，而高峰期的误差虽然绝对数值大，但在 MAPE 里占比反而不高。所以模型会倾向把精力放在预测准夜间平峰，高峰期则“平均化”。解决方法是把评估按高峰时段分开：早高峰（7-10 点）、晚高峰（17-20 点）、平峰时段各自统计 RMSE。如果高峰期的 MAPE 明显更差，说明特征缺失或者模型容量不够，此时再去加卷积层或调 dropout 才有意义，否则只是拿着一个混合指标瞎调参。
5.5 现象：只用单断面历史流量，早晚高峰的峰值总被削平
很多初学者做的只用了当前断面的流量时间序列，模型预测出来的曲线在高峰顶点处总是偏小。原因是高速公路上的流量不是孤立的，上游断面半小时前的拥堵会传导到下游；反过来，下游排队也会回堵。单断面模型完全靠历史自回归，等价于假设未来是过去的平滑延续，自然学不会突变。解决方法是扩特征：把上游 2 到 3 个断面的流量、速度拼进输入特征。拼接方式可以简单粗暴，直接把上游序列拼到特征维度上，也可以在数据预处理时构建一个断面之间的邻接矩阵，按空间距离做特征加权，再把加权结果喂给模型。对于只有普通 python 环境、不想引入复杂图模型的团队，先做特征拼接是最稳妥的。
6. 让预测模型真的能上勤：高峰误差、滞后修正与重训练节奏
模型在离线验证集上跑得再好，离“能上岗”还有一段路。我最后集中说三个让模型可用的细节：高峰分段评估、滞后修正、重训练节奏。这三个技巧在公开数据集上不显眼，但在真实系统里非常关键。
6.1 用高峰时段分段指标做验收
离线评估不能只看一组总误差。我会把一天按早高峰、晚高峰、平峰、夜间分成四段，每段单独算 RMSE、MAPE 和最大误差。实测里最常见的验收标准是：高峰期 MAPE 不超过 15%，夜间不超过 25%。如果高峰段过不了，就针对高峰段补特征，而不是盲目改模型结构。
6.2 输出结果做一个滞后修正
即使采用直接多步预测，模型在峰顶时刻还是容易偏缓。一个轻量技巧是记录最近 N 天的预测残差，用指数加权平均修正当前输出。注意这个修正要在评估指标计算完之后再做，否则等于拿过去的信息偷看当前结果。我做这个修正时一般取 alpha=0.25，修正幅度超过 8% 就停手，说明原始预测偏差太大，问题在模型不在修正。
6.3 重训练节奏与告警阈值
交通流预测模型不是训练一次就能长期用的，收费站改造、周边路网开通都会改变流量模式。我一般保留最近 60 天数据，每周触发一次增量训练，同时把在线误差滚动和最近 30 天分布对比，如果误差超过历史 P95 持续三天，就发告警并回溯特征是否发生变化。重训练不用完全重来，用之前的权重继续训练 10 个 epoch 就够了。
我自己做这个方向时，最深的教训是：把一堆模型代码跑通很容易，把评估指标定义清楚、把归一化切分做干净，反而要花掉三分之二的时间。后来我不管拿到什么数据，都先做一张按小时切分的误差热力图，再决定改模型还是改特征。希望你也能少走这些弯路，祝你一次跑通。
本文还有配套的精品资源，点击获取
