这次我们来看一个在量化研究里被反复讨论的问题：金融时序预测中，Transformer 模型能否真正超越 LSTM。

这个问题的答案并不只是“Transformer 更强”这么简单。金融数据本身噪声大、非平稳、样本量有限，LSTM 作为循环网络在这类数据上统治了很多年。而 Transformer 靠全局注意力机制，天然适合捕捉长距离依赖和多变量之间的交叉影响。近几年的大量实验表明，在合适的特征工程和训练策略下，Transformer 在金融预测任务上的表现确实可以超过 LSTM，尤其在中长期趋势预测和多标的联合建模场景中。

这篇文章会从模型原理对比讲起，然后给出一套完整的 Transformer 金融时序预测方案的搭建思路，包括数据预处理、模型实现、训练验证、批量预测和 API 服务封装。如果你正在做股票价格预测、收益率预测、宏观经济指标预测，或者想给自己的量化策略加一个更现代的基线模型，这篇文章可以直接参考。

1. 核心能力速览

能力项	说明
模型类型	Transformer Encoder 时序回归模型，可扩展为多变量输入
对比基准	LSTM、BiLSTM、ARIMA 等传统时序模型
核心优势	全局注意力机制、并行计算、长序列依赖建模、多变量交叉特征提取
支持任务	单变量时序预测、多变量时序预测、多步预测、分类/回归
硬件要求	训练端建议 NVIDIA GPU，显存 6G 以上；推理端 CPU 即可
开发框架	PyTorch 2.x + pandas + numpy + scikit-learn
数据要求	标准化后的数值型时序数据，建议样本量在数千条以上
启动方式	命令行训练脚本 + Python 推理脚本 + FastAPI 接口服务
是否支持批量任务	支持，可通过批量预测脚本或 API 批量提交
是否支持 API	支持，提供 POST /predict 接口
适合场景	股票/基金/商品价格预测、收益率预测、波动率建模、宏观经济指标预测

这里要提前说明：Transformer 并不是一个“开箱即用就能赚钱”的工具。它的优势在于特征提取能力，但金融预测的最终效果高度依赖数据质量、特征工程、训练策略和评估方式。后面每个环节都会提到。

2. 为什么金融预测需要 Transformer 而不是继续用 LSTM

2.1 LSTM 的核心瓶颈

LSTM 通过门控机制解决长序列训练中的梯度消失问题，但它本质上还是逐步递推的结构。当前时间步的计算依赖前一步的隐状态，这意味着：

• 训练无法完全并行，序列越长训练越慢。
• 长期依赖信息在经过多步传递后仍会被稀释，虽然比 RNN 好，但依然有上限。
• 对多变量之间复杂的交叉依赖关系，LSTM 需要靠堆叠层数和增加隐单元来隐式建模，效率不高。

在金融数据中，不同资产、不同因子之间往往存在滞后相关性。比如美债收益率变化可能滞后影响 A 股成长股估值，这个滞后期可能是不固定的。LSTM 对这种“不定长滞后交叉影响”的建模能力有限。

2.2 Transformer 的注意力机制解决了什么

Transformer 的核心是自注意力机制。给定输入序列，模型会计算序列中任意两个位置之间的注意力权重。这带来几个直接优势：

• 任意距离的依赖都能直接建模，不需要逐层传递。
• 所有位置可以并行计算，训练速度明显快于 LSTM。
• 可以同时关注多个维度的特征，比如开盘价、成交量、技术指标和外盘数据之间的交互关系。
• 位置编码为模型提供了时间顺序信息，不依赖递归结构。

以金融预测为例，假设我们要用过去 60 个交易日的数据预测未来 5 天的走势。对于 LSTM，第 60 天的信息需要经过 60 步更新才能影响输出，虽然门控机制让信息衰减没那么严重，但终究有损耗。而 Transformer 直接计算第 1 天和第 60 天之间的注意力权重，可以精准捕捉“60 天前的某个宏观事件仍然在影响当前走势”这种长周期规律。

2.3 实验对比中的常见结论

从材料中可以看到，Transformer 在时间序列预测任务上的表现并不是绝对优于所有模型，但在以下情况下通常能体现出优势：

• 序列长度较长（一般超过 50 个时间步）。
• 多变量之间存在复杂的交叉相关性。
• 需要同时预测多个标的或多个目标字段。
• 训练数据量相对充足（数千条以上）。

在短序列、小样本、强噪声场景下，LSTM 甚至 ARIMA 仍然可能是更稳的选择。这提醒我们：选模型不是追新，而是看数据和任务形态。更稳妥的判断是，将 Transformer 作为一个更强的基线，与 LSTM、XGBoost 等模型做对比实验，而不是默认它一定更好。

3. 金融数据预处理与特征工程

金融时序预测中，数据预处理的质量往往比模型结构更重要。下面给出一个通用处理流程。

3.1 数据来源与基础清洗

建议使用 pandas 加载数据，基础字段至少包括日期、开盘价、最高价、最低价、收盘价、成交量。数据清洗包括：

• 去除重复日期的记录。
• 处理停牌导致的缺失值，可以选择前向填充或删除。
• 剔除明显异常的交易数据（比如成交量骤变为 0 且非停牌日）。
• 统一时区和交易日历。

import pandas as pd

df = pd.read_csv("stock_data.csv", parse_dates=["date"])
df = df.drop_duplicates(subset="date")
df = df.sort_values("date").reset_index(drop=True)
df = df.fillna(method="ffill")

3.2 特征构造

原始 OHLCV 数据直接输入模型往往效果一般，建议构造以下特征：

• 收益率序列： close.pct_change()
• 对数收益率： np.log(close / close.shift(1))
• 技术指标：移动平均线、布林带、RSI、MACD
• 市场情绪特征：成交额变化率、换手率
• 外部特征：指数涨跌、行业板块涨跌、宏观利率变化

需要特别注意的是：所有特征必须在时间序列上“向前看”以避免未来函数。也就是说，构造第 t 天的特征时，只能用第 t 天及之前的数据。

3.3 数据标准化与序列化

金融数据数值范围差异很大，建议对每个特征分别做标准化。标准化器必须只在训练集上拟合，然后用同一套参数转换验证集和测试集，否则会造成数据泄漏。

from sklearn.preprocessing import StandardScaler

feature_cols = ["open", "high", "low", "close", "volume", "pct_change", "ma5", "ma20"]
scaler = StandardScaler()
scaled = scaler.fit_transform(df[feature_cols])

然后构造滑窗样本。常用配置是：用过去 60 天的数据预测未来 5 天的收益率方向或数值。为了让代码可复现，需要固定随机种子。

import numpy as np

def create_sequences(data, seq_len=60, pred_len=5):
    X, y = [], []
    for i in range(len(data) - seq_len - pred_len + 1):
        X.append(data[i : i + seq_len])
        y.append(data[i + seq_len : i + seq_len + pred_len, target_idx])
    return np.array(X), np.array(y)

4. 环境准备与依赖安装

4.1 推荐环境

• 操作系统：Windows 10/11、Ubuntu 20.04 及以上
• Python 版本：3.9 到 3.11
• 深度学习框架：PyTorch 2.x
• GPU：建议 NVIDIA 显卡，显存 6G 以上
• CPU 推理：支持，但批量预测时速度会有明显差距
• 磁盘空间：数据量不大时，预留 10G 以上足够

4.2 创建虚拟环境

conda create -n fin_transformer python=3.10
conda activate fin_transformer

4.3 安装依赖

pip install torch --index-url https://download.pytorch.org/whl/cu118
pip install pandas numpy scikit-learn matplotlib tqdm fastapi uvicorn

如果机器没有 NVIDIA GPU，可以安装 CPU 版 PyTorch：

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

注意：实际安装命令需要根据本机 CUDA 版本和 PyTorch 官方索引调整，建议先访问官方页面确认最新的安装命令。

5. Transformer 金融预测模型设计与实现

5.1 模型结构

整个模型由以下几部分组成：

• 输入嵌入层：将每个时间步的特征向量映射到 d_model 维。
• 位置编码：为输入添加时间顺序信息。
• TransformerEncoder：堆叠多层编码器，每层包含多头自注意力机制和前馈网络。
• 输出层：将最终编码结果映射为预测目标。

代码实现如下：

import torch
import torch.nn as nn
import math

class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len=512):
        super().__init__()
        pe = torch.zeros(max_len, d_model)
        position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
        div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))
        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)
        self.register_buffer("pe", pe.unsqueeze(0))

    def forward(self, x):
        return x + self.pe[:, : x.size(1)]

class TimeSeriesTransformer(nn.Module):
    def __init__(self, input_dim, d_model=128, nhead=8, num_layers=4, pred_len=5, dropout=0.1):
        super().__init__()
        self.input_proj = nn.Linear(input_dim, d_model)
        self.pos_encoder = PositionalEncoding(d_model)
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=d_model,
            nhead=nhead,
            dim_feedforward=512,
            dropout=dropout,
            batch_first=True,
        )
        self.transformer_encoder = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)
        self.decoder = nn.Sequential(
            nn.Linear(d_model, 64),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(64, pred_len),
        )

    def forward(self, x):
        # x: [batch, seq_len, input_dim]
        x = self.input_proj(x)
        x = self.pos_encoder(x)
        x = self.transformer_encoder(x)
        # 取最后一个时间步的编码结果
        out = self.decoder(x[:, -1, :])
        return out

5.2 训练配置

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = TimeSeriesTransformer(
    input_dim=feature_cols_num,
    d_model=128,
    nhead=8,
    num_layers=4,
    pred_len=5,
    dropout=0.1,
).to(device)

criterion = nn.MSELoss()
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4, weight_decay=1e-5)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=50)

训练时建议使用早停机制，防止过拟合。金融数据噪声大，模型很容易在训练集上拟合得很好但在测试集上失效。

from tqdm import tqdm

for epoch in range(100):
    model.train()
    train_loss = 0.0
    for X_batch, y_batch in train_loader:
        X_batch = X_batch.float().to(device)
        y_batch = y_batch.float().to(device)

        optimizer.zero_grad()
        pred = model(X_batch)
        loss = criterion(pred, y_batch)
        loss.backward()
        torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
        optimizer.step()
        train_loss += loss.item() * len(X_batch)

    val_loss = evaluate(model, val_loader, criterion, device)
    scheduler.step()
    print(f"Epoch {epoch + 1}: train_loss={train_loss / len(train_loader.dataset):.4f}, val_loss={val_loss:.4f}")

6. 模型训练与效果验证

6.1 训练集 / 验证集 / 测试集划分

金融时序数据不能随机打乱划分。必须按时间顺序切分，例如前 70% 训练，中间 15% 验证，最后 15% 测试。这是金融预测与图像分类任务最大的不同点。

train_size = int(len(sequences) * 0.7)
val_size = int(len(sequences) * 0.15)
train_sequences = sequences[:train_size]
val_sequences = sequences[train_size : train_size + val_size]
test_sequences = sequences[train_size + val_size:]

6.2 评价指标

常用的评价指标包括：

• MSE / RMSE：衡量预测值与真实值之间的平均误差，数值越小越好。
• MAE：平均绝对误差，对异常值更稳健。
• 方向准确率：预测涨跌方向与真实涨跌方向一致的比例，对交易策略更直观。
• IC / Rank IC：信息系数，衡量预测值与未来收益的相关性。

from sklearn.metrics import mean_squared_error, mean_absolute_error

def evaluate_direction(y_true, y_pred):
    true_direction = np.diff(y_true, axis=0)
    pred_direction = np.diff(y_pred, axis=0)
    return np.mean(np.sign(true_direction) == np.sign(pred_direction))

6.3 Transformer vs LSTM 对比实验设计

要验证“Transformer 超越 LSTM”，必须保证两组实验使用完全相同的数据、特征和评估方式。推荐做法是把训练脚本抽成同一个框架，只替换模型类。

models = {
    "transformer": TimeSeriesTransformer(...),
    "lstm": LSTMModel(...),
}

results = {}
for name, model in models.items():
    print(f"Training {name}...")
    history = train_model(model, train_loader, val_loader, device)
    test_pred = predict(model, test_loader, device)
    results[name] = {
        "mse": mean_squared_error(test_y, test_pred),
        "mae": mean_absolute_error(test_y, test_pred),
        "direction_acc": evaluate_direction(test_y, test_pred),
    }

for name, metrics in results.items():
    print(f"{name}: {metrics}")

实际实验时建议记录以下信息：

• 训练轮数和最终损失
• 验证集最优 epoch
• 测试集 RMSE / MAE / 方向准确率
• 单次训练耗时和显存占用
• 推理耗时

6.4 判断训练是否成功的标准

• 训练损失和验证损失都持续下降，没有出现明显 divergence。
• 验证集损失不再下降时触发早停，模型进入收敛状态。
• 测试集方向准确率显著高于 0.5，如果是回归任务则 RMSE 低于基线模型（如直接用历史均值预测）。
• 如果训练损失下降但验证损失大幅回升，说明过拟合，需要增加 dropout、减小模型容量或增加数据量。

7. 批量预测与 API 接口封装

训练完成后的模型可以用于两个主要场景：离线批量预测和在线接口服务。

7.1 保存模型

torch.save({
    "model_state_dict": model.state_dict(),
    "feature_cols": feature_cols,
    "scaler": scaler,
    "config": {
        "input_dim": feature_cols_num,
        "d_model": 128,
        "nhead": 8,
        "num_layers": 4,
        "pred_len": 5,
    },
}, "transformer_finance.pt")

7.2 批量预测脚本

批量预测适合处理多只股票、多个时间窗口的预测任务。输入目录下放多只股票的数据文件，输出目录保存各自的预测结果。

python batch_predict.py --input-dir ./data/stocks --output-dir ./output/predictions --model ./checkpoints/transformer_finance.pt

batch_predict.py 核心逻辑：

import os
import glob
import pandas as pd

model, scaler, config = load_model("checkpoints/transformer_finance.pt")

input_files = glob.glob(os.path.join(args.input_dir, "*.csv"))
for file in input_files:
    df = pd.read_csv(file)
    processed = preprocess(df, scaler, config["feature_cols"])
    pred = model.predict(processed)
    output_path = os.path.join(args.output_dir, os.path.basename(file).replace(".csv", "_pred.csv"))
    pd.DataFrame(pred, columns=[f"pred_{i+1}" for i in range(config["pred_len"])]).to_csv(output_path, index=False)

批量任务需要注意：

• 每只股票单独标准化，不能混用其他股票的统计量，除非有明确依据。
• 输出文件建议带上生成时间戳，便于后续回测和复盘。
• 增加日志记录，逐文件打印成功或失败原因，避免中途卡住无法定位。

7.3 FastAPI 接口服务

接口调用是生产环境最常见的集成方式。使用 FastAPI 封装预测服务：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class PredictRequest(BaseModel):
    features: list  # 标准化后的特征序列

@app.post("/predict")
def predict(data: PredictRequest):
    if len(data.features) < 60:
        raise HTTPException(status_code=400, detail="Feature sequence too short")
    X = torch.tensor([data.features], dtype=torch.float32).to(device)
    with torch.no_grad():
        pred = model(X)
    return {"prediction": pred.cpu().tolist()}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

启动接口服务：

python api_server.py

调用接口：

import requests

payload = {
    "features": [
        [1.2, 0.5, 10.3, 20.1],
        [1.3, 0.6, 10.5, 20.3],
    ]
}
response = requests.post("http://127.0.0.1:8000/predict", json=payload, timeout=30)
print(response.json())

接口服务上线时要加访问限制，最好只绑定内网地址，并通过网关鉴权。如果担心并发压力，可以把模型加载放到全局变量中，避免每个请求都重新加载模型。

8. 资源占用与性能观察

8.1 显存占用观察方法

训练时可通过以下方式观察显存占用：

nvidia-smi -l 2

如果使用 PyTorch，也可以在训练脚本中加入显存日志：

if torch.cuda.is_available():
    print(f"CUDA memory allocated: {torch.cuda.memory_reserved() / 1024**3:.2f} GB")

显存占用取决于模型维度、层数、批次大小和序列长度。实际占用需以本机测试为准，无法给出固定数字。

8.2 影响资源消耗的关键因素

• 序列长度：序列越长，注意力矩阵越大，显存和计算量呈平方增长。
• 批次大小：批次越大，梯度估计越稳定，但显存占用同步上升。
• d_model 和 num_layers：模型容量越大，参数量越多，训练越慢。
• 预测长度：影响输出层和损失计算，但相对影响较小。

8.3 降低资源消耗的方法

• 减少序列长度，例如从 120 天降到 60 天。
• 减小 d_model，从 128 降到 64。
• 使用梯度累积模拟更大的批次。
• 推理时开启 torch.no_grad() 并切换到 eval 模式。
• CPU 推理时限制线程数，避免资源被全部占满。

torch.set_num_threads(4)
model.eval()
with torch.no_grad():
    pred = model(X)

8.4 长时间训练时的稳定性

金融数据训练通常需要多次实验调参。建议：

• 固定随机种子，保证实验可复现。
• 定期保存 checkpoint 和最优模型。
• 记录每次实验的配置和指标，用 CSV 或 MLflow 保存。
• 如果训练崩溃，优先检查数据中是否存在 NaN 或 Inf。

9. 常见问题与排查方法

问题现象	可能原因	排查方式	解决方案
训练损失不下降	学习率过大或过小	打印每轮损失变化	调整学习率，改用 warmup 策略
验证集损失大幅上升	过拟合	对比训练集和验证集损失	增大 dropout、降低模型容量、早停
预测结果全部接近同一个值	数据标准化不当或模型容量不足	查看预测值分布	检查 scaler 拟合方式，增加模型层数
注意力矩阵显存溢出	序列过长或批次过大	查看报错信息和显存占用	减小批次或序列长度，开启梯度累积
方向准确率接近 0.5	数据噪声大或特征信息不足	检查训练集方向准确率	增加特征、改用分类任务、尝试多模型集成
API 请求超时	推理耗时过长或并发过高	查看服务日志和耗时	使用异步接口、增加批处理、模型量化
数据中含有 NaN	除零、缺失值未处理	检查数据摘要	填充缺失值或删除对应样本
不同股票预测效果差异大	各标的波动特征不同	分组评估指标	对不同标的分开训练或加入标的特征

10. 最佳实践与合规使用建议

10.1 工程化建议

• 第一次训练先跑小参数，确认数据管线和训练流程没问题再放大模型。
• 保留一套最小可运行配置，方便快速复现问题。
• 模型文件、训练日志、预测结果分目录管理，文件名带时间戳。
• 批量任务必须加日志和失败重试，单文件失败不应中断整个任务。
• 模型服务要限制访问范围，绑定内网地址并加鉴权。

10.2 金融预测与合规边界

金融预测模型的结果不能作为投资决策的唯一依据。以下几点需要特别强调：

• 本文所有模型和代码仅用于技术研究和教育目的，不构成任何投资建议。
• 如果使用真实股票数据，确保数据来源合法，并且符合数据使用协议。
• 模型预测存在很大的不确定性，特别是在市场极端行情下，预测结果可能完全失效。
• 如果模型用于实盘策略，必须经过充分的回测、样本外验证和模拟交易测试。
• 涉及他人数据或非公开数据时，必须确认获得了合法授权。

10.3 模型之外的关键环节

Transformer 在金融预测中能否超越 LSTM，取决于很多模型之外的因素：

• 训练数据的质量和长度。
• 特征工程的有效性。
• 预测目标的定义是回归还是分类。
• 评估指标是否真实反映策略价值。
• 是否做了样本外测试和稳健性检验。

模型结构只是整个预测系统中的一个环节。即便是最先进的 Transformer，面对强噪声、低信噪比的金融数据，也需要配合严格的数据处理、测试和风控流程才能落地上线。

11. 总结与下一步

Transformer 在金融时序预测中的价值，主要体现在长序列建模、多变量交叉特征捕捉和并行训练效率上。和 LSTM 相比，它在合适的条件下确实能取得更好的预测效果，但这种优势不是无条件的。

如果现在要开始实践，建议按以下顺序推进：

1. 先准备一份干净的日线数据，做好缺失值处理和特征工程。
2. 分别实现 Transformer 和 LSTM 基线模型，固定相同的训练配置。
3. 在测试集上对比 RMSE、MAE 和方向准确率。
4. 确认 Transformer 有优势后，再考虑调参、加入更多特征或拓展到多标的预测。
5. 最后再考虑封装 API 服务和批量预测流程。

最容易踩的坑有两个：

• 没有按时间顺序划分数据集，导致数据泄漏，测试结果虚高。
• 忽略金融数据的强噪声特性，模型过拟合到训练集噪声上，实盘完全失效。

后续可以继续扩展的方向包括：

• 用 Transformer 做多步预测时添加误差修正机制。
• 在注意力机制中加入交易日期和宏观经济事件的先验信息。
• 对比 Transformer 与 XGBoost、LightGBM 等树模型在不同时间尺度上的表现。
• 将预测结果接入回测框架，评估真实策略收益和风险指标。

建议把整套训练、评估和部署流程沉淀成模板，后续换数据、换标的时可以快速复用。先用一种资产验证流程，再横向扩展到更多资产，这是最稳妥的推进方式。