轨迹预测完整代码实战：滑动窗口、GRU基线到插件化推理

简介：面向具备Python与深度学习基础、希望实践轨迹预测并具备一定代码阅读能力的学生和算法开发者，这套压缩包提供可直接研读的完整工程代码及配套数据。资源共11个文件，以7个Python脚本为核心，辅以模型检查点、一份CSV轨迹数据以及TensorFlow数据分片与索引文件，整体约723KB，体量轻便但模块相对完整。代码围绕轨迹数据处理、LSTM与GRU循环神经网络建模展开，并加入逻辑回归作为对比基线，覆盖数据读取、模型定义、训练状态保存与结果分析等环节，能够帮助读者理解序列预测项目的组织方式与实验流程。已有231人学习，适合用作课程设计、毕业项目或小规模实验参考，也可替换数据、调整网络结构并复现对比实验，进一步掌握从预处理、训练到评估的完整思路。

1. 拆开完整代码.zip轨迹预测：先跑通哪条最小链路

完整代码.zip轨迹预测这个压缩包最容易被误当成“跑通就完事”的课程作业，实际拆开后，真正决定效果的是数据窗口、坐标系和评估口径，而不是模型层数。它解决的是从历史轨迹推未来若干帧的完整闭环：原始坐标清洗、滑动窗口采样、序列模型训练、ADE/FDE 评估、可视化，再到推理软件/插件入口。适合手里有轨迹表、想快速搭一个可复现基线的人；也适合已经跑过模型、但卡在归一化泄漏和上线接口上的熟手。先别急着改网络，按数据形状、scaler、模型输出、反归一化、指标这五步跑通最小链路，后面换模型才有对照。

2. 轨迹数据预处理：把原始坐标压成滑动窗口的四个参数

2.1 先约定字段、坐标系与张量形状

轨迹预测代码包翻车，十有八九不是模型结构，而是字段顺序和坐标系没对齐。原始表里常见字段是object_id、frame_id、timestamp、x、y、scene_id、valid，但有的包把x/y写成lon/lat，有的把像素坐标和米制坐标混在一起。进模型前我一般统一成局部米制坐标：以场景起点或场景中心做平移，保留同一场景内的相对位置，不直接把经纬度喂给序列模型。输入张量固定为[B, T_obs, 2]，输出为[B, T_pred, 2]，另带[B, T_obs]的有效性 mask；如果后续要包成软件/插件，推理入口的字段名必须和训练脚本完全一致，不能靠“到时候再映射”。

先跑一个字段检查脚本，确认排序、重复帧和缺失比例，再动转换脚本。下面这段只读原始表，不生成样本，目的是把黑匣子打开。

import pandas as pd # 读取原始轨迹表，列名按实际文件调整 df = pd.read_csv("data/raw.csv") # 必要字段检查，缺一个就先补字段映射 required = ["object_id", "frame_id", "timestamp", "x", "y", "scene_id"] missing = [c for c in required if c not in df.columns] assert not missing, f"缺少字段: {missing}" # 同一对象按时间排序，检查重复帧和坐标缺失 df = df.sort_values(["scene_id", "object_id", "timestamp"]).reset_index(drop=True) dup = df.duplicated(["scene_id", "object_id", "frame_id"]).sum() na_ratio = df[["x", "y"]].isna().mean().to_dict() print("重复帧数量:", dup) print("坐标缺失比例:", na_ratio) print(df[["scene_id", "object_id", "frame_id", "x", "y"]].head())

这段检查的逻辑很直接：先保证字段齐全，再按场景、对象、时间排序，重复帧和缺失比例决定后面是丢弃、插值还是重采样。required里的字段名按压缩包内实际 CSV 替换，但语义不要改；scene_id必须参与排序和切分，否则同一场景的轨迹会被随机拆到训练集和验证集，指标会虚高。

2.2 用转换脚本生成滑动窗口样本并固化 scaler

轨迹预测的样本不是“一行一个未来点”，而是“一段历史 + 一段未来”。我一般用obs_len=8、pred_len=12、stride=1起步，按对象切滑动窗口，窗口内有效点比例低于 0.8 就丢弃。归一化只使用训练集的均值和方差，验证集、测试集和推理插件都复用同一份 scaler；这一步如果偷懒对全量数据fit，后面评估会好看得离谱，上线就露出原形。

下面脚本把原始表转成npz，同时保存scaler.json和meta.json。代码里按场景切分，避免同一轨迹泄漏。

import json import numpy as np import pandas as pd OBS_LEN, PRED_LEN, STRIDE = 8, 12, 1 MIN_VALID_RATIO = 0.8 def build_windows(group): group = group.sort_values("timestamp") xy = group[["x", "y"]].to_numpy(dtype=np.float32) valid = group["valid"].to_numpy(dtype=np.float32) if "valid" in group else np.ones(len(group), dtype=np.float32) samples = [] for start in range(0, len(group) - OBS_LEN - PRED_LEN + 1, STRIDE): obs = xy[start:start + OBS_LEN] pred = xy[start + OBS_LEN:start + OBS_LEN + PRED_LEN] obs_mask = valid[start:start + OBS_LEN] if obs_mask.mean() < MIN_VALID_RATIO: continue # 以最后一个观测点为原点做相对位移，减少绝对坐标漂移 origin = obs[-1:].copy() samples.append((obs - origin, pred - origin, obs_mask)) return samples df = pd.read_csv("data/raw.csv").sort_values(["scene_id", "timestamp"]) train_scenes = sorted(df["scene_id"].unique())[: int(df["scene_id"].nunique() * 0.7)] train_df = df[df["scene_id"].isin(train_scenes)] all_samples = [] for _, g in train_df.groupby(["scene_id", "object_id"]): all_samples.extend(build_windows(g)) obs_arr = np.stack([s[0] for s in all_samples]) pred_arr = np.stack([s[1] for s in all_samples]) mask_arr = np.stack([s[2] for s in all_samples]) # 只用训练集统计量，保存给验证、测试和插件复用 mean = obs_arr.reshape(-1, 2).mean(axis=0) std = obs_arr.reshape(-1, 2).std(axis=0) + 1e-6 np.savez("data/processed/train.npz", obs=(obs_arr - mean) / std, pred=(pred_arr - mean) / std, mask=mask_arr) with open("data/processed/scaler.json", "w") as f: json.dump({"mean": mean.tolist(), "std": std.tolist()}, f) with open("data/processed/meta.json", "w") as f: json.dump({"obs_len": OBS_LEN, "pred_len": PRED_LEN, "stride": STRIDE}, f)

这里的参数要按采集频率和任务时限调整：obs_len决定模型能看到多长历史，pred_len决定要推多远，stride控制样本重叠度。origin取最后一个观测点做相对平移，是为了让模型学“接下来往哪走”，而不是记绝对坐标；mask保留无效点，训练时不要直接当零坐标用。mean/std必须写入文件，推理插件加载模型时一起加载，否则脚本和插件会出现同一段输入两个结果。

2.3 四个参数怎么定：窗口、步长、归一化和缺失值

窗口长度不是越长越好。历史太短，模型看不到转向意图；历史太长，早期帧和未来帧相关性弱，还会放大噪声。我一般从 8 帧历史、12 帧未来起步，采样频率固定后，先看匀速基线能不能跑出合理误差，再决定是否加长。步长stride在训练集可以取 1 增加样本，验证集不要重叠采样，否则指标会被相似窗口拉平。

缺失值处理要分场景：短缺口可以线性插值，长缺口宁可丢窗口，不要用均值硬填。归一化统计量只来自训练集，验证和测试按同一份mean/std变换；如果包里的脚本对全量数据fit，先改掉再谈模型。自检时随机抽三个窗口，做一次反归一化，确认预测点和真实点还在同一坐标系里，这一步比看 loss 更能提前发现翻车。

提示：转换完成后先打印obs_arr.shape、pred_arr.shape和mask_arr.mean()，形状对不上时不要继续训练，后面所有指标都会错位。

3. 模型训练与损失函数：GRU 基线怎么接进完整代码

3.1 为什么先用编码器-解码器基线

轨迹预测的模型选型很容易一上来就追新，结果连数据管线都没跑顺。我的习惯是先接一个 GRU 编码器-解码器：编码器吃历史轨迹，解码器逐帧输出未来位移，输入输出形状和后面的 Transformer、时序卷积都能对齐。它训练快、显存低、可解释性强，适合当完整代码包的第一条基线。模型预测相对最后一个观测点的位移，而不是绝对坐标，这样换场景时漂移小；解码器输出[B, T_pred, 2]，损失只在有效未来点上计算。

import torch import torch.nn as nn class GRUTrajectoryBaseline(nn.Module): def __init__(self, hidden_dim=64, pred_len=12): super().__init__() self.pred_len = pred_len self.encoder = nn.GRU(input_size=2, hidden_size=hidden_dim, batch_first=True) self.decoder = nn.GRU(input_size=2, hidden_size=hidden_dim, batch_first=True) self.head = nn.Linear(hidden_dim, 2) def forward(self, obs, pred_gt=None): # obs: [B, T_obs, 2]，已经做过相对平移和归一化 _, hidden = self.encoder(obs) dec_input = obs[:, -1:, :] # 用最后一帧启动解码 outputs = [] for _ in range(self.pred_len): out, hidden = self.decoder(dec_input, hidden) step = self.head(out) # [B, 1, 2] outputs.append(step) dec_input = step return torch.cat(outputs, dim=1)

这段模型只做最小闭环：编码器提取历史状态，解码器逐步生成未来位移，head把隐藏状态映射回二维坐标。hidden_dim从 64 或 128 起步，pred_len必须和预处理保存的meta.json一致；推理时不要传pred_gt，训练时也不要用真实未来帧喂给解码器，否则验证阶段会高估效果。

3.2 训练循环、端点加权和模型保存

轨迹预测的平均损失容易掩盖终点漂移，我一般在主损失外加一个端点加权项，让最后一帧的误差多承担一些权重。训练循环里保留梯度裁剪、验证集反归一化评估和最佳模型保存；保存的不只是state_dict，还要带config和scaler，否则推理插件加载后还得靠人脑记参数。

import json import numpy as np import torch from torch.utils.data import DataLoader, TensorDataset def train_one_epoch(model, loader, optimizer, device, endpoint_weight=0.5): model.train() total = 0.0 for obs, pred, mask in loader: obs, pred = obs.to(device), pred.to(device) optimizer.zero_grad() out = model(obs) loss_all = torch.nn.functional.smooth_l1_loss(out, pred, reduction="none") loss = loss_all.mean() # 最后一帧误差加权，缓解终点漂移 loss = loss + endpoint_weight * loss_all[:, -1].mean() loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=5.0) optimizer.step() total += loss.item() * obs.size(0) return total / len(loader.dataset) # 数据加载和训练参数集中在一处，方便插件复用 cfg = {"hidden_dim": 64, "batch_size": 128, "lr": 1e-3, "epochs": 50, "endpoint_weight": 0.5} data = np.load("data/processed/train.npz") loader = DataLoader(TensorDataset(torch.tensor(data["obs"]), torch.tensor(data["pred"]), torch.tensor(data["mask"])), batch_size=cfg["batch_size"], shuffle=True) model = GRUTrajectoryBaseline(hidden_dim=cfg["hidden_dim"], pred_len=data["pred"].shape[1]) optimizer = torch.optim.Adam(model.parameters(), lr=cfg["lr"]) for epoch in range(cfg["epochs"]): loss = train_one_epoch(model, loader, optimizer, "cpu", cfg["endpoint_weight"]) print(epoch, round(loss, 6)) torch.save({"state_dict": model.state_dict(), "cfg": cfg}, "checkpoints/gru_baseline.pt")

endpoint_weight控制终点误差的权重，太大容易让中间帧欠拟合，太小则终点漂移明显，0.3 到 1.0 之间先扫一轮。clip_grad_norm_是序列模型的后悔药，梯度爆炸时先看它有没有生效。保存字典里带cfg，是为了让推理软件/插件加载时不用猜hidden_dim和pred_len；如果包里只保存了裸state_dict，建议补一个config.json，后面排错会省很多时间。

3.3 超参和显存不够时怎么改

显存不够时优先降batch_size和hidden_dim，不要先砍pred_len，因为未来长度是业务契约，训练和评估必须一致。学习率从1e-3起步，验证损失震荡就降到5e-4或加 warmup；过拟合表现为训练损失继续降、验证 ADE 不降，先加 dropout 或减少历史窗口，再考虑换模型。下面这张表是我在轨迹预测基线里常用的调参顺序。

训练脚本跑通后，先用匀速基线对比一次：拿最后一个观测点的速度外推未来，算同一个 ADE/FDE。如果 GRU 基线连匀速都打不过，问题通常不在模型容量，而在坐标、窗口或评估口径。把这条对照写进日志，后面换更复杂的模型才有意义。

4. 评估与可视化：ADE/FDE 之外的三个验收指标

4.1 指标口径先统一，再谈模型好坏

ADE 是未来各帧位移误差的平均，FDE 是最后一帧误差，两者都必须在反归一化之后算，单位要和业务坐标一致。很多代码包在归一化空间里算指标，数值看着很小，但没有物理意义；还有的把无效帧也平均进去，缺失越多指标越好看。我一般再补三个验收项：Miss Rate（终点误差超过阈值的样本比例）、终点误差 P90、速度/加速度越界率。前者看极端失败，后两者看轨迹是否物理合理。

阈值不要拍脑袋，按场景的采样间隔和对象类型定；同一个包里训练和评估的阈值必须来自同一份配置。评估集按场景切分，不能把同一对象的相邻窗口拆到两边，否则指标会虚高。若包内脚本只输出一个平均 ADE，先补上按场景、按对象、按预测步长的拆分，否则你无法判断模型是整体差还是某几帧崩。

4.2 评估脚本：反归一化后再累计误差

评估脚本要先把预测和真实值反归一化，再按样本累计误差，不能先算 batch 平均再平均 batch。下面的写法保留每个样本的 ADE/FDE，方便后面算 P90 和 Miss Rate。

import json import numpy as np import torch def evaluate(model, loader, scaler, device="cpu", miss_threshold=2.0): model.eval() mean = torch.tensor(scaler["mean"], dtype=torch.float32, device=device) std = torch.tensor(scaler["std"], dtype=torch.float32, device=device) ade_list, fde_list = [], [] with torch.no_grad(): for obs, pred, mask in loader: obs, pred = obs.to(device), pred.to(device) out = model(obs) # 反归一化回米制坐标，再算物理误差 out = out * std + mean pred = pred * std + mean dist = torch.norm(out - pred, dim=-1) # [B, T_pred] ade_list.extend(dist.mean(dim=1).cpu().numpy()) fde_list.extend(dist[:, -1].cpu().numpy()) ade = np.array(ade_list) fde = np.array(fde_list) return { "ADE": float(ade.mean()), "FDE": float(fde.mean()), "MR": float((fde > miss_threshold).mean()), "FDE_P90": float(np.percentile(fde, 90)), } with open("data/processed/scaler.json") as f: scaler = json.load(f) metrics = evaluate(model, loader, scaler) print(metrics)

miss_threshold要按业务容差设，单位是米或像素，取决于坐标系；FDE_P90比平均值更能暴露长尾失败。model.eval()和torch.no_grad()不是装饰，推理插件里漏掉它们，dropout 和 BN 会让同一输入两次结果不同。指标输出建议写进 JSON 日志，和模型文件、scaler、数据版本放在一起，后面复盘才有依据。

4.3 可视化和基线对比：别只看 loss 曲线

可视化不是画着好看，而是检查轨迹方向、尺度和平移有没有错。我一般抽三类样本：正常直行、转弯、终点误差最大的失败样本。每张图画历史轨迹、真实未来、模型预测和匀速基线，坐标轴保持等比例，否则转弯会被视觉拉直。若预测整体偏移一个固定向量，优先查相对原点和平移；若方向反了，查x/y顺序和坐标系旋转。

import matplotlib.pyplot as plt def plot_case(obs, pred, out, path): # obs/pred/out 均为反归一化后的 [T, 2] plt.figure(figsize=(5, 5)) plt.plot(obs[:, 0], obs[:, 1], "o-", label="history") plt.plot(pred[:, 0], pred[:, 1], "s-", label="ground truth") plt.plot(out[:, 0], out[:, 1], "x--", label="prediction") plt.axis("equal") plt.legend() plt.savefig(path, dpi=150, bbox_inches="tight") plt.close()

画图前先反归一化，axis("equal")必须加，不然横向纵向比例不同会误导判断。把匀速基线画在同一张图上，能快速看出模型是否只是在学“沿最后方向直走”。如果模型和基线几乎重合，先别加注意力，回去检查历史窗口里有没有速度变化信息；如果模型在转弯样本上明显滞后，再考虑加转向特征或换时序结构。

5. 避坑与排查：轨迹预测代码包落地时的五类翻车

5.1 五条踩坑记录：现象、原因、解决

第一条，训练 loss 很低但评估 ADE 爆炸。现象是训练日志漂亮，验证集误差大一个数量级；原因通常是对全量数据做了归一化，或者评估时忘了反归一化。解决是把scaler固化到训练集，评估脚本先乘std加mean，再算米制误差，脚本和插件共用同一份scaler.json。

第二条，验证集好得不真实。现象是验证 ADE 接近训练集，换个场景就崩；原因是按样本随机切分，同一对象的相邻窗口同时进了训练和验证。解决是按scene_id或object_id切分，验证集不重叠采样，测试集单独留场景。

第三条，推理插件和脚本结果对不上。现象是同一段轨迹在脚本里正常，包成软件/插件后偏移或抖动；原因多是字段顺序、eval()、no_grad()、相对原点不一致。解决是把预处理、模型前向、反归一化封成一个类，脚本和插件都调这个类，输入输出用同一份契约测试。

第四条，batch 推理出现 NaN 或形状错误。现象是单条推理正常，多条一起就报错；原因是轨迹长度不一、空轨迹、mask 没对齐。解决是推理前按有效长度排序或 padding，空输入直接返回明确错误码，不要用零坐标硬凑。

第五条，可视化和指标错位。现象是图上看着还行，指标却很差；原因是frame_id没排序、重复帧没去重、时间戳和帧号混用。解决是转换阶段固定排序键，重复帧按规则合并，评估和画图都从同一份npz读数据。

注意：这五类问题里，前三类会直接让结论失真，后两类会让排错方向跑偏。先修数据契约，再调模型。

5.2 排查顺序与日志字段

排查不要从模型结构开始，我一般按“形状 → 坐标系 → scaler → 模型状态 → 反归一化 → 指标”顺序走。每一步打一条结构化日志，字段包括scene_id、object_id、obs_shape、pred_shape、scaler_mean、scaler_std、model_mode、ade_raw、ade_denorm。这样脚本和插件对不上时，能一眼看出是哪一步分叉。

import json import logging logging.basicConfig(level=logging.INFO, format="%(message)s") def log_once(tag, obs, pred, scaler, model): record = { "tag": tag, "obs_shape": list(obs.shape), "pred_shape": list(pred.shape), "scaler_mean": scaler["mean"], "scaler_std": scaler["std"], "model_training": model.training, "has_nan": bool(obs.isnan().any() or pred.isnan().any()), } logging.info(json.dumps(record, ensure_ascii=False))

model_training能直接暴露插件里漏掉eval()的问题，has_nan放在预处理之后、模型之前，能区分是输入坏还是前向坏。日志不要只打 loss，形状、scaler、模式这三个字段最能定位分叉。每次改预处理或换模型，先用固定小样本跑一遍脚本和插件的一致性对比，误差超过1e-5就停下来查，不要带病上线。

6. 插件化推理与验收：把预测器包成可调用组件的具体技巧

软件/插件形态落地时，我最看重三件事：输入契约、预处理与模型绑定、失败可观测。输入契约固定为obs: [T,2]、mask: [T]、meta，输出为pred: [T_pred,2]和status；预处理、模型前向、反归一化封进同一个类，脚本和插件都调它，避免两套代码各走各的。下面这个最小类可以直接嵌到服务入口里，加载时把scaler和config一起读进来。

import json import torch class TrajectoryPredictor: def __init__(self, ckpt_path, scaler_path, device="cpu"): ckpt = torch.load(ckpt_path, map_location=device) self.model = GRUTrajectoryBaseline(**ckpt["cfg"]) self.model.load_state_dict(ckpt["state_dict"]) self.model.eval() self.device = device with open(scaler_path) as f: s = json.load(f) self.mean = torch.tensor(s["mean"], dtype=torch.float32, device=device) self.std = torch.tensor(s["std"], dtype=torch.float32, device=device) @torch.no_grad() def predict(self, obs): x = torch.tensor(obs, dtype=torch.float32, device=self.device).unsqueeze(0) if torch.isnan(x).any(): return None, "INPUT_NAN" out = self.model(x)[0] out = out * self.std + self.mean # 回到米制坐标 return out.cpu().numpy(), "OK"

ckpt["cfg"]必须包含模型构造参数，scaler_path必须和训练时同一份文件；@torch.no_grad()和eval()是插件稳定性的底线。验收时至少跑空输入、NaN 输入、单条、批量、脚本与插件同输入对比五项，终点最大差值超过1e-5就回查字段顺序和反归一化。从那以后我每次把轨迹预测包成软件/插件，都强制走一遍脚本与插件同输入对比，差一点就停工。希望帮到你。

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