简介这份资源面向交通工程、智能交通与深度学习方向的学习者和研究者提供一套基于CNN与LSTM的高速公路交通流预测完整实现方案并融合SAE堆栈自编码器进行特征提取可用于课程设计、毕业课题或科研实验中的短时交通流预测任务。压缩包共20个文件、约31.48MB包含7个Python脚本、6个h5模型权重、4个json参数配置及2张png结果图覆盖数据预处理、模型训练、对比实验与结果可视化等环节另附PeMS数据集压缩包便于直接复现实验。资源已有2441人学习下载热度较高。读者可据此掌握CNN-LSTM混合模型的搭建思路、SAE特征降维方法、多模型对比评估流程并借助已训练好的权重文件快速验证预测效果适合具备一定Python与深度学习基础、希望深入理解交通流预测建模全流程的读者参考。1. 交通流预测这套代码为什么高速场景下LSTMSAECNN的组合值得认真跑一遍高速公路上跑交通流预测最怕的不是模型不收敛而是你拿到的数据根本撑不起一个能上路的预测器。很多做智慧交通的团队手里攥着收费站流水、门架交易记录、线圈检测器数据却卡在同一个地方单点LSTM预测短时流量还行一遇到节假日突增、事故封路、天气突变就集体翻车。原因不复杂——纯时序模型只学到了“过去怎么变”没学到“路网怎么连”。而标题里这套LSTMSAECNN的组合本质上是在做三件事SAE先把高维、含噪的原始交通参数压成干净的低维表征CNN从路网空间拓扑里抽路段之间的相互影响LSTM再在时间轴上建模演化规律。三者叠起来才是高速交通流预测比较稳的落地路径。这套代码适合两类人一是手里有高速门架或线圈数据、想快速搭一个可复现baseline的算法工程师二是做交通信息化项目、需要向业务方解释“为什么预测值可信”的技术负责人。下面我按数据准备、SAE降维、CNN空间特征、LSTM时序建模、调参避坑的顺序把这条链路拆开讲透。2. 数据准备与预处理从原始交通参数到模型可吃的张量2.1 高速交通流数据的典型字段与清洗逻辑高速场景下的原始数据常见来源是门架交易流水、线圈检测器、微波雷达。不管哪种落到CSV里通常长这样时间戳、路段ID或门架编号、车道号、流量veh/h、平均速度km/h、占有率%有时还有车型分类计数。做预测之前第一步不是急着喂模型而是把“脏”处理干净。我一般会按下面这个顺序走时间对齐把不同采样频率的数据统一到5分钟粒度。门架数据可能是逐车交易线圈可能是30秒聚合先重采样到5分钟用sum聚合流量、用mean聚合速度和占有率。缺失值处理高速数据缺失很常见夜间低峰期检测器可能无数据。不要直接填0那会制造虚假的“流量骤降”。我一般用前后各两个时间窗的线性插值如果连续缺失超过6个点即30分钟就标记为无效段训练时mask掉。异常值剔除流量为负、速度超过120km/h除非是特殊路段、占有率超过100%的直接置为NaN再插值。归一化流量和速度量纲差很大必须做Min-Max归一化到[0,1]。注意归一化参数只能从训练集算再应用到验证和测试集否则就是数据泄露。下面这段代码是我常用的清洗骨架直接可跑import pandas as pd import numpy as np def clean_traffic_data(df, freq5T): df: 原始交通数据含 timestamp, segment_id, flow, speed, occupancy freq: 重采样频率默认5分钟 df[timestamp] pd.to_datetime(df[timestamp]) df df.set_index(timestamp) # 按路段分组重采样 cleaned [] for seg_id, group in df.groupby(segment_id): resampled group.resample(freq).agg({ flow: sum, speed: mean, occupancy: mean }) # 线性插值最多连续补6个点 resampled resampled.interpolate(methodlinear, limit6) # 异常值处理 resampled.loc[resampled[flow] 0, flow] np.nan resampled.loc[resampled[speed] 120, speed] np.nan resampled.loc[resampled[occupancy] 100, occupancy] np.nan resampled resampled.interpolate(methodlinear, limit6) resampled[segment_id] seg_id cleaned.append(resampled) result pd.concat(cleaned).reset_index() # 删除仍然缺失的行 result result.dropna() return result逻辑说明先按路段分组再重采样避免不同路段数据混在一起导致聚合错误。limit6控制插值范围超过就保留NaN后续训练时跳过。异常值先置NaN再插值比直接删除行更温和能保留时间连续性。参数说明freq5T表示5分钟如果数据本身是15分钟粒度就改成15T。limit根据你的采样频率调整原则是不要补超过30分钟的连续缺失。2.2 构建时空张量的两种切法清洗完的数据还是二维表模型要吃的是三维张量。高速路网天然有空间结构常见两种切法第一种按路段ID排成“伪图像”。假设你选了N个连续路段每个时间步有F个特征流量、速度、占有率那一个样本就是(T, N, F)T是历史时间窗长度。这种切法适合CNN直接卷因为路段顺序隐含了空间相邻关系。第二种按图结构构建邻接矩阵。如果你有路段之间的拓扑连接关系比如门架上下游可以构造邻接矩阵A然后用图卷积。但标题里是CNNSAELSTM我建议先用第一种实现简单效果在高速场景下已经够用。滑动窗口切样本的代码def create_sequences(data, n_steps_in, n_steps_out, n_segments, n_features): data: 清洗后的DataFrame按时间排序 n_steps_in: 历史时间步数比如12即1小时 n_steps_out: 预测时间步数比如3即15分钟 n_segments: 路段数量 n_features: 特征数量 X, y [], [] # 先转成 (总时间步, 路段数, 特征数) values data[[flow, speed, occupancy]].values total_steps len(values) // n_segments values values.reshape(total_steps, n_segments, n_features) for i in range(total_steps - n_steps_in - n_steps_out 1): X.append(values[i:in_steps_in, :, :]) # 预测目标只取流量 y.append(values[in_steps_in:in_steps_inn_steps_out, :, 0]) return np.array(X), np.array(y)逻辑说明先把一维表reshape成(时间, 路段, 特征)再滑动窗口。y只取流量作为预测目标因为速度和占有率在预测任务里通常作为辅助特征输入不作为输出。参数说明n_steps_in12对应1小时历史n_steps_out3对应15分钟预测。高速场景下预测窗口太长误差会累积15到30分钟是比较实用的范围。3. SAE降维把高维交通参数压成干净表征3.1 为什么高速场景需要SAE而不是直接PCA交通数据维度不算特别高但噪声大、非线性强。PCA是线性降维遇到流量和速度之间的非线性关系就力不从心。SAE稀疏自编码器在普通自编码器基础上加了稀疏惩罚强迫隐层只有少量神经元激活学到的表征更鲁棒。高速场景下SAE能起到两个作用一是把原始特征里的检测器噪声、短时波动滤掉二是把多个相关特征流量、速度、占有率高度相关压缩成低维独立因子减少后续LSTM和CNN的输入维度。我一般把SAE的隐层维度设成8到16之间。太小会丢信息太大起不到降维效果。稀疏系数一般取0.05到0.1具体看重构误差曲线。3.2 SAE网络结构与训练代码下面是一个三层的SAE输入维度是n_segments * n_features隐层逐步压缩import torch import torch.nn as nn class SparseAutoEncoder(nn.Module): def __init__(self, input_dim, hidden_dims[64, 16], sparsity0.05): super().__init__() # 编码器 self.encoder nn.Sequential( nn.Linear(input_dim, hidden_dims[0]), nn.ReLU(), nn.Linear(hidden_dims[0], hidden_dims[1]), nn.ReLU() ) # 解码器 self.decoder nn.Sequential( nn.Linear(hidden_dims[1], hidden_dims[0]), nn.ReLU(), nn.Linear(hidden_dims[0], input_dim), nn.Sigmoid() # 输入已归一化到[0,1] ) self.sparsity sparsity def forward(self, x): encoded self.encoder(x) decoded self.decoder(encoded) return encoded, decoded def sparse_loss(self, encoded): # 计算平均激活值 rho_hat torch.mean(encoded, dim0) rho self.sparsity # KL散度稀疏惩罚 kl rho * torch.log(rho / (rho_hat 1e-8)) \ (1 - rho) * torch.log((1 - rho) / (1 - rho_hat 1e-8)) return torch.sum(kl)逻辑说明编码器两层第一层从输入维度降到64第二层降到16。解码器对称还原。sparse_loss用KL散度约束隐层平均激活值接近sparsity让表征稀疏化。参数说明input_dim等于n_segments * n_features比如20个路段3个特征就是60。hidden_dims[64,16]是经验值数据量小可以改成[32,8]。sparsity0.05表示期望只有5%的神经元激活这个值太小会导致欠拟合太大会失去稀疏效果。训练SAE时损失函数是重构误差加稀疏惩罚def train_sae(model, dataloader, epochs50, lr1e-3, sparse_weight0.01): optimizer torch.optim.Adam(model.parameters(), lrlr) mse nn.MSELoss() for epoch in range(epochs): total_loss 0 for batch in dataloader: x batch.float() encoded, decoded model(x) recon_loss mse(decoded, x) sparse_penalty model.sparse_loss(encoded) loss recon_loss sparse_weight * sparse_penalty optimizer.zero_grad() loss.backward() optimizer.step() total_loss loss.item() if (epoch1) % 10 0: print(fEpoch {epoch1}, Loss: {total_loss/len(dataloader):.6f})逻辑说明sparse_weight控制稀疏惩罚的强度一般取0.01到0.1。训练完后只保留编码器部分把原始输入转成低维表征再喂给后面的CNNLSTM。参数说明epochs50通常够用看重构误差不再下降就可以停。lr1e-3是Adam的常用起点。4. CNNLSTM联合建模空间特征与时序依赖怎么串4.1 CNN提取路段空间关联的卷积核设计高速路网的空间关联直观理解就是相邻路段的流量会相互影响。上游堵了下游很快也会堵。CNN的卷积核在路段维度上滑动能捕捉这种局部空间依赖。输入张量是(batch, n_segments, n_features)我一般用一维卷积在路段维度上卷class SpatialCNN(nn.Module): def __init__(self, n_features, n_segments, out_channels32): super().__init__() # 一维卷积在路段维度上滑动 self.conv1 nn.Conv1d(in_channelsn_features, out_channelsout_channels, kernel_size3, padding1) self.bn1 nn.BatchNorm1d(out_channels) self.conv2 nn.Conv1d(out_channels, out_channels, kernel_size3, padding1) self.bn2 nn.BatchNorm1d(out_channels) self.relu nn.ReLU() def forward(self, x): # x: (batch, n_segments, n_features) - 转成 (batch, n_features, n_segments) x x.permute(0, 2, 1) x self.relu(self.bn1(self.conv1(x))) x self.relu(self.bn2(self.conv2(x))) # 转回 (batch, n_segments, out_channels) x x.permute(0, 2, 1) return x逻辑说明Conv1d的输入通道是特征数输出通道是卷积核数量。kernel_size3表示每次看相邻3个路段。两层卷积堆叠感受野扩大到5个路段。padding1保持路段维度不变方便后续和LSTM拼接。参数说明out_channels32是空间特征维度可以调到64但要注意过拟合。kernel_size根据路段密度调整路段间距大就用3密集就用5。4.2 LSTM时序建模与多步预测输出CNN输出的是每个时间步的空间特征接下来要沿时间轴建模。把CNN输出按时间步展开喂给LSTMclass CNNLSTM(nn.Module): def __init__(self, n_features, n_segments, cnn_channels32, lstm_hidden64, n_steps_out3): super().__init__() self.cnn SpatialCNN(n_features, n_segments, cnn_channels) self.lstm nn.LSTM(input_sizecnn_channels * n_segments, hidden_sizelstm_hidden, num_layers2, batch_firstTrue, dropout0.2) self.fc nn.Linear(lstm_hidden, n_segments * n_steps_out) self.n_segments n_segments self.n_steps_out n_steps_out def forward(self, x): # x: (batch, n_steps_in, n_segments, n_features) batch, n_steps, n_seg, n_feat x.shape # 每个时间步过CNN cnn_out [] for t in range(n_steps): out self.cnn(x[:, t, :, :]) # (batch, n_seg, cnn_channels) cnn_out.append(out) cnn_out torch.stack(cnn_out, dim1) # (batch, n_steps, n_seg, cnn_channels) # 展平特征维度 cnn_out cnn_out.reshape(batch, n_steps, -1) # LSTM lstm_out, _ self.lstm(cnn_out) # 取最后一个时间步 last_out lstm_out[:, -1, :] # 全连接输出多步预测 pred self.fc(last_out) pred pred.reshape(batch, self.n_steps_out, self.n_segments) return pred逻辑说明每个时间步单独过CNN提取空间特征再按时间顺序喂LSTM。LSTM最后时间步的隐状态包含整个历史窗口的信息通过全连接层映射到多步预测输出。输出形状是(batch, n_steps_out, n_segments)对应每个未来时间步每个路段的流量预测。参数说明lstm_hidden64可以调到128但要注意训练时间。num_layers2比单层效果好但超过3层容易过拟合。dropout0.2在LSTM层间做正则化。训练时用MSE损失优化器选Adam学习率用余弦退火def train_model(model, train_loader, val_loader, epochs100, lr1e-3): optimizer torch.optim.Adam(model.parameters(), lrlr) scheduler torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_maxepochs) mse nn.MSELoss() best_val_loss float(inf) for epoch in range(epochs): model.train() train_loss 0 for X, y in train_loader: pred model(X.float()) loss mse(pred, y.float()) optimizer.zero_grad() loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm5.0) optimizer.step() train_loss loss.item() model.eval() val_loss 0 with torch.no_grad(): for X, y in val_loader: pred model(X.float()) val_loss mse(pred, y.float()).item() scheduler.step() if val_loss best_val_loss: best_val_loss val_loss torch.save(model.state_dict(), best_model.pth) if (epoch1) % 10 0: print(fEpoch {epoch1}, Train: {train_loss/len(train_loader):.6f}, Val: {val_loss/len(val_loader):.6f})逻辑说明clip_grad_norm_防止LSTM梯度爆炸max_norm5.0是经验值。余弦退火让学习率从初始值平滑降到接近0有助于收敛到更优解。保存验证集损失最低的模型避免过拟合。参数说明epochs100配合早停实际可能50轮就收敛。lr1e-3是起点如果损失震荡就降到5e-4。5. 避坑与排查高速交通流预测里最容易翻车的5个地方5.1 数据泄露归一化参数从验证集算现象训练集损失很低验证集损失也低但一上测试集误差翻倍。原因归一化时用了全量数据的min/max验证集和测试集的信息泄露到了训练过程。解决严格用训练集的min/max保存下来应用到验证和测试集。代码里fit_transform只在训练集调用其他用transform。5.2 路段顺序错乱CNN卷了个寂寞现象模型训练不收敛或者收敛后预测值在所有路段上几乎一样。原因构建张量时路段ID没有按空间顺序排列CNN在路段维度上卷积时相邻位置其实是相距很远的路段空间关联完全错位。解决按门架或路段的实际上下游顺序排序确保相邻索引对应相邻路段。如果路网有分支考虑用图结构替代一维卷积。5.3 预测窗口太长误差累积到不可用现象预测1步5分钟时MAPE在8%左右预测6步30分钟时MAPE飙到25%以上。原因多步预测直接输出所有步长LSTM最后时间步的隐状态要承载太长的预测任务信息不够用。解决缩短预测窗口到3步15分钟或者改用seq2seq结构让解码器逐步输出。高速场景下15分钟预测对诱导和管控已经够用。5.4 SAE稀疏系数设太小隐层几乎全死现象SAE训练完后编码器输出大部分是0后续模型学不到东西。原因sparsity设成0.01甚至更小KL散度惩罚太强把隐层激活压得太狠。解决从0.1开始试观察编码器输出的平均激活值控制在0.05到0.15之间比较合理。如果还是死把sparse_weight降到0.001。5.5 批次大小与LSTM状态别在训练时重置隐状态现象训练损失波动很大每个batch之间损失跳变。原因LSTM的隐状态在batch之间没有detach梯度跨batch传播导致不稳定。解决每个batch前调用model.lstm.reset_parameters()或者手动hidden None。另外batch_size不要太小32到64比较稳太小会让LSTM难以捕捉长程依赖。6. 进阶技巧用注意力加权和滚动预测把MAPE再压两个点基础版CNNLSTM跑通后如果想把预测精度再往上提我一般会加两个东西注意力机制和滚动预测。注意力机制加在LSTM输出后面让模型自动学习哪些历史时间步对当前预测更重要。高速场景下早高峰的预测可能更依赖前30分钟的数据而平峰期可能前1小时都差不多。注意力权重能动态调整这个窗口。实现上在LSTM输出后加一个nn.Linear算注意力分数再softmax归一化加权求和后送全连接class Attention(nn.Module): def __init__(self, hidden_dim): super().__init__() self.attn nn.Linear(hidden_dim, 1) def forward(self, lstm_out): # lstm_out: (batch, n_steps, hidden_dim) scores self.attn(lstm_out) # (batch, n_steps, 1) weights torch.softmax(scores, dim1) context torch.sum(weights * lstm_out, dim1) # (batch, hidden_dim) return context, weights逻辑说明attn把每个时间步的隐状态映射成一个分数softmax后得到权重加权求和得到上下文向量。这个向量替代原来LSTM最后时间步的输出送全连接层。参数说明hidden_dim和LSTM的lstm_hidden一致。注意力层不需要额外调参训练时自动学。滚动预测是另一个实用技巧。不要一次性预测未来3步而是每次预测1步然后把预测值拼回输入再预测下一步。这样每一步都有最新的历史信息误差不会一次性累积。代价是推理时间变长但高速场景下15分钟预测滚动3次完全可接受。实现上写个循环def rolling_predict(model, initial_input, n_steps_out): initial_input: (1, n_steps_in, n_segments, n_features) model.eval() preds [] current_input initial_input.clone() with torch.no_grad(): for _ in range(n_steps_out): pred model(current_input.float()) # (1, 1, n_segments) preds.append(pred) # 把预测的流量拼到输入末尾去掉最早的时间步 # 这里简化处理只更新流量特征速度和占有率沿用上一时刻 new_step current_input[:, -1, :, :].clone() new_step[:, :, 0] pred.squeeze(1) current_input torch.cat([current_input[:, 1:, :, :], new_step.unsqueeze(1)], dim1) return torch.cat(preds, dim1)逻辑说明每次预测一步把预测流量替换掉输入窗口最后一个时间步的流量特征然后窗口向前滑动。速度和占有率没有预测沿用上一时刻的值这在短时预测里是合理近似。参数说明n_steps_out和训练时一致。滚动预测的误差比直接多步输出低但要注意如果第一步误差大后续会累积所以第一步的精度是关键。最后说个我自己的习惯每次跑完模型不要只看MAPE一定要把预测曲线和真实曲线叠在一起看。高速场景下模型在平峰期往往预测得很准但早高峰的峰值经常被削平。如果发现峰值预测偏低检查训练集里高峰样本的权重或者加一个峰值敏感的损失函数比如对高流量样本加大MSE权重。这个技巧比调网络结构往往更管用。希望帮到你。本文还有配套的精品资源点击获取