企业官网建设流程全解析

1. 先厘清一个问题：LOB 建模到底在建模什么

很多人在刚接触限价订单簿（Limit Order Book，下文简称 LOB）建模时，会下意识把它归类为“时间序列预测”的一个分支，然后直接搬 LSTM、GRU、TCN 那一套上来跑。跑完之后发现准确率上不去，于是换模型、加层数、调学习率，陷入一种“模型不够强”的循环。但实际做过一轮完整的实验流程之后，更常见的结论是：真正卡住结果的往往不是模型结构，而是数据管线。

具体来说，LOB 建模做的事情，是给定过去一段时间内订单簿的挂单快照序列，去预测未来一个短窗口内中间价的移动方向。注意这里有两个限定词：“短窗口”和“方向”。短窗口意味着预测目标本身噪声极大，方向意味着你输出的三类标签里，中间那一类（价格基本不动）在真实数据里占比常常很高。这两个特点叠加在一起，导致一个很尴尬的现象：模型精度看起来不低，但把它放到真实的交易成本、延迟和滑点约束下，收益会迅速被消耗掉。

这也是为什么 DeepLOB 这篇 2019 年的工作到今天仍然值得拿出来讲。它不是一个“最强”的模型，但它是一个非常干净的基线：结构不复杂、参数量可控、复现路径清晰，而且它的设计里包含了几个对 LOB 数据特性的明确判断。理解这些判断，比无脑堆 Transformer 有用得多。

本文会从 LOB 的数据形态讲起，拆解 DeepLOB 的结构动机，再对比 Transformer 方案的适用边界，最后给出一套可以直接跑起来的 PyTorch 代码，包括数据管线、两个模型、训练脚本和排查思路。读完之后，你应该能独立完成一次 LOB 方向预测的最小实验，并且知道哪些地方在错误处理上代价最高。

需要先说清楚边界：本文的内容是机器学习建模与工程实践，不构成任何交易建议。LOB 相关的实验应当只在历史数据和模拟环境中进行，任何真实资金操作都需要在合规前提下、经过独立的风控评估，这一点后面还会再强调。

2. LOB 的核心概念、数据表示与标签构造

2.1 订单簿的原始形态与信息量

限价订单簿本质上是某只标的当前所有未成交限价单的集合，按价格分档组织。买侧（bid）按价格从高到低排列，卖侧（ask）按价格从低到高排列，最优买价和最优卖价之间的差额就是价差（spread）。每一档记录两个数字：价格和该价位上的挂单量。

这个结构的信息量远比“最新成交价”丰富。价格只是一个结果，而订单簿反映了形成价格的供需结构。比如买侧前十档累计量远大于卖侧，说明短期买压更强；比如最优卖价上突然挂出巨量，可能意味着有人在刻意压制价格；再比如价差突然拉大，通常意味着流动性提供方在撤单，市场处于不确定状态。

对于建模来说，关键点是：订单簿是一个高维、强噪声、非平稳的截面数据，同时又是随时间连续变化的面板数据。这两重属性决定了它的预处理方式和普通时序任务不一样，后面会展开。

2.2 三种常见的 LOB 表示方式

实际项目里，同一份原始逐笔数据可以整理成完全不同的三种形态，选哪种直接决定模型输入长什么样。

DeepLOB 和绝大多数公开工作都采用第二种：定长快照序列。原因很实际——只有固定形状的输入才是可批量训练的输入。做了这个选择，就必须接受信息损失，因此采样频率的选取成了一个需要认真对待的超参数：太密，相邻快照高度冗余；太疏，会错过关键的流动性变化。

一个常见的特征构造是取买卖各 10 档的价格与挂单量，共 40 维。如果再补上价差、中间价、买卖累计量差等衍生量，维度会更高，但要注意衍生量和原始量之间高度相关，直接堆进去会让归一化的尺度更混乱。

2.3 标签构造：最容易埋雷的一步

标签怎么定义，比模型怎么设计更影响最终结论。公开数据集（比如 FI-2010）常用的做法是平滑标签：取未来 k 个时间步的中间价均值，与当前均值比较，按阈值分成上涨、持平、下跌三类。“平滑”的作用是降低单点噪声，避免标签被一个异常 tick 带偏。

自己构造标签时，有三个点必须想清楚。

第一，阈值怎么定。如果用固定的绝对价格差（比如 0.01），在不同价格水平和不同波动率阶段完全不等价。更稳妥的做法是用相对阈值，比如当前价差的一定比例，或者用过去一段时间的波动率做自适应。

第二，未来窗口有多长。k 越大，信号越强但可交易性越差；k 越小，噪声越大。这个参数要和你的决策频率对齐，不能孤立地调。

第三，也是最容易出错的一点：任何涉及“未来”的计算，都不能污染训练集。滚动归一化、标签平滑、波动率估计，只要用到了未来窗口内的数据，测试集上的指标就会虚假偏高。这条在 LOB 任务里尤其致命，因为数据本身的自相关性极强，一点泄漏就能让 F1 涨十几个点。

3. DeepLOB 的结构拆解与设计动机

3.1 整体结构：三块卷积加一个循环层

DeepLOB 的输入是一个形状为(1, T, F)的张量，T 是时间步数（论文中取 100），F 是特征维（论文中取 40）。整个网络分成三段：

第一段是三个卷积块，每个块内部先沿特征方向做(1, 2)卷积，再沿时间方向做(4, 1)卷积。前两块后面接(2, 2)池化，第三块接(1, 2)池化，把特征维压到 1。

第二段是一个 Inception 模块，用1×1、3×3、5×5三种感受野并行提取时间维上的多尺度模式，最后拼接通道。

第三段是一个单层 LSTM，隐藏维度 64，取最后时刻的输出，接全连接层输出三类概率。

这个结构里，真正有意思的不是参数多少，而是每个设计选择背后的判断。

3.2 为什么先卷积再循环

很多人第一反应是“LOB 是时间序列，那当然直接上 LSTM”。DeepLOB 的作者选择了相反的顺序：先用卷积处理，再交给 LSTM。原因在于数据维度的语义不对称。

输入矩阵的两个轴含义完全不同。特征轴（40 维）是有内部结构的：前几维是买侧档位，中间几维是卖侧档位，每一档内部又是价格和数量的配对。这个结构是空间性的、局部相关的。而时间轴上是序列性的、长程相关的。

(1, 2)的卷积核正好横跨“价格-数量”这一对，它在特征轴上滑动，做的其实是跨档位的局部模式提取。(4, 1)的卷积核在时间轴上滑动，提取的是短窗口内的时序形态。这两步做完，特征轴的信息已经被压扁并抽象，剩下的才交给 LSTM 去建模时间依赖。

如果反过来直接上 LSTM，模型要同时学“特征轴的结构”和“时间轴的依赖”，在小数据上很难收敛好。这个思路其实很通用：当输入是多维张量且各轴语义不同时，先用卷积吃掉结构性强的那一维，再用序列模型处理剩下那一维。

3.3 Inception 模块在 LOB 上解决了什么

Inception 的经典作用是提供多尺度感受野。放到 LOB 语境下，这个需求是真实的：订单簿上的模式有快慢之分。挂单量的瞬时抖动是秒级的，而流动性枯竭、大单堆积这类状态可能持续几十个 tick。单一尺寸的卷积核只能覆盖一个尺度。

用3×3和5×5并行，配合1×1的降维分支，等于让模型自己决定当前样本更依赖哪个尺度的模式。代价是通道数膨胀（拼接后到 256），参数量上升，但相比后面接一个大 LSTM，这个代价是划算的。

需要提醒的是，Inception 分支里的padding必须小心设置，否则时间维长度会在每层之后变化，最终导致 LSTM 输入形状对不上。用(1, 0)这种不对称 padding 是保持时间维不变、只在特征维上补零的常见写法。

4. Transformer 方案：能替代 DeepLOB 吗

4.1 直接套标准 Transformer 会遇到的问题

Transformer 在长序列建模上的优势是公认的，把它搬到 LOB 上看起来顺理成章。但如果直接把nn.TransformerEncoder套在(T, F)序列上，通常会遇到几个具体问题。

第一个是位置编码与时间间隔不匹配。标准正弦位置编码假设相邻 token 的间隔相等，但真实 LOB 快照在事件时间上是不均匀的。高频时段一秒几十条，低频时段几分钟一条。等间隔编码等于把这个信息抹掉了。可行的补救是把真实时间差作为额外的位置特征拼进输入，但这要求原始数据里保留精确时间戳。

第二个是有效信号过于局部。LOB 的方向预测窗口通常只有几十个 tick，真正决定方向的往往是最近几个快照的流动性变化，而不是 100 步之前的形态。全注意力在长序列上会把注意力摊薄，反而不如卷积的局部归纳偏置有效。

第三个是参数量和过拟合。标准的 4 层 encoder、d_model=128 的配置，参数量轻松超过百万级。而 LOB 数据集即便有几十万样本，有效独立样本数也远少于表面数量（因为样本间高度重叠）。这种情况下大模型的泛化往往不如小模型稳。

4.2 三种务实的改造思路

基于上面的问题，公开工作里出现了几种比较务实的改造方向，值得参考。

第一种是卷积 stem + Transformer encoder。先用几层一维卷积把原始 40 维特征投影到d_model维的 token 序列，同时压缩时间维长度（比如 100 → 25）。这让 Transformer 处理的序列变短，同时卷积本身提供了局部归纳偏置。Wallbridge 在 2020 年提出的 TransLOB 就是这一类思路，用空洞卷积做前置特征提取，再交给 encoder。具体实现细节建议以论文公开描述为准。

第二种是轴向注意力（Axial Attention）。把 40 维特征和 100 步时间当作两个独立的注意力轴，先在特征轴上做注意力（建模档位间关系），再在时间轴上做注意力（建模时序依赖）。这样注意力矩阵的规模从(T×F)²降到T² + F²，计算量大幅下降，同时保留了二维结构信息。

第三种是相对位置编码 + 时间差嵌入。用相对位置代替绝对位置，把相邻 token 的实际时间差作为一个偏置项加到注意力分数上。这个改法改动量小，但对不规则采样的数据效果显著。

4.3 DeepLOB 与 Transformer 的选型对比

把两者放在一起比较，结论其实不复杂：它们不是替代关系，而是适用条件不同。

更实际的建议是：先用 DeepLOB 把整条管线跑通，确认数据处理、标签构造、评估方式都没有问题，再考虑上 Transformer。如果基线本身只有 40% 的 F1，换模型救不回来；如果基线能到合理水平，换模型的收益才有意义。

5. 环境准备与项目结构

5.1 运行环境与依赖

本文代码基于 PyTorch，GPU 不是必须的，CPU 上跑小规模实验完全可行。版本方面不写死具体号，以你本机实际安装的稳定版为准，关键是保证torch和numpy版本互相兼容。

# 创建并激活虚拟环境 python -m venv lob-env source lob-env/bin/activate # Windows 下用 lob-env\Scripts\activate # 安装依赖（版本以实际兼容为准） pip install torch numpy pandas scikit-learn matplotlib

如果你的机器有可用的 CUDA，装对应版本的 torch 即可；如果没有，默认的 CPU 版本足够跑通本文的示例。

5.2 目录结构约定

代码组织建议按下面这个结构来，后面所有文件路径都基于它。这样做的目的是把“数据处理”和“模型训练”彻底分开，避免在一堆脚本里互相 import 造成循环依赖。

lob-project/ ├── configs/ │ └── base.yaml ├── data/ │ └── (数据文件放这里) ├── src/ │ ├── dataset.py │ ├── models/ │ │ ├── deeplob.py │ │ └── lob_transformer.py │ ├── train.py │ └── evaluate.py └── README.md

5.3 数据格式约定

不同数据源的字段名差异很大，为了避免绑定到某一个特定数据集，这里统一约定一个中间格式：一个.npz文件，包含三个数组。

X: shape (N, T, F)，float32，已按时间顺序排列 y: shape (N,)，int64，取值 0/1/2 分别代表下跌/持平/上涨 t: shape (N,)，int64，每条样本的结束时间戳（用于按时间切分）

有了这个中间格式，无论原始数据来自哪家供应商、是 csv 还是二进制，只需要写一个转换脚本落到这个格式，后面的训练代码就不用改。这是工程上一个很实用的习惯：把数据源的多样性收敛到一个内部契约上。

6. 完整代码实现

6.1 构造一份可运行的模拟数据

真实 LOB 数据不易获取，为了让你先把流程跑通，这里写一个合成数据生成器。它用带均值回复的随机游走模拟中间价，再用与价格变化相关的随机过程模拟买卖两侧挂单量，从而制造出“订单簿状态与未来方向存在弱相关”的效果。注意这只是为了验证管线正确，不代表真实市场规律。

# 文件路径：src/make_synthetic_data.py import numpy as np def generate_lob_series(n_steps=60000, n_levels=10, seed=42): rng = np.random.default_rng(seed) # 中间价：带均值回复的随机游走 price = np.zeros(n_steps) price[0] = 100.0 for i in range(1, n_steps): shock = rng.normal(0, 0.01) revert = -0.02 * (price[i - 1] - 100.0) price[i] = price[i - 1] + shock + revert # 订单簿不平衡度：与下一期价格变化弱相关 future_move = np.diff(price, append=price[-1]) imbalance = np.tanh(future_move * 30.0) + rng.normal(0, 0.55, n_steps) # 生成 40 维特征：买卖各 10 档的 (价格偏移, 挂单量) feats = np.zeros((n_steps, n_levels * 4), dtype=np.float32) for lvl in range(n_levels): spread_step = 0.01 * (lvl + 1) base_vol = 100 + 40 * lvl bid_vol = base_vol * (1 + 0.3 * imbalance) + rng.normal(0, 8, n_steps) ask_vol = base_vol * (1 - 0.3 * imbalance) + rng.normal(0, 8, n_steps) feats[:, lvl * 2] = price - spread_step # 买价 feats[:, lvl * 2 + 1] = np.clip(bid_vol, 1, None) # 买量 feats[:, n_levels * 2 + lvl * 2] = price + spread_step feats[:, n_levels * 2 + lvl * 2 + 1] = np.clip(ask_vol, 1, None) return feats, price def build_dataset(feats, price, window=100, horizon=10, thr=0.0002): X, y, t = [], [], [] for end in range(window, len(price) - horizon): future_mid = price[end: end + horizon].mean() cur_mid = price[end - window: end].mean() ret = (future_mid - cur_mid) / cur_mid if ret > thr: label = 2 elif ret < -thr: label = 0 else: label = 1 X.append(feats[end - window: end]) y.append(label) t.append(end) return (np.asarray(X, dtype=np.float32), np.asarray(y, dtype=np.int64), np.asarray(t, dtype=np.int64)) if __name__ == "__main__": feats, price = generate_lob_series() X, y, t = build_dataset(feats, price) np.savez_compressed("data/synthetic_lob.npz", X=X, y=y, t=t) print("X:", X.shape, "y:", y.shape) print("类别分布:", np.bincount(y))

这段代码里有几个点值得注意。阈值thr用的是相对收益率而非绝对价差，这样在不同价格水平下语义一致。horizon取 10 表示用未来 10 步的均值判断方向，而不是单点比较，这正是前面说的“平滑标签”思想。最后保存时把时间戳t一起存下来，是为了后面能按时间切分。

6.2 Dataset 与归一化：把防泄漏写进代码

归一化是最容易引入未来信息的地方。正确做法是只用训练集统计均值和标准差，然后应用到验证集和测试集。下面的实现把这个约束固化到了类的设计里。

# 文件路径：src/dataset.py import numpy as np import torch from torch.utils.data import Dataset, DataLoader class LOBDataset(Dataset): def __init__(self, X, y): # X: (N, T, F) float32, y: (N,) int64 self.X = torch.from_numpy(X) self.y = torch.from_numpy(y) def __len__(self): return len(self.y) def __getitem__(self, idx): # 返回 (1, T, F)，与 DeepLOB 的 Conv2d 输入维度对齐 return self.X[idx].unsqueeze(0), self.y[idx] def split_by_time(X, y, t, train_ratio=0.7, val_ratio=0.15): order = np.argsort(t) X, y = X[order], y[order] n = len(y) n_train = int(n * train_ratio) n_val = int(n * (train_ratio + val_ratio)) return (X[:n_train], y[:n_train], X[n_train:n_val], y[n_train:n_val], X[n_val:], y[n_val:]) def fit_scaler(X_train): # 逐特征统计，shape: (F,) mean = X_train.reshape(-1, X_train.shape[-1]).mean(axis=0) std = X_train.reshape(-1, X_train.shape[-1]).std(axis=0) std = np.where(std < 1e-8, 1.0, std) return mean.astype(np.float32), std.astype(np.float32) def apply_scaler(X, mean, std): return ((X - mean) / std).astype(np.float32) def build_loaders(npz_path, batch_size=128, num_workers=0): d = np.load(npz_path) X, y, t = d["X"], d["y"], d["t"] Xtr, ytr, Xva, yva, Xte, yte = split_by_time(X, y, t) mean, std = fit_scaler(Xtr) # 只用训练集统计 Xtr = apply_scaler(Xtr, mean, std) Xva = apply_scaler(Xva, mean, std) Xte = apply_scaler(Xte, mean, std) return ( DataLoader(LOBDataset(Xtr, ytr), batch_size=batch_size, shuffle=True, num_workers=num_workers), DataLoader(LOBDataset(Xva, yva), batch_size=batch_size, shuffle=False, num_workers=num_workers), DataLoader(LOBDataset(Xte, yte), batch_size=batch_size, shuffle=False, num_workers=num_workers), )

split_by_time强制按时间顺序切分，而不是随机切分。这一步看似平常，实际上是整个流程里最重要的一行——随机切分会让训练集里包含时间上晚于验证集的样本，由于 LOB 数据的高自相关性，指标会虚高得离谱，而且这种虚高在错误排查时极难发现。

6.3 DeepLOB 模型实现

下面按照前面拆解的结构实现。注意每一层的形状变化，尤其是池化之后特征维的收敛过程：40 → 20 → 10 → 5 → 2 → 1。

# 文件路径：src/models/deeplob.py import torch import torch.nn as nn class DeepLOB(nn.Module): def __init__(self, num_classes=3, hidden_size=64): super().__init__() self.conv_block1 = nn.Sequential( nn.Conv2d(1, 32, kernel_size=(1, 2), stride=(1, 2)), nn.LeakyReLU(0.01), nn.BatchNorm2d(32), nn.Conv2d(32, 32, kernel_size=(4, 1)), nn.LeakyReLU(0.01), nn.BatchNorm2d(32), nn.MaxPool2d(kernel_size=(2, 2), stride=(2, 2)), ) self.conv_block2 = nn.Sequential( nn.Conv2d(32, 32, kernel_size=(1, 2), stride=(1, 2)), nn.LeakyReLU(0.01), nn.BatchNorm2d(32), nn.Conv2d(32, 32, kernel_size=(4, 1)), nn.LeakyReLU(0.01), nn.BatchNorm2d(32), nn.MaxPool2d(kernel_size=(2, 2), stride=(2, 2)), ) self.conv_block3 = nn.Sequential( nn.Conv2d(32, 32, kernel_size=(1, 2), stride=(1, 2)), nn.LeakyReLU(0.01), nn.BatchNorm2d(32), nn.Conv2d(32, 32, kernel_size=(4, 1)), nn.LeakyReLU(0.01), nn.BatchNorm2d(32), nn.MaxPool2d(kernel_size=(1, 2), stride=(1, 2)), ) # Inception 模块，多尺度时间感受野 self.inception = nn.Sequential( nn.Conv2d(32, 64, kernel_size=(1, 1)), nn.LeakyReLU(0.01), nn.BatchNorm2d(64), InceptionBlock(64), nn.Conv2d(256, 128, kernel_size=(1, 1)), nn.LeakyReLU(0.01), nn.BatchNorm2d(128), ) self.lstm = nn.LSTM(input_size=128, hidden_size=hidden_size, num_layers=1, batch_first=True) self.fc = nn.Linear(hidden_size, num_classes) def forward(self, x): # x: (B, 1, T, F) x = self.conv_block1(x) x = self.conv_block2(x) x = self.conv_block3(x) x = self.inception(x) # (B, C, T', 1) -> (B, T', C) b, c, t, f = x.shape x = x.reshape(b, c, t).permute(0, 2, 1) out, _ = self.lstm(x) last = out[:, -1, :] return self.fc(last) class InceptionBlock(nn.Module): def __init__(self, in_ch=64, out_ch=64): super().__init__() self.b1 = nn.Sequential(nn.Conv2d(in_ch, out_ch, (1, 1)), nn.LeakyReLU(0.01), nn.BatchNorm2d(out_ch)) self.b2 = nn.Sequential(nn.Conv2d(in_ch, out_ch, (1, 1)), nn.LeakyReLU(0.01), nn.BatchNorm2d(out_ch), nn.Conv2d(out_ch, out_ch, (3, 3), padding=(1, 0)), nn.LeakyReLU(0.01), nn.BatchNorm2d(out_ch)) self.b3 = nn.Sequential(nn.Conv2d(in_ch, out_ch, (1, 1)), nn.LeakyReLU(0.01), nn.BatchNorm2d(out_ch), nn.Conv2d(out_ch, out_ch, (5, 5), padding=(2, 0)), nn.LeakyReLU(0.01), nn.BatchNorm2d(out_ch)) self.b4 = nn.Sequential(nn.MaxPool2d((3, 1), stride=(1, 1), padding=(1, 0)), nn.Conv2d(in_ch, out_ch, (1, 1)), nn.LeakyReLU(0.01), nn.BatchNorm2d(out_ch)) def forward(self, x): return torch.cat([self.b1(x), self.b2(x), self.b3(x), self.b4(x)], dim=1)

Inception 里的分支都是padding=(k, 0)的形式，第一个维度（时间）补零保持长度，第二个维度（特征）此时已经是 1，不补零也不会变负。这个细节如果写错，reshape那一步会直接报形状不匹配。

6.4 Transformer 版本实现

采用“卷积 stem + encoder”的思路，把 100×40 的输入先用一维卷积压缩到 25 个 token，再交给 encoder。这样既控制了计算量，也保留了局部归纳偏置。

# 文件路径：src/models/lob_transformer.py import torch import torch.nn as nn import math class PositionalEncoding(nn.Module): def __init__(self, d_model, max_len=512, dropout=0.1): super().__init__() self.dropout = nn.Dropout(dropout) pe = torch.zeros(max_len, d_model) pos = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1) 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=40, d_model=128, nhead=4, num_layers=3, num_classes=3, dropout=0.1): super().__init__() # 卷积 stem：时间维下采样 4 倍，特征维升到 d_model self.stem = nn.Sequential( nn.Conv1d(n_features, d_model, kernel_size=5, stride=2, padding=2), nn.GELU(), nn.BatchNorm1d(d_model), nn.Conv1d(d_model, d_model, kernel_size=5, stride=2, padding=2), nn.GELU(), nn.BatchNorm1d(d_model), ) self.pos_enc = 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, num_classes) def forward(self, x): # x: (B, 1, T, F) -> (B, F, T) x = x.squeeze(1) x = self.stem(x) # (B, d_model, T/4) x = x.permute(0, 2, 1) # (B, T/4, d_model) x = self.pos_enc(x) x = self.encoder(x) x = self.norm(x.mean(dim=1)) # 平均池化代替取最后一位 return self.head(x)

这里用平均池化而非取最后一位，是因为经过 encoder 之后每个 token 都在一定程度上聚合了全局信息，取均值比取末端更稳定。norm_first=True在中小规模数据上通常比原始顺序更容易训稳，这是一个经验性选择，如果你的数据量很小，可以尝试关掉。

6.5 训练与评估脚本

训练循环本身不复杂，但有几个细节会影响结果的可比性：固定随机种子、用 macro-F1 作为主要指标（因为类别不平衡）、保存验证集最优的权重。

# 文件路径：src/train.py import argparse import numpy as np import torch import torch.nn as nn from sklearn.metrics import f1_score, classification_report from src.dataset import build_loaders from src.models.deeplob import DeepLOB from src.models.lob_transformer import LOBTransformer def set_seed(seed): np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) @torch.no_grad() def evaluate(model, loader, device): model.eval() preds, labels = [], [] for xb, yb in loader: xb = xb.to(device) logits = model(xb) preds.append(logits.argmax(dim=1).cpu().numpy()) labels.append(yb.numpy()) preds = np.concatenate(preds) labels = np.concatenate(labels) return f1_score(labels, preds, average="macro"), preds, labels def main(): parser = argparse.ArgumentParser() parser.add_argument("--data", default="data/synthetic_lob.npz") parser.add_argument("--model", choices=["deeplob", "transformer"], default="deeplob") parser.add_argument("--epochs", type=int, default=15) parser.add_argument("--lr", type=float, default=1e-3) parser.add_argument("--batch-size", type=int, default=128) parser.add_argument("--seed", type=int, default=42) args = parser.parse_args() set_seed(args.seed) device = torch.device("cuda" if torch.cuda.is_available() else "cpu") train_loader, val_loader, test_loader = build_loaders( args.data, batch_size=args.batch_size) model = DeepLOB() if args.model == "deeplob" else LOBTransformer() model = model.to(device) criterion = nn.CrossEntropyLoss() optimizer = torch.optim.AdamW(model.parameters(), lr=args.lr, weight_decay=1e-4) scheduler = torch.optim.lr_scheduler.CosineAnnealingLR( optimizer, T_max=args.epochs) best_f1, best_state = -1.0, None for epoch in range(1, args.epochs + 1): model.train() total_loss = 0.0 for xb, yb in train_loader: xb, yb = xb.to(device), yb.to(device) optimizer.zero_grad() loss = criterion(model(xb), yb) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) optimizer.step() total_loss += loss.item() * xb.size(0) scheduler.step() val_f1, _, _ = evaluate(model, val_loader, device) avg_loss = total_loss / len(train_loader.dataset) print(f"epoch {epoch:02d} | loss {avg_loss:.4f} | val macro-F1 {val_f1:.4f}") if val_f1 > best_f1: best_f1 = val_f1 best_state = {k: v.detach().cpu().clone() for k, v in model.state_dict().items()} if best_state is not None: model.load_state_dict(best_state) test_f1, preds, labels = evaluate(model, test_loader, device) print(f"best val macro-F1: {best_f1:.4f} | test macro-F1: {test_f1:.4f}") print(classification_report(labels, preds, target_names=["down", "flat", "up"], digits=4)) if __name__ == "__main__": main()

clip_grad_norm_这一行不是可有可无的。LOB 输入在某些时段会出现极端值（比如价差瞬间拉大），梯度爆炸的概率比普通时序任务高，加上这个约束能显著降低训练中断的风险。

7. 运行结果与验证方法

先跑数据生成，再跑训练。命令和预期输出大致如下：

python -m src.make_synthetic_data python -m src.train --model deeplob --epochs 15 python -m src.train --model transformer --epochs 15

数据生成阶段应该看到类似X: (59890, 100, 40) y: (59890,)的输出，以及三类样本的分布。如果某类样本数接近 0，说明阈值thr设得不合适，需要调整。

训练阶段每个 epoch 打印一行，验证集 macro-F1 应该在前几个 epoch 内快速上升，之后趋于平稳。在本文的合成数据上，DeepLOB 的 test macro-F1 大概落在 0.4 到 0.6 之间波动（具体数值受随机种子影响），三类的召回率差异会很明显——这正是真实 LOB 任务里的典型现象。

判断训练是否正常，看三个信号。第一，训练 loss 持续下降但验证 F1 不涨，说明过拟合，需要加正则或减容量。第二，验证 F1 从第一个 epoch 就很高，比如 0.9 以上，几乎可以确定存在数据泄漏，回去检查切分和归一化。第三，三类里“flat”类的 F1 明显高于其他两类，而“down”和“up”接近 0，说明模型退化成了多数类预测器，需要看类别分布并考虑类别权重。

如果运行失败，第一步先看报错发生在哪一层。形状相关的错误九成出在reshape和permute，把x.shape在 forward 里打出来一路追下去最快。CUDA 内存不足则优先降batch_size，而不是改模型。

8. 常见问题与排查思路

还有一类问题不体现在报错上，但更影响结论：模型在验证集和测试集上表现差异很大。这通常意味着两个集合的数据分布不同（比如测试集处于不同的波动率阶段），而不是模型有问题。遇到这种情况，按时间段做分组评估，看看是哪一段拖低了整体指标，比盲目调参更有价值。

9. 最佳实践与工程建议

先把评估协议定死，再动模型。很多实验做了一轮又一轮却得不出结论，根源是每换一次模型就顺手改一下切分方式或评估指标，导致结果不可比。建议在开始之前就固定：切分方式、归一化策略、标签定义、主指标、随机种子。这五项不动，任何模型改动才有意义。

归一化的粒度值得单独设计。本文用的是全局 z-score，但真实 LOB 数据是非平稳的，全局统计量在不同时段差异巨大。更贴近实践的做法是用滚动窗口的统计量做归一化，窗口长度要小于预测窗口，确保不引入未来信息。这个改动对结果的影响，往往比换模型更大。

延迟和吞吐要提前估算。如果你的目标场景对推理时延敏感，DeepLOB 这类小模型有明显优势。一个 100×40 输入的 DeepLOB 在单核 CPU 上的推理时间是毫秒级的，而 3 层 Transformer 会高出数倍。在建模阶段就把这个约束考虑进去，能避免后期推倒重来。

日志要记录到能复现的程度。除了超参数，还要记录数据版本、特征构造代码的 commit、随机种子、环境版本。LOB 实验的可复现性很差，没有这些信息，两周后你自己也复现不出当时的结果。

关于实盘迁移，务必保持克制。模型在测试集上的 macro-F1 是一个纯粹的统计指标，它不包含交易成本、滑点、撮合延迟、撤单失败这些真实约束。把统计指标直接等价于策略收益，是这类工作里最危险的认知跳跃。任何涉及真实资金的环节，都必须在合规框架内、经过独立风控评估，并且先在模拟环境中长期验证。本文所有代码和结论仅用于研究与学习。

版本管理与协作。建议把数据管线、模型定义、训练脚本分成三个独立模块，用配置文件串起来。团队协作时，数据管线的任何改动都要经过 review，因为这里的 bug 最隐蔽，而且一旦出错，后续所有实验结论全部作废。

10. 写在最后：这份工作的边界

回到开头那个问题：LOB 建模到底难在哪。把整条链路走一遍之后，答案会比较清楚——难的不是把 DeepLOB 或者 Transformer 的代码敲出来，这两份实现加起来不到三百行；难的是让这条链路里的每一个环节都不骗自己。

标签构造时那个阈值是不是合理，归一化有没有偷看未来，切分是不是真的按时间，评估指标选得对不对，模型在测试集上的优势是不是只来自于某个特定时段——这些问题没有标准答案，但每一个都会实质性地改变结论。DeepLOB 的价值在于它给了一个足够简单、足够透明的起点，让你能快速判断“问题出在数据还是模型”。Transformer 的价值在于它提供了更强的表达能力，但前提是你已经有一份可信的基线可以对照。

如果要继续往下走，几个方向值得深入：一是用真实的高频数据替换合成数据，重点观察滚动归一化和事件时间对齐带来的差异；二是试试轴向注意力那种改法，把 40 维特征的结构显式建模进去；三是把评估从单一的分类指标扩展到考虑成本的分层回测，哪怕只是在模拟环境里，也能让你对模型的真实可用性有更清醒的认识。这套代码建议收藏下来，改数据源、改标签定义都能直接复用，比每次从零搭一遍要省事得多。