PyTorch实现LSTM+Transformer高频交易预警混合模型

简介这是一份面向金融科技从业者与深度学习研究者的技术文档聚焦金融风控场景下的高频交易预警难题系统讲解如何借助PyTorch搭建LSTM-Transformer混合模型适合具备一定神经网络基础、希望把时序建模能力落到风控实战的开发者与研究生阅读也可作为相关课题的参考材料。资源为单个PDF文件压缩包约2.18MB正文共40页支持目录章节跳转、阅读器左侧大纲显示与章节快速定位查阅方便。内容自金融风控与高频交易风险概述讲起逐层展开RNN与LSTM、Transformer注意力机制的原理并给出混合模型的设计思路随后落到PyTorch环境搭建、高频交易数据采集与清洗标准化、LSTM层与Transformer编码器层的代码实现、训练验证循环、超参数调优与正则化再延伸到预警系统的集成部署、数据采集传输、模型推理性能优化与预警规则制定最后通过案例对比传统方法评估预警准确性与及时性。目前已有118人学习。1. 混合模型在高频交易预警中的定位毫秒级行情里单笔 tick 的信噪比低得离谱纯 LSTM 容易把局部抖动当成趋势纯 Transformer 又容易在只有几百根 K 线的窗口上把噪声记成模式。这份 40 页资料给出的思路是把两者串起来先让 LSTM 沿时间轴做一次带门控的压缩把局部依赖和长记忆收进隐状态序列再交给 Transformer 编码器用多头自注意力看全局最后接一个线性头输出风险概率。它要解决的不是预测涨跌而是未来 N 个 tick 内是否出现异常波动或流动性骤降属于典型的事件预警任务评价口径天然偏向召回率和误报率的权衡。适合已经能独立写 PyTorch 训练循环、想把时序异常检测落到真实数据上的工程师也适合做量化风控、需要一套可复现基线的人。2. PyTorch 环境搭建与 tick 级数据的窗口化预处理2.1 环境安装与设备可用性核验高频数据动辄上千万行CPU 上跑一轮训练够泡两杯咖啡所以环境第一步就是把 GPU 链路确认清楚。我一般用 conda 建独立环境避免和系统里的 NumPy 版本打架。# 建一个专用环境Python 3.10 对 torch 2.x 兼容最好 conda create -n hft-risk python3.10 -y conda activate hft-risk # 按本机驱动选 wheel 源驱动支持 CUDA 12.1 就用 cu121否则退到 cu118 pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121 pip install numpy pandas scikit-learn pyarrow matplotlib装完必须核验不然跑到一半才发现模型在 CPU 上白白浪费半天。import torch print(torch.__version__, torch.version.cuda) # 版本号 编译时 CUDA 版本 print(torch.cuda.is_available()) # 必须是 True print(torch.cuda.get_device_name(0) if torch.cuda.is_available() else cpu only) # 设备一致性检查模型和输入张量必须在同一个 device 上 x torch.randn(4, 16, 8) dev cuda if torch.cuda.is_available() else cpu print(x.to(dev).device, x.dtype) print(torch.backends.cudnn.version())参数说明torch.version.cuda是编译时链接的 CUDA 版本和nvidia-smi显示的驱动版本不是一回事前者不高于后者即可。输入形状(batch, seq_len, n_feat)是后面所有层的约定先记住这个三元组。窗口长度固定时打开torch.backends.cudnn.benchmark True卷积和矩阵乘的算法会自动选最快的那套。2.2 tick 数据的特征构造与去极值原始逐笔数据直接喂模型效果很差因为价格量纲、成交量量级、订单簿厚度的分布差异太大。我的习惯是先构造量纲无关的派生特征再做稳健去极值。import numpy as np import pandas as pd def build_features(df: pd.DataFrame) - pd.DataFrame: out pd.DataFrame(indexdf.index) # 中间价买卖一档均值比最新成交价抗单笔异常成交 out[mid] (df[bid1] df[ask1]) / 2.0 # 相对价差无量纲跨标的可复用 out[spread] (df[ask1] - df[bid1]) / out[mid] # 订单簿不平衡买卖挂量之差占总量比例落在 [-1, 1] depth df[bid_vol1] df[ask_vol1] out[obi] (df[bid_vol1] - df[ask_vol1]) / depth.replace(0, np.nan) # 对数成交量压掉尖峰 out[log_vol] np.log1p(df[volume]) # 撤单率反映做市方情绪 out[cancel_ratio] df[cancel_cnt] / (df[order_cnt] 1) # 一阶收益用于后续算波动率 out[ret1] out[mid].pct_change() return out.replace([np.inf, -np.inf], np.nan).dropna() def mad_clip(s: pd.Series, n: float 5.0) - pd.Series: med s.median() mad (s - med).abs().median() scale 1.4826 * mad # 还原到正态标准差尺度 if scale 0: # 一字板等极端情况回退到分位数 return s.clip(s.quantile(0.001), s.quantile(0.999)) return s.clip(med - n * scale, med n * scale)逻辑说明obi用了.replace(0, np.nan)防止除零mad_clip里的常数 1.4826 是 MAD 到标准差的换算系数n5大致对应正态分布下的 3.4 倍标准差。用 Z-score 去极值在这里不合适因为金融收益的峰度极高均值和标准差本身就被离群点污染了MAD 的中位数口径更抗造。2.3 滑窗切片、标签对齐与按时间切分这一节是整个流程里最容易翻车的地方。很多人顺手写train_test_split(X, y, test_size0.3, random_state42)滑窗之间的重叠会让测试集的未来信息回灌到训练集验证指标虚高得离谱。def make_windows(feat: np.ndarray, labels: np.ndarray, win: int, horizon: int): xs, ys [], [] # 窗口终点之后 horizon 步才产生标签保证标签在特征窗口之外 for end in range(win, len(feat) - horizon): xs.append(feat[end - win:end]) ys.append(labels[end horizon - 1]) return np.asarray(xs, dtypenp.float32), np.asarray(ys, dtypenp.float32) # 标签未来 horizon 步内中间价绝对收益超过 k 倍滚动波动率 horizon, k 20, 3.0 fwd_ret feat[mid].shift(-horizon) / feat[mid] - 1 vol feat[ret1].rolling(500, min_periods100).std() labels (fwd_ret.abs() k * vol).astype(np.float32).values参数说明win是回看窗口高频场景 100 到 300 足够再长边际收益很低horizon20表示预警要比事件提前约 20 个 tick 给出信号太短来不及撤单太长正样本会被噪声淹没k3.0控制正样本占比通常调到 0.5% 到 3% 之间比较合理具体看业务容忍度。划分方式是否可用原因随机train_test_split不可用相邻窗口高度重叠未来信息泄漏按时间顺序 7:1.5:1.5可用训练在前、验证居中、测试在后模拟线上推进留出最后 N 个交易日首选与真实部署形态最接近能暴露跨日分布漂移标准化的 scaler 只能在训练段上fit然后 transform 验证段和测试段顺序颠倒同样属于泄漏。DataLoader 的配置也有讲究from torch.utils.data import DataLoader, TensorDataset loader DataLoader( TensorDataset(torch.from_numpy(Xtr), torch.from_numpy(ytr)), batch_size256, shuffleTrue, drop_lastTrue, num_workers2, pin_memoryTrue) # pin_memory 让 H2D 拷贝走异步通道drop_lastTrue是为了丢掉最后那个不满的 batch避免 BatchNorm 层在单样本上统计量失真num_workers在 Windows 上偶尔会因为多进程 pickle 报错遇到就设成 0。3. LSTM 编码层与 Transformer 编码器的串联实现3.1 串联顺序为什么是先 LSTM 后 Transformer顺序反过来的方案我也试过效果普遍差一截。原因是 Transformer 的自注意力复杂度对序列长度是平方级的如果把几千个原始 tick 特征直接丢进去显存和计算都不划算而 LSTM 先用常数级的代价把序列压到一个语义更密的隐状态序列再让注意力层在这个压缩表示上找跨位置关系等于把降采样和建模全局两件事拆给了两个模块。另一个考虑是局部性价格序列的短程依赖比如连续几笔主动买单用门控结构表达最自然注意力机制反而要花更多头数才能学到。3.2 LSTM 层的实现与关键参数import torch import torch.nn as nn class LSTMTransformer(nn.Module): def __init__(self, n_feat, d_model64, lstm_layers2, nhead4, ff_dim256, trans_layers2, dropout0.2): super().__init__() # 第一段门控循环把局部时序压成隐状态序列 self.lstm nn.LSTM( n_feat, d_model, num_layerslstm_layers, batch_firstTrue, dropoutdropout if lstm_layers 1 else 0.0, bidirectionalFalse) self.norm nn.LayerNorm(d_model) # 稳定注意力层的输入分布 # 第二段自注意力编码器捕捉跨时间步的全局依赖 layer nn.TransformerEncoderLayer( d_modeld_model, nheadnhead, dim_feedforwardff_dim, dropoutdropout, activationgelu, batch_firstTrue, norm_firstTrue) self.encoder nn.TransformerEncoder(layer, num_layerstrans_layers) # 注意力池化让模型自己挑重要的时间步比只取最后一步稳 self.attn_pool nn.Linear(d_model, 1) self.head nn.Sequential( nn.Linear(d_model, d_model // 2), nn.GELU(), nn.Dropout(dropout), nn.Linear(d_model // 2, 1)) def forward(self, x, pad_maskNone): h, _ self.lstm(x) # (B, T, d_model) h self.norm(h) h self.encoder(h, src_key_padding_maskpad_mask) # 仍是 (B, T, d_model) w torch.softmax(self.attn_pool(h).squeeze(-1), dim1) # (B, T) z torch.einsum(bt,btd-bd, w, h) # 加权求和 return self.head(z).squeeze(-1) # (B,) logits参数说明d_model同时是 LSTM 的 hidden_size 和 Transformer 的 d_model两者必须相等才能直接对接nhead必须整除d_model64/4 是最常用的组合。batch_firstTrue让输入输出都是(B, T, F)不用来回 transpose。norm_firstTrue在浅层小模型上收敛更稳代价是显存略高。输出层不接 Sigmoid因为后面用的是BCEWithLogitsLoss它内部已经做了数值稳定的 sigmoid。3.3 三个模块的分工与调参方向组件负责的依赖类型关键超参调整方向LSTM 层局部、短中程依赖hidden_size / num_layers窗口 200 以内不建议超 2 层Transformer 编码器全局、跨位置依赖nhead / dim_feedforwardff_dim 从 4×d_model 起步注意力池化时间步权重分配无比取最后一步的 AUC 通常高 1~2 个点模型规模不要一上来就堆大。我一般先用d_model64, nhead4, trans_layers2跑通确认训练损失能正常下降到平台期再加宽。高频预警的正样本本来就少参数量一大立刻过拟合验证集 PR-AUC 会在两三个 epoch 后掉头往下。4. 训练循环、类别不平衡与调优排错4.1 正样本只有百分之一时的损失函数选择预警任务的标签极度不平衡1% 的正样本配 99% 的负样本直接上普通 BCE 会让模型学会全预测为负准确率 99% 但一个风险都报不出来。pos float(ytr.sum()) neg float(len(ytr) - pos) pos_weight torch.tensor([neg / max(pos, 1.0)], devicedevice) criterion nn.BCEWithLogitsLoss(pos_weightpos_weight) print(f正样本 {pos:.0f}负样本 {neg:.0f}pos_weight{pos_weight.item():.1f})逻辑说明pos_weight把正样本的损失按负正比例放大让模型不敢忽视少数类。但也不能无限放大我通常会在理论值的基础上取sqrt(neg/pos)做一次收缩因为完全按比例加权会让模型过度激进误报率爆炸。监控指标上准确率在这里没有意义要看 PR-AUC 和固定误报率下的召回光看 loss 曲线判断不了模型好坏。4.2 优化器、学习率调度与梯度裁剪from torch.optim.lr_scheduler import OneCycleLR opt torch.optim.AdamW(model.parameters(), lr3e-4, weight_decay1e-2) sched OneCycleLR(opt, max_lr1e-3, total_stepsepochs * len(loader), pct_start0.1, anneal_strategycos) scaler torch.amp.GradScaler(cuda, enableduse_amp) for epoch in range(epochs): model.train() for xb, yb in loader: xb, yb xb.to(device, non_blockingTrue), yb.to(device, non_blockingTrue) opt.zero_grad(set_to_noneTrue) with torch.autocast(cuda, dtypetorch.float16, enableduse_amp): logits model(xb) loss criterion(logits, yb) scaler.scale(loss).backward() # 时序模型梯度容易炸先 unscale 再裁剪顺序反了裁剪的是缩放后的梯度 scaler.unscale_(opt) torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) scaler.step(opt); scaler.update(); sched.step()参数说明weight_decay1e-2是 AdamW 的解耦权重衰减对小样本时序模型正则化效果明显pct_start0.1表示前 10% 步数做 warmupLSTM 加注意力这种组合在冷启动阶段学习率太大会直接发散max_norm1.0是梯度范数上限超过就等比缩放。混合精度训练在显存吃紧时几乎是必选项能省下四成左右的显存。4.3 高频报错与指标异常的定位路径现象可能原因定位手段loss 卡在 0.69 附近不动输出层又加了 Sigmoid 再进 BCE检查 head 末尾有没有多写一层激活验证 PR-AUC 远高于测试窗口重叠导致信息泄漏打印各段数据的时间范围确认无交集训练 loss 抖动剧烈学习率过大或未做梯度裁剪每个 step 打印 grad norm 的分位数显存 OOMseq_len × batch 乘积过大减半 batch 或用梯度累积凑等效批量推理结果全为同一类scaler 用了测试集统计量确认 scaler 只在训练段 fit还有一类隐性问题是设备不一致模型在 GPU、标签张量在 CPU报错信息通常是Expected all tensors to be on the same device。建议在训练入口处统一做一次.to(device)不要指望框架自动搬。5. 推理侧提速与预警阈值的验证方法训练完只是半成品真正上线时推理延迟和阈值选择才是决定系统好不好用的两件事。先说延迟。行情推送是持续流式的单次推理必须压到几毫秒以内否则预警出来行情已经走完了。# 推理引擎inference_mode 比 no_grad 更彻底关掉版本计数和梯度相关元数据 torch.inference_mode() def predict_proba(model, loader, device): model.eval() out [] for xb, _ in loader: logits model(xb.to(device, non_blockingTrue)) out.append(torch.sigmoid(logits).float().cpu()) return torch.cat(out).numpy() # 固定输入形状后启用 compile常见能拿到 20%~40% 的吞吐提升 model torch.compile(model, modereduce-overhead)torch.compile的首次编译有几秒到几十秒的开销一定要在服务预热阶段完成别等到第一笔行情来了才开始编译。如果目标环境不方便装完整 PyTorch用torch.jit.trace导出 TorchScript或者导出 ONNX 交给onnxruntime跑延迟通常还能再降一截。半精度推理在 A 卡上收益明显但要先拿一批历史数据比对 fp32 和 fp16 的输出分布确认阈值含义没有漂移。阈值这一步比调模型更容易被忽视。很多人直接写if prob 0.5而在 1% 正样本比例下0.5 这个阈值几乎不可能被触发。import numpy as np from sklearn.metrics import roc_curve, average_precision_score probs predict_proba(model, val_loader, device) print(PR-AUC:, average_precision_score(y_val, probs)) # 按业务能承受的误报率上限反查阈值而不是拍脑袋定 0.5 budget 0.01 # 误报率红线1% fpr, tpr, thr roc_curve(y_val, probs) idx np.searchsorted(fpr, budget, sideright) - 1 idx max(idx, 0) print(f阈值{thr[idx]:.4f} 误报率{fpr[idx]:.4f} 召回{tpr[idx]:.4f})逻辑说明roc_curve返回的fpr单调不减用searchsorted找到刚好不超过误报预算的位置取那一点的阈值。这个阈值必须写在配置里并且按周或按波动率状态滚动重估——市场从低波动切到高波动时同一模型输出的概率分布整体会平移固定阈值会在两个方向上分别失效。验证环节我强烈建议做滚动回测而不是单次留出把测试段按交易日切成若干折每折只用该折之前的数据重估标准化参数和阈值然后在这折上算召回和误报。这样跑出来的曲线才接近线上表现。另外给预警加一层冷却时间比如同一标的在 60 秒内只报一次能把重复告警压掉大半同时不影响真正的事件召回。上线前最后一件小事把模型版本号、scaler 的均值方差、阈值、窗口长度这四项一起存进 checkpoint加载时一并校验避免换了模型忘了换阈值这种低级故障。本文还有配套的精品资源点击获取