简介：这份资源面向智能交通、深度学习方向的学习者与开发者，围绕卡口实时过车数据展开交通流量预测实践，核心采用LSTM循环神经网络并引入融合预测思路，宣称准确率可达90%以上。内容覆盖时间序列预测的完整链路：卡口数据清洗与归一化、小时与星期等特征工程、LSTM输入层与隐藏层门控机制构建、基于均方误差的训练与超参数调优，以及多模型集成的融合预测策略，可帮助读者理解从原始过车记录到流量预测结果的全过程。压缩包共62个文件，约20.23MB，以csv数据文件、py脚本、TensorFlow模型checkpoint相关文件（meta、index、data-00000-of-00001）为主，另含少量md说明与pyc缓存，模型权重与训练数据齐备，便于直接复现与二次实验。目前已有297人学习下载，适合作为交通流量预测的课程设计、毕业设计或工程原型参考。
1. 卡口过车数据做实时流量预测：LSTM 融合模型怎么落地到 90% 准确率
卡口实时过车数据是交通领域最容易拿到、也最容易被低估的一类时序数据。每辆车经过卡口时，会产生一条带时间戳、车牌、车道、车型的过车记录，按分钟或 5 分钟聚合后，就是一条天然的交通流量时间序列。问题在于，这份数据既有强周期性（早晚高峰、工作日/周末），又有突发性（事故、天气、临时管制），单靠移动平均或 ARIMA 这类线性模型，预测精度通常卡在 75%～85%，遇到流量突变就明显滞后。LSTM 循环神经网络之所以在这类场景被反复使用，是因为它的门控结构能同时记住长期周期模式和短期波动，配合合理的特征融合策略，把分钟级流量预测的准确率推到 90% 以上是可达的。这篇内容面向已经拿到卡口过车数据、想用 LSTM 做实时预测的工程师，从数据聚合、特征构造、模型搭建到部署推理，把每一步的参数和坑讲清楚，新手能照着跑通，熟手能对照边界条件做取舍。
2. 从过车记录到 LSTM 输入张量：数据聚合与特征工程
2.1 卡口数据的三个原始字段与聚合粒度选择
卡口过车数据最核心的字段只有三个：过车时间戳（精确到秒）、卡口编号、车道编号。部分系统还会带车型和车牌，但做流量预测时，车牌只用于去重，车型用于分车型流量。原始记录是事件级的，不能直接喂给 LSTM，必须先聚合成等间隔时间序列。
聚合粒度的选择直接决定预测任务的定义。常见做法是 5 分钟粒度，原因是：1 分钟粒度噪声太大，单辆车抖动就会造成流量跳变；15 分钟粒度又太粗，早晚高峰的爬升过程被抹平，预测失去调度价值。5 分钟粒度下，一天 288 个点，一周 2016 个点，既能体现高峰形态，又不会让序列过长导致训练慢。
聚合时要注意一个容易翻车的点：空窗补零。某些卡口在凌晨可能连续几个 5 分钟没有过车记录，如果直接 drop 掉这些时间段，序列就断了，LSTM 的时间步会错位。正确做法是按完整时间轴 reindex，缺失段填 0，并额外加一列掩码标记该点是否为真实观测。
import pandas as pd import numpy as np # 原始过车记录：timestamp, station_id, lane_id df = pd.read_csv("kakou_pass.csv", parse_dates=["timestamp"]) # 按 5 分钟聚合单卡口总流量 df["time_bin"] = df["timestamp"].dt.floor("5min") flow = df.groupby(["station_id", "time_bin"]).size().reset_index(name="flow") # 构造完整时间轴，缺失补 0 并加掩码 full_idx = pd.date_range(flow["time_bin"].min(), flow["time_bin"].max(), freq="5min") station_ids = flow["station_id"].unique() full = pd.MultiIndex.from_product([station_ids, full_idx], names=["station_id", "time_bin"]) flow = flow.set_index(["station_id", "time_bin"]).reindex(full, fill_value=0).reset_index() flow["is_observed"] = (flow["flow"] > 0).astype(int)这段代码的逻辑是先把事件级记录按 5 分钟窗口计数，再用 MultiIndex 笛卡尔积补齐每个卡口的完整时间轴。fill_value=0保证缺失段流量为 0，is_observed列让模型知道这个 0 是真实无车还是数据缺失。参数上，freq="5min"可按业务改成"1min"或"15min"，但改完要同步调整后续滑窗长度。
2.2 时间特征、周期特征与上游卡口融合
只给 LSTM 喂一列流量，模型能学到趋势，但学不到「现在是周五晚高峰」这种上下文。必须显式构造时间特征。最有效的几类：小时的正余弦编码（避免 23 点和 0 点数值上相距很远但实际相邻）、星期几的 one-hot、是否节假日标记。
更关键的是上游卡口融合。城市路网里，一个卡口的流量往往受上游 1～3 个卡口影响，单纯预测单点会忽略这种空间关联。常见做法有两种：一是把上游卡口同时刻流量作为额外特征拼进输入向量；二是用图神经网络先做空间聚合再送 LSTM。前者实现简单、落地快，后者精度更高但工程复杂度大。我一般先用前者验证 baseline，确认单点 LSTM 能到 88% 左右，再考虑上空间融合补最后几个点。
def build_features(flow_df, upstream_map): flow_df = flow_df.sort_values(["station_id", "time_bin"]) flow_df["hour"] = flow_df["time_bin"].dt.hour flow_df["minute"] = flow_df["time_bin"].dt.minute # 正余弦编码，周期设为一天 288 个 5 分钟点 flow_df["sin_hour"] = np.sin(2 * np.pi * (flow_df["hour"] * 12 + flow_df["minute"] // 5) / 288) flow_df["cos_hour"] = np.cos(2 * np.pi * (flow_df["hour"] * 12 + flow_df["minute"] // 5) / 288) flow_df["dayofweek"] = flow_df["time_bin"].dt.dayofweek # 上游卡口同时刻流量 for up in upstream_map: up_flow = flow_df[flow_df["station_id"] == up][["time_bin", "flow"]] up_flow = up_flow.rename(columns={"flow": f"up_{up}_flow"}) flow_df = flow_df.merge(up_flow, on="time_bin", how="left") return flow_df.fillna(0)正余弦编码里hour * 12 + minute // 5是把一天的时间折算成 0～287 的序号，再除以 288 映射到 [0, 2π]，这样 23:55 和 00:00 的编码值几乎连续。上游流量用 merge 左连接，缺失填 0。参数上，upstream_map需要根据路网拓扑人工配置或从卡口关联表读取，一般选直接相邻的 1～2 跳卡口。
2.3 滑窗构造与标准化：别在划分前做归一化
LSTM 的输入是三维张量(样本数, 时间步, 特征数)。用过去 12 个点（1 小时）预测未来 1 个点（5 分钟）是最常见的配置，也可以预测未来 6 个点做多步预测。滑窗时要注意：训练集、验证集、测试集必须按时间顺序切分，不能随机打乱，否则未来信息泄漏，验证准确率虚高。
标准化是另一个高频翻车点。很多人对整个数据集算均值和方差再归一化，这等于把测试集的分布信息提前泄露给训练。正确做法是只用训练集的均值和方差，应用到验证和测试集。
from sklearn.preprocessing import StandardScaler def make_windows(data, feature_cols, target_col, lookback=12, horizon=1): X, y = [], [] arr = data[feature_cols].values target = data[target_col].values for i in range(len(data) - lookback - horizon + 1): X.append(arr[i:i + lookback]) y.append(target[i + lookback + horizon - 1]) return np.array(X), np.array(y) # 按时间 7:1:2 切分 n = len(feature_df) train_df = feature_df.iloc[:int(n * 0.7)] val_df = feature_df.iloc[int(n * 0.7):int(n * 0.8)] test_df = feature_df.iloc[int(n * 0.8):] scaler = StandardScaler().fit(train_df[feature_cols]) for d in [train_df, val_df, test_df]: d[feature_cols] = scaler.transform(d[feature_cols]) X_train, y_train = make_windows(train_df, feature_cols, "flow") X_val, y_val = make_windows(val_df, feature_cols, "flow") X_test, y_test = make_windows(test_df, feature_cols, "flow")lookback=12对应 1 小时历史，horizon=1表示预测下一个 5 分钟。如果要做 30 分钟预测，把horizon改成 6。标准化器只在训练集 fit，这是保证线上推理一致性的前提——线上部署时也要用同一个 scaler 的参数，不能重新 fit。
3. LSTM 模型搭建与训练：层数、隐藏单元和损失函数怎么定
3.1 PyTorch LSTM 网络结构：两层就够，别堆深
交通流量序列的复杂度远低于 NLP，LSTM 不需要堆很多层。实践中 1～2 层 LSTM 加一个全连接输出层就能到很好的效果，层数加到 3 层以上，验证损失反而上升，训练时间翻倍。隐藏单元数一般取 64 或 128，输入特征维度在 10～20 之间时，64 个隐藏单元足够捕捉周期模式。
import torch import torch.nn as nn class TrafficLSTM(nn.Module): def __init__(self, input_dim, hidden_dim=64, num_layers=2, dropout=0.2): super().__init__() self.lstm = nn.LSTM( input_size=input_dim, hidden_size=hidden_dim, num_layers=num_layers, batch_first=True, dropout=dropout if num_layers > 1 else 0 ) self.fc = nn.Sequential( nn.Linear(hidden_dim, 32), nn.ReLU(), nn.Linear(32, 1) ) def forward(self, x): # x: (batch, seq_len, input_dim) out, _ = self.lstm(x) # 取最后一个时间步的输出 return self.fc(out[:, -1, :]).squeeze(-1)batch_first=True让输入维度是(batch, seq, feature)，符合大多数人的直觉。dropout只在多层 LSTM 之间生效，单层时设为 0。输出层用两层全连接加 ReLU，比直接 Linear 到 1 维更稳。参数上，hidden_dim=64是起点，如果欠拟合（训练损失降不下去）再调到 128；num_layers=2是上限，不要再加。
3.2 损失函数选择：MSE 还是 MAE，取决于你要什么
流量预测里，MSE 对大误差惩罚重，会让模型偏向预测高峰值，低谷段误差被放大；MAE 对异常值更鲁棒，但梯度在 0 附近不稳定。常见做法是用 HuberLoss，它在误差小于 delta 时表现为 MSE，大于 delta 时表现为 MAE，兼顾两者。
from torch.utils.data import DataLoader, TensorDataset device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model = TrafficLSTM(input_dim=X_train.shape[2]).to(device) criterion = nn.HuberLoss(delta=1.0) optimizer = torch.optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-5) scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(optimizer, patience=5, factor=0.5) train_loader = DataLoader( TensorDataset(torch.FloatTensor(X_train), torch.FloatTensor(y_train)), batch_size=64, shuffle=True ) for epoch in range(100): model.train() for xb, yb in train_loader: xb, yb = xb.to(device), yb.to(device) pred = model(xb) loss = criterion(pred, yb) optimizer.zero_grad() loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() # 验证集评估，调度器按验证损失降学习率 model.eval() with torch.no_grad(): val_pred = model(torch.FloatTensor(X_val).to(device)) val_loss = criterion(val_pred, torch.FloatTensor(y_val).to(device)) scheduler.step(val_loss)HuberLoss(delta=1.0)里的 delta 是切换阈值，流量标准化后一般在 0.5～2.0 之间调。clip_grad_norm_是 LSTM 训练的后悔药，梯度爆炸时能把范数截到 1.0，避免 loss 变 NaN。ReduceLROnPlateau在验证损失 5 个 epoch 不降时把学习率减半，比固定学习率更容易收敛到好点。
3.3 准确率怎么算才不是自欺欺人
标题里说准确率 90% 以上，必须定义清楚这个 90% 是什么。回归任务没有分类准确率，常见替代指标有三个：MAE、MAPE、以及自定义的「误差在阈值内算对」的准确率。如果按 MAPE 算，低谷时段流量接近 0，分母极小，MAPE 会爆炸，所以低谷段要单独看 MAE。
我一般用两个指标同时汇报：整体 MAE，以及「预测误差小于真实值 15% 的样本占比」。后者更接近业务方理解的准确率。如果这个占比能到 90%，说明模型在绝大多数时间点都够用。
def accuracy_within(pred, true, ratio=0.15): # 误差在真实值 ratio 以内算预测正确 mask = true > 1 # 过滤掉流量为 0 或极小的点 correct = np.abs(pred[mask] - true[mask]) <= ratio * true[mask] return correct.mean() model.eval() with torch.no_grad(): pred_test = model(torch.FloatTensor(X_test).to(device)).cpu().numpy() # 反标准化 pred_test = scaler.inverse_transform( np.concatenate([pred_test.reshape(-1, 1), np.zeros((len(pred_test), len(feature_cols) - 1))], axis=1) )[:, 0] true_test = scaler.inverse_transform( np.concatenate([y_test.reshape(-1, 1), np.zeros((len(y_test), len(feature_cols) - 1))], axis=1) )[:, 0] print("MAE:", np.mean(np.abs(pred_test - true_test))) print("Acc@15%:", accuracy_within(pred_test, true_test))反标准化时构造了一个和 feature_cols 同宽度的矩阵，只把第一列替换成预测值，再取回第一列，这是为了复用同一个 scaler。mask = true > 1过滤掉流量极小的点，避免比例误差失真。参数ratio=0.15可按业务容忍度调整，要求严格就降到 0.1。
4. 实时推理与部署：从离线模型到分钟级预测服务
4.1 模型导出与推理延迟控制
离线训练完的模型要上线做实时预测，核心约束是推理延迟。LSTM 单次前向在 CPU 上通常 5～20ms，GPU 上 2～5ms，对 5 分钟粒度的预测完全够用。但如果卡口数量多（几百个），逐个推理会累积延迟，常见做法是把同一时刻所有卡口的输入拼成一个 batch 一次前向。
导出时用torch.jit.trace或torch.save保存 state_dict 都行。生产环境我更倾向 TorchScript，因为它不依赖 Python 源码，部署到 C++ 服务也方便。
# 保存 TorchScript model.eval() example = torch.randn(1, 12, X_train.shape[2]).to(device) traced = torch.jit.trace(model, example) traced.save("traffic_lstm.pt") # 推理服务里加载 loaded = torch.jit.load("traffic_lstm.pt") loaded.eval() with torch.no_grad(): batch_input = torch.FloatTensor(recent_windows).to(device) # (N, 12, F) preds = loaded(batch_input).cpu().numpy()torch.jit.trace需要一个示例输入，形状必须和线上一致。recent_windows是最近 12 个点的滑窗，N 是卡口数量。注意 TorchScript 不支持动态控制流，如果模型里有 if 分支要改成torch.where。
4.2 数据管道与特征一致性
线上推理最容易翻车的地方不是模型，是特征不一致。离线训练时用的正余弦编码、上游流量、标准化参数，线上必须一模一样。我见过太多案例：离线 MAE 3.2，上线后 MAE 飙到 8，排查半天发现线上小时编码用的是hour而不是hour*12 + minute//5。
解决办法是把特征工程逻辑封装成一个函数，离线和线上共用同一份代码。标准化参数存成 json，服务启动时加载。
import json # 保存标准化参数 scaler_params = { "mean": scaler.mean_.tolist(), "scale": scaler.scale_.tolist(), "feature_cols": feature_cols } with open("scaler.json", "w") as f: json.dump(scaler_params, f) # 线上加载并手动标准化 def online_transform(raw_features, params): mean = np.array(params["mean"]) scale = np.array(params["scale"]) return (raw_features - mean) / scale手动标准化比 pickle 加载 sklearn scaler 更可控，也避免了 sklearn 版本不一致导致的兼容问题。feature_cols的顺序必须和训练时完全一致，顺序错了模型不会报错，但预测会离谱。
4.3 滚动预测与误差累积
实时预测通常是滚动进行的：每 5 分钟来一批新数据，更新滑窗，预测下一个点。如果做多步预测（比如预测未来 30 分钟），有两种策略：直接多输出（模型一次输出 6 个点）和滚动单步（把预测值当输入继续预测）。直接多输出误差不累积，但训练难度大；滚动单步实现简单，但 6 步之后误差会明显放大。
我的经验是：预测 1～2 步用滚动单步，3 步以上用直接多输出。直接多输出只需把输出层改成nn.Linear(32, horizon)，损失函数对每个时间步分别算再平均。
class MultiStepLSTM(nn.Module): def __init__(self, input_dim, hidden_dim=64, num_layers=2, horizon=6): super().__init__() self.lstm = nn.LSTM(input_dim, hidden_dim, num_layers, batch_first=True, dropout=0.2) self.fc = nn.Linear(hidden_dim, horizon) def forward(self, x): out, _ = self.lstm(x) return self.fc(out[:, -1, :]) # (batch, horizon)horizon=6对应未来 30 分钟。训练时y的形状要改成(样本数, 6)，损失函数用HuberLoss对 6 个输出求平均。这样模型一次前向就出 6 个点，没有误差累积问题。
5. 避坑与排查：LSTM 流量预测最常见的 5 个翻车现场
5.1 验证损失下降但测试 MAE 很高
现象：训练时验证损失一路降到很低，测试集 MAE 却是验证集的两三倍。原因通常是数据泄漏——滑窗构造时训练集和验证集之间有重叠，或者标准化用了全量数据。解决：按时间切分后，在切分点留出至少lookback个点的 gap，确保验证集第一个样本的输入窗口完全来自验证集时段。标准化严格只用训练集 fit。
5.2 预测曲线整体滞后一个时间步
现象：预测值看起来形状对，但总是比真实值晚 5～10 分钟，高峰爬升段尤其明显。原因是模型学到了「用当前值预测下一个值」的恒等映射，LSTM 退化成滞后器。解决：在损失函数里加大对大误差的惩罚（提高 HuberLoss 的 delta 或改用 MSE），同时在输入里去掉当前时刻的流量值，强制模型依赖历史窗口和周期特征。另一个办法是预测流量变化量（diff）而不是绝对值。
5.3 凌晨时段预测全是 0 或负值
现象：白天预测正常，凌晨流量低谷段预测出负数或恒定 0。原因是标准化后凌晨流量接近均值减两三个标准差，模型输出经过反标准化后可能落到负区间。解决：输出层加softplus激活保证非负，或者在损失函数里对负预测加惩罚。更根本的办法是对流量做 log1p 变换再标准化，让分布更接近正态。
5.4 上游卡口特征缺失导致线上预测漂移
现象：离线评估很好，上线后某些卡口预测明显偏差。排查发现这些卡口的上游卡口在线上数据里没有实时上报，merge 后填 0，模型把「上游无车」当成了真实信号。解决：上游特征除了流量值，再加一列「上游是否在线」的掩码，缺失时掩码为 0，让模型学会区分「上游真的没车」和「上游数据没来」。
5.5 批量推理时显存溢出
现象：单卡口推理正常，一次推理几百个卡口时 CUDA out of memory。原因是 batch 太大，LSTM 的中间状态占用显存。解决：把 batch 切成 64 或 128 一批，循环推理再拼接；或者用torch.no_grad()包住推理（这个最容易忘，忘了会保留计算图，显存直接翻倍）。如果还是不够，把模型转成 half 精度推理。
6. 把 90% 准确率守住：在线监控与模型迭代的一个实用技巧
模型上线不是终点，流量模式会随季节、道路施工、新开商圈而漂移。我吃过最大的亏是模型上线三个月没管，某条路旁边开了个商场，晚高峰流量翻倍，模型预测还是按老模式走，MAE 从 3.1 涨到 9.7，业务方直接打电话来问是不是服务挂了。后来我养成了一个习惯：给每个卡口维护一个滚动 7 天的误差监控，MAE 超过训练时基线 1.5 倍就触发告警，自动把最近 7 天数据加入训练集做增量微调。
具体做法不复杂。每天凌晨低峰期跑一次增量训练，只 fine-tune 最后一层和 LSTM 的最后一段，学习率设成初始值的十分之一，训练 5～10 个 epoch 就够。这样既跟上了分布漂移，又不会把之前学到的周期模式冲掉。
# 增量微调：只更新部分参数 for name, param in model.named_parameters(): if "lstm" in name and "weight_ih_l1" not in name: param.requires_grad = False # 冻结第一层 LSTM optimizer = torch.optim.Adam( filter(lambda p: p.requires_grad, model.parameters()), lr=1e-4 # 初始学习率的 1/10 ) # 用最近 7 天数据训练 5 个 epoch冻结第一层 LSTM 是因为底层学的是通用周期模式，不需要频繁更新；上层和输出层学的是当前流量水平，需要跟着漂移走。lr=1e-4是经验值，太大容易把模型带偏，太小跟不上变化。
监控指标除了 MAE，我还建议加一个「预测方向准确率」——预测涨跌方向和真实一致的比例。这个指标对调度业务比绝对误差更敏感，方向错了，调度就会反向操作。方向准确率低于 80% 时，即使 MAE 还行，也要考虑重新训练。
最后说一个我自己的习惯：每次模型更新前，先把旧模型在最近 7 天数据上的表现存一份 baseline，新模型必须在这份 baseline 上不差于旧模型才允许上线。这个「后悔药」机制帮我挡过好几次越训越差的增量更新。LSTM 做交通流量预测，模型结构本身不复杂，难的是把数据管道、特征一致性、线上监控这三件事做扎实。希望帮到你。
本文还有配套的精品资源，点击获取
