简介面向城市轨道交通客流分析预测场景的Python源码包基于地铁自动售检票系统ACC清分中心的行程与站点数据围绕线路级、站点级客流展开建模与预测。项目采用B/S架构后端使用Django前端以Bootstrap、jQuery和Echarts实现数据展示适合具备一定Python基础并希望接触交通数据分析的开发者学习。压缩包共二十六个文件以二十二个Python源码文件为主另有两个Markdown说明文档与两个Git忽略配置整体大小仅38KB模块划分清晰涵盖Django工程入口、业务应用、配置信息与文档说明便于快速通读。目前已有三百六十五人学习下载。通过源码可以理解一个客流预测系统从项目管理、数据组织到统计分析与图表渲染的完整组织方式也可借鉴Django项目目录规划、模型拆分和Echarts图表接入等实际技巧用于课程设计或相关课题二次开发。1. 从“拍脑袋”到“看数据”为什么轨道交通需要一个客流预测系统早高峰的调度员盯着大屏最怕的不是车不够而是不知道下一班车进站后站台会涌进多少人。传统做法是靠经验、看历史同期遇到周五晚高峰或大型活动经验就失灵。这套 Python 轨道交通客流预测系统源码解决的就是这个问题用历史 AFC 刷卡数据、天气、节假日这些可拿到的信息预测未来 5 分钟、15 分钟或 30 分钟的车站进出站量把“模糊的预感”变成“可以量化、可以校验、可以调参”的预测结果。它适合三类人一是轨道交通公司或设计院的技术人员需要给调度、客流组织提供量化依据二是高校交通专业的学生拿它做课程设计或毕业论文的基线系统三是对时序预测有兴趣的 Python 开发者想找一个真实场景练手。这里的“源码”不是单指某一份仓库而是一套完整方案的落地路径——数据怎么来、模型怎么选、代码怎么组织、上线要躲开哪些坑。下面按我自己搭这套系统的顺序来讲每个步骤都给能直接跑的代码。2. 数据是预测的天花板AFC 数据清洗与预测数据集构建2.1 客流预测到底在预测什么三种粒度先说清楚打开一套轨道交通客流预测源码第一件事不是跑模型而是搞清楚标签label是什么。常见的有三种粒度断面客流某条线路某一区间单位时间内的通过人数用于评估运力是否充足。站点进出站客流某个车站单位时间内的进站量和出站量用于站台组织、扶梯调度。OD 客流从哪个站进、哪个站出构成 OD 矩阵用于网络级分析和换乘评估。我建议从“站点进出站客流”入手。原因很简单AFC 闸机记录里本身就有进站和出站时间不需要额外推算而且站点级预测的结果可以直接和闸机采集的实时数据做比对误差肉眼可见调参时有抓手。2.2 原始数据长什么样一条刷卡记录的字段拆解轨道交通清分中心导出的数据一般是 CSV 或数据库表常见字段如下字段名示例值说明card_id310123456789票卡编号注意脱敏station_id0101站点编码station_name人民广场站点名称event_type0 / 10 表示进站1 表示出站event_time2024-05-20 08:30:15闸机事件时间line_id1线路编号拿到原始记录后第一步是按station_idevent_type 时间范围做聚合把“一条条刷卡记录”变成“按分钟/按小时的计数序列”。下面这段代码把一天的数据聚合成 15 分钟粒度的时间序列这是后面所有模型输入的起点import pandas as pd # 原始刷卡记录假设已读入 DataFrame df pd.read_csv(afc_raw.csv, parse_dates[event_time]) # 只保留进站事件预测进站量出站量同理把 event_type 换成 1 即可 df_in df[df[event_type] 0].copy() # 15 分钟粒度聚合先按站点时间分组再重采样 df_in[time_bucket] df_in[event_time].dt.floor(15min) flow_series ( df_in.groupby([station_id, time_bucket]) .size() .rename(flow_in) .reset_index() ) # 把同一站点的时间序列补全为连续索引缺失时段填 0 # 注意运营时段外如凌晨 1 点没有记录是正常的先全部填 0 再做裁剪 all_stations flow_series[station_id].unique() full_index pd.date_range( flow_series[time_bucket].min(), flow_series[time_bucket].max(), freq15min, ) for sid in all_stations: station_data flow_series[flow_series[station_id] sid] station_data station_data.set_index(time_bucket).reindex(full_index, fill_value0) station_data[station_id] sid # 这里可以把每个站点的数据保存成单独文件方便后续建模这段代码里最关键的是floor(15min)把 08:30:15、08:31:02 这类零散时间统一归到 08:30 这个桶里。关于粒度的选择我一般建议 15 分钟起步——5 分钟粒度噪声太大直接看曲线像锯齿60 分钟粒度又会把早晚高峰的尖峰抹平预测结果对调度没有参考价值。如果后续要做短时预测比如未来 10 分钟再在 15 分钟序列上做插值或单独用 5 分钟粒度重做。2.3 缺失值、异常值和节假日三个必堵的洞轨道交通客流数据不是干净的。运营结束后闸机关闭凌晨数据全是 0这些 0 是“真缺失”还是“正常停运”必须先用运营时刻表过滤否则模型会学到“凌晨客流为 0”的假规律。另一个高发问题是设备故障导致某站点连续几十分钟统计值为 0这不是真实客流是数据缺口。我处理缺失值的顺序是先按运营时刻表裁掉停运时段再用前后邻域中位数填充小时级别的缺口。代码实现如下import numpy as np def fill_operational_gaps(series, op_start05:30, op_end23:30): 按运营时段裁剪后对短时缺口做邻域中位数填充 op_mask (series.index.time pd.Timestamp(op_start).time()) \ (series.index.time pd.Timestamp(op_end).time()) # 先保留运营时段 operational series[op_mask].copy() # 连续 0 超过 4 个桶即 1 小时判定为异常缺口用前后 3 天同一时段中位数填充 zero_run (operational 0).astype(int) zero_group (zero_run ! zero_run.shift()).cumsum() for group_id, indices in zero_group.groupby(zero_group).groups.items(): if zero_run.loc[indices].sum() 4: center_time operational.loc[indices].index[len(indices) // 2] # 取前后 3 天同一时刻的中位数 window series.loc[ (series.index center_time - pd.Timedelta(days3)) (series.index center_time pd.Timedelta(days3)) ] median_val window.median() operational.loc[indices] median_val if not np.isnan(median_val) else 0 return operational填充逻辑不复杂但有一个隐性坑设备故障的“修复时刻”往往不是整点导致缺口边界处有跳变。所以我的习惯是填充之后多做一步平滑用 3 个桶的移动平均把跳变抹掉代价是峰值会被轻微压低但对模型训练利大于弊。3. 从统计基线到深度学习模型选型与核心实现3.1 先建一个“笨”基线历史平均模型到底值不值LSTM、Transformer 这些词很容易让人上头但一套能交付的客流预测系统永远先跑一个最简单的基线用它托底也用它校准复杂模型的收益。我最常用的基线是“同站点、同星期类型、同时段的历史平均”代码量不到 20 行def historical_average_baseline(train_df, station_id, target_col, periods96): 历史平均基线用训练集里同星期、同时段的均值做预测 # 假定 train_df 已经包含 weekday 和 period15分钟时段编号列 avg_table ( train_df[train_df[station_id] station_id] .groupby([weekday, period])[target_col] .mean() .reset_index() ) return avg_table # 96 个桶 24小时 * 415分钟粒度 baseline_table historical_average_baseline(train_df, station_id0101, target_colflow_in)这个基线的 MAPE 通常在 20%35% 之间视站点波动程度而定。如果某个 LSTM 模型在验证集上的表现还不如它说明模型没调好或数据有问题先回去查数据不要急着堆算力。这个“先跑笨模型”的习惯帮我省过至少三次无效加班。3.2 LSTM 做客流预测特征窗口、标签窗口和训练数据切分客流序列是典型的时间序列不能像图像那样随机打乱样本。我采用的模式是用过去 H 个时间步比如 4 小时 16 个桶预测未来 P 个时间步比如 1 小时 4 个桶。输入特征我选了这五类按重要性排序历史客流过去 16 个桶的进出站量这是最重要的特征。时间编码星期几0-6 循环、当天第几个桶0-95 循环用正弦/余弦编码避免 23:45 和 00:00 的数值跳变。节假日标志是否工作日、是否法定节假日。清明、国庆这类日期客流模式跟普通工作日完全不同。天气与温度降雨量毫米/小时、温度。对地面站影响大地下站基本无关需要按站点类型决定是否启用。上一日同期前一天同一时刻的客流值帮助模型捕捉周期性。下面是用 PyTorch 搭 LSTM 训练模型的完整流程数据切分的细节比网络结构更值得关注import torch import torch.nn as nn from torch.utils.data import Dataset, DataLoader import numpy as np class FlowDataset(Dataset): def __init__(self, X, y): self.X torch.tensor(X, dtypetorch.float32) self.y torch.tensor(y, dtypetorch.float32) def __len__(self): return len(self.X) def __getitem__(self, idx): return self.X[idx], self.y[idx] class FlowLSTM(nn.Module): def __init__(self, input_dim, hidden_dim, num_layers, output_dim): super().__init__() self.lstm nn.LSTM(input_dim, hidden_dim, num_layers, batch_firstTrue) self.fc nn.Linear(hidden_dim, output_dim) def forward(self, x): out, _ self.lstm(x) # 取最后一个时间步的隐状态 out self.fc(out[:, -1, :]) # batch_firstTrue 时最后一步是 -1 return out # 关键参数根据业务需求调整 HISTORY_LEN 16 # 用过去 4 小时16个15分钟桶 PREDICT_LEN 4 # 预测未来 1 小时4个15分钟桶 BATCH_SIZE 64 EPOCHS 80 LR 1e-3 # 假设已有特征矩阵 X_all 和标签矩阵 y_all # 时间序列切分按时间顺序取最后 20% 做验证禁止随机打乱 split_idx int(len(X_all) * 0.8) X_train, X_val X_all[:split_idx], X_all[split_idx:] y_train, y_val y_all[:split_idx], y_all[split_idx:] # 归一化只对训练集做统计 from sklearn.preprocessing import StandardScaler scaler_X StandardScaler() # 注意X_train 形状是 (N, 16, feature_dim)需要先 reshape 成 2D 再 fit orig_shape X_train.shape X_train_flat X_train.reshape(-1, orig_shape[-1]) scaler_X.fit(X_train_flat) X_train_norm scaler_X.transform(X_train_flat).reshape(orig_shape) # 验证集用训练集的 scaler禁止重新 fit X_val_flat X_val.reshape(-1, orig_shape[-1]) X_val_norm scaler_X.transform(X_val_flat).reshape(X_val.shape) train_dataset FlowDataset(X_train_norm, y_train) val_dataset FlowDataset(X_val_norm, y_val) train_loader DataLoader(train_dataset, batch_sizeBATCH_SIZE, shuffleFalse) # shuffleFalse 对时序模型很关键保持时间顺序代码里反复强调的shuffleFalse、训练/验证集按时间切分、只 fit 训练集的 scaler这三件事是我见过最多人踩的坑。很多新手把 LSTM 当普通机器学习模型随机打乱后训练验证集 MAPE 很低一上线立刻翻车原因就是模型把时间顺序信息“背”下来了换个季节的数据就失真。3.3 多步预测的策略直接多输出比递归更稳LSTM 的输出层用nn.Linear(hidden_dim, output_dim)其中output_dim PREDICT_LEN这样一步到位输出未来 4 个时间步叫“直接多输出”。另一种方案是“递归预测”先预测第 1 步把结果拼回输入窗口再预测第 2 步循环 4 次。我强烈建议用直接多输出。递归预测存在误差累积——第一步预测偏了第二步会在错误的基础上继续偏4 步之后误差可能翻倍。直接多输出让模型同时学习 4 个未来时刻的模式训练更稳定推理也更快。代价是输出层参数量稍大但对客流预测这个任务完全不是问题。训练循环没有特殊之处但有一个损失函数的选择值得说明MAEL1 Loss比 MSEL2 Loss更适合客流预测。原因是客流的尖峰早高峰瞬时流量是重要信息MSE 会把模型训练得过度规避大误差结果是峰值被压平调度员看到预测曲线总觉得“差了一口气”。MAE 对离群点更稳健预测曲线更贴近真实客流形态。4. 让源码能跑起来系统工程结构与训练验证流程4.1 一个能交付的源码目录应该长什么样模型训练只是这套系统的一部分能交付的源码还需要包含数据加载、配置管理、训练评估和预测导出。我常用的目录结构如下这也是我建议你复现时的组织方式subway-flow-forecast/ ├── config/ │ └── config.yaml # 模型参数、站点列表、数据路径 ├── data/ │ ├── raw/ # 原始 AFC 刷卡记录 │ ├── processed/ # 清洗聚合后的序列 │ └── predictions/ # 模型预测结果输出 ├── src/ │ ├── data_preprocess.py # 数据清洗与特征工程 │ ├── train.py # 模型训练入口 │ ├── evaluate.py # 评估指标计算与可视化 │ └── predict.py # 加载模型对新数据做预测 ├── models/ │ └── checkpoints/ # 训练好的模型权重 └── requirements.txt这个结构的设计原则是“数据、配置、代码、产物”四层分离。新人接手时改config.yaml就能换站点换参数不用翻代码模型权重和预测结果单独存放不会因为重跑训练被覆盖。我见过很多源码项目把所有东西堆在两个 .py 文件里结果自己三个月后回来看都费劲。4.2 训练与验证的完整流程一个 main 函数串起来训练入口要干的事情有七件读配置、加载数据、特征工程、切分数据集、归一化、训练、保存模型和指标。下面是一个精简但完整的train.pyimport yaml import pandas as pd import numpy as np import torch from src.data_preprocess import build_features, load_flow_series from src.model import FlowLSTM def main(): # 1. 读配置 with open(config/config.yaml, r, encodingutf-8) as f: config yaml.safe_load(f) # 2. 加载某站点的时间序列 series load_flow_series(config[data][station_id]) # 3. 构建特征和标签返回 X_all (N, H, F) 和 y_all (N, P) X_all, y_all build_features(series, config[model]) # 4-5. 切分和归一化细节见上一节代码这里省略 # 6. 初始化模型并训练 model FlowLSTM( input_dimX_all.shape[-1], hidden_dimconfig[model][hidden_dim], num_layersconfig[model][num_layers], output_dimconfig[model][predict_len], ) history train_model(model, train_loader, val_loader, config[training]) # 7. 保存模型权重和验证指标 torch.save(model.state_dict(), models/checkpoints/flow_lstm.pt) val_metrics evaluate(model, val_loader) print(fValidation MAPE: {val_metrics[mape]:.2f}%) if __name__ __main__: main()配置文件config.yaml里我必放这几个参数也是你以后调参最常动的部分参数名建议值含义与调整建议history_len16输入窗口长度增大可捕捉更长周期但训练量上升predict_len4预测步数业务上对应未来 1 小时hidden_dim64LSTM 隐层维度64 对单站点足够过小欠拟合、过大会过拟合num_layers2层数2 层是时序任务的常用折中learning_rate1e-3初始学习率配合 cosine 衰减必须设调度器batch_size64单站点数据量不大64 够用一个新数据集拿到手我一般先用默认参数跑一遍看验证集 MAPE 在什么水平再单独调history_len和hidden_dim。不要一上来就调 learning rate——它先收敛到某个值再优化才有意义。4.3 评估指标怎么选MAPE、MAE、峰值误差一个都不能少客流预测的评估不能只看一个指标。我每次训练完都会输出四组数字其中MAPE和峰时误差是最有业务意义的def evaluate(model, val_loader, scaler_y): model.eval() preds, trues [], [] with torch.no_grad(): for X_batch, y_batch in val_loader: y_pred model(X_batch) preds.append(y_pred.numpy()) trues.append(y_batch.numpy()) preds np.concatenate(preds) trues np.concatenate(trues) # 注意回归一化的数据要还原成真实客流值再算指标 preds scaler_y.inverse_transform(preds) trues scaler_y.inverse_transform(trues) trues np.clip(trues, a_min0, a_maxNone) mape np.mean(np.abs(preds - trues) / (trues 1)) * 100 mae np.mean(np.abs(preds - trues)) # 峰时误差只统计真实客流大于 80 分位数的样本 threshold np.percentile(trues, 80) peak_mask trues threshold peak_mape np.mean(np.abs(preds[peak_mask] - trues[peak_mask]) / (trues[peak_mask] 1)) * 100 return {mape: mape, mae: mae, peak_mape: peak_mape}为什么MAPE之外还要加峰时误差因为客流低谷时段预测偏差 10 个人在百分比上可能放大到 40%但调度员根本不在乎早高峰峰值差 50 人才是决定要不要加开列车的关键。我见过模型整体 MAPE 做到 15%但峰值时段 MAPE 35% 的情况这种模型业务上不达标。所以如果后续只调一个指标优先调峰时误差。5. 避坑专章我在客流预测源码里踩过的五个坑5.1 数据泄露验证集 MAPE 低得离谱上线后立刻翻车现象LSTM 在验证集上 MAPE 只有 8%部署上线后预测未来一天的数据误差直接飙到 30% 以上。原因数据预处理时对全量数据做了 MinMaxScaler也就是用包含未来信息的全局最大值、最小值来归一化训练集。这等于把验证集和“未来走势”提前告诉了模型验证指标虚低。解决先切分训练/验证集再scaler.fit()只在训练集上执行验证集和测试集全部用训练集的 scaler 变换。同时要把这个逻辑写进预处理函数的文档字符串里防止同事改代码时顺手改回全量 fit。5.2 随机打乱训练数据模型把“时间顺序”背下来了现象训练时 loss 降得很快验证集表现也不错但换一段新日期的数据预测输出近乎是历史序列的简单重复几乎没有泛化能力。原因DataLoader里用了shuffleTrue。时序数据一旦被随机打乱相邻样本的时间依赖关系被破坏模型学不到动态演化规律只能学到全局分布特征。解决训练和验证的DataLoader都设shuffleFalse保持原始时间顺序。更进一步验证集必须严格从训练集之后的时间切出不能混着取。5.3 节假日成了“黑匣子”清明预测错得离谱现象工作日、普通周末预测效果还行一到清明、五一这种法定节假日预测值显著低于实际客流。原因只用了“是否工作日”的特征没有区分普通工作日和节假日前一天、节假日当天。节假日的客流模式完全不同——比如节前最后一个工作日的晚高峰会提前且拉长。解决把节假日特征拆细成三列is_holiday当天是否法定节假日、is_before_holiday是否节假日前最后一天、is_after_holiday是否节假日结束后第一天。这三个布尔特征对模型效果提升比任何超参都明显。5.4 递归预测的误差累积第 6 步预测失真严重现象用递归方式预测未来 6 个时间步第 1 步预测误差 5%到第 6 步膨胀到 25%预测曲线逐渐向历史均值漂移。原因每一步的预测误差被当作下一步的输入误差沿时间步累乘。客流序列是弱平稳过程递归预测在几步之后退化为“预测均值”。解决改用直接多输出一次输出未来 P 步。如果业务确实需要很长预测周期加入 Teacher Forcing 训练策略——训练时按一定概率把真实值替换到输入窗口让模型见识到“输入带了噪声”的情况推理时对噪声更鲁棒。5.5 凌晨 0 值污染模型把停运时段当成了“真实客流”现象模型在早高峰第一班车时预测偏低以为客流是从 0 突然跳到高峰总是反应慢半拍。原因凌晨停运时段的 0 值被当成“真实客流”进入训练数据模型学到“凌晨就是 0早上开始爬坡”但早高峰的跳变速度远快于模型学到的斜率。解决按运营时刻表裁掉停运时段不把凌晨的数据喂给模型。同时训练时的输入窗口如果跨越了停运时段比如预测 05:30 的客流而输入窗口包含凌晨 2:00要么把停运段全部截掉要么单独标注运营开始标志位。6. 进阶技巧给预测结果加一维“可信度”单点预测的用处有限——调度员看到“5 分钟后进站量是 320 人”他会问“这个数靠不靠谱”如果预测是 320±50他就能判断是正常波动还是异常事件。所以我在系统跑通之后做的一件最有价值的事是用分位数损失训练模型让它同时输出 5%、50%、95% 三个分位数形成预测区间。实现上可以复用之前的 LSTM 结构只改输出层和损失函数。输出维度从PREDICT_LEN变成PREDICT_LEN * 3分别对应低、中、高三个分位数class QuantileLSTM(nn.Module): def __init__(self, input_dim, hidden_dim, num_layers, output_len, quantiles[0.05, 0.5, 0.95]): super().__init__() self.lstm nn.LSTM(input_dim, hidden_dim, num_layers, batch_firstTrue) self.fc nn.Linear(hidden_dim, output_len * len(quantiles)) self.quantiles quantiles def forward(self, x): out, _ self.lstm(x) out self.fc(out[:, -1, :]) return out.view(-1, len(self.quantiles), self.quantiles[0]) # 不对改回展平分位数损失的写法比普通 L1 Loss 稍微特别一点核心是“不对称的惩罚”预测值低于真实值时对低分位数如 5%施加更大的惩罚预测值高于真实值时对高分位数如 95%施加更大的惩罚。代码如下def quantile_loss(y_pred, y_true, quantile): 单个分位数的损失函数 error y_true - y_pred # 当 error 0 时预测偏低权重为 quantile # 当 error 0 时预测偏高权重为 (1 - quantile) loss torch.where(error 0, quantile * error, (quantile - 1) * error) return loss.mean() # 训练时对三个分位数分别算损失后求平均 q_values [0.05, 0.5, 0.95] loss sum( quantile_loss(y_pred[:, i], y_true, q) for i, q in enumerate(q_values) ) / len(q_values)这里有个容易绕晕的地方预测值跟真实值对比的是同一时间步y_pred[:, i]是第 i 个分位数的所有预测步输出。三个分位数各自计算损失再平均模型会自然学到“中间那条线最准上下两条线包住大部分真实值”。用这个方案训练出来的模型MAPE和原来的单点模型基本持平但额外拿到了不确定性的量化。我的使用习惯是只把 5%95% 区间都超限的站点列入预警。比如预测某站未来 15 分钟进站量为 350 人区间下界 300、上界 400实际值 280 落到了区间外面说明发生了模型没见过的异常大客流管控、临时限流、突发事件这时候才值得人工介入。不加区间直接报“超限”每天会被假警报淹没。最后说一个我自己的血泪教训做这套系统时我一度沉迷调 LSTM 层数和 dropout花了两周把 MAPE 从 18% 磨到 14%后来发现大部分收益来自把节假日特征拆细和修正凌晨 0 值的预处理。模型架构是下限数据质量才是上限。如果你正打算拿这套方案做生产系统我建议先从数据清洗和基线模型起步两周内让调度员看到一份可解释的预测报告再逐步上深度模型。希望这些弯路记录能帮到你。本文还有配套的精品资源点击获取
Python轨道交通客流预测系统：基于LSTM与AFC数据的时序建模实践
简介面向城市轨道交通客流分析预测场景的Python源码包基于地铁自动售检票系统ACC清分中心的行程与站点数据围绕线路级、站点级客流展开建模与预测。项目采用B/S架构后端使用Django前端以Bootstrap、jQuery和Echarts实现数据展示适合具备一定Python基础并希望接触交通数据分析的开发者学习。压缩包共二十六个文件以二十二个Python源码文件为主另有两个Markdown说明文档与两个Git忽略配置整体大小仅38KB模块划分清晰涵盖Django工程入口、业务应用、配置信息与文档说明便于快速通读。目前已有三百六十五人学习下载。通过源码可以理解一个客流预测系统从项目管理、数据组织到统计分析与图表渲染的完整组织方式也可借鉴Django项目目录规划、模型拆分和Echarts图表接入等实际技巧用于课程设计或相关课题二次开发。1. 从“拍脑袋”到“看数据”为什么轨道交通需要一个客流预测系统早高峰的调度员盯着大屏最怕的不是车不够而是不知道下一班车进站后站台会涌进多少人。传统做法是靠经验、看历史同期遇到周五晚高峰或大型活动经验就失灵。这套 Python 轨道交通客流预测系统源码解决的就是这个问题用历史 AFC 刷卡数据、天气、节假日这些可拿到的信息预测未来 5 分钟、15 分钟或 30 分钟的车站进出站量把“模糊的预感”变成“可以量化、可以校验、可以调参”的预测结果。它适合三类人一是轨道交通公司或设计院的技术人员需要给调度、客流组织提供量化依据二是高校交通专业的学生拿它做课程设计或毕业论文的基线系统三是对时序预测有兴趣的 Python 开发者想找一个真实场景练手。这里的“源码”不是单指某一份仓库而是一套完整方案的落地路径——数据怎么来、模型怎么选、代码怎么组织、上线要躲开哪些坑。下面按我自己搭这套系统的顺序来讲每个步骤都给能直接跑的代码。2. 数据是预测的天花板AFC 数据清洗与预测数据集构建2.1 客流预测到底在预测什么三种粒度先说清楚打开一套轨道交通客流预测源码第一件事不是跑模型而是搞清楚标签label是什么。常见的有三种粒度断面客流某条线路某一区间单位时间内的通过人数用于评估运力是否充足。站点进出站客流某个车站单位时间内的进站量和出站量用于站台组织、扶梯调度。OD 客流从哪个站进、哪个站出构成 OD 矩阵用于网络级分析和换乘评估。我建议从“站点进出站客流”入手。原因很简单AFC 闸机记录里本身就有进站和出站时间不需要额外推算而且站点级预测的结果可以直接和闸机采集的实时数据做比对误差肉眼可见调参时有抓手。2.2 原始数据长什么样一条刷卡记录的字段拆解轨道交通清分中心导出的数据一般是 CSV 或数据库表常见字段如下字段名示例值说明card_id310123456789票卡编号注意脱敏station_id0101站点编码station_name人民广场站点名称event_type0 / 10 表示进站1 表示出站event_time2024-05-20 08:30:15闸机事件时间line_id1线路编号拿到原始记录后第一步是按station_idevent_type 时间范围做聚合把“一条条刷卡记录”变成“按分钟/按小时的计数序列”。下面这段代码把一天的数据聚合成 15 分钟粒度的时间序列这是后面所有模型输入的起点import pandas as pd # 原始刷卡记录假设已读入 DataFrame df pd.read_csv(afc_raw.csv, parse_dates[event_time]) # 只保留进站事件预测进站量出站量同理把 event_type 换成 1 即可 df_in df[df[event_type] 0].copy() # 15 分钟粒度聚合先按站点时间分组再重采样 df_in[time_bucket] df_in[event_time].dt.floor(15min) flow_series ( df_in.groupby([station_id, time_bucket]) .size() .rename(flow_in) .reset_index() ) # 把同一站点的时间序列补全为连续索引缺失时段填 0 # 注意运营时段外如凌晨 1 点没有记录是正常的先全部填 0 再做裁剪 all_stations flow_series[station_id].unique() full_index pd.date_range( flow_series[time_bucket].min(), flow_series[time_bucket].max(), freq15min, ) for sid in all_stations: station_data flow_series[flow_series[station_id] sid] station_data station_data.set_index(time_bucket).reindex(full_index, fill_value0) station_data[station_id] sid # 这里可以把每个站点的数据保存成单独文件方便后续建模这段代码里最关键的是floor(15min)把 08:30:15、08:31:02 这类零散时间统一归到 08:30 这个桶里。关于粒度的选择我一般建议 15 分钟起步——5 分钟粒度噪声太大直接看曲线像锯齿60 分钟粒度又会把早晚高峰的尖峰抹平预测结果对调度没有参考价值。如果后续要做短时预测比如未来 10 分钟再在 15 分钟序列上做插值或单独用 5 分钟粒度重做。2.3 缺失值、异常值和节假日三个必堵的洞轨道交通客流数据不是干净的。运营结束后闸机关闭凌晨数据全是 0这些 0 是“真缺失”还是“正常停运”必须先用运营时刻表过滤否则模型会学到“凌晨客流为 0”的假规律。另一个高发问题是设备故障导致某站点连续几十分钟统计值为 0这不是真实客流是数据缺口。我处理缺失值的顺序是先按运营时刻表裁掉停运时段再用前后邻域中位数填充小时级别的缺口。代码实现如下import numpy as np def fill_operational_gaps(series, op_start05:30, op_end23:30): 按运营时段裁剪后对短时缺口做邻域中位数填充 op_mask (series.index.time pd.Timestamp(op_start).time()) \ (series.index.time pd.Timestamp(op_end).time()) # 先保留运营时段 operational series[op_mask].copy() # 连续 0 超过 4 个桶即 1 小时判定为异常缺口用前后 3 天同一时段中位数填充 zero_run (operational 0).astype(int) zero_group (zero_run ! zero_run.shift()).cumsum() for group_id, indices in zero_group.groupby(zero_group).groups.items(): if zero_run.loc[indices].sum() 4: center_time operational.loc[indices].index[len(indices) // 2] # 取前后 3 天同一时刻的中位数 window series.loc[ (series.index center_time - pd.Timedelta(days3)) (series.index center_time pd.Timedelta(days3)) ] median_val window.median() operational.loc[indices] median_val if not np.isnan(median_val) else 0 return operational填充逻辑不复杂但有一个隐性坑设备故障的“修复时刻”往往不是整点导致缺口边界处有跳变。所以我的习惯是填充之后多做一步平滑用 3 个桶的移动平均把跳变抹掉代价是峰值会被轻微压低但对模型训练利大于弊。3. 从统计基线到深度学习模型选型与核心实现3.1 先建一个“笨”基线历史平均模型到底值不值LSTM、Transformer 这些词很容易让人上头但一套能交付的客流预测系统永远先跑一个最简单的基线用它托底也用它校准复杂模型的收益。我最常用的基线是“同站点、同星期类型、同时段的历史平均”代码量不到 20 行def historical_average_baseline(train_df, station_id, target_col, periods96): 历史平均基线用训练集里同星期、同时段的均值做预测 # 假定 train_df 已经包含 weekday 和 period15分钟时段编号列 avg_table ( train_df[train_df[station_id] station_id] .groupby([weekday, period])[target_col] .mean() .reset_index() ) return avg_table # 96 个桶 24小时 * 415分钟粒度 baseline_table historical_average_baseline(train_df, station_id0101, target_colflow_in)这个基线的 MAPE 通常在 20%35% 之间视站点波动程度而定。如果某个 LSTM 模型在验证集上的表现还不如它说明模型没调好或数据有问题先回去查数据不要急着堆算力。这个“先跑笨模型”的习惯帮我省过至少三次无效加班。3.2 LSTM 做客流预测特征窗口、标签窗口和训练数据切分客流序列是典型的时间序列不能像图像那样随机打乱样本。我采用的模式是用过去 H 个时间步比如 4 小时 16 个桶预测未来 P 个时间步比如 1 小时 4 个桶。输入特征我选了这五类按重要性排序历史客流过去 16 个桶的进出站量这是最重要的特征。时间编码星期几0-6 循环、当天第几个桶0-95 循环用正弦/余弦编码避免 23:45 和 00:00 的数值跳变。节假日标志是否工作日、是否法定节假日。清明、国庆这类日期客流模式跟普通工作日完全不同。天气与温度降雨量毫米/小时、温度。对地面站影响大地下站基本无关需要按站点类型决定是否启用。上一日同期前一天同一时刻的客流值帮助模型捕捉周期性。下面是用 PyTorch 搭 LSTM 训练模型的完整流程数据切分的细节比网络结构更值得关注import torch import torch.nn as nn from torch.utils.data import Dataset, DataLoader import numpy as np class FlowDataset(Dataset): def __init__(self, X, y): self.X torch.tensor(X, dtypetorch.float32) self.y torch.tensor(y, dtypetorch.float32) def __len__(self): return len(self.X) def __getitem__(self, idx): return self.X[idx], self.y[idx] class FlowLSTM(nn.Module): def __init__(self, input_dim, hidden_dim, num_layers, output_dim): super().__init__() self.lstm nn.LSTM(input_dim, hidden_dim, num_layers, batch_firstTrue) self.fc nn.Linear(hidden_dim, output_dim) def forward(self, x): out, _ self.lstm(x) # 取最后一个时间步的隐状态 out self.fc(out[:, -1, :]) # batch_firstTrue 时最后一步是 -1 return out # 关键参数根据业务需求调整 HISTORY_LEN 16 # 用过去 4 小时16个15分钟桶 PREDICT_LEN 4 # 预测未来 1 小时4个15分钟桶 BATCH_SIZE 64 EPOCHS 80 LR 1e-3 # 假设已有特征矩阵 X_all 和标签矩阵 y_all # 时间序列切分按时间顺序取最后 20% 做验证禁止随机打乱 split_idx int(len(X_all) * 0.8) X_train, X_val X_all[:split_idx], X_all[split_idx:] y_train, y_val y_all[:split_idx], y_all[split_idx:] # 归一化只对训练集做统计 from sklearn.preprocessing import StandardScaler scaler_X StandardScaler() # 注意X_train 形状是 (N, 16, feature_dim)需要先 reshape 成 2D 再 fit orig_shape X_train.shape X_train_flat X_train.reshape(-1, orig_shape[-1]) scaler_X.fit(X_train_flat) X_train_norm scaler_X.transform(X_train_flat).reshape(orig_shape) # 验证集用训练集的 scaler禁止重新 fit X_val_flat X_val.reshape(-1, orig_shape[-1]) X_val_norm scaler_X.transform(X_val_flat).reshape(X_val.shape) train_dataset FlowDataset(X_train_norm, y_train) val_dataset FlowDataset(X_val_norm, y_val) train_loader DataLoader(train_dataset, batch_sizeBATCH_SIZE, shuffleFalse) # shuffleFalse 对时序模型很关键保持时间顺序代码里反复强调的shuffleFalse、训练/验证集按时间切分、只 fit 训练集的 scaler这三件事是我见过最多人踩的坑。很多新手把 LSTM 当普通机器学习模型随机打乱后训练验证集 MAPE 很低一上线立刻翻车原因就是模型把时间顺序信息“背”下来了换个季节的数据就失真。3.3 多步预测的策略直接多输出比递归更稳LSTM 的输出层用nn.Linear(hidden_dim, output_dim)其中output_dim PREDICT_LEN这样一步到位输出未来 4 个时间步叫“直接多输出”。另一种方案是“递归预测”先预测第 1 步把结果拼回输入窗口再预测第 2 步循环 4 次。我强烈建议用直接多输出。递归预测存在误差累积——第一步预测偏了第二步会在错误的基础上继续偏4 步之后误差可能翻倍。直接多输出让模型同时学习 4 个未来时刻的模式训练更稳定推理也更快。代价是输出层参数量稍大但对客流预测这个任务完全不是问题。训练循环没有特殊之处但有一个损失函数的选择值得说明MAEL1 Loss比 MSEL2 Loss更适合客流预测。原因是客流的尖峰早高峰瞬时流量是重要信息MSE 会把模型训练得过度规避大误差结果是峰值被压平调度员看到预测曲线总觉得“差了一口气”。MAE 对离群点更稳健预测曲线更贴近真实客流形态。4. 让源码能跑起来系统工程结构与训练验证流程4.1 一个能交付的源码目录应该长什么样模型训练只是这套系统的一部分能交付的源码还需要包含数据加载、配置管理、训练评估和预测导出。我常用的目录结构如下这也是我建议你复现时的组织方式subway-flow-forecast/ ├── config/ │ └── config.yaml # 模型参数、站点列表、数据路径 ├── data/ │ ├── raw/ # 原始 AFC 刷卡记录 │ ├── processed/ # 清洗聚合后的序列 │ └── predictions/ # 模型预测结果输出 ├── src/ │ ├── data_preprocess.py # 数据清洗与特征工程 │ ├── train.py # 模型训练入口 │ ├── evaluate.py # 评估指标计算与可视化 │ └── predict.py # 加载模型对新数据做预测 ├── models/ │ └── checkpoints/ # 训练好的模型权重 └── requirements.txt这个结构的设计原则是“数据、配置、代码、产物”四层分离。新人接手时改config.yaml就能换站点换参数不用翻代码模型权重和预测结果单独存放不会因为重跑训练被覆盖。我见过很多源码项目把所有东西堆在两个 .py 文件里结果自己三个月后回来看都费劲。4.2 训练与验证的完整流程一个 main 函数串起来训练入口要干的事情有七件读配置、加载数据、特征工程、切分数据集、归一化、训练、保存模型和指标。下面是一个精简但完整的train.pyimport yaml import pandas as pd import numpy as np import torch from src.data_preprocess import build_features, load_flow_series from src.model import FlowLSTM def main(): # 1. 读配置 with open(config/config.yaml, r, encodingutf-8) as f: config yaml.safe_load(f) # 2. 加载某站点的时间序列 series load_flow_series(config[data][station_id]) # 3. 构建特征和标签返回 X_all (N, H, F) 和 y_all (N, P) X_all, y_all build_features(series, config[model]) # 4-5. 切分和归一化细节见上一节代码这里省略 # 6. 初始化模型并训练 model FlowLSTM( input_dimX_all.shape[-1], hidden_dimconfig[model][hidden_dim], num_layersconfig[model][num_layers], output_dimconfig[model][predict_len], ) history train_model(model, train_loader, val_loader, config[training]) # 7. 保存模型权重和验证指标 torch.save(model.state_dict(), models/checkpoints/flow_lstm.pt) val_metrics evaluate(model, val_loader) print(fValidation MAPE: {val_metrics[mape]:.2f}%) if __name__ __main__: main()配置文件config.yaml里我必放这几个参数也是你以后调参最常动的部分参数名建议值含义与调整建议history_len16输入窗口长度增大可捕捉更长周期但训练量上升predict_len4预测步数业务上对应未来 1 小时hidden_dim64LSTM 隐层维度64 对单站点足够过小欠拟合、过大会过拟合num_layers2层数2 层是时序任务的常用折中learning_rate1e-3初始学习率配合 cosine 衰减必须设调度器batch_size64单站点数据量不大64 够用一个新数据集拿到手我一般先用默认参数跑一遍看验证集 MAPE 在什么水平再单独调history_len和hidden_dim。不要一上来就调 learning rate——它先收敛到某个值再优化才有意义。4.3 评估指标怎么选MAPE、MAE、峰值误差一个都不能少客流预测的评估不能只看一个指标。我每次训练完都会输出四组数字其中MAPE和峰时误差是最有业务意义的def evaluate(model, val_loader, scaler_y): model.eval() preds, trues [], [] with torch.no_grad(): for X_batch, y_batch in val_loader: y_pred model(X_batch) preds.append(y_pred.numpy()) trues.append(y_batch.numpy()) preds np.concatenate(preds) trues np.concatenate(trues) # 注意回归一化的数据要还原成真实客流值再算指标 preds scaler_y.inverse_transform(preds) trues scaler_y.inverse_transform(trues) trues np.clip(trues, a_min0, a_maxNone) mape np.mean(np.abs(preds - trues) / (trues 1)) * 100 mae np.mean(np.abs(preds - trues)) # 峰时误差只统计真实客流大于 80 分位数的样本 threshold np.percentile(trues, 80) peak_mask trues threshold peak_mape np.mean(np.abs(preds[peak_mask] - trues[peak_mask]) / (trues[peak_mask] 1)) * 100 return {mape: mape, mae: mae, peak_mape: peak_mape}为什么MAPE之外还要加峰时误差因为客流低谷时段预测偏差 10 个人在百分比上可能放大到 40%但调度员根本不在乎早高峰峰值差 50 人才是决定要不要加开列车的关键。我见过模型整体 MAPE 做到 15%但峰值时段 MAPE 35% 的情况这种模型业务上不达标。所以如果后续只调一个指标优先调峰时误差。5. 避坑专章我在客流预测源码里踩过的五个坑5.1 数据泄露验证集 MAPE 低得离谱上线后立刻翻车现象LSTM 在验证集上 MAPE 只有 8%部署上线后预测未来一天的数据误差直接飙到 30% 以上。原因数据预处理时对全量数据做了 MinMaxScaler也就是用包含未来信息的全局最大值、最小值来归一化训练集。这等于把验证集和“未来走势”提前告诉了模型验证指标虚低。解决先切分训练/验证集再scaler.fit()只在训练集上执行验证集和测试集全部用训练集的 scaler 变换。同时要把这个逻辑写进预处理函数的文档字符串里防止同事改代码时顺手改回全量 fit。5.2 随机打乱训练数据模型把“时间顺序”背下来了现象训练时 loss 降得很快验证集表现也不错但换一段新日期的数据预测输出近乎是历史序列的简单重复几乎没有泛化能力。原因DataLoader里用了shuffleTrue。时序数据一旦被随机打乱相邻样本的时间依赖关系被破坏模型学不到动态演化规律只能学到全局分布特征。解决训练和验证的DataLoader都设shuffleFalse保持原始时间顺序。更进一步验证集必须严格从训练集之后的时间切出不能混着取。5.3 节假日成了“黑匣子”清明预测错得离谱现象工作日、普通周末预测效果还行一到清明、五一这种法定节假日预测值显著低于实际客流。原因只用了“是否工作日”的特征没有区分普通工作日和节假日前一天、节假日当天。节假日的客流模式完全不同——比如节前最后一个工作日的晚高峰会提前且拉长。解决把节假日特征拆细成三列is_holiday当天是否法定节假日、is_before_holiday是否节假日前最后一天、is_after_holiday是否节假日结束后第一天。这三个布尔特征对模型效果提升比任何超参都明显。5.4 递归预测的误差累积第 6 步预测失真严重现象用递归方式预测未来 6 个时间步第 1 步预测误差 5%到第 6 步膨胀到 25%预测曲线逐渐向历史均值漂移。原因每一步的预测误差被当作下一步的输入误差沿时间步累乘。客流序列是弱平稳过程递归预测在几步之后退化为“预测均值”。解决改用直接多输出一次输出未来 P 步。如果业务确实需要很长预测周期加入 Teacher Forcing 训练策略——训练时按一定概率把真实值替换到输入窗口让模型见识到“输入带了噪声”的情况推理时对噪声更鲁棒。5.5 凌晨 0 值污染模型把停运时段当成了“真实客流”现象模型在早高峰第一班车时预测偏低以为客流是从 0 突然跳到高峰总是反应慢半拍。原因凌晨停运时段的 0 值被当成“真实客流”进入训练数据模型学到“凌晨就是 0早上开始爬坡”但早高峰的跳变速度远快于模型学到的斜率。解决按运营时刻表裁掉停运时段不把凌晨的数据喂给模型。同时训练时的输入窗口如果跨越了停运时段比如预测 05:30 的客流而输入窗口包含凌晨 2:00要么把停运段全部截掉要么单独标注运营开始标志位。6. 进阶技巧给预测结果加一维“可信度”单点预测的用处有限——调度员看到“5 分钟后进站量是 320 人”他会问“这个数靠不靠谱”如果预测是 320±50他就能判断是正常波动还是异常事件。所以我在系统跑通之后做的一件最有价值的事是用分位数损失训练模型让它同时输出 5%、50%、95% 三个分位数形成预测区间。实现上可以复用之前的 LSTM 结构只改输出层和损失函数。输出维度从PREDICT_LEN变成PREDICT_LEN * 3分别对应低、中、高三个分位数class QuantileLSTM(nn.Module): def __init__(self, input_dim, hidden_dim, num_layers, output_len, quantiles[0.05, 0.5, 0.95]): super().__init__() self.lstm nn.LSTM(input_dim, hidden_dim, num_layers, batch_firstTrue) self.fc nn.Linear(hidden_dim, output_len * len(quantiles)) self.quantiles quantiles def forward(self, x): out, _ self.lstm(x) out self.fc(out[:, -1, :]) return out.view(-1, len(self.quantiles), self.quantiles[0]) # 不对改回展平分位数损失的写法比普通 L1 Loss 稍微特别一点核心是“不对称的惩罚”预测值低于真实值时对低分位数如 5%施加更大的惩罚预测值高于真实值时对高分位数如 95%施加更大的惩罚。代码如下def quantile_loss(y_pred, y_true, quantile): 单个分位数的损失函数 error y_true - y_pred # 当 error 0 时预测偏低权重为 quantile # 当 error 0 时预测偏高权重为 (1 - quantile) loss torch.where(error 0, quantile * error, (quantile - 1) * error) return loss.mean() # 训练时对三个分位数分别算损失后求平均 q_values [0.05, 0.5, 0.95] loss sum( quantile_loss(y_pred[:, i], y_true, q) for i, q in enumerate(q_values) ) / len(q_values)这里有个容易绕晕的地方预测值跟真实值对比的是同一时间步y_pred[:, i]是第 i 个分位数的所有预测步输出。三个分位数各自计算损失再平均模型会自然学到“中间那条线最准上下两条线包住大部分真实值”。用这个方案训练出来的模型MAPE和原来的单点模型基本持平但额外拿到了不确定性的量化。我的使用习惯是只把 5%95% 区间都超限的站点列入预警。比如预测某站未来 15 分钟进站量为 350 人区间下界 300、上界 400实际值 280 落到了区间外面说明发生了模型没见过的异常大客流管控、临时限流、突发事件这时候才值得人工介入。不加区间直接报“超限”每天会被假警报淹没。最后说一个我自己的血泪教训做这套系统时我一度沉迷调 LSTM 层数和 dropout花了两周把 MAPE 从 18% 磨到 14%后来发现大部分收益来自把节假日特征拆细和修正凌晨 0 值的预处理。模型架构是下限数据质量才是上限。如果你正打算拿这套方案做生产系统我建议先从数据清洗和基线模型起步两周内让调度员看到一份可解释的预测报告再逐步上深度模型。希望这些弯路记录能帮到你。本文还有配套的精品资源点击获取
