企业官网建设流程全解析

简介：面向城市轨道交通客流分析预测场景的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 或数据库表，常见字段如下：

拿到原始记录后，第一步是按station_id+event_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(), freq="15min", ) 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_value=0) 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_start="05:30", op_end="23: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(days=3)) & (series.index <= center_time + pd.Timedelta(days=3)) ] 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, periods=96): """历史平均基线：用训练集里同星期、同时段的均值做预测""" # 假定 train_df 已经包含 weekday 和 period（15分钟时段编号）列 avg_table = ( train_df[train_df["station_id"] == station_id] .groupby(["weekday", "period"])[target_col] .mean() .reset_index() ) return avg_table # 96 个桶 = 24小时 * 4（15分钟粒度） baseline_table = historical_average_baseline(train_df, station_id="0101", target_col="flow_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, dtype=torch.float32) self.y = torch.tensor(y, dtype=torch.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_first=True) self.fc = nn.Linear(hidden_dim, output_dim) def forward(self, x): out, _ = self.lstm(x) # 取最后一个时间步的隐状态 out = self.fc(out[:, -1, :]) # batch_first=True 时最后一步是 -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_size=BATCH_SIZE, shuffle=False) # shuffle=False 对时序模型很关键，保持时间顺序

代码里反复强调的shuffle=False、训练/验证集按时间切分、只 fit 训练集的 scaler，这三件事是我见过最多人踩的坑。很多新手把 LSTM 当普通机器学习模型，随机打乱后训练，验证集 MAPE 很低，一上线立刻翻车，原因就是模型把时间顺序信息“背”下来了，换个季节的数据就失真。

3.3 多步预测的策略：直接多输出比递归更稳

LSTM 的输出层用nn.Linear(hidden_dim, output_dim)，其中output_dim = PREDICT_LEN，这样一步到位输出未来 4 个时间步，叫“直接多输出”。另一种方案是“递归预测”：先预测第 1 步，把结果拼回输入窗口，再预测第 2 步，循环 4 次。

我强烈建议用直接多输出。递归预测存在误差累积——第一步预测偏了，第二步会在错误的基础上继续偏，4 步之后误差可能翻倍。直接多输出让模型同时学习 4 个未来时刻的模式，训练更稳定，推理也更快。代价是输出层参数量稍大，但对客流预测这个任务完全不是问题。

训练循环没有特殊之处，但有一个损失函数的选择值得说明：MAE（L1 Loss）比 MSE（L2 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.py：

import 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", encoding="utf-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_dim=X_all.shape[-1], hidden_dim=config["model"]["hidden_dim"], num_layers=config["model"]["num_layers"], output_dim=config["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(f"Validation MAPE: {val_metrics['mape']:.2f}%") if __name__ == "__main__": main()

配置文件config.yaml里我必放这几个参数，也是你以后调参最常动的部分：

一个新数据集拿到手，我一般先用默认参数跑一遍，看验证集 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_min=0, a_max=None) 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里用了shuffle=True。时序数据一旦被随机打乱，相邻样本的时间依赖关系被破坏，模型学不到动态演化规律，只能学到全局分布特征。

解决：训练和验证的DataLoader都设shuffle=False，保持原始时间顺序。更进一步，验证集必须严格从训练集之后的时间切出，不能混着取。

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_first=True) 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 值的预处理。模型架构是下限，数据质量才是上限。如果你正打算拿这套方案做生产系统，我建议先从数据清洗和基线模型起步，两周内让调度员看到一份可解释的预测报告，再逐步上深度模型。希望这些弯路记录能帮到你。

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