☰
CNN+LSTM交通流量预测实战:LCTFP模型从数据到滚动预测
2026/10/2 14:47:31 网站建设 项目流程

简介:这份资源提供了一套基于 CNN 与 LSTM 组合结构的高速公路短时交通流量预测完整 Python 实现,面向具备一定深度学习基础、关注智能交通与时空序列预测的开发者与研究人员。模型以 1D CNN 提取交通流的空间特征,LSTM 捕捉时间依赖,可用于站点级短时流量预测实验与复现。压缩包共 21 个文件,约 31MB,包含 7 个 py 脚本、6 个 h5 权重文件、4 个 json 模型配置、2 张 png 结构示意图及说明文档,覆盖数据预处理、模型训练、超参数搜索与多模型对比等环节。其中 cnn_lstm_param.py 支持基于 hyperas 的超参数搜索,便于调优。资源还附带 PeMS 数据预处理脚本与训练好的模型权重,读者可据此快速跑通从数据归一化、时序处理到预测评估的完整流程,并对照 CNN、LSTM、SAE 等模型结果理解组合结构的优势。目前已有 5334 人学习下载,适合作为交通流预测课题的实践参考与代码模板。

1. LCTFP 到底在解决什么问题:从一条高速的 15 分钟粒度说起

高速公路上每 15 分钟就会产生一条流量记录,一天 96 条,一年三万五千条。单看某一条毫无意义,但把它们连起来,早高峰的爬升、午间的塌陷、晚高峰的二次冲顶、凌晨的谷底,构成了一条有强周期、强突变、还带随机扰动的曲线。LCTFP 这个标题里的核心,就是拿 CNN 和 LSTM 两套结构去啃这条曲线:CNN 负责从相邻时段里抠出局部突变模式,LSTM 负责把一天、一周的长期依赖记住,最后拼成一个能滚动预测未来若干时段的模型。它适合两类人——一类是刚学完 CNN、LSTM 想找个真实序列练手的 Python 学习者,另一类是手里有收费站或门架数据、想先跑通一个基线再谈落地的交通数据从业者。下面我按“数据怎么进、模型怎么搭、参数怎么调、坑在哪”的顺序,把这条链路完整走一遍,代码可以直接抄。

2. 数据准备与特征工程:把原始流量表变成模型能吃的张量

2.1 先搞清楚你的数据长什么样

交通流量预测的输入通常是一张按时间排序的表,最少要有两列:时间戳和流量值。真实门架数据还会带车道号、车型、方向,但做单点流量预测时,我一般先按“断面 + 时间”聚合,把多车道加总成一条序列。这一步不做,后面模型学到的就是车道间的噪声。

拿到数据先做三件事:确认时间间隔是否均匀、检查缺失和异常、看周期性是否明显。间隔不均匀(比如某段缺了半小时)会直接毁掉滑动窗口的语义,必须重采样。缺失值不要一上来就填 0,凌晨的 0 和丢数据的 0 是两回事,前者是真实谷底,后者是黑匣子。我一般用前向填充加线性插值组合处理短缺口,超过两小时的缺口直接标记并考虑丢弃该天。

import pandas as pd import numpy as np # 读取原始数据,假设列名为 timestamp 和 flow df = pd.read_csv("traffic_raw.csv", parse_dates=["timestamp"]) df = df.sort_values("timestamp").set_index("timestamp") # 重采样到 15 分钟,缺失用时间插值,长缺口先看比例 df = df.resample("15min").mean() missing_ratio = df["flow"].isna().mean() print(f"缺失比例: {missing_ratio:.2%}") # 短缺口线性插值,首尾用前向/后向填充兜底 df["flow"] = df["flow"].interpolate(method="time", limit=8) df["flow"] = df["flow"].ffill().bfill()

这段代码的关键参数是limit=8,意思是连续缺失超过 8 个点(即 2 小时)就不插值,保留 NaN 交给后续判断。resample("15min").mean()里如果原始数据本身就是 15 分钟粒度,这步等于去重对齐,不会改变数值。缺失比例超过 5% 时,我会回头查采集端,而不是硬着头皮训练。

2.2 构造监督学习样本:滑动窗口是核心

LSTM 吃的是序列,但训练时需要“输入一段、预测下一段”的样本对。假设用过去 8 个时段(2 小时)预测未来 1 个时段(15 分钟),窗口就是 8 进 1 出。窗口长度的选择直接决定模型能看到的上下文:太短抓不住趋势,太长引入过多噪声且训练变慢。高速流量我一般从 8 起步,再试 16 和 96(一整天)。

def make_windows(series, input_len=8, output_len=1): X, y = [], [] values = series.values for i in range(len(values) - input_len - output_len + 1): X.append(values[i : i + input_len]) y.append(values[i + input_len : i + input_len + output_len]) X = np.array(X).reshape(-1, input_len, 1) # (样本, 时间步, 特征) y = np.array(y).reshape(-1, output_len) return X, y X, y = make_windows(df["flow"], input_len=8, output_len=1) print(X.shape, y.shape) # 例如 (34900, 8, 1) (34900, 1)

reshape(-1, input_len, 1)里的最后一个 1 是特征维度。如果你还想加节假日标记、温度、是否雨天,就把它变成多特征,比如np.stack([flow, is_holiday], axis=-1),此时最后一维变成 2,模型输入维度也要同步改。这一步是新手最容易翻车的地方:特征加了但模型input_shape没改,报错信息还特别绕。

2.3 归一化与数据集切分:别让未来信息泄漏

流量值动辄几千,直接喂给网络会让梯度爆炸。归一化到 [0,1] 是常规操作,但必须只用训练集的 min/max 去变换验证集和测试集,否则就是典型的数据泄漏,验证指标会虚高,上线就崩。时间序列不能随机打乱切分,要按时间先后切,通常 7:2:1。

from sklearn.preprocessing import MinMaxScaler split_train = int(len(X) * 0.7) split_val = int(len(X) * 0.9) scaler = MinMaxScaler() # 只用训练段拟合,避免泄漏 scaler.fit(X[:split_train].reshape(-1, 1)) X_all = scaler.transform(X.reshape(-1, 1)).reshape(X.shape) y_all = scaler.transform(y.reshape(-1, 1)).reshape(y.shape) X_train, y_train = X_all[:split_train], y_all[:split_train] X_val, y_val = X_all[split_train:split_val], y_all[split_train:split_val] X_test, y_test = X_all[split_val:], y_all[split_val:]

注意scaler.fit只传了训练段,transform才作用于全体。预测结束后要用scaler.inverse_transform还原回真实流量,否则你评估的 MAE 是归一化尺度下的,没有物理意义。这个还原步骤我在早期项目里漏过,结果模型“看起来”误差只有 0.02,实际是 200 辆车,血泪经验。

3. LCTFP 模型结构:CNN 提局部、LSTM 记长期,怎么拼才不打架

3.1 为什么是 CNN + LSTM 而不是单独用一个

单独用 LSTM 能记住长期趋势,但对局部突变(比如事故导致的骤降、节假日的尖峰)反应偏慢,因为它的门控是逐时间步更新的,短窗口内的形态特征提取不如卷积直接。单独用 CNN(比如一维卷积堆叠)感受野有限,想覆盖一整天需要很深的网络,训练成本高还容易过拟合。CNN + LSTM 的分工是:一维卷积先在时间轴上滑动,把相邻几个时段的局部模式压成更紧凑的特征序列,LSTM 再在这个特征序列上建模长期依赖。常见做法是 Conv1D 提取通道特征,接池化降采样,再送进 LSTM,最后全连接输出预测值。

3.2 用 PyTorch 搭出 LCTFP 主干

下面是一个可以直接跑的结构,输入形状(batch, input_len, features)。Conv1D 的kernel_size我设 3,表示每次看相邻 3 个时段;padding=1保持长度不变,方便后面接池化。LSTM 隐藏单元从 64 起步,层数 1 到 2 层足够,再深对小数据集就是灾难。

import torch import torch.nn as nn class LCTFP(nn.Module): def __init__(self, input_len=8, n_features=1, hidden=64, output_len=1): super().__init__() # 一维卷积:提取局部时段模式 self.conv = nn.Sequential( nn.Conv1d(in_channels=n_features, out_channels=32, kernel_size=3, padding=1), nn.ReLU(), nn.MaxPool1d(kernel_size=2), # 时间步减半 ) # LSTM:在卷积特征上建模长期依赖 self.lstm = nn.LSTM(input_size=32, hidden_size=hidden, num_layers=1, batch_first=True) self.fc = nn.Linear(hidden, output_len) def forward(self, x): # x: (batch, input_len, n_features) -> Conv1d 需要 (batch, feat, len) x = x.permute(0, 2, 1) x = self.conv(x) x = x.permute(0, 2, 1) # 换回 (batch, len, feat) out, (h, c) = self.lstm(x) last = out[:, -1, :] # 取最后一个时间步 return self.fc(last) model = LCTFP(input_len=8, n_features=1, hidden=64, output_len=1) print(sum(p.numel() for p in model.parameters()), "参数")

几个参数要盯住:out_channels=32是卷积核数量,太小欠拟合,太大在小数据上过拟合;MaxPool1d(2)会把 8 个时间步压成 4 个,如果input_len是奇数要处理边界;out[:, -1, :]取最后时间步,是因为 LSTM 已经把历史信息累积到末态,这也是预测任务最常见的取法。batch_first=True让输入维度顺序符合直觉,忘了设会导致维度对不上,报错信息很难读。

3.3 训练循环与损失选择

回归任务用 MSE 或 MAE。MSE 对大误差敏感,适合你更在意高峰时段预测准;MAE 对整体更稳,适合流量波动大、异常值多的场景。我一般先用 MSE 跑通,再看残差分布决定要不要换 Huber。优化器 Adam,学习率 1e-3,批次 64,早停看验证集损失。

from torch.utils.data import TensorDataset, DataLoader train_ds = TensorDataset(torch.tensor(X_train, dtype=torch.float32), torch.tensor(y_train, dtype=torch.float32)) train_loader = DataLoader(train_ds, batch_size=64, shuffle=True) optimizer = torch.optim.Adam(model.parameters(), lr=1e-3) criterion = nn.MSELoss() for epoch in range(50): model.train() total = 0 for xb, yb in train_loader: optimizer.zero_grad() pred = model(xb) loss = criterion(pred, yb) loss.backward() optimizer.step() total += loss.item() print(f"epoch {epoch}, loss {total/len(train_loader):.5f}")

shuffle=True在训练集上可以打乱,因为样本之间已经通过窗口构造保留了时序,打乱不会泄漏未来信息;但验证和测试集绝不能 shuffle,评估时要保持时间顺序。学习率 1e-3 是 Adam 的常用起点,如果 loss 前几个 epoch 就震荡,降到 1e-4;如果下降极慢,试 3e-3。训练轮数不要死设 50,加个验证集早停,连续 5 轮不降就停,能省不少时间。

4. 训练完怎么评估:指标、可视化与滚动预测

4.1 三个必看指标和它们的陷阱

MAE、RMSE、MAPE 是交通预测的标配。MAE 直观,单位是车辆数;RMSE 放大大误差,能暴露高峰时段的崩盘;MAPE 是百分比,但凌晨流量接近 0 时分母极小,会算出爆炸值,所以 MAPE 我只在白天时段看。评估前务必inverse_transform还原真实尺度。

model.eval() with torch.no_grad(): pred = model(torch.tensor(X_test, dtype=torch.float32)).numpy() pred_inv = scaler.inverse_transform(pred) true_inv = scaler.inverse_transform(y_test) mae = np.mean(np.abs(pred_inv - true_inv)) rmse = np.sqrt(np.mean((pred_inv - true_inv) ** 2)) print(f"MAE={mae:.2f}, RMSE={rmse:.2f}")

如果 MAE 是 30 辆车、RMSE 是 90,说明大部分时段误差可控,但少数高峰点错得离谱,这时候要回去看是不是窗口太短没抓住爬升段。反过来 MAE 和 RMSE 都大,多半是归一化或特征出了问题。

4.2 滚动多步预测:别只测一步就下结论

一步预测好看不代表能用。真实场景要预测未来 1 小时甚至 4 小时,做法是把预测值填回输入窗口,滚动往前推。滚动步数越多误差累积越明显,这是 LSTM 类模型的通病,也是评估时必须暴露的。

def rolling_forecast(model, init_window, steps=4): model.eval() window = init_window.copy() # (1, input_len, 1) preds = [] with torch.no_grad(): for _ in range(steps): x = torch.tensor(window, dtype=torch.float32) p = model(x).item() preds.append(p) # 把预测值接到窗口尾部,去掉最老的一个点 window = np.append(window[:, 1:, :], [[[p]]], axis=1) return scaler.inverse_transform(np.array(preds).reshape(-1, 1))

steps=4就是预测未来 1 小时。滚动时窗口的更新逻辑是“去头接尾”,顺序不能反。实测中 4 步滚动 MAE 通常比一步高 30% 到 60%,如果高得离谱,说明模型过度依赖最近一个点,长期记忆没学好,可以考虑加深 LSTM 或加长输入窗口。

4.3 把预测画出来,肉眼比指标更诚实

指标是平均数,会掩盖结构性错误。把真实值和预测值按时间画在同一张图上,重点看早高峰起点、晚高峰峰值、凌晨谷底这三处。如果模型总是慢半拍,是输入窗口不够;如果峰值被削平,是 MSE 让模型偏向均值,换 Huber 或对高峰样本加权。这一步不写代码也行,matplotlib 两行搞定,但一定要做。

5. 避坑与排查:LCTFP 落地时最容易翻车的 5 个地方

5.1 现象:验证 loss 很低,测试集一塌糊涂

原因:切分时随机打乱或归一化用了全体数据,未来信息泄漏。解决:严格按时间切分,scaler 只在训练段 fit,检查split_train之前的代码有没有提前 transform 全体。

5.2 现象:训练 loss 不降,一直在高位震荡

原因:学习率过大或输入没归一化。解决:先确认X_train的数值范围在 [0,1],再把学习率从 1e-3 降到 1e-4 试。如果还不降,检查 Conv1d 的输入维度有没有 permute 错,维度错了有时不报错但学不到东西。

5.3 现象:预测曲线整体平移,形状对但数值差一截

原因:inverse_transform用错 scaler,或者 y 的归一化和 X 用了不同尺度。解决:X 和 y 如果都是流量,用同一个 scaler;如果 y 是差分后的值,要单独还原再累加。这个坑我踩过,模型本身没问题,纯粹是还原公式写错。

5.4 现象:多步滚动预测几步之后迅速发散

原因:把预测值当真实值回填,误差累积。解决:这是滚动预测的固有缺陷,缓解办法是训练时就用多步目标(output_len 设成 4),让模型直接学多步输出,而不是靠单步模型硬滚。代价是样本数减少,需要更多数据。

5.5 现象:换了新路段数据,模型完全失效

原因:不同路段的流量量级和周期形态差异大,归一化参数不通用。解决:新路段要么重新训练,要么做迁移学习——冻结 CNN 层只微调 LSTM 和全连接。直接套用旧 scaler 是最常见的误用,量级差一倍,预测全废。

6. 让 LCTFP 更实用:多特征输入与残差修正的一个技巧

单变量流量预测能跑通,但真实场景里节假日、天气、上游事故都会让曲线偏离历史模式。我的习惯是在输入里加两个低成本特征:一是“是否节假日”的 0/1 标记,二是“前一天同时段流量”作为参照列。前者让模型知道今天周期特殊,后者给它一个强基线。加特征时把n_features从 1 改成 3,make_windows里用np.stack拼起来,模型第一层 Conv1d 的in_channels同步改 3,其余不动。

# 假设 df 已有 flow, is_holiday, flow_yesterday 三列 feat = df[["flow", "is_holiday", "flow_yesterday"]].values # 构造窗口时每个时间步带 3 个特征 def make_multi_windows(arr, input_len=8, output_len=1): X, y = [], [] for i in range(len(arr) - input_len - output_len + 1): X.append(arr[i : i + input_len]) # (input_len, 3) y.append(arr[i + input_len : i + input_len + output_len, 0]) # 只预测 flow return np.array(X), np.array(y)

注意 y 只取第 0 列,因为节假日标记和昨日流量不需要预测。归一化时三列尺度不同,is_holiday是 0/1 不用动,另外两列各自归一化,别用一个 scaler 硬套。加完特征后如果验证 MAE 没降,说明特征没信息量或者引入噪声,果断删掉,特征不是越多越好。

另一个技巧是残差修正:先用 LCTFP 预测,再用一个轻量模型(比如线性回归)去拟合预测残差,把系统性偏差补回来。做法是把验证集上的真实值 - 预测值当新目标,用“时段序号 + 是否高峰”做特征训一个小模型,测试时叠加。这个思路在高峰时段常能再压 5% 到 10% 的误差,代价是多维护一个模型。我一般只在基线误差已经压不动、又不想大改网络结构时才上这招。

最后说个习惯:每次改完结构或特征,固定跑一遍同一段测试集,把 MAE、RMSE 和滚动 4 步误差记在一张表里,别凭感觉判断“好像变好了”。我早期就是改完直接看训练 loss,结果验证集悄悄变差,白跑两天。希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询