阅读路线

说明

第一章 问题介绍：什么是城市时空预测

城市每时每刻都在产生带有"位置 + 时间"标签的数据。时空预测就是利用过去的时空观测，推断未来的时空状态。

1.1 问题定义

给定过去 T 个时刻的观测，预测未来 T' 个时刻的状态：

• X：历史观测张量。网格型数据形状通常为 [T, C, H, W]，图型数据为 [T, N, C]。
• 𝒢：空间结构，如路网邻接矩阵；网格数据的空间结构是隐式的。
• C：外部上下文，如时间编码、节假日、天气、POI、事件。
• F_θ：时空模型，例如 SimVP、PredRNN、Graph WaveNet，或时空大模型。

1.2 典型城市场景

1.3 核心难点

1.4 为什么需要时空大模型

传统做法是"一个城市、一个任务、一个模型"。时空大模型（时空基础模型）尝试在多城市、多任务、多模态数据上统一预训练，得到可迁移的时空表征，再通过少量微调或零样本方式服务新场景。 学术界已有相关探索，如面向交通的 OpenCity、面向城市网格的 UniST 等。它们的具体结论请以论文原文为准。

第二章 技术路线：五步法

无论用哪个开源项目，城市时空预测的工程链路都可以抽象为下面五步。

2.1 第一步：数据收集

• 来源：交通卡口和线圈、GPS 轨迹、手机信令（需合规授权）、气象再分析数据、开放基准数据集。
• 处理：时间对齐（统一采样间隔）、空间映射（栅格化或建图）、缺失值处理、归一化（Z-Score / Min-Max）。
• 切分：必须按时间顺序切分训练 / 验证 / 测试集，常用比例为 7:1:2 或 6:2:2。随机打乱会造成信息泄漏。
• 样本构造：用滑动窗口生成 (历史 T 步, 未来 T' 步) 样本对。

2.2 第二步：模型选择

2.3 第三步：进行训练

统一的做法是用配置文件驱动训练，固定随机种子，记录日志与检查点，用验证集做早停和超参搜索。规模大时，使用分布式数据并行（DDP）和混合精度。

2.4 第四步：进行评测

• 网格 / 视频类：MSE、MAE、SSIM、PSNR、LPIPS。
• 交通类：MAE、RMSE、MAPE。通常按预测步长 Horizon 3 / 6 / 12 分别报告，并对缺失值做 masked 计算。
• 工程类：推理延迟、显存、吞吐、参数量与 FLOPs。

2.5 第五步：部署上线

包括模型导出（TorchScript / ONNX）、在线推理服务、数据流水线接入、监控漂移与定期重训，详见第七章。

2.6 本教程中各项目的定位

第三章 OpenSTL：网格型时空预测基准

OpenSTL 是一个时空预测学习（Spatio-Temporal Predictive Learning）的综合基准与代码库，相关论文发表于 NeurIPS 2023 Datasets & Benchmarks。 它把"基于循环"的方法（如 PredRNN）和"无循环"的方法（如 SimVP）统一在同一套训练与评测框架下。

3.0 源码结构速览

OpenSTL/
├── configs/              # 按 数据集/方法 组织的配置文件，如 configs/mmnist/simvp/SimVP_gSTA.py
├── openstl/
│   ├── api/              # 实验入口（BaseExperiment）：组装数据、方法、训练循环
│   ├── core/             # 评测指标（metrics）、优化器与调度器等
│   ├── datasets/         # 各数据集的 dataloader_*.py
│   ├── methods/          # 方法层：封装每种算法的前向与损失（SimVP、PredRNN、TAU...）
│   ├── models/           # 网络结构定义
│   ├── modules/          # 可复用的基础模块（各类时空翻译器 Block）
│   └── utils/            # 参数解析、日志、分布式工具
├── tools/
│   ├── prepare_data/     # 数据下载脚本
│   ├── train.py          # 训练入口
│   └── test.py           # 测试入口
└── environment.yml

# 安装（以仓库 README 为准）
git clone https://github.com/chengtan9907/OpenSTL.git
cd OpenSTL
conda env create -f environment.yml
conda activate OpenSTL
python setup.py develop

OpenSTL 内置的基准覆盖合成数据、人体运动、驾驶场景、交通、气象等，其中与城市预测最相关的是 TaxiBJ（北京出租车流入/流出网格）和 WeatherBench（气象再分析网格）。

# 下载数据（脚本名以 tools/prepare_data/ 实际内容为准）
bash tools/prepare_data/download_mmnist.sh
bash tools/prepare_data/download_taxibj.sh
# 数据默认放在 ./data/ 下，如 data/moving_mnist、data/taxibj

接入自有城市数据

如果要用自己城市的网格数据，可以参照 openstl/datasets/dataloader_taxibj.py，写一个返回 (X, Y) 的 Dataset。约定形状为 [B, T, C, H, W]。

import numpy as np
import torch
from torch.utils.data import Dataset

class CityGridDataset(Dataset):
    """自有城市网格数据：raw 形状 [总时长, C, H, W]，已按时间顺序切分好"""
    def __init__(self, raw: np.ndarray, pre_seq=4, aft_seq=4, mean=None, std=None):
        self.mean = raw.mean() if mean is None else mean   # 只用训练集统计量
        self.std = raw.std() if std is None else std
        self.data = ((raw - self.mean) / (self.std + 1e-8)).astype(np.float32)
        self.pre, self.aft = pre_seq, aft_seq

    def __len__(self):
        return len(self.data) - self.pre - self.aft + 1

    def __getitem__(self, i):
        x = self.data[i : i + self.pre]                          # [T, C, H, W]
        y = self.data[i + self.pre : i + self.pre + self.aft]    # [T', C, H, W]
        return torch.from_numpy(x), torch.from_numpy(y)

SimVP 的结构是：空间编码器把每帧压缩成隐特征，时空翻译器把时间维折叠进通道后做卷积或注意力，最后由解码器还原。论文基准的一个重要发现是：无循环方法在效率和性能之间往往有更好的平衡。

# 单卡训练：Moving MNIST + SimVP(gSTA)
python tools/train.py -d mmnist --lr 1e-3 \
    -c configs/mmnist/simvp/SimVP_gSTA.py --ex_name mmnist_simvp_gsta

# 城市人流：TaxiBJ + SimVP(gSTA)
python tools/train.py -d taxibj \
    -c configs/taxibj/simvp/SimVP_gSTA.py --ex_name taxibj_simvp_gsta

# 多卡分布式训练（脚本参数以仓库为准）
PORT=29001 CUDA_VISIBLE_DEVICES=0,1,2,3 bash tools/dist_train.sh \
    configs/taxibj/simvp/SimVP_gSTA.py 4 -d taxibj --ex_name taxibj_simvp_gsta_ddp

配置文件是一个 Python 文件，核心字段示例如下（字段名以实际配置为准）：

# configs/taxibj/simvp/SimVP_gSTA.py（示意）
method = 'SimVP'
model_type = 'gSTA'     # 时空翻译器类型
hid_S = 32              # 空间隐通道
hid_T = 256             # 时间翻译器隐通道
N_S = 2                 # 编码/解码层数
N_T = 8                 # 翻译器层数
lr = 1e-3
batch_size = 16
sched = 'onecycle'

训练日志、检查点和测试结果默认写入 work_dirs/<ex_name>/。

python tools/test.py -d taxibj \
    -c configs/taxibj/simvp/SimVP_gSTA.py --ex_name taxibj_simvp_gsta

第四章 LibCity：城市交通时空预测框架

LibCity 是面向城市时空任务的统一开源库，覆盖交通状态预测、轨迹下一跳预测、到达时间估计、路网匹配等任务，并提供大量复现模型。 本章聚焦交通状态预测（traffic_state_pred）。

git clone https://github.com/LibCity/Bigscity-LibCity.git
cd Bigscity-LibCity
pip install -r requirements.txt

LibCity 用统一的原子文件描述各类城市数据，把不同来源的数据集转换为同一格式：

常用数据集包括 METR_LA、PEMS_BAY、PEMSD4、PEMSD8 和 NYC 出租车 / 单车类数据。数据按 README 指引下载后，放到 raw_data/<数据集名>/。

自有数据转换示例

import pandas as pd

# 传感器点位 -> .geo
geo = pd.DataFrame({"geo_id": [0, 1, 2], "type": "Point",
                    "coordinates": ["[116.39,39.90]", "[116.40,39.91]", "[116.41,39.92]"]})
geo.to_csv("raw_data/MyCity/MyCity.geo", index=False)

# 邻接关系 -> .rel（cost 可用道路距离）
rel = pd.DataFrame({"rel_id": [0, 1], "type": "geo",
                    "origin_id": [0, 1], "destination_id": [1, 2], "cost": [520.0, 610.0]})
rel.to_csv("raw_data/MyCity/MyCity.rel", index=False)

# 观测 -> .dyna（time 使用 ISO 格式，间隔需统一，如 5 分钟）
# 列：dyna_id, type, time, entity_id, traffic_speed
# 字段规范请对照 LibCity 文档中的原子文件说明

# 统一入口：任务 + 模型 + 数据集
python run_model.py --task traffic_state_pred --model GWNET --dataset METR_LA

# 使用自定义配置覆盖超参（config_file 指向一个 json，不含扩展名）
python run_model.py --task traffic_state_pred --model GWNET \
    --dataset METR_LA --config_file my_gwnet

{
  "max_epoch": 100,
  "learning_rate": 0.001,
  "batch_size": 64,
  "input_window": 12,
  "output_window": 12,
  "gpu": true,
  "gpu_id": 0
}

LibCity 也提供超参搜索脚本（如 hyper_tune.py），具体参数请查阅文档。

训练结束后会自动在测试集上评测，并把结果写到 libcity/cache/<exp_id>/evaluate_cache/ 等目录。常用指标为 masked 版本的 MAE、MAPE、RMSE，并按每个预测步分别给出。

# 自行复核：masked MAE（忽略缺失值 0）
import numpy as np
def masked_mae(pred, true, null_val=0.0):
    mask = ~np.isclose(true, null_val)
    return np.abs(pred[mask] - true[mask]).mean()

第五章 BasicTS：公平的时空预测基准

BasicTS 关注"公平、可复现"的比较：统一的数据预处理、切分、归一化和评测协议，让不同模型可以真正横向对比。 它同时支持时空预测（交通）和长序列时序预测。

git clone https://github.com/GestaltCogTeam/BasicTS.git
cd BasicTS
pip install -r requirements.txt   # PyTorch 版本请按 README 单独安装

• 时空数据集：METR-LA、PEMS-BAY、PEMS03/04/07/08。
• 时序数据集：ETTh1/ETTm1、Electricity、Weather 等。
• 按 README 下载预处理后的数据包，解压到 datasets/。旧版本需要运行 scripts/data_preparation/ 下的生成脚本。

接入新城市数据时，应遵循与内置数据相同的切分比例和归一化方式，否则结果不可比。

# 以 STID 在 PEMS04 上为例（配置路径以实际仓库为准）
python experiments/train.py -c baselines/STID/PEMS04.py -g 0

每个配置文件集中声明数据集、模型参数、优化器、调度器、输入/输出长度和评测 Horizon，方便复现。修改超参时，建议复制一份配置后再改，不要直接改原文件。

建议同时记录：多随机种子下的均值和标准差、单 epoch 训练耗时、推理耗时。这些信息会直接用于部署阶段的选型。

第六章 DeepOpen 时空大模型

前三章是"一个数据集训练一个模型"。时空大模型的目标是：在多城市、多任务、多模态数据上统一预训练，再快速适配新的城市和任务。

• 规模与多样性优先：汇聚多城市、多类型（人流、车速、需求、气象）数据。可以复用第三至五章的数据集作为预训练语料。
• 统一表示：网格数据切成 p×p×Δt 的时空 Patch；图数据按节点和时间窗口生成 Token。同时附带位置编码（经纬度）、时间编码（时刻、星期）和数据源标识。
• 分城市归一化：不同城市的量纲差异很大，通常对每个城市、每个变量单独做标准化。
• 严格防泄漏：用于零样本评测的目标城市或时间段，必须完全排除在预训练数据之外。
• 合规：轨迹、信令类数据需要脱敏、聚合，并取得授权。

# 示意：把不同来源的数据统一成 (tokens, meta)
import numpy as np

def grid_to_patches(x, p=4, dt=2):
    """x: [T, C, H, W] -> [N_tokens, dt*C*p*p]；要求 T、H、W 分别能被 dt、p 整除"""
    T, C, H, W = x.shape
    x = x.reshape(T // dt, dt, C, H // p, p, W // p, p)
    x = x.transpose(0, 3, 5, 1, 2, 4, 6)          # [T', H', W', dt, C, p, p]
    return x.reshape(-1, dt * C * p * p)

sample = np.random.rand(8, 2, 32, 32).astype("float32")   # 如 TaxiBJ 片段
tokens = grid_to_patches(sample)
print(tokens.shape)   # (4*8*8, 2*2*4*4) = (256, 64)

3.1 预训练（示意接口）

# 伪代码：请替换为 DeepOpen 官方 SDK 的真实接口
from deepopen import STFoundationModel, PretrainConfig   # 示意

cfg = PretrainConfig(
    corpora=["taxibj", "metr_la", "pems04", "weatherbench_t2m", "mycity_flow"],
    objective="masked_reconstruction", mask_ratio=0.75,
    patch_size=4, time_patch=2,
    precision="bf16", strategy="ddp",            # 或 fsdp / deepspeed
)
model = STFoundationModel(size="base")
model.pretrain(cfg, output_dir="ckpt/pretrain")

3.2 下游微调（示意接口）

from deepopen import FinetuneConfig   # 示意

ft = FinetuneConfig(
    dataset="mycity_flow", input_len=12, output_len=12,
    method="lora", lora_rank=8, lr=1e-4, epochs=20,
)
model = STFoundationModel.from_pretrained("ckpt/pretrain")
model.finetune(ft, output_dir="ckpt/mycity_lora")

如果暂时没有 DeepOpen SDK，可以先用 PyTorch 自建一个小规模时空 Transformer，在 OpenSTL 的 TaxiBJ 与 BasicTS 的 PEMS 数据上做联合预训练，验证整条路线是否可行。

第七章 部署上线：让预测进入城市业务

模型只有接入实时数据流、稳定产出预测并持续被监控，才算真正落地。

7.1 模型导出

# 以 OpenSTL 的 SimVP 为例（构造参数与权重路径以实际训练配置和 work_dirs 为准）
import torch
from openstl.models import SimVP_Model

model = SimVP_Model(in_shape=(4, 2, 32, 32), hid_S=32, hid_T=256,
                    N_S=2, N_T=8, model_type="gSTA")
state = torch.load("work_dirs/taxibj_simvp_gsta/checkpoint.pth", map_location="cpu")
model.load_state_dict(state.get("state_dict", state), strict=False)  # 检查 missing/unexpected keys
model.eval()

dummy = torch.randn(1, 4, 2, 32, 32)
torch.onnx.export(model, dummy, "simvp_taxibj.onnx",
                  input_names=["x"], output_names=["y"],
                  dynamic_axes={"x": {0: "batch"}, "y": {0: "batch"}},
                  opset_version=17)

7.2 推理服务（FastAPI + onnxruntime）

# serve.py  依赖：pip install fastapi uvicorn onnxruntime numpy
import numpy as np
import onnxruntime as ort
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

MEAN, STD = 0.0, 1.0          # 必须替换为训练集的归一化统计量
EXPECTED = (4, 2, 32, 32)     # [T, C, H, W]

sess = ort.InferenceSession("simvp_taxibj.onnx", providers=["CPUExecutionProvider"])
app = FastAPI(title="City Flow Forecast")

class Req(BaseModel):
    history: list            # 形状 [T, C, H, W] 的嵌套列表，原始量纲

@app.post("/predict")
def predict(req: Req):
    x = np.asarray(req.history, dtype=np.float32)
    if x.shape != EXPECTED:
        raise HTTPException(400, f"history shape must be {EXPECTED}, got {x.shape}")
    x = ((x - MEAN) / STD)[None]                       # 加 batch 维
    y = sess.run(["y"], {"x": x})[0][0] * STD + MEAN   # 反归一化
    return {"forecast": np.clip(y, 0, None).tolist()}  # 流量非负

# 启动：uvicorn serve:app --host 0.0.0.0 --port 8000

7.3 运维与持续迭代

7.4 全书回顾