简介面向道路通行时间预测场景的LSTM回归项目适合学习时序预测与深度神经网络回归的中高级开发者。方案采用LSTM3层全连接层结构将各link的历史旅行时间作为特征拟合上下游道路通行时间的复杂非线性关系并给出数据预处理思路包括交通事故时段剔除与空值前向填充等可继续优化的细节。压缩包共12个文件以4个Python脚本为主体覆盖数据读取、LSTM模型定义、单文件训练与全量训练流程另有4个txt数据说明与样例数据、工程配置文件及README文档整体约61.78MB结构清晰便于直接运行和二次修改。已有132人学习下载适合作为天池道路通行时间预测赛题的入门基线与扩展参考。1. 道路通行时间预测为什么选 LSTM序列回归不是瞎套模型每天早晚高峰导航给出的通行时间总是和实际差出十几分钟问题不在于地图数据不准而是通行时间本身是一条强时序相关的曲线当前时刻的路况高度依赖过去半小时甚至一小时的变化趋势。LSTM神经网络天然适合这类序列回归任务它能通过门控机制记住早高峰的拥堵从几点开始酝酿、几点达到峰值再把这些记忆交给全连接层做数值输出。标题里的“LSTM3层全连接层”是这类预测问题里最常见的落地方案前面的循环层负责从历史序列里抽特征后面的全连接层负责把特征回归成具体分钟数。这篇文章面向的是想自己动手跑通这个方案的工程师——不管你是要做智慧交通项目还是给园区做通勤预测接下来的数据构造、模型搭建和参数调整都能直接照搬。2. 数据准备把通行时间历史记录改造成 LSTM 能吃的样本2.1 LSTM 输入长什么样三维张量与滑动窗口很多人第一次写 LSTM 就卡在输入形状上。全连接网络的输入是二维的(样本数, 特征数)而 LSTM 要求三维(batch_size, seq_len, input_size)。其中seq_len是时间步数也就是你用过去多少个时间点的数据来预测下一个时间点input_size是每个时间点上有几个特征。道路通行时间预测里最常见的原始数据长这样time, segment_id, travel_time 2024-01-15 08:00:00, seg_001, 23.5 2024-01-15 08:05:00, seg_001, 26.8每条记录是一个路段在某个时刻的平均通行时间间隔固定5分钟或15分钟。要做预测得先把这些平铺的记录转成“过去 N 个点预测下一点”的监督学习样本。假设用过去 6 个时间点半小时预测未来 1 个时间点5 分钟后那么模型看到的是 6 步的序列每一步上只有 1 个特征通行时间本身。这里有个关键问题seq_len怎么选。选太短模型看不到早高峰的完整形成过程选太长训练样本量骤减而且 LSTM 对超长序列的记忆会衰减。常见做法是先用 6 或 12 起步——对应 30 分钟到 1 小时的上下文基本能覆盖一个完整的路况变化周期。2.2 构建数据集滑动窗口切割与归一化的顺序我一般会把数据准备写成独立函数核心逻辑就是用一个固定大小的窗口在时间序列上滑动每滑一步产出一个样本。下面这段代码是纯 Python 实现不依赖 Pandas 以外的东西import numpy as np import pandas as pd def create_sequences(data, seq_len6, pred_len1): 把一维时间序列切成 (X, y) 监督学习样本 data: 一维数组通行时间序列按时间升序排列 seq_len: 用过去多少个时间步 pred_len: 预测未来多少个时间步 X, y [], [] for i in range(len(data) - seq_len - pred_len 1): X.append(data[i : i seq_len]) y.append(data[i seq_len : i seq_len pred_len]) return np.array(X), np.array(y) # 读取原始数据按时间排序 df pd.read_csv(travel_time.csv, parse_dates[time]) df df.sort_values(time).reset_index(dropTrue) # 这里只取单个路段的通行时间列做演示 raw_series df[travel_time].values.astype(float) # 先归一化再切窗口 from sklearn.preprocessing import MinMaxScaler scaler MinMaxScaler(feature_range(0, 1)) scaled_series scaler.fit_transform(raw_series.reshape(-1, 1)).flatten() X, y create_sequences(scaled_series, seq_len6, pred_len1) print(f样本总量: {X.shape[0]}, 输入形状: {X.shape}, 输出形状: {y.shape})逻辑说明create_sequences里的循环是从序列头部一路滑到尾部每个位置取seq_len个历史点作为X紧接着的pred_len个点作为y。pred_len1就是单步预测——用过去 6 个时刻预测下一个时刻。这里要特别强调归一化和切窗口的顺序。以上代码是先归一化再切窗口这是正确做法。如果先切窗口再分别对每个窗口做归一化等于把不同时间段的相对大小关系抹掉了——早高峰的 30 分钟和凌晨的 5 分钟会被各自缩放到同一个区间模型完全失去判断“现在到底堵不堵”的能力。另外MinMaxScaler只能fit在训练集上验证集和测试集要用同一个scaler.transform这一点在第 5 章会展开说。2.3 数据划分训练、验证、测试要按时间切分类任务里习惯用train_test_split随机打乱数据但时间序列绝对不能这么做。通行时间的预测本质上是“用昨天和前天推测今天”如果测试集里混进了训练集相邻时间段的数据模型等于开卷考试——它见过那个时段的上下文预测准确率会虚高得离谱。# 按时间顺序切分禁止 shuffle train_ratio, val_ratio 0.7, 0.15 n len(X) train_end int(n * train_ratio) val_end int(n * (train_ratio val_ratio)) 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:] # 转成 PyTorch 需要的格式Tensor 且形状为 (样本数, 时间步, 特征数) import torch X_train torch.tensor(X_train, dtypetorch.float32).unsqueeze(-1) y_train torch.tensor(y_train, dtypetorch.float32).unsqueeze(-1) print(f训练集输入形状: {X_train.shape}) # (样本数, 6, 1)参数说明unsqueeze(-1)这里把形状从(样本数, 6)变成(样本数, 6, 1)补上的最后一个维度就是input_size1。如果你后面要加入星期几、天气等额外特征这个维度会变成 2、3 甚至更多。验证集的比例 I 一般给 15% 左右留太少看不出过拟合趋势留太多挤占训练数据。3. 模型搭建LSTM 层 3 层全连接层的结构与参数3.1 网络结构设计LSTM 提取时序特征全连接做回归输出标题给了明确的结构指引LSTM 3 层全连接层。这是序列回归任务里特别顺手的一套组合。LSTM 的作用是读入整段历史序列把每个时间步的信息压缩成隐藏状态全连接层的作用是把最后一个隐藏状态映射成具体的通行时间数值。有个设计细节值得留意LSTM 在每个时间步都会输出一个隐藏状态但做回归预测时一般只取最后一个时间步的输出——因为它看过完整的序列携带的信息最全。比如输入 6 个时间步前 5 步的隐藏状态是中间产物第 6 步的隐藏状态才是“看完整个序列之后的理解”。import torch.nn as nn class TravelTimeLSTM(nn.Module): def __init__(self, input_size1, hidden_size32, num_layers1, dropout0.2): super().__init__() # LSTM 层提取时序特征 self.lstm nn.LSTM( input_sizeinput_size, hidden_sizehidden_size, num_layersnum_layers, batch_firstTrue, dropoutdropout if num_layers 1 else 0 ) # 3 层全连接把 LSTM 输出的特征回归成通行时间 self.fc_layers nn.Sequential( nn.Linear(hidden_size, 64), nn.ReLU(), nn.Dropout(dropout), nn.Linear(64, 32), nn.ReLU(), nn.Linear(32, 1) # 单步预测输出 1 个值 ) def forward(self, x): # x 形状: (batch, seq_len, input_size) lstm_out, (h_n, c_n) self.lstm(x) # 取最后一个时间步的隐藏状态作为特征 last_hidden lstm_out[:, -1, :] # 形状: (batch, hidden_size) output self.fc_layers(last_hidden) return output逻辑说明nn.LSTM需要指定batch_firstTrue这样输入形状直接是(batch, seq_len, input_size)不用在 forward 里做转置。lstm_out[:, -1, :]取出序列最后一个时间步的隐藏状态形状是(batch, hidden_size)。之后接 3 层全连接中间两层带ReLU激活最后一层是纯线性层因为回归任务输出连续数值不需要sigmoid或softmax压缩到 0-1 区间。这里有个容易纠结的点为什么 3 层全连接中间要夹Dropout通行时间预测的输入特征维度很低有时只有 1 个特征模型容量很容易过剩Dropout是防止它死记硬背训练集里某几天的特殊波动。设0.2已经够用太高会让训练收敛变慢。3.2 关键参数hidden_size、num_layers、dropout 怎么定这些参数没有标准答案但有合理的起步区间。hidden_size是 LSTM 隐藏状态的维度相当于模型用多少个数值来“记住”这个序列的特征。通行时间预测的特征本来就少hidden_size设 16 到 64 之间都合理——32 是常见起步值。太小记不住早晚高峰的变化模式太大容易过拟合且训练变慢。num_layers是 LSTM 堆叠层数。1 层是默认选项2 层能捕捉更复杂的层级特征比如先识别“当前处于拥堵上升期”再判断“这个上升期有多陡”3 层以上在小数据集上几乎必然过拟合。我自己的习惯是先跑 1 层看基线效果不够再试 2 层不要一上来就堆深度。dropout在 LSTM 层和全连接层里的行为不同。nn.LSTM的dropout参数只在num_layers 1时生效——它作用于层与层之间不会作用于时间步之间。如果只有 1 层 LSTM传dropout0.5也不会有任何效果这是个很容易让人误会的点。3.3 损失函数与优化器回归任务标配回归任务最常用的损失函数是MSELoss均方误差它对大误差的惩罚更重——预测偏差 10 分钟和偏差 2 分钟损失差距是 25 倍。这在通行时间场景下是合理的因为严重误判会导致用户行程规划完全失效。如果想更温和一些L1Loss平均绝对误差对异常值的敏感度更低但收敛速度会稍慢。model TravelTimeLSTM(input_size1, hidden_size32, num_layers1, dropout0.2) criterion nn.MSELoss() optimizer torch.optim.Adam(model.parameters(), lr0.001, weight_decay1e-5) # 训练参数 epochs 100 batch_size 64参数说明lr0.001是 Adam 优化器在中小规模回归任务上的安全起步值比这大容易损失震荡比这小收敛太慢。weight_decay1e-5是 L2 正则化虽然视觉上很不起眼但能压制模型对训练集中极端值的过度拟合。需要提前说清楚的是PyTorch 的nn.LSTM默认会随机初始化权重你每次跑的初始条件都不一样。第一次训练前最好固定随机种子def set_seed(seed42): torch.manual_seed(seed) np.random.seed(seed) set_seed()这样至少保证同一个模型结构跑出来的结果可复现——在实际项目里结果不可复现会被同事当成黑匣子后续调参会变得非常痛苦。4. 训练与评估从损失曲线到 MAE、RMSE4.1 训练循环怎么写梯度裁剪与早停LSTM 训练有个著名的坑梯度爆炸。因为时间步之间是链式求导序列一长梯度可能指数级增长损失直接变成nan。解决手段是梯度裁剪——给梯度的范数设个上限超过就按比例缩回去。from torch.utils.data import TensorDataset, DataLoader dataset TensorDataset(X_train, y_train) dataloader DataLoader(dataset, batch_sizebatch_size, shuffleFalse) def train_model(model, dataloader, criterion, optimizer, epochs, clip1.0): model.train() for epoch in range(epochs): epoch_loss 0.0 for batch_X, batch_y in dataloader: # batch_X: (batch, seq_len, input_size) # batch_y: (batch, pred_len, input_size1) optimizer.zero_grad() predictions model(batch_X) # 输出: (batch, 1) loss criterion(predictions, batch_y[:, -1, :]) loss.backward() # 梯度裁剪防止梯度爆炸 nn.utils.clip_grad_norm_(model.parameters(), clip) optimizer.step() epoch_loss loss.item() * batch_X.size(0) avg_loss epoch_loss / len(dataloader.dataset) if (epoch 1) % 20 0: print(fEpoch {epoch1}/{epochs}, Loss: {avg_loss:.6f})逻辑说明batch_y[:, -1, :]是因为我构造数据时y的形状是(样本数, pred_len1, 1)这里只取它的最后一个时间步形状变成(batch, 1)才能和模型的输出predictions对齐。clip_grad_norm_设置clip1.0意味着所有参数的梯度按全局范数归一化到不超过 1.0这是我在时间序列训练里的默认值。另外一个必须做的操作是早停Early Stopping。做法是每个 epoch 结束后在验证集上算一次损失如果连续多个 epoch 没有下降就停止训练。实测里未加早停的模型通常在 80 到 100 个 epoch 就开始在验证集上稳步上升——典型的过拟合信号。4.2 评估指标MAE、RMSE、MAPE 各看什么训练结束后光看损失曲线不够业绩汇报时要给业务方讲人话的指标。回归预测有三个标配指标各有用处指标公式用途MAE预测值与真实值绝对误差的平均最直观误差 5 分钟就是 5 分钟RMSE均方误差开根号放大重大误差对高峰期的极端拥堵更敏感MAPE绝对误差占比的平均相对误差凌晨低流量时段容易被异常干扰def evaluate_model(model, X_tensor, y_tensor): model.eval() with torch.no_grad(): predictions model(X_tensor).numpy().flatten() y_true y_tensor[:, -1, 0].numpy().flatten() # 反归一化回真实的通行时间分钟 pred_real scaler.inverse_transform(predictions.reshape(-1, 1)).flatten() y_real scaler.inverse_transform(y_true.reshape(-1, 1)).flatten() mae np.mean(np.abs(pred_real - y_real)) rmse np.sqrt(np.mean((pred_real - y_real) ** 2)) mape np.mean(np.abs((pred_real - y_real) / y_real)) * 100 return {MAE: mae, RMSE: rmse, MAPE: mape} # 注意这里传的 scaler 必须是训练集上 fit 的那个 train_metrics evaluate_model(model, X_train, y_train) test_metrics evaluate_model(model, X_test, y_test) print(f训练集指标: {train_metrics}) print(f测试集指标: {test_metrics})参数说明evaluate_model函数里最关键的是反归一化。模型输出的是 0 到 1 之间的归一化值直接拿它和真实分钟数比没有意义。必须用同一个scaler.inverse_transform还原后再算指标。model.eval()会关闭Dropout和BatchNorm的训练行为——如果忘了加这一行验证时的预测结果会因为 Dropout 随机失活而产生抖动。4.3 基线对比为什么拿历史平均做对照很多团队做完 LSTM 直接报 MAE 数值却发现业务方不买账。原因是没有对照。通行时间预测有一个零成本基线——用过去 N 天同一时刻的均值做预测。如果 LSTM 的误差比这个基线还大说明模型学到的时序规律还不如“周期性平均”来得可靠。# 基线预测用训练集每个时间位置的历史均值 def baseline_mae(y_real, seq_len6): # y_real 是测试集真实值这里用最朴素的上周同时刻均值做基线 # 实际场景基线可以更多样这里演示最简单的历史均值 return np.mean(y_real) # 占位示例 # 更合理的做法对测试集每个时间点取训练集相同时段均值 test_times df[time].values[val_end seq_len:] # 测试集对应时间戳 train_series raw_series[:train_end] # 按小时-分钟分组取均值这是一个粗略但有效的基线 from collections import defaultdict hour_minute_mean defaultdict(list) for i, t in enumerate(range(seq_len, train_end)): ts df[time].values[t] key f{ts.hour}:{ts.minute} hour_minute_mean[key].append(raw_series[t]) baseline_preds [] for ts in test_times: key f{ts.hour}:{ts.minute} if hour_minute_mean[key]: baseline_preds.append(np.mean(hour_minute_mean[key])) else: baseline_preds.append(np.mean(raw_series)) test_mae_baseline np.mean(np.abs(np.array(baseline_preds) - raw_series[val_end seq_len:]))逻辑说明这个基线的思想非常简单——如果 8 点的平均通行时间是 35 分钟那测试集里所有 8 点都预测 35 分钟。它能吃掉周期性规律带来的“基础分”LSTM 的 MAE 必须明显低于这个值才说明模型捕捉到了非周期性的路况变化。实测里LSTM 的优势主要体现在突发事件后的恢复过程而基线的优势是稳定——模型在这一点上翻车的例子非常多。5. 避坑指南LSTM 预测通行时间的 5 个常见翻车点5.1 数据泄露归一化用了全量数据统计量现象测试集指标好得离谱MAE 只有 1 分钟但一到线上预测就彻底崩掉。原因预处理时对整条序列做了MinMaxScaler().fit_transform()min和max是从未来数据里学到的。测试集里的每个数值都被“剧透”了它在全序列中的相对位置。解决对训练集单独fit验证集和测试集只做transform。# 错误写法 scaler MinMaxScaler() scaled_all scaler.fit_transform(raw_series.reshape(-1, 1)) # 正确写法 scaler MinMaxScaler() scaled_train scaler.fit_transform(raw_series[:train_end].reshape(-1, 1)) scaled_val scaler.transform(raw_series[train_end:val_end].reshape(-1, 1)) scaled_test scaler.transform(raw_series[val_end:].reshape(-1, 1))5.2 序列窗口长度拍脑袋窗口太短丢信息太长样本稀疏现象窗口选 315 分钟时模型完全学不出早晚高峰的趋势预测曲线几乎是平的。原因15 分钟的上下文太短LSTM 只看到了拥堵的瞬间看不到拥堵从 30 分钟前就开始积累的渐变过程。解决用自相关分析ACF看序列在多长的滞后阶数上仍然相关再决定窗口长度。实测里 5 分钟间隔的数据窗口 6 到 24 都是可接受范围。5.3 训练集随机打乱时间序列不能 shuffle现象验证集和测试集的指标比训练集还低。原因有人引用了完整的train_test_split默认参数——shuffleTrue结果模型在训练时见过测试集的时间片段。这在分类任务里没问题但在时间序列预测里等于把答案提前给了模型。解决所有划分脚本里显式传shuffleFalse或者完全手动按索引切分。5.4 预测值滞后模型学到的是“拷贝上一时刻”现象预测曲线和真实曲线形状几乎一致但整体向右平移了几个时间步看起来像“滞后了一条线”。原因通行时间序列的自相关性非常强LSTM 发现最简单的降损失策略就是把上一时刻的值稍作修改直接输出——这样 MAE 也不会太差。解决看残差的自相关性如果残差在低滞后阶数上仍然显著相关说明模型没有学到真正的动态变化。对策是增加特征天气、时段、星期几让模型有更多依据来做出“非拷贝”的判断或者改用序列差分作为训练目标。5.5 损失不降学习率和梯度问题现象训练了几十个 epoch损失纹丝不动或者直接变成nan。原因lr设置过大导致梯度在最优解附近震荡跳脱变成nan则是典型的梯度爆炸。解决先用小学习率0.0001跑 10 个 epoch 看损失是否下降同时检查输入数据里有没有NaN值——缺失通行时间如果直接填 0LSTM 会把它当成“道路空闲”污染整个训练序列。缺失值用前向填充或用前后均值插值都比填 0 合理。6. 进阶把单步预测扩展成多步预测并验证模型是否真的值得用单步预测能告诉用户“5 分钟后到路口要多久”但出行决策更需要“接下来 30 分钟的路况趋势”。扩展方向有两个一是递归预测——把模型的输出当输入再喂回去预测下一步二是直接多输出——修改全连接层最后一层让它一次输出 N 个未来时间步。递归预测实现简单但误差会随时间步累积实测预测 6 步以后的误差就可能不可接受直接多输出更稳定代价是最后一层全连接的参数量略增。特征扩展方面in 通行时间预测里只喂历史通行时间序列是“裸奔版本”。把星期几周一早高峰和周三早高峰明显不同、节假日标记、天气状况降雨量、温度拼进每个时间步的input_size维度模型才有依据区分“同一时刻但不同状态的堵”。这些特征需要和通行时间同步归一化并在滑动窗口构造时一起切进去——不能只切时间序列那一列。验证模型是否真的能用我建议做滚动回测把数据按时间分成多段每段都做一次完整的“训练 验证 测试”而不是只跑一次固定划分。比如用 8 周数据训练预测第 9 周然后把第 9 周并入训练集再预测第 10 周。这样得到的 MAE 分布比单次划分更能说明模型的稳定性。我自己的教训是单次划分测试集 MAE 3.8 分钟看起来很漂亮滚动回测一跑变成 6.2 分钟——中间差的 2.4 分钟就是模型在不同时间段上的泛化能力差距。做完这一步你才有底气把这个方案投入到真实的路况预测流程里希望帮到你。本文还有配套的精品资源点击获取