ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

基于深度学习的交通流量预测算法设计与实战源码

基于深度学习的交通流量预测算法设计与实战源码 简介本资源为基于深度学习的交通流量预测算法设计源码面向交通工程、智慧城市与机器学习方向的研究者及开发者用于构建高精度流量预测模型、优化城市交通管理与实时决策。压缩包共267个文件约45.52MB其中212个PNG图像用于数据可视化与预测结果展示7个CSV文件承载时间、地点、流量等原始交通数据7个Python源码覆盖数据处理、模型定义、训练与预测全流程另有pth权重、npy数组、xml配置及ipynb笔记等辅助文件。已有484人学习下载。读者可获得一套完整可复现的预测方案从数据加载与预处理、模型结构设计到训练日志、权重保存与预测评估目录结构清晰便于对照理解深度学习在交通流量场景中的落地路径也可作为课程设计或科研实验的参考模板。1. 从一段环路感应数据说起交通流量预测到底在预测什么早高峰的环路感应线圈每 30 秒回传一次过车数一天下来就是 2880 个点。很多人拿到这份数据的第一反应是套个 LSTM 跑一遍结果 MAPE 卡在 15% 上下怎么调都下不去。问题往往不在模型而在于没想清楚预测的到底是「下一个 30 秒」还是「未来 15 分钟」——前者靠惯性就能猜个八九不离十后者才真正考验模型对周期性和突发拥堵的建模能力。基于深度学习的交通流量预测算法设计源码核心就是把历史流量序列、时间戳特征和路网拓扑关系喂给神经网络输出未来若干时间步的流量估计。它解决的是信号配时、路径诱导和拥堵预警的输入问题适合做智慧交通方向的学生、算法工程师以及手上有感应线圈或卡口数据想跑通一套 baseline 的从业者。这一篇不讲空泛的「深度学习入门」只讲怎么把一份交通流量数据从原始 CSV 推到可复现的预测结果中间每一步的参数和坑都摊开说。2. 数据决定上限交通流量序列的清洗、切分与特征工程2.1 先搞清楚你的数据是「流量」还是「占有率」感应线圈和卡口返回的字段名五花八门常见的有 flow、volume、occupancy、speed、headway。做流量预测目标列必须是单位时间内的过车数veh/30s 或 veh/5min而不是占有率。占有率是时间占比和流量之间是非线性关系直接拿来当标签训练模型学到的曲线形状会完全跑偏。我一般会先做一次字段审计import pandas as pd df pd.read_csv(loop_detector.csv, parse_dates[timestamp]) # 字段审计确认目标列语义 print(df.columns.tolist()) print(df[[flow, occupancy, speed]].describe()) # 流量列不应有负值占有率应在 0~1 之间 assert (df[flow] 0).all(), 流量出现负值检查传感器翻转 assert df[occupancy].between(0, 1).all(), 占有率越界检查单位这段代码做两件事一是把字段语义确认清楚二是用断言把明显异常挡在建模之前。参数上parse_dates必须指定时间列否则后面做滑动窗口时时间索引是字符串重采样会报错。describe()看的是分布如果 flow 的 75 分位数和最大值差了两个数量级说明有尖峰异常需要单独处理。2.2 缺失值不是填 0 就完事交通流量数据的缺失分两种传感器离线造成的整段缺失和偶发的单点丢包。整段缺失如果填 0模型会学到「这段路半夜没车」的假模式。常见做法是按「同星期几同时刻」的历史均值填充而不是全局均值。单点丢包可以用前后向线性插值但连续缺失超过 3 个点就必须走历史均值路线。# 标记缺失 df[is_missing] df[flow].isna() # 单点缺失线性插值限制最多连续填 2 个 df[flow] df[flow].interpolate(methodlinear, limit2, limit_directionboth) # 长段缺失用同星期几同时刻的中位数回填 df[dow] df[timestamp].dt.dayofweek df[slot] df[timestamp].dt.hour * 2 df[timestamp].dt.minute // 30 profile df.groupby([dow, slot])[flow].transform(median) df[flow] df[flow].fillna(profile)limit2是关键参数它保证插值不会跨越太长的空缺。slot的计算假设数据是 30 秒粒度如果你的是 5 分钟粒度分母要改成 5。transform(median)而不是mean是因为交通流量在早晚高峰有尖峰均值容易被拉高。2.3 滑动窗口切分与「未来多步」标签构造深度学习做时间序列输入通常是过去 N 个时间步输出未来 M 个时间步。N 和 M 的选择直接决定任务难度。N 太小比如 6即 3 分钟模型看不到周期性N 太大比如 576即 48 小时参数量和显存都吃不消。我一般从 N121 小时、M630 分钟起步再根据数据周期调整。import numpy as np def make_windows(series, n_in12, n_out6): X, y [], [] for i in range(len(series) - n_in - n_out 1): X.append(series[i : i n_in]) y.append(series[i n_in : i n_in n_out]) return np.array(X), np.array(y) # 按时间顺序切分禁止随机打乱 split int(len(df) * 0.7) train_X, train_y make_windows(df[flow].values[:split]) val_X, val_y make_windows(df[flow].values[split:])这里最容易被忽视的是切分方式。时间序列绝不能train_test_split(shuffleTrue)否则验证集的信息会泄漏到训练集。make_windows里n_in和n_out是唯一两个需要调的窗口参数建议先用n_in12, n_out6跑通再网格搜索。另外如果要做多路段预测series应该是一个二维数组时间 × 路段窗口切分逻辑不变只是每个窗口多了一个维度。3. 模型选型从 LSTM 到时空图网络什么时候该上哪一种3.1 单路段 baselineLSTM 和 GRU 的差别没你想的大如果只有一条路的数据LSTM 和 GRU 的预测精度差距通常在 1% 以内。GRU 参数少、训练快我一般先用 GRU 跑 baseline。关键不在选哪个门控单元而在输入特征的构造——把小时、星期几做 one-hot 编码拼在流量序列后面比单纯堆 LSTM 层数有效得多。import torch import torch.nn as nn class GRUForecaster(nn.Module): def __init__(self, n_in12, n_out6, hidden64, n_feat1): super().__init__() self.gru nn.GRU(n_feat, hidden, batch_firstTrue) self.head nn.Linear(hidden, n_out) def forward(self, x): # x: (batch, n_in, n_feat) out, _ self.gru(x) return self.head(out[:, -1, :]) # 取最后时间步hidden64是单路段任务的常用起点超过 128 在小数据上容易过拟合。out[:, -1, :]取的是最后一个时间步的隐状态这意味着模型把所有历史信息压缩进一个向量再解码未来 6 步。如果未来步数 M 较大比如 12 以上这种「编码-解码」结构会丢信息需要换成 seq2seq。3.2 多路段不上图神经网络就浪费了路网结构当你有几十上百个路段时路段之间的流量是相互影响的——上游堵了下游 10 分钟后流量会下降。这种空间依赖用 CNN 只能捕捉欧氏距离而路网是图结构。常见做法是用 GCN 聚合邻居信息再串 GRU 捕捉时间依赖也就是 DCRNN 那一类架构。如果你不想引入复杂的图卷积一个简化方案是用邻接矩阵做一次线性聚合class GraphGRU(nn.Module): def __init__(self, n_nodes, n_in12, n_out6, hidden64): super().__init__() self.adj nn.Parameter(torch.eye(n_nodes), requires_gradFalse) # 预定义邻接 self.gru nn.GRU(n_nodes, hidden, batch_firstTrue) self.head nn.Linear(hidden, n_nodes * n_out) self.n_out n_out self.n_nodes n_nodes def forward(self, x): # x: (batch, n_in, n_nodes) x torch.einsum(bij,jk-bik, x, self.adj) # 空间聚合 out, _ self.gru(x) return self.head(out[:, -1, :]).view(-1, self.n_out, self.n_nodes)self.adj用requires_gradFalse固定住是因为邻接矩阵应该来自路网拓扑或相关性分析而不是让模型自己学——让模型学邻接在小数据上会退化成全连接。einsum那行做的是「每个节点用邻居的加权和替换自己」权重就是邻接矩阵的值。如果你的路段之间没有明确拓扑可以用历史流量的皮尔逊相关系数阈值比如 0.6来构造邻接。3.3 损失函数MSE 不是唯一选择但别一上来就上 Quantile Loss交通流量预测的评估指标通常是 MAE、RMSE、MAPE。训练时用 MSE 对应 RMSE用 L1 对应 MAE。如果你的数据尖峰多MSE 会被大误差主导这时候换 Huber Loss 更稳。Quantile Loss 适合做区间预测但如果你只要点预测上它只会让调参复杂度翻倍。# Huber Loss对异常值比 MSE 鲁棒 criterion nn.HuberLoss(delta1.0) # delta 控制「多大误差以内用平方以外用线性」delta1.0意味着误差小于 1 辆车时梯度是平方的大于 1 时变成线性的。这个值应该根据你数据的流量量级来定——如果流量均值是 200 veh/30sdelta 设 1.0 太小模型几乎全程走线性分支收敛会慢。我一般设成流量标准差的 0.5 倍左右。4. 训练与调参学习率、批大小和早停的实操参数4.1 学习率用 OneCycle别手工调时间序列模型的损失曲面通常比较平滑OneCycle 策略能在前期快速下降、后期精细收敛。比 ReduceLROnPlateau 省心不用盯着验证集手动降。from torch.optim import Adam from torch.optim.lr_scheduler import OneCycleLR optimizer Adam(model.parameters(), lr1e-3) scheduler OneCycleLR( optimizer, max_lr3e-3, epochs50, steps_per_epochlen(train_loader), pct_start0.3, # 前 30% 步数用于升温 )max_lr3e-3是 Adam 在中小型序列模型上的常用上限pct_start0.3表示前 30% 的训练步把学习率从低升到 max_lr之后余弦下降到接近 0。如果你的 loss 在前几个 epoch 就炸成 NaN先把 max_lr 降到 1e-3。4.2 批大小和序列长度的显存权衡批大小受显存限制但交通流量数据的 batch 不宜太大——同一 batch 内的样本如果来自不同时段梯度方向会互相干扰。我一般用 batch_size64配合 n_in12在单卡 8G 显存上跑 GRU 绰绰有余。如果是 GraphGRU 且节点数超过 200batch_size 要降到 16 或 32。from torch.utils.data import DataLoader, TensorDataset train_ds TensorDataset(torch.FloatTensor(train_X), torch.FloatTensor(train_y)) train_loader DataLoader(train_ds, batch_size64, shuffleTrue) # 注意这里的 shuffle 是在窗口样本之间打乱不是打乱时间顺序这里的shuffleTrue和前面说的「不能随机切分」不矛盾——切分必须在窗口构造之前按时间做窗口构造完之后样本之间打乱是为了让梯度更稳定不会造成信息泄漏。4.3 早停看验证集 MAE耐心值设 8早停的 patience 设太小会欠训练设太大浪费算力。交通流量数据的验证集 MAE 通常在 10 到 20 个 epoch 后进入平台期patience8 比较合适。best_mae float(inf) patience, wait 8, 0 for epoch in range(50): model.train() for xb, yb in train_loader: optimizer.zero_grad() loss criterion(model(xb), yb) loss.backward() optimizer.step() scheduler.step() model.eval() with torch.no_grad(): pred model(torch.FloatTensor(val_X)) val_mae torch.mean(torch.abs(pred - torch.FloatTensor(val_y))).item() if val_mae best_mae: best_mae, wait val_mae, 0 torch.save(model.state_dict(), best.pt) else: wait 1 if wait patience: break保存best.pt而不是最后一个 epoch 的权重是因为验证集 MAE 在后期可能反弹。scheduler.step()放在 batch 循环里因为 OneCycleLR 是按步更新的不是按 epoch。5. 避坑与排查那些让 MAPE 突然翻车的细节5.1 现象验证集 MAE 正常但 MAPE 超过 30%原因MAPE 的分母是真实流量夜间流量接近 0 时微小的绝对误差会被放大成巨大的百分比。解决要么在评估时过滤掉真实流量小于某阈值比如 5 veh/30s的样本要么改用 SMAPE对称平均绝对百分比误差它的分母是真实值和预测值的平均。5.2 现象模型在训练集上表现很好验证集一塌糊涂原因最常见的是数据泄漏——做归一化时用了全量数据的均值和方差而不是只用训练集。解决scaler.fit(train)之后用同一个 scaler 去transform(val)绝不能fit_transform(val)。另一个可能是窗口切分时训练集和验证集有重叠检查split的索引是否留了n_in n_out的缓冲。5.3 现象预测曲线整体滞后于真实曲线原因模型学到了「用上一时刻预测下一时刻」的恒等映射这在一步预测里 MAE 很低但多步预测会累积滞后。解决在损失函数里对多步输出加权越远的步权重越大或者改用 seq2seq 结构让解码器在每一步都接收真实的前一步作为输入teacher forcing但验证时必须关掉 teacher forcing。5.4 现象换了随机种子结果波动超过 5%原因小数据集上模型初始化对结果影响很大。解决跑 3 到 5 个随机种子取平均报告均值±标准差。如果波动仍然很大说明数据量不够考虑减少模型参数量或增加数据增强比如对流量序列做小幅时间抖动。5.5 现象GPU 利用率低训练速度上不去原因DataLoader的num_workers默认是 0数据加载在主进程里串行执行。解决设num_workers4并确保pin_memoryTrue。如果数据集不大把整个数据集预加载成 tensor 放到 GPU 上连 DataLoader 都省了。6. 进阶技巧用残差修正把 MAPE 再压两个点跑通 baseline 之后如果想把 MAPE 从 12% 压到 10% 以内一个性价比很高的做法是残差修正。思路是先用 GRU 预测一个粗结果再用一个轻量模型去预测「真实值减去粗预测」的残差。残差的分布通常比原始流量更平稳更容易学。# 第一步训练基础模型拿到训练集和验证集的预测 base_pred_train model(torch.FloatTensor(train_X)).detach().numpy() base_pred_val model(torch.FloatTensor(val_X)).detach().numpy() # 第二步构造残差标签 residual_train train_y - base_pred_train residual_val val_y - base_pred_val # 第三步用同样的窗口结构训练残差模型 res_model GRUForecaster(n_in12, n_out6, hidden32) # 训练代码与基础模型一致只是标签换成 residual残差模型的hidden可以比基础模型小一半因为它要学的模式更简单。训练完之后最终预测是base_pred residual_pred。注意残差模型必须用基础模型在训练集上的预测来构造标签如果用真实值构造残差里会混入基础模型没见过的信息验证时会虚高。另一个技巧是「多模型加权」。GRU、TCN 和 LightGBM 在交通流量上的误差模式不同简单平均就能降 1 到 2 个点。权重可以用验证集上的 MAE 倒数来定不用搞复杂的 stacking。# 加权融合权重与验证集 MAE 成反比 mae_gru, mae_tcn, mae_lgb 11.2, 12.5, 13.1 w np.array([1/mae_gru, 1/mae_tcn, 1/mae_lgb]) w w / w.sum() final_pred w[0]*pred_gru w[1]*pred_tcn w[2]*pred_lgb这套组合拳我在多个城市的卡口数据上试过baseline GRU 的 MAPE 大约 13%加残差修正后到 11%再加三模型融合能到 9.5% 左右。再往下压就需要引入外部特征了比如天气、节假日标记、周边 POI 密度但这些特征的获取和清洗成本远高于模型本身的调参。我的习惯是先把单模型和残差修正做到位确认数据里没有可挖的周期模式之后再考虑上外部特征。希望帮到你。本文还有配套的精品资源点击获取
返回列表