LCTFP实战：用CNN+LSTM预测高速公路交通流量

简介：LCTFP是一个针对高速公路短时交通流预测的深度学习项目，采用1D CNN与LSTM组合结构，分别提取历史车流数据的空间与时间特征，解决传统模型难以同时刻画两类特征的不足，适合智能交通领域研究者、时间序列建模初学者以及相关课题开发者使用。资源共21个文件、压缩包31MB，包含7个Python脚本、6个H5模型权重、4个JSON配置、2张网络结构图等：脚本覆盖原始数据读取、缺失值处理、时间归一化、滑动窗口构造、模型训练与评估等多个环节；权重与配置文件可直接加载用于复现实验，图片能辅助理解网络结构。随包提供PeMS真实高速公路交通流数据及完整处理流程，支持通过hyperas自动搜索超参数，并内置SAE、LSTM、CNN等基准模型用于横向对比。已有5334人学习，适合需要快速运行完整交通流预测流程并做扩展实验的进阶开发者。整体代码模块化程度较高，便于替换数据集或调整网络参数，也可迁移至其他交通流预测场景。

1. 先说结论：LCTFP 不是黑匣子，它就是把 CNN 和 LSTM 串起来的高速公路交通流量预测模型

LCTFP 这个缩写第一次看到确实容易懵，但拆开看它做的事情并不玄学：用 CNN 提取流量序列里的局部特征，再用 LSTM 记住跨时段的依赖，最后用 Python 输出未来半小时到一小时的高速公路交通流量预测结果。我在做断面流量短时预测时，最看重的就是它能不能用一份普通收费站 5 分钟粒度数据直接跑通。

它适合谁？适合手里已经有流量、速度和占有率数据，想做短时预测落地的人。比如匝道汇入预警、施工窗口安排、服务区饱和度预判，都可以拿这个结构当基线。它不挑显卡，普通 CPU 也能训练，真正的难点不在模型有多复杂，而在数据切分、滑动窗口构造和评估指标怎么设。下面按我平时从数据到上线的顺序讲。

2. 数据准备才是 LCTFP 的胜负手：从断面流量到滑动窗口样本

用 LCTFP 之前，先把交通流量预测问题定义清楚。高速公路交通流量预测的目标通常是“未来 N 个时间段的流量”，不是一个点。如果数据是 5 分钟聚合一次，未来 30 分钟就是 6 个点，模型要一次输出 6 个值。把任务定义成多步回归，后面评估才不会因为只看单点误差被带偏。

2.1 先定特征、时间粒度和预测窗口

我一般会保留下面四个字段。flow 是预测目标，speed 和 occupancy 是辅助特征，time 用来排序和构造时间窗口。如果收费流水能折算出车型比例，也可以加进去，但第一条原则是：特征必须和预测目标来自同一断面，不能把下游收费站的数据挪过来。否则模型学到的只是“下游流量传到上游”的假关系，换个断面就失效。

时间粒度选 5 分钟比较折中。1 分钟噪声太大，15 分钟会抹平早晚高峰的尖峰。预测窗口我常用 lookback=12、horizon=6，也就是“看过去 1 小时，预测未来半小时”。这个组合对收费站排队预警足够，也不会让 LSTM 背太长的记忆负担。

2.2 清洗两步走：排序、去重、按时间插值

拿到 CSV 后不要急着训练，先做两件事：把时间轴理干净，把缺失值补得尽量不像假数据。

import pandas as pd import numpy as np df = pd.read_csv("flow_5min.csv", parse_dates=["time"]) # 1) 时间有序且唯一，这是后续滑窗的前提 df = df.sort_values("time").drop_duplicates(subset="time") # 2) 重采样，保证每个 5 分钟都有记录 df = df.set_index("time").resample("5min").mean().reset_index() # 3) 流量用线性插值，不要用 0 填充 df["flow"] = df["flow"].interpolate(method="linear") df["speed"] = df["speed"].interpolate(method="linear") df["occupancy"] = df["occupancy"].interpolate(method="linear") print(df.head()) print(df.isna().sum())

这段代码先排序去重，再用 resample 补出缺失时间戳。为什么不用 fillna(0) 填流量？高速夜间断面流量本身可能接近 0，用 0 填充会把设备故障和真实低峰混在一起，模型会误以为 0 是常态。线性插值是保守做法；但如果一段断网超过两三个小时，插值会插出一条平滑的假曲线，我建议直接把这段标记出来，不参与训练。

提示：Python 环境里装好 pandas、numpy、scikit-learn、tensorflow 这套常用库，上面代码就能直接跑。关键不是库的版本号，而是插值之后再看一次 isna()，确保没有残留的 NaN。

如果数据是按车道给的流量，我建议先合并成断面总流量再训练，而不是把每个车道当作独立序列。车辆换道会让单车道流量看起来像噪声，LCTFP 很难从里面学到稳定规律。

2.3 按时间切分数据集，再写滑动窗口函数

很多人习惯用 train_test_split 做随机切分，但时间序列绝对不行。车流有明显的早晚高峰、工作日和周末差异，随机切分等于把未来信息混进训练集，验证集指标看着漂亮，上线以后立刻露馅。我习惯按时间比例切：前 70% 训练，中间 15% 验证，最后 15% 测试，验证集紧跟在训练集后面。

from sklearn.preprocessing import MinMaxScaler def split_by_time(df, train_ratio=0.7, val_ratio=0.15): n = len(df) train_end = int(n * train_ratio) val_end = int(n * (train_ratio + val_ratio)) return df.iloc[:train_end].copy(), df.iloc[train_end:val_end].copy(), df.iloc[val_end:].copy() feature_cols = ["flow", "speed", "occupancy"] target_col = "flow" train_df, val_df, test_df = split_by_time(df) # 只对训练集拟合 scaler，防止数据泄漏 feat_scaler = MinMaxScaler() targ_scaler = MinMaxScaler() train_feat = feat_scaler.fit_transform(train_df[feature_cols]) train_targ = targ_scaler.fit_transform(train_df[[target_col]]) val_feat = feat_scaler.transform(val_df[feature_cols]) val_targ = targ_scaler.transform(val_df[[target_col]]) test_feat = feat_scaler.transform(test_df[feature_cols]) test_targ = targ_scaler.transform(test_df[[target_col]])

这里两个 scaler 分开 fit 是有意为之：特征矩阵是三列一起缩放，目标值单独用一个 scaler，后续做反归一化才不会被特征维度干扰。测试集在训练阶段只能碰 transform，不能碰 fit_transform，这个原则后面避坑部分会再提。

接着构造滑动窗口：

def make_windows(features, target, lookback=12, horizon=6): X, y = [], [] for i in range(len(features) - lookback - horizon + 1): X.append(features[i:i + lookback]) y.append(target[i + lookback:i + lookback + horizon]) return np.array(X), np.array(y) X_train, y_train = make_windows(train_feat, train_targ, lookback=12, horizon=6) X_val, y_val = make_windows(val_feat, val_targ, lookback=12, horizon=6) X_test, y_test = make_windows(test_feat, test_targ, lookback=12, horizon=6) print(X_train.shape, y_train.shape)

X_train 的形状是（样本数, 12, 3），12 是 lookback，3 是特征数。y_train 的形状是（样本数, 6），代表未来 6 个 5 分钟流量。这里最容易错的地方是窗口滑动的终点：循环上限必须减掉 lookback + horizon，否则最后一个窗口会滑到序列末尾之外。这一步形状不对，后面模型结构再复杂也跑不出可解释的结果。

3. 模型搭建：用 Keras 写一个 Conv1D 提特征、LSTM 做记忆的 LCTFP

模型结构本身不神秘，难的是理解“为什么 CNN 在前、LSTM 在后”。我见过不少同事上来就堆两层 LSTM，结果训练很慢还过拟合；也见过有人用 Conv2D 处理一维序列，把维度搞得很别扭。LCTFP 这个方向更适合用一维卷积。

3.1 为什么 Conv1D 在前、LSTM 在后

流量序列里最值得关注的模式是局部突变。连续三个 5 分钟流量突然下降，可能是事故或者匝道管制；连续 20 分钟缓慢上升，可能是进入高峰的前兆。这种相邻时间步之间的相关关系，用 Conv1D 提取最直接，卷积核只在时间轴上滑动，参数共享让模型不用为每一时刻单独学一套权重。

LSTM 的职责是接住更长的依赖。同样是早上 7 点，昨天的 7 点和今天的 7 点相关性很强，但中间隔了 144 个 5 分钟时间步，Conv1D 的局部感受野覆盖不到这么远。LSTM 用门控状态把这种跨时段的记忆保留下来，和 CNN 提取的局部特征互补。

为什么不把两层并联？并联结构适合一个分支处理原始序列、另一个分支处理外部特征的场景。如果只有流量、速度、占有率三列数据，串联结构更简单，也更容易排查问题。先让 CNN 把序列缩短并抽象，再让 LSTM 处理压缩后的特征，训练稳定性更好。

3.2 张量形状一路变化

定义模型之前，先算一下每一层输出是什么形状。以 lookback=12、特征数=3 为例：

None 是批次维度，实际训练时等于 batch_size。注意两个 MaxPooling 之后，时间步从 12 变成了 3，这个 3 就是 LSTM 的输入时间步长。只要时间步大于 1，LSTM 就能工作；如果池化层再多加一层，时间步变成 1，LSTM 就退化成全连接了，这是很多人忽略的细节。

3.3 完整 build 函数：参数、激活、Dropout 的一次性选择

下面是我常用的一个最小可跑版本，函数名直接叫 build_lctfp。

import tensorflow as tf from tensorflow.keras.layers import Input, Conv1D, MaxPooling1D, LSTM, Dropout, Dense from tensorflow.keras.models import Model def build_lctfp(lookback=12, n_features=3, horizon=6): inputs = Input(shape=(lookback, n_features)) # 第一层卷积：提取相邻 3 个 5 分钟内的局部趋势 x = Conv1D(filters=64, kernel_size=3, padding="same", activation="relu")(inputs) x = MaxPooling1D(pool_size=2)(x) # 第二层卷积：在 6 个时间步上做更高层抽象 x = Conv1D(filters=128, kernel_size=3, padding="same", activation="relu")(x) x = MaxPooling1D(pool_size=2)(x) # LSTM 接收 (batch, 3, 128)，记住跨时段依赖 x = LSTM(units=64, return_sequences=False)(x) x = Dropout(0.2)(x) # 输出前加一层全连接做非线性组合 x = Dense(units=32, activation="relu")(x) outputs = Dense(units=horizon)(x) model = Model(inputs, outputs, name="LCTFP") return model model = build_lctfp() model.compile( optimizer=tf.keras.optimizers.Adam(learning_rate=0.001), loss="mse", metrics=[tf.keras.metrics.RootMeanSquaredError()] ) model.summary()

几个参数我说一下自己的默认选择。Conv1D 的 kernel_size 在时间序列里一般取 3 或 5，取 3 意味着每次看相邻三个 5 分钟，取 5 就能覆盖 25 分钟，但要小心后面池化后时间步不够。filters 先给 64 和 128，是希望模型在小数据上不至于立刻过拟合；如果数据量很大，再翻倍也不迟。Dropout 放在 LSTM 之后而不是之前，是因为 LSTM 的输出维度较高，这里做随机丢弃对抑制过拟合最有效。

损失函数用 mse 而不是 mae，是因为 mse 对大偏差更敏感，流量预测里最怕的就是高峰时刻突然差一大截。如果后续发现结果太平滑，可以切换到 mae 再对比一轮。这段代码在 CPU 上也能跑，epoch 不是越大越好，后面配合早停看验证集再说。

4. 训练与评估：预测曲线先回答“早晚高峰准不准”，再看总误差

很多深度学习教程只到模型训练就停了，但对交通流量预测来说，评估才是真正决定能不能上线的部分。训练集 loss 降到很低没有意义，我要的是测试集上早晚高峰时段误差可控。

4.1 按时间顺序给模型喂数据：回调与超参

训练 LCTFP 时，我会用三个回调：ModelCheckpoint 保存最优权重，EarlyStopping 防止训练过度，ReduceLROnPlateau 在验证 loss 停滞时自动降低学习率。

from tensorflow.keras.callbacks import ModelCheckpoint, EarlyStopping, ReduceLROnPlateau checkpoint = ModelCheckpoint("lctfp_best.h5", monitor="val_loss", save_best_only=True) early_stop = EarlyStopping(monitor="val_loss", patience=15, restore_best_weights=True) lr_scheduler = ReduceLROnPlateau(monitor="val_loss", factor=0.5, patience=5, min_lr=1e-5) history = model.fit( X_train, y_train, validation_data=(X_val, y_val), epochs=100, batch_size=128, callbacks=[checkpoint, early_stop, lr_scheduler], shuffle=True, verbose=1 )

batch_size 我一般取 128。LSTM 对批量大小比纯 CNN 敏感：批量太大，梯度更新方向太平滑，会把早晚高峰的突变抹掉；批量太小，训练又慢又容易震荡。128 在多数路测数据上是个稳妥起点。

shuffle=True 只在每个 epoch 内打乱样本顺序，没有打乱单个样本内部的时间顺序，所以可以开。它能让训练更稳，避免模型总按同一个时间顺序被喂数据。如果数据量很小，也可以用 shuffle=False，但容易过拟合，我不推荐。

4.2 反归一化以后算指标，不要直接拿 0-1 的数值汇报

模型的输出是归一化后的流量，不能直接拿来汇报。要先用 target_scaler 做 inverse_transform，把预测值和真实值还原成车辆数，再算 RMSE 和 MAPE。

y_pred = model.predict(X_test, verbose=0) # 反归一化回真实车辆数 y_pred_inv = targ_scaler.inverse_transform(y_pred) y_test_inv = targ_scaler.inverse_transform(y_test) rmse = np.sqrt(np.mean((y_pred_inv - y_test_inv) ** 2)) mae = np.mean(np.abs(y_pred_inv - y_test_inv)) # 防止流量为 0 时除零，给分母加一个保护 mape = np.mean(np.abs((y_test_inv - y_pred_inv) / np.maximum(y_test_inv, 1))) * 100 print("RMSE:", rmse) print("MAE:", mae) print("MAPE:", mape)

这里要注意，targ_scaler 是在训练集的目标值上 fit 的，不能拿测试集重新 fit。否则还原出来的预测值只是“看起来合理”，其实包含了未来 min/max 信息，评估结果会偏乐观。np.maximum(y_test_inv, 1)是给除零兜底；更严格的做法是只在 flow 大于某个阈值的时段算 MAPE，低流量时段不参与。

4.3 按小时和峰值分桶看误差

整体 RMSE 只能告诉你大方向对不对，不能回答“早高峰能不能用”。我的习惯是把测试集按时间拆成几个桶：早高峰 7:00-9:00、晚高峰 17:00-19:00、平峰、夜间，分别计算 MAE。

分桶评估做起来不复杂，把测试集的 time 列和 y_test_inv 拼成一个 DataFrame，然后用 pandas 的 between_time 筛出对应时段再算误差。关键是注意预测的 6 个时间点如何归属时段：我一般取预测窗口的第一个时间点作为标签，比如预测 8:00-8:30，就归到早高峰桶。如果混着归，误差表会变得很难读。

5. 避坑清单：LCTFP 实测里最容易翻车的 5 个原点

这一章是我最想让你先看的部分。模型跑不出来通常不是网络结构的问题，而是下面这些看似不起眼的位置出了问题。

5.1 时间错位：预测曲线比真实晚一个时段

现象：把预测值和真实值画在同一张图上，预测曲线明显比真实曲线向右平移了一格。

原因：滑动窗口构造时把输出索引写错。常见写法是 y.append(target[i + lookback - 1])，这里只取了窗口结束后第一个点，但那时刚好是“当前时刻”，而模型要预测的是未来。正确做法是 y.append(target[i + lookback: i + lookback + horizon])。

解决：打印前三条样本的索引，确认 y 的第一个值确实晚于 X 的最后一个时间步。我调试时会给每个窗口加上时间戳列，对比“窗口最后时间”和“预测目标开始时间”是否相差一个时段。差一个点不是小事，它会让所有评估指标都看着不错，但上线后每次都慢一拍。

5.2 归一化泄漏：训练集用了未来的 min/max

现象：训练集和验证集 loss 都很好，到了测试现场却差一大截，甚至模型输出的曲线被压在一个很窄的范围里。

原因：数据还没切分就直接调用 MinMaxScaler 对整个 DataFrame 做了 fit_transform。scaler 在计算 min 和 max 时已经看到了未来数据，相当于把测试集的数值范围提前告诉了模型。

解决：严格按时间切分，先切训练、验证、测试，再在训练集上 fit scaler，验证集和测试集只允许 transform。如果特征和目标分开两个 scaler，也要保证两个 scaler 都只见过训练集。这个错误在实验阶段不会报错，所以特别隐蔽。

5.3 时间步被 pool 没了：看到 Negative dimension size 时别急着调层

现象：模型 summary 之前还能跑，加上池化层后直接报类似 Negative dimension size caused by subtracting 2 from 1 的错。

原因：lookback 太小，或者 MaxPooling1D 层数太多，把时间维压缩成了 0 或负数。比如 lookback=4，池化一次变 2，池化两次就出问题。

解决：先算好每一层 shape，再搭模型。我习惯把 lookback 设成 12 或 24，保证经过两层池化后时间步至少还有 3。如果确实想用很小的 lookback，就用 padding="same"，并且少放池化层。打印中间层 shape 的成本很低，比训练报错了再猜快多了。

5.4 缺失值救过头：把流量缺失补成 0

现象：验证集整体 loss 不高，但周末早上的预测明显偏低，尤其节假日前后误差很大。

原因：数据清洗阶段用了 fillna(0) 或者直接把缺失行删掉后没做插值。周末凌晨流量低，缺失如果被 0 填进去，模型学到“低流量时段输出 0”，白天也会把偏小值当真。

解决：流量缺失用相邻时间的线性插值，或者按同一个星期几、同一个时间段的中位数填充。更稳妥的做法是额外生成一列 is_missing 标记，把缺失时刻告诉模型，这样模型能区分“设备没回数据”和“路上确实没车”。如果缺失超过两小时，建议把这天标记成异常样本，不进训练集。

5.5 夜间低流量让 MAPE 失真

现象：MAPE 一算 60%，看起来模型没法用，但把早高峰单独拿出来看误差又不大。

原因：夜间流量绝对值小，比如真实值是 5 辆/5 分钟，预测值是 8 辆，绝对误差只有 3，MAPE 却到了 60%。低流量时段对 MAPE 的贡献被放大，会掩盖真正需要关注的早晚高峰问题。

解决：评估时分桶看，或者只在流量超过 30 辆/5 分钟的时间段计算 MAPE。对外汇报时先给 MAE，再给高峰时段的 MAPE，不要拿一个整体数字下结论。模型是给运营用的，运营关心的永远是高峰时段准不准，而不是凌晨有多么精确。

6. 上线前的一个习惯：用滚动预测代替整段预测，少被数据漂移坑

模型在测试集上表现不错之后，我建议立刻写一个滚动预测函数。测试集的评价是“一次性预测未来 6 步”，但线上真实场景是每 5 分钟来一条新数据，模型要反复读取最新观测值，再重新预测未来 30 分钟。

def rolling_predict(model, latest_df, lookback=12, horizon=6): # 只取最近 lookback 个时间点的真实观测值 recent = latest_df[feature_cols].tail(lookback).copy() recent_scaled = feat_scaler.transform(recent) X = recent_scaled.values.reshape(1, lookback, len(feature_cols)) y_pred_norm = model.predict(X, verbose=0) return targ_scaler.inverse_transform(y_pred_norm).ravel()

滚动预测的关键是不能把模型上一步的预测值当成输入继续传下去，否则误差会累积。每次新数据到达，把窗口往前挪一格，丢掉最旧的一条，再让模型重新算。这样预测结果永远基于真实观测，误差不会滚动放大。我在部署这类模型时，还会保留最近 2 周真实流量做盲测，每两天用最新数据微调一次，观察 MAE 是否漂移。

另一个值得做的习惯是把时间特征加进 LCTFP 的输入。最简单的是加一列 is_weekend，再复杂一点可以加小时的正弦编码。虽然标题里 CNN 和 LSTM 是主角，但高速公路流量的周模式很强，没有时间特征时模型只能硬记。这个技巧看起来不起眼，却往往比调整 LSTM units 更有效。

我现在的做法是：所有新模型先跑一版滚动预测，再和旧模型在早高峰和节假日两个场景上并行对比两周，确认误差没有再切换。这条路我走过不少弯路，最深的感受是模型结构只占三成，数据切分和滚动评估占七成。希望帮到你。

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