企业官网建设流程全解析

限价订单簿（Limit Order Book，简称 LOB）建模是高频量化方向里少有的、既有公开基准又能用单卡跑通的方向。DeepLOB 是这条线上最经典的起点：它把卷积核当成跨档位的特征提取器，用 Inception 结构做多尺度感受野，再交给两层 LSTM 建模时间依赖，最终输出未来一小段时间内中间价的涨、跌、不变三分类。整个模型参数量很小，输入是固定形状的张量，训练和推理都不挑硬件，这让它非常适合作为 LOB 建模的入门基线。

Transformer 方案是另一条路线。把每个时间步的 LOB 快照看成一个 token，用自注意力代替循环结构，好处是能直接建模长程依赖，也能通过注意力权重看到模型在关注哪几档盘口、哪一个时刻。代价是数据量要求更高，对数据划分和正则化更敏感，稍不注意就会在验证集上表现得很好、到测试集上却完全失效。

这篇文章会从零走一遍完整流程：环境准备、FI-2010 数据加载、DeepLOB 复现、Transformer 实现、训练与评估、推理接口封装、性能观察和踩坑排查。目标读者是已经会 PyTorch、想进入高频时序建模、但还没把整条链路跑通的工程同学。所有指标都以你自己的复现结果为准，文中给出的公开论文区间只作为对照参考，模型精度也不等同于可交易收益。

1. DeepLOB 与 Transformer 的 LOB 建模核心能力速览

先把关键规格摆出来。LOB 建模和图像、语音那类模型最大的区别是，它对硬件的需求其实很低，真正的门槛在数据和划分方法，而不是显卡。

把这张表看完，基本能判断自己要不要继续。如果你的目的是学习时序建模和高频数据处理的完整链路，这套东西性价比很高；如果你指望训完就能拿去赚钱，那大概率会失望，原因后面会展开。

2. LOB 建模的适用场景与使用边界

2.1 这个方向适合谁

第一类是刚接触量化时序的算法同学。LOB 数据的结构非常规整，特征维度固定，标签构造规则明确，比起处理非结构化文本或者多模态数据，它更容易验证「我的代码到底对不对」。跑通 DeepLOB 之后，你对序列建模、类别不平衡、时间序列划分这些概念的直觉会强很多。

第二类是想做基线对比的研究者。DeepLOB 在公开论文里的指标区间是清楚的，你复现出来的结果如果和论文差了十几个点，那一定是流程里有问题，而不是模型不行。这种可对照性在深度学习中并不常见。

第三类是工程侧想把模型封装成服务的开发者。这个模型的推理接口非常轻，输入一段 LOB 快照，输出三个类别的概率，包成 HTTP 接口或者消息队列消费者都很简单，适合拿来练手模型服务化。

2.2 不适合什么场景

最典型的一个误区是把分类精度当成策略收益。三分类模型在 FI-2010 上做到接近 0.8 的加权 F1，听起来不错，但这里面有几个陷阱：标签是按固定阈值划分的，中间价变动小于阈值的样本被归为「不变」，而实盘里这类样本恰恰是最难处理的；分类精度不区分预测的置信度分布，模型可能在高置信度时全错；更关键的是，论文的评估完全不含交易成本、滑点、排队延迟和撤单率，这些在真实环境中足以把任何微弱优势抹平。

第二个误区是忽略数据窥探。LOB 是强非平稳序列，如果你用随机打乱的方式划分训练集和测试集，同一段时间的相邻样本会同时出现在两边，模型等于在背答案。这种情况下验证集指标可以高到离谱，但没有任何参考价值。正确做法必须按时间切分，而且标准化参数只能用训练集拟合。

第三个边界是数据合规。不同交易所的 Level-2 数据授权范围差别很大，有些只允许自用不允许再分发，有些对存储时长有限制。使用任何非公开数据前，先确认自己有没有相应的授权，别等到做出东西来了才发现数据来源有问题。

2.3 法律与合规提醒

这一节必须单独说清楚。本文所有内容仅用于技术研究与教学演示，不构成任何投资建议，也不承诺任何收益。基于 LOB 的预测模型在真实市场中面临监管、风控、资金容量等多重约束，任何将其用于实盘的行为都需要自行承担全部风险，并确保符合所在地的法律法规与交易所规则。涉及真实账户和历史成交数据时，务必确认数据获取的合法性和隐私保护要求。

3. 环境准备与目录结构

3.1 依赖清单

这套东西的依赖非常干净，核心就是 PyTorch、NumPy、scikit-learn 三件套。版本上不需要追新，PyTorch 1.13 以上的稳定版都能跑，CUDA 版本跟着显卡驱动走就行。如果你只是做推理不做训练，CPU 版本的 PyTorch 完全够用。

# 建议用独立环境，避免和其他项目冲突 conda create -n lob python=3.10 -y conda activate lob # 训练用（按自己的 CUDA 版本选择对应索引） pip install torch torchvision --index-url https://download.pytorch.org/whl/cu121 # 通用依赖 pip install numpy pandas scikit-learn matplotlib tqdm tensorboard pip install fastapi uvicorn # 需要封装推理接口时再装

如果显卡比较老或者干脆没有独显，把第一条换成 CPU 版本即可：

pip install torch torchvision --index-url https://download.pytorch.org/whl/cpu

3.2 目录规划

LOB 项目最容易乱的地方是数据、模型权重、日志混在一起。建议一开始就把目录定好，后面批量实验会省很多事。

lob_project/ ├── configs/ │ └── deeplob.yaml # 实验配置 ├── data/ │ ├── raw/ # 原始 FI-2010 文件，只读 │ ├── processed/ # 切分好的 npy │ └── scaler.pkl # 训练集拟合的标准化参数 ├── models/ │ ├── deeplob.py │ └── lob_transformer.py ├── datasets/ │ └── fi2010.py ├── train.py ├── evaluate.py ├── serve.py # FastAPI 推理服务 ├── checkpoints/ └── logs/

把raw目录设成只读是个好习惯，可以避免预处理脚本不小心覆盖原始数据。scaler.pkl单独存盘也很重要，线上推理时必须用训练时同一份标准化参数，否则输入分布会漂移。

3.3 配置文件示例

把超参写在配置文件里，比在代码里改常量要靠谱得多。YAML 就够了，不需要上 Hydra。

# configs/deeplob.yaml data: feature_dim: 40 window: 100 num_classes: 3 train_days: [1, 2, 3, 4, 5, 6, 7] val_days: [8, 9] test_days: [10] model: name: deeplob lstm_hidden: 64 dropout: 0.0 train: batch_size: 64 epochs: 60 lr: 0.001 weight_decay: 0.0001 patience: 10 runtime: device: cuda seed: 42 num_workers: 4

配置文件里把天数划分写死，是为了防止后面有人随手改了划分却不自知。这个细节在复现论文时特别关键。

4. 数据准备与特征工程

4.1 FI-2010 数据集的基本情况

FI-2010 是 LOB 方向最常被引用的公开基准。它包含 5 只股票在连续 10 个交易日内的限价订单簿快照，每只股票的盘口取 10 档，每档记录买价、买量、卖价、卖量，经过归一化后正好是 40 维特征。样本按窗口长度切分，常见设定是 100 个时间步，也就是每个样本覆盖 100 个快照的盘口演化过程。

标签的构造方式是基于未来一段时间的中间价变化：如果未来 k 个 tick 的中间价均值相对当前上涨超过阈值，标为「涨」；下跌超过阈值标为「跌」；其余标为「不变」。这个阈值设计直接决定了三类的样本分布，也决定了模型任务的难度。需要说明的是，这个数据集需要通过官方渠道提交申请后获取，不随公开代码仓库分发，所以第一步是先拿到数据。

划分方式在数据集说明里是给定的：前 7 天作为训练集，第 8 到 9 天作为验证集，第 10 天作为测试集。千万不要自己重新划分，否则和公开论文的指标就没法对比了。

4.2 加载与检查数据

拿到数据后第一件事不是急着训练，而是先做一遍统计检查。很多复现失败的原因，是数据本身在加载阶段就错位了。

import numpy as np def load_fi2010(prefix: str): """加载 FI-2010 的 npy 文件，返回特征和标签""" x = np.load(f"data/raw/{prefix}_X.npy") y = np.load(f"data/raw/{prefix}_Y.npy") print(f"[{prefix}] X shape={x.shape}, dtype={x.dtype}") print(f"[{prefix}] Y shape={y.shape}, dtype={y.dtype}") print(f"[{prefix}] 类别分布={np.bincount(y.argmax(axis=1))}") print(f"[{prefix}] NaN 数量={np.isnan(x).sum()}, Inf 数量={np.isinf(x).sum()}") return x, y

检查项至少有四个：样本数是否与预期一致、类别分布是否严重失衡、是否存在 NaN 或 Inf、特征取值范围是否在合理区间。如果输出里出现 Inf，基本可以断定原始数据里有价格为 0 或者除零的地方，需要单独处理。

4.3 标准化只能用训练集拟合

这是最容易出错的一步。正确做法是先在训练集上统计每个特征的均值和标准差，保存下来，然后训练集、验证集、测试集统一用这组参数做变换。

import numpy as np import pickle def fit_scaler(x_train: np.ndarray, eps: float = 1e-8): # x_train: (N, T, F) -> 按特征维度统计 mean = x_train.mean(axis=(0, 1), keepdims=True) std = x_train.std(axis=(0, 1), keepdims=True) + eps return {"mean": mean, "std": std} def transform(x: np.ndarray, scaler: dict) -> np.ndarray: return (x - scaler["mean"]) / scaler["std"] x_train, y_train = load_fi2010("train") scaler = fit_scaler(x_train) with open("data/scaler.pkl", "wb") as f: pickle.dump(scaler, f)

注意统计的轴。你统计的是「每个特征在全部时间和全部样本上的分布」，所以 axis 应该是(0, 1)，保留特征维度。如果写成axis=0，得到的就是每个时间步的均值，形状对不上，后面的广播会悄悄产生错误结果，而且不会报错，这种 bug 最难查。

4.4 构造 PyTorch Dataset

DeepLOB 期望的输入是四维张量，所以 Dataset 里要多做一次 unsqueeze，把通道维度补上。

import torch from torch.utils.data import Dataset class LOBDataset(Dataset): def __init__(self, x: np.ndarray, y: np.ndarray, scaler: dict): # x: (N, T, F), y: (N, 3) one-hot x = transform(x, scaler).astype(np.float32) self.x = torch.from_numpy(x).unsqueeze(1) # (N, 1, T, F) self.y = torch.from_numpy(y.argmax(axis=1)).long() def __len__(self): return self.x.shape[0] def __getitem__(self, idx): return self.x[idx], self.y[idx]

unsqueeze(1)这一步对应卷积层的通道维。如果你的模型是 Transformer，就不需要这一维，可以直接用(N, T, F)。两种模型的输入约定不一样，切换的时候记得改 Dataset，别指望模型自己兼容。

4.5 特征增强的常见做法

除了 40 维原始盘口特征，工程上还常做一些额外构造：中间价与微价格（microprice）、买卖量不平衡度、各档位的加权累计深度、价差（spread）、订单流不平衡（OFI）。这些派生特征能把模型的上限往上抬一截，尤其在真实数据上比在 FI-2010 这种已经归一化过的数据上效果更明显。

不过要提醒一句：第一步复现论文时不要加特征，就老老实实用 40 维输入。等基线跑通、指标对上了，再去做特征工程。否则一旦指标不对，你没办法判断是模型实现错了还是特征出了问题。

5. DeepLOB 模型复现与训练

5.1 结构拆解

DeepLOB 的核心思路可以概括成三句话：用二维卷积在「时间 × 档位」平面上提取局部模式，用 Inception 结构并行多个卷积核尺寸来覆盖不同的感受野，用 LSTM 把提取到的特征在时间维度上串起来。最后接几层全连接做三分类。

它和普通 CNN 处理图像的区别在于，卷积核的形状是精心设计的。第一层用(1, 2)这种形状，意味着只在档位维度上滑动，不跨时间步，目的是把相邻档位的买卖信息做配对；后面几层才用(4, 1)这种形状，开始在时间维度上聚合。这个设计背后有明确的市场直觉，不是随手堆的。

5.2 可运行的实现

下面这份实现是常见的等价复现版本，能直接跑通，结构和原论文的思路一致。参数量会与原始代码有差异，因为通道数和 Inception 分支的具体配置在不同复现版本中略有不同，但不影响作为基线使用。

import torch import torch.nn as nn class DeepLOB(nn.Module): def __init__(self, n_features: int = 40, n_classes: int = 3, hidden: int = 64): super().__init__() # 阶段一：跨档位卷积，压缩宽度维度 self.cnn = nn.Sequential( nn.Conv2d(1, 16, kernel_size=(1, 2), stride=(1, 2)), # (B,16,T,20) nn.LeakyReLU(0.1), nn.Conv2d(16, 32, kernel_size=(4, 1)), # (B,32,T-3,20) nn.LeakyReLU(0.1), nn.Conv2d(32, 32, kernel_size=(4, 1)), # (B,32,T-6,20) nn.LeakyReLU(0.1), ) # 阶段二：时间维度建模 self.lstm = nn.LSTM( input_size=32 * (n_features // 2), hidden_size=hidden, num_layers=2, batch_first=True, ) self.head = nn.Sequential( nn.Linear(hidden, 64), nn.ReLU(), nn.Linear(64, n_classes), ) def forward(self, x): # x: (B, 1, T, F) feat = self.cnn(x) # (B, 32, T', 20) b, c, t, w = feat.shape feat = feat.permute(0, 2, 1, 3).reshape(b, t, c * w) out, _ = self.lstm(feat) # (B, T', hidden) return self.head(out[:, -1, :]) # 取最后时间步

这里取out[:, -1, :]而不是对时间维度做平均，是因为 LSTM 的最后一个隐状态已经聚合了整段序列的信息，平均反而会稀释掉最近时刻的信号。如果你在自己的数据上发现平均池化更好，那说明序列里的长程信息比短期信息更重要，这本身也是一个值得记录的观察。

5.3 Inception 模块的可选增强

原论文里的 Inception 模块有多个并行分支，用1×1、3×1、5×1等不同尺寸的卷积核并行提取，再拼接起来。如果你的显存和时间允许，加上它会带来一些提升；如果只是想要一个能跑通的基线，上面那份简化实现已经足够。

class InceptionBlock(nn.Module): def __init__(self, in_ch: int = 32, out_ch: int = 32): super().__init__() mid = out_ch // 4 self.b1 = nn.Sequential(nn.Conv2d(in_ch, mid, (1, 1)), nn.LeakyReLU(0.1)) self.b2 = nn.Sequential(nn.Conv2d(in_ch, mid, (3, 1), padding=(1, 0)), nn.LeakyReLU(0.1)) self.b3 = nn.Sequential(nn.Conv2d(in_ch, mid, (5, 1), padding=(2, 0)), nn.LeakyReLU(0.1)) self.b4 = nn.Sequential(nn.MaxPool2d((3, 1), stride=(1, 1), padding=(1, 0)), nn.Conv2d(in_ch, mid, (1, 1)), nn.LeakyReLU(0.1)) def forward(self, x): return torch.cat([self.b1(x), self.b2(x), self.b3(x), self.b4(x)], dim=1)

拼接的时候要注意dim=1，也就是通道维。如果写成dim=-1，拼的是宽度维，通道数不变但特征含义全乱了，而且形状可能还是合法的，不会报错。

5.4 训练循环

训练逻辑本身不复杂，但有两点必须做对：一是用交叉熵损失配合 L2 正则，二是做早停并且按验证集加权 F1 挑模型，而不是按 loss。

import torch import torch.nn as nn from sklearn.metrics import f1_score def train_one_epoch(model, loader, optimizer, criterion, device): model.train() total_loss = 0.0 for xb, yb in loader: xb, yb = xb.to(device), yb.to(device) optimizer.zero_grad() logits = model(xb) loss = criterion(logits, yb) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() total_loss += loss.item() * xb.size(0) return total_loss / len(loader.dataset) @torch.no_grad() def evaluate(model, loader, device): model.eval() preds, gts = [], [] for xb, yb in loader: xb = xb.to(device) logits = model(xb) preds.append(logits.argmax(dim=1).cpu()) gts.append(yb) preds = torch.cat(preds).numpy() gts = torch.cat(gts).numpy() return f1_score(gts, preds, average="weighted"), preds, gts

梯度裁剪这一行不是必须的，但 LSTM 在长序列上偶尔会出现梯度爆炸，加上它几乎没成本，建议保留。早停的 patience 设 10 左右比较合适，太大容易过拟合，太小又会在验证集抖动时提前终止。

5.5 一个完整的训练入口

把上面这些拼起来，就是一个可以跑的 train.py 骨架：

def main(): cfg = load_config("configs/deeplob.yaml") device = torch.device(cfg["runtime"]["device"]) x_tr, y_tr = load_fi2010("train") x_va, y_va = load_fi2010("val") scaler = fit_scaler(x_tr) tr_ds = LOBDataset(x_tr, y_tr, scaler) va_ds = LOBDataset(x_va, y_va, scaler) tr_loader = DataLoader(tr_ds, batch_size=cfg["train"]["batch_size"], shuffle=True) va_loader = DataLoader(va_ds, batch_size=256, shuffle=False) model = DeepLOB().to(device) optimizer = torch.optim.Adam( model.parameters(), lr=cfg["train"]["lr"], weight_decay=cfg["train"]["weight_decay"], ) criterion = nn.CrossEntropyLoss() best_f1, best_state, wait = 0.0, None, 0 for epoch in range(cfg["train"]["epochs"]): loss = train_one_epoch(model, tr_loader, optimizer, criterion, device) f1, _, _ = evaluate(model, va_loader, device) print(f"epoch={epoch} loss={loss:.4f} val_f1={f1:.4f}") if f1 > best_f1: best_f1, wait = f1, 0 best_state = {k: v.clone() for k, v in model.state_dict().items()} else: wait += 1 if wait >= cfg["train"]["patience"]: print("early stop") break torch.save(best_state, "checkpoints/deeplob_best.pt") print(f"best val f1 = {best_f1:.4f}")

注意shuffle=True只用在训练集上。验证集和测试集绝对不能打乱，虽然对指标计算本身没影响，但如果你后面要做时序相关的分析（比如画出预测随时间的分布），顺序就是必需的。

6. Transformer 方案实现与对比

6.1 为什么要在 LOB 上用 Transformer

LSTM 的问题在于它是串行计算的，长序列上的信息传递会衰减，而且训练时没法并行。Transformer 用自注意力一步到位地把任意两个时间步连起来，理论上能捕捉到更长时间跨度的盘口形态，比如「大单挂出后经过若干 tick 才被吃掉」这类跨期的模式。

另一个实际的好处是可解释性。注意力权重矩阵可以直接可视化，你能看到模型在预测某个时刻时，主要参考了前面哪几个时刻的盘口状态。这个在调试阶段非常有用，比 LSTM 的隐状态好理解得多。

但代价也要说清楚。Transformer 没有卷积那样的局部性先验，在数据量有限的情况下更容易过拟合。LOB 数据虽然样本数看起来很大，但有效信息高度集中在少部分时段，实际可学习的信息量远小于样本数。所以在这类任务上，Transformer 往往需要更强的正则化才能和 DeepLOB 打平。

6.2 实现代码

import math import torch import torch.nn as nn class PositionalEncoding(nn.Module): def __init__(self, d_model: int, max_len: int = 512, dropout: float = 0.1): super().__init__() self.dropout = nn.Dropout(dropout) pe = torch.zeros(max_len, d_model) pos = torch.arange(0, max_len).unsqueeze(1).float() div = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model)) pe[:, 0::2] = torch.sin(pos * div) pe[:, 1::2] = torch.cos(pos * div) self.register_buffer("pe", pe.unsqueeze(0)) def forward(self, x): return self.dropout(x + self.pe[:, :x.size(1)]) class LOBTransformer(nn.Module): def __init__(self, n_features: int = 40, d_model: int = 64, nhead: int = 4, num_layers: int = 2, n_classes: int = 3, dropout: float = 0.1): super().__init__() self.proj = nn.Linear(n_features, d_model) self.pos = PositionalEncoding(d_model, dropout=dropout) layer = nn.TransformerEncoderLayer( d_model=d_model, nhead=nhead, dim_feedforward=d_model * 4, dropout=dropout, batch_first=True, norm_first=True, ) self.encoder = nn.TransformerEncoder(layer, num_layers=num_layers) self.norm = nn.LayerNorm(d_model) self.head = nn.Linear(d_model, n_classes) def forward(self, x, return_attn: bool = False): # x: (B, T, F) h = self.pos(self.proj(x)) h = self.encoder(h) h = self.norm(h[:, -1, :]) logits = self.head(h) if return_attn: return logits, h return logits

这里用了norm_first=True，也就是 Pre-LN 结构。在中小规模数据上，Pre-LN 比 Post-LN 更稳定，不容易在训练初期发散。如果你发现 loss 一上来就变成 NaN，先检查这个参数。

6.3 因果掩码的必要性

这一条极其重要。LOB 预测是从历史推未来，模型在任何时候都不能看到当前时刻之后的信息。标准的 Transformer Encoder 是双向的，每个位置都能看到全序列，用在这里就是严重的信息泄露。

解决办法是加因果掩码：

def build_causal_mask(seq_len: int, device): # 上三角为 -inf，表示不可见 mask = torch.triu(torch.ones(seq_len, seq_len, device=device), diagonal=1) return mask.masked_fill(mask == 1, float("-inf")) # 使用时 mask = build_causal_mask(T, x.device) h = self.encoder(h, mask=mask)

如果忘了这一步，你会发现 Transformer 的验证指标高得离谱，甚至能到 0.95 以上，然后兴冲冲地拿去测试集上跑，结果直接掉到 0.5 左右。这个现象在 LOB 任务上特别典型，因为相邻时间步的盘口状态高度相关，看到未来等于直接看到答案。

6.4 两种方案的对比

从工程角度看，DeepLOB 的优势是训练快、显存小、对数据量不敏感，缺点是感受野受卷积核尺寸限制，长程依赖要靠 LSTM 硬扛。Transformer 的优势是全局建模能力强、可解释性好，缺点是需要更多数据和更强正则化，训练时间也更长。

表格里的指标区间来自公开论文的报告，不同复现之间通常有几个点的波动，具体以你自己跑出来的结果为准。如果你的 Transformer 比 DeepLOB 低了十几个点，先别怀疑模型，去检查掩码、位置编码和正则化强度。

7. 评估验证与接口调用

7.1 评估不能只看准确率

三分类任务里，如果某一类样本占比很高，模型全预测这一类也能拿到不错的准确率。所以必须看加权 F1，最好再把混淆矩阵打出来。

from sklearn.metrics import classification_report, confusion_matrix def full_report(model, loader, device): f1, preds, gts = evaluate(model, loader, device) print(f"weighted F1 = {f1:.4f}") print(classification_report(gts, preds, target_names=["down", "stable", "up"], digits=4)) print("confusion matrix:") print(confusion_matrix(gts, preds)) return f1

重点看两个地方。第一是「不变」类的召回率，如果这一类几乎全错，说明模型在噪声上过拟合了，可以考虑加标签平滑或者调整类别权重。第二是「涨」和「跌」的对称性，如果模型严重偏向某一侧，多半是训练数据的分布问题，而不是模型本身的问题。

7.2 标签平滑与类别权重

FI-2010 的三类分布并不严格相等，具体比例以你拿到的版本统计为准。如果发现明显失衡，有两种常用处理方式。

# 方式一：类别权重 counts = torch.bincount(torch.from_numpy(y_train.argmax(axis=1))) weights = (1.0 / counts.float()) / (1.0 / counts.float()).sum() criterion = nn.CrossEntropyLoss(weight=weights) # 方式二：标签平滑 criterion = nn.CrossEntropyLoss(label_smoothing=0.1)

标签平滑在 LOB 任务上通常比类别权重更稳，因为「不变」类本身就带有很强的模糊性，硬标签等于强迫模型在噪声上给出极端置信度。用 0.05 到 0.1 之间的平滑系数比较常见，太大会让模型失去区分度。

7.3 封装推理服务

模型训完之后，最常见的需求是包成一个接口，让外部系统可以传一段 LOB 快照进来拿预测结果。用 FastAPI 写起来很短。

import pickle import numpy as np import torch from fastapi import FastAPI from pydantic import BaseModel app = FastAPI(title="LOB Predictor") with open("data/scaler.pkl", "rb") as f: SCALER = pickle.load(f) device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model = DeepLOB().to(device) model.load_state_dict(torch.load("checkpoints/deeplob_best.pt", map_location=device)) model.eval() class LOBRequest(BaseModel): snapshot: list # 形状为 [T, 40] 的嵌套列表 @app.post("/predict") def predict(req: LOBRequest): x = np.asarray(req.snapshot, dtype=np.float32) if x.ndim != 2 or x.shape[1] != 40: return {"error": f"expect shape (T, 40), got {x.shape}"} x = (x - SCALER["mean"][0]) / SCALER["std"][0] tensor = torch.from_numpy(x).unsqueeze(0).unsqueeze(0).to(device) with torch.no_grad(): prob = torch.softmax(model(tensor), dim=1)[0].cpu().numpy() labels = ["down", "stable", "up"] return { "probs": {labels[i]: float(prob[i]) for i in range(3)}, "pred": labels[int(prob.argmax())], }

注意标准化那一步用的是SCALER["mean"][0]，因为保存的时候保留了批维和通道维。如果索引写错，形状广播会静默出错，输出看起来还是三个概率，但分布完全不对。上线前一定要用训练集里的一条样本做端到端校验，确认接口输出和离线推理结果一致。

7.4 批量任务的组织

批量推理常见的场景是按天回放历史数据，产出整段的预测序列，用于研究分析。建议按「股票 + 日期」的粒度切分子任务，每个任务单独落盘，这样中途失败了可以只重跑出错的那一片。

import os import json import numpy as np import torch def batch_infer(model, x_all, scaler, out_dir, window=100, batch_size=512): os.makedirs(out_dir, exist_ok=True) device = next(model.parameters()).device model.eval() results = [] for start in range(0, len(x_all) - window + 1, batch_size): chunk = x_all[start:start + batch_size + window - 1] chunks, valid = [], 0 for i in range(len(chunk) - window + 1): if start + i >= len(x_all) - window + 1: break chunks.append(chunk[i:i + window]) valid += 1 if not chunks: break batch = np.stack(chunks) batch = (batch - scaler["mean"]) / scaler["std"] tensor = torch.from_numpy(batch.astype(np.float32)).unsqueeze(1).to(device) with torch.no_grad(): probs = torch.softmax(model(tensor), dim=1).cpu().numpy() for i in range(valid): results.append({ "index": start + i, "probs": probs[i].round(6).tolist(), }) with open(os.path.join(out_dir, "preds.jsonl"), "w") as f: for r in results: f.write(json.dumps(r) + "\n") return len(results)

写成 JSONL 而不是一个大的 JSON，是因为可以流式追加，中途崩了也不会丢掉已经算完的部分。这个模式在长时间批处理任务里很实用。

8. 资源占用与性能观察

8.1 显存占用的观察方法

这个模型的显存需求其实很低。参数量在百万以内，输入张量也不大，真正的显存消耗主要来自 batch 大小。观察方式很直接：

import torch def report_memory(device): if device.type != "cuda": print("cpu mode") return allocated = torch.cuda.memory_allocated(device) / 1024 ** 2 reserved = torch.cuda.memory_reserved(device) / 1024 ** 2 print(f"allocated={allocated:.1f} MB, reserved={reserved:.1f} MB")

在训练循环里每隔几十个 batch 打一次，就能看出峰值出现在哪里。通常峰值出现在反向传播那一瞬间，因为要保存中间激活。如果发现 reserved 远大于 allocated，说明碎片比较多，可以试着把 batch 调大一点或者重启进程。

从量级上看，batch 开到 1024 的 100×40 输入，显存占用也就在几百 MB 以内，普通消费级显卡完全没有压力。真正需要关注的反而是数据加载，如果 num_workers 设得太大，CPU 内存和句柄会成为瓶颈。

8.2 CPU 与 GPU 的差异

训练阶段 GPU 的优势很明显，因为 LSTM 的串行计算和卷积都能吃到并行度。但推理阶段就未必了：单条样本的推理在 CPU 上也就毫秒级，如果你的服务 QPS 要求不高，用 CPU 部署反而更省事，不用管驱动和 CUDA 版本兼容。

判断标准可以这样定：如果单次请求只处理一条快照，CPU 足够；如果要批量处理几千条甚至几万条，再上 GPU。别一开始就为了「看起来专业」把服务搞得特别复杂。

8.3 影响性能的几个参数

按影响从大到小排：窗口长度、batch 大小、LSTM 隐藏维度、Transformer 的层数和 d_model。窗口从 100 加到 200，计算量差不多翻倍；隐藏维度从 64 加到 128，参数量大概翻四倍；Transformer 层数从 2 加到 4，训练时间接近翻倍。

反过来，如果显存或者时间不够，优先降的是 batch 大小和窗口长度，最后才动模型宽度。因为模型宽度一旦降下来，表达能力就受限了，指标会明显变差。

还有一个容易被忽略的点：num_workers设成 0 和设成 4，训练速度能差出一截，尤其在数据预处理比较重的时候。但设太大了反而会因为进程切换变慢，一般设成 CPU 核心数的一半左右比较合适。

8.4 进程与端口管理

如果用 FastAPI 起服务，默认端口常用的有 8000 和 8080，很容易撞车。启动前先查一下：

# Linux / macOS lsof -i :8000 # Windows netstat -ano | findstr :8000

如果端口被占了，换个端口启动就行，不用去杀进程：

uvicorn serve:app --host 127.0.0.1 --port 8010

另外提醒一点，用--reload参数会启动子进程，有时候 Ctrl+C 之后子进程还在，导致端口一直被占。生产环境不要开 reload，开发时如果遇到端口占用，先确认是不是上次的残留进程。

9. 常见问题与排查方法

这一类项目遇到的问题，八成集中在数据泄露、形状不匹配和训练不稳定三个方向。下面这张表覆盖了最常见的现象。

举一个具体的排查例子。如果你发现模型在验证集上 F1 是 0.82，测试集是 0.51，这个差距基本可以断定是泄露。最常见的两个来源：一是 Dataset 里对全量数据做了标准化，二是划分的时候不小心把相邻时间段的样本分到了两边。前者改起来容易，后者需要回头检查数据切片的索引逻辑。

再举一个。如果训练了 30 个 epoch，loss 一直在 1.09 附近不动，也就是ln(3)的水平，说明模型什么都没学到，输出接近均匀分布。这种情况下先别调模型，去看输入：打印一个 batch 的x.mean()和x.std()，如果均值是几十甚至上百，那就是标准化没生效，模型被巨大的输入值压死了。

10. 最佳实践与总结

10.1 一套能长期用的最小配置

反复做实验的过程中，最值钱的东西不是最好的那个模型，而是一套稳定可复现的最小配置。建议固定下这几样：一份写死的天数划分、一份保存到磁盘的标准化参数、一个固定的随机种子、一套完整的评估脚本。这四样东西固定之后，任何一次改动带来的指标变化，你都能确定是改动本身导致的。

随机种子的设置要覆盖到 Python、NumPy 和 PyTorch 三层，只设torch.manual_seed是不够的，DataLoader 的 shuffle 用的是另一套随机源。

import random import numpy as np import torch def set_seed(seed: int = 42): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) torch.backends.cudnn.deterministic = True torch.backends.cudnn.benchmark = False

benchmark=False会牺牲一点速度，但能保证结果可复现。研究阶段建议关掉，等确认最终配置了再打开提速。

10.2 实验记录怎么留

至少记录这几项：完整配置文件、git commit、训练日志、验证集 F1 曲线、测试集混淆矩阵。如果嫌麻烦，用 TensorBoard 把 loss 和 F1 打进去就够了，但配置文件和 commit 一定要单独存，因为 TensorBoard 里看不到这些。

命名上建议带上关键信息，比如deeplob_w100_h64_lr1e-3_seed42，比exp_003这种好查太多。半年后回头看实验，你会感谢当时认真起名的自己。

10.3 从研究到应用的边界

这里必须再说一次。LOB 建模的研究价值和商业价值之间隔着巨大的鸿沟。学术基准上的三分类准确率，和真实市场中的可执行收益之间，至少还差着交易成本建模、订单执行模拟、风控约束、资金容量评估这几道关。任何跳过这些环节直接谈收益的说法都不可信。

如果你的目标是做研究，那这篇文章的流程已经够用了，接下来可以往特征工程、多任务学习、跨股票迁移这些方向走。如果你的目标是做工程落地，那建议先把离线回测框架搭起来，把成本和滑点算进去，再看模型还有没有剩余价值。很多时候算完会发现，模型的那点微弱优势直接被手续费吃掉了。

10.4 接下来值得尝试的方向

第一个方向是特征增强。40 维原始盘口之外，加上订单流不平衡、微价格偏离、累计深度斜率这些派生特征，通常在真实数据上能带来明显提升。FI-2010 已经做过归一化，特征的增量空间有限，换成自己的数据之后差异会很明显。

第二个方向是多任务学习。除了预测方向，同时预测未来波动幅度或者成交量的变化，辅助任务能起到正则化作用，主任务的泛化性往往会更好。

第三个方向是跨标的学习。把多只股票的数据混在一起训练，用股票 ID 做 embedding，能显著扩大有效样本量，缓解单只股票数据不足的问题。这在 Transformer 上做起来比 LSTM 更自然，因为注意力机制本身就能处理变长的上下文。

最后一个建议：先把 DeepLOB 跑通、指标对上、流程固化，再去动 Transformer。顺序反了的话，你会同时面对两个不确定的东西，出问题的时候根本无从下手。这条经验适用于几乎所有深度学习复现工作。