如何用神经网络赢得每个交易

让我们直奔主题。

我是 Roan,一名从事系统设计、HFT 风格执行和量化交易系统的后端开发者。我的工作重点是研究预测市场在负载下的实际表现。欢迎通过 DM 提出建议、进行有深度的合作或洽谈伙伴关系。

大多数交易者亏钱的原因与他们的市场观点无关。

他们对方向的判断往往比自己以为的更准确。问题在于,他们只依赖单一信号、单一指标或单一直觉。他们在没有概率框架的情况下做概率决策,而市场会反复且可靠地惩罚这种错误。

神经网络解决的是一个根本不同的问题。它不预测未来。它所做的是学习隐藏在历史数据中的期望函数——即当前可观察到的信息与市场统计上最可能下一步做什么之间的数学关系。它能同时处理数千个变量,这是人类肉眼无法看到、单一指标无法捕捉的。

最大的基金早已深谙此道。Two Sigma 同时运行超过 10,000 个实时信号通过机器学习模型。Citadel 的量化研究部门在股票、期权和宏观策略中部署深度学习架构。Renaissance Technologies 早在其他人理解他们在做什么的几十年前,就完全基于这个框架建立了 Medallion 基金。该基金在 30 年内年化回报 66%(扣费前)。这不是运气,而是统计学习在大规模上的系统性应用。

在这些公司构建这些模型的入门级量化研究员总薪酬在 35 万至 65 万美元之间。不是因为工作光鲜,而是因为用数学在金融数据中找到可靠优势,是应用科学中最难的未解问题之一。

如果你还没读过我之前那篇涵盖本文所有数学基础的完整量化路线图,请先从那里开始。

2025 年,量化金融领域的 AI 和机器学习招聘显著加速。每个主要的系统化基金都在构建由深度学习驱动的信号生成基础设施。然而,大多数尝试将神经网络融入自己策略的交易者,都会在同一个可预测的步骤上失败,原因也相同。

本文将精确告诉你这个步骤是什么,以及如何避免它。然后给出从数据准备到实时信号部署的完整实现框架。

读完本文,你将彻底理解:神经网络如何从数据中学习,以及它做出预测时实际在计算什么;为什么直接应用于价格图表必然失败,以及正确的输入应该是什么;哪种架构最适合序列市场数据及其原因;区分真实优势与自我欺骗的完整严谨训练框架;以及部署你的第一个神经网络交易信号的完整流水线。

注意:本文刻意写得较长。每一部分都建立在前一部分基础上。如果你认真想为交易增加真正的量化优势,请逐字阅读。如果你只想找捷径,这篇文章不适合你。

1、神经网络到底在计算什么

在构建任何东西之前,你必须理解神经网络训练时实际发生了什么。大多数解释跳过这一步直接给代码,这就是大多数实现失败的原因。

神经网络是一个参数化的复合函数。你输入一个向量,它应用一系列线性变换,中间穿插非线性激活函数,产生输出。其形式化结构为:

其中 W_i 是权重矩阵,b_i 是偏置向量,σ 是逐元素应用的非线性激活函数。这就是无论如何可视化,每一个神经网络底层的数学本质。

其强大之处在于训练过程。你有一个输入-输出对的数据集。网络从随机参数开始,做出预测,计算预测与实际观测输出的误差,然后向减少误差的方向调整参数,重复数百万次。

形式化目标是最小化损失函数。对于回归任务,标准选择是均方误差:

其中 θ 代表所有可学习参数,yᵢ 是实际观测值,ŷᵢ 是模型预测。优化过程通过计算损失对参数的梯度,然后沿负梯度方向迈出一小步:θ ← θ - α · ∇_θ L(θ)

其中 α 是学习率。这就是随机梯度下降。所有现代深度学习优化器都是在此基础上加入自适应步长、动量项或二阶近似等变体。

现在是改变你对神经网络产生的一切结果的解读的关键洞见。

当你训练网络最小化平方误差来匹配目标变量时,它学到了什么?它学到的是给定输入下该变量的条件期望。最小化 E[(Y - f(X))²] 的函数正是 E[Y|X]。这不是日常语义上的“预测”,而是数学上的期望——所有与观测输入一致的情景下的加权平均结果。

这个结论的证明很简单。展开平方损失:

第一项是不可约的方差。第二项通过将 f(X) 设置为 E[Y|X] 来最小化。在平方误差下的最优预测器就是条件均值。

这有深刻的实践意义。当你掷一万次公平骰子并训练神经网络最小化平方误差时,它会预测 3.5。骰子能落在 3.5 上吗?不能。但 3.5 是最小化期望平方误差的值。网络计算的是期望,而非下一次实现值。

为什么神经网络在学习这种条件期望上能优于更简单的模型?通用逼近定理。具有足够神经元的单隐藏层前馈网络可以在 Rⁿ 的紧致子集上以任意精度逼近任何连续函数。对于具有多层的深度网络,参数效率呈指数级提升。

这意味着:如果你的输入特征与目标变量之间存在任何光滑的数学关系,设计得当的神经网络就能找到它。问题从来不是网络能不能学到函数,而是你让它学的函数是否在时间上足够稳定,值得去学。

2、为什么价格预测必然失败,以及正确的做法

这是本文最重要的部分。这里描述的错误几乎每一个第一次尝试用神经网络的交易者都会犯。理解它为什么失败是后续一切有效方法的前提。

典型的失败模式是:取 500 天的收盘价,喂给 LSTM,让它预测第 501 天。

样本内,预测看起来平滑且接近实际价格。样本外,模型预测出几乎恒定或回归到近期均值的值,而价格却走向完全不同的方向。模型似乎什么有用的东西都没学到。

这不是模型失败,而是数据分布失败。

回顾第一部分:神经网络学习 E[Y|X],即给定 X 下 Y 的条件期望。只有当数据生成分布 P(Y|X) 在时间上稳定时,这个期望才有意义。如果分布发生漂移,模型从历史数据学到的条件期望就不再描述当前的期望。

形式上,这就是非平稳性问题。时间序列 {Xₜ} 如果其联合分布 P(Xₜ₁, Xₜ₂, ..., Xₜₙ) 对时间平移不变,则为平稳的。金融价格序列在最强的意义上是非平稳的。2008 年股票回报的分布与 2021 年的结构完全不同。均值、方差、自相关和尾部行为都会随 regime、流动性、波动率聚类和宏观条件而变化。

后果在任何价格预测神经网络的回测中都可见。基于 2015-2019 年数据训练的模型学到了那个 regime 的期望结构。当 2020 年分布发生漂移时,模型就在拟合一个移动的目标。它的预测总是以系统性的方式错误,表明它在对过去的分布建模,而非当前的。

解决方案是什么?

通过特征工程产生平稳或接近平稳的输入。目标变量本身也应构造为近似平稳。

对于输入,在不同市场 regime 中表现出合理平稳性的特征包括:

多窗口对数收益率:r_t = ln(P_t / P_{t-k}),k 取 {1, 5, 20}

波动率比率:σ_short / σ_long,其中每个 σ 是不同窗口的滚动实现波动率

经波动率归一化的动量信号:r_t / σ_t,即按实现风险缩放的回报

成交量 Z 分数:(V_t - μ_V) / σ_V,在滚动窗口上计算

基于价差的信号:买卖价差相对于其历史分布,或基于成交价与中间价的有效价差

Regime 指标:VWAP 偏差、距滚动高低点的距离、隐含与实现波动率比率

对于目标变量,不要预测下一期价格,而是预测下一期方向作为二分类问题:风险调整后回报是否为正?或者预测下一期回报相对于近期分布的 Z 分数。这两种目标都比原始价格甚至原始回报更稳定。

判断特征是否有用的实用检验:对其运行 Augmented Dickey-Fuller 检验。p 值低于 0.05 表明序列是平稳的。对于未通过检验的特征,先进行一阶差分或用滚动标准差归一化。

实现说明:在 Python 中,这看起来像:

from statsmodels.tsa.stattools import adfuller
import numpy as np

def check_stationarity(series, name):
    result = adfuller(series.dropna())
    print(f"{name}: ADF statistic={result[0]:.4f}, p-value={result[1]:.4f}")
    return result[1] < 0.05

# Example feature engineering
returns = df['close'].pct_change()
vol_20 = returns.rolling(20).std()
vol_5 = returns.rolling(5).std()

features = {
    'return_1d': returns,
    'return_5d': df['close'].pct_change(5),
    'vol_ratio': vol_5 / vol_20,
    'momentum': returns / vol_20,
    'volume_zscore': (df['volume'] - df['volume'].rolling(20).mean()) / df['volume'].rolling(20).std()
}

for name, series in features.items():
    is_stationary = check_stationarity(series, name)

每一个未通过平稳性检验的特征,都会让你的模型学到错误的分布。

3、适用于序列市场数据的 LSTM 架构

市场数据是序列性的。事件的顺序携带了快照无法提供的信息。过去五分钟发生的事通过自相关、动量、均值回归和微观结构动态与接下来五分钟发生的事相连。你使用的架构必须能够从这种序列结构中学习。

标准前馈网络将每个输入独立处理。当你向它呈现第 100 天的特征向量时,它对 1 到 99 天的情况一无所知。对于非序列数据(如第一部分中的扔足球例子),这没问题。但对于金融市场的时间序列数据,它丢掉了数据集中最重要的信息。

循环神经网络通过在时间上向前传递隐藏状态来解决这个问题。在每个时间步 t,隐藏状态 h_t 计算为:

隐藏状态编码了网络从第 1 到 t-1 步学到的一切,并与当前输入 x_t 结合。这就是记忆机制。但基础 RNN 存在梯度消失问题:梯度在时间上反向传播时指数级衰减,导致实际上无法学习超过几个时间步的依赖关系。

LSTM 用门控记忆架构解决了这个问题。LSTM 不仅维护隐藏状态 h_t,还维护细胞状态 c_t。细胞状态是长期记忆,隐藏状态是工作记忆。三个学习的门控制信息流:

遗忘门:f_t = σ(W_f · [h_{t-1}, x_t] + b_f) 该门决定要丢弃前一个细胞状态的多少比例。当市场在 regime 间转换时,遗忘门允许模型释放过时的模式并重置理解。

输入门:i_t = σ(W_i · [h_{t-1}, x_t] + b_i),g_t = tanh(W_g · [h_{t-1}, x_t] + b_g) 这些门决定要将什么新信息写入细胞状态。

输出门:o_t = σ(W_o · [h_{t-1}, x_t] + b_o),h_t = o_t ⊙ tanh(c_t) 该门决定从细胞状态输出什么。

细胞状态更新结合遗忘和新写入:c_t = f_t ⊙ c_{t-1} + i_t ⊙ g_t

这种架构允许 LSTM 在 50、100 或更多时间步的序列上维持相关上下文,而不会出现梯度消失问题。对于日频交易策略,这意味着模型能学到三周前特定的波动率压缩模式会预测今天某种特定的突破行为。没有基于指标的系统能捕捉到这种复杂度的关系。

PyTorch 中完整的 LSTM 实现:

import torch
import torch.nn as nn
import numpy as np
from sklearn.preprocessing import StandardScaler

class TradingLSTM(nn.Module):
    def __init__(self, input_size, hidden_size=64, num_layers=2, dropout=0.2):
        super(TradingLSTM, self).__init__()
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        
        self.lstm = nn.LSTM(
            input_size=input_size,
            hidden_size=hidden_size,
            num_layers=num_layers,
            dropout=dropout,
            batch_first=True
        )
        
        self.dropout = nn.Dropout(dropout)
        self.fc = nn.Linear(hidden_size, 1)
        self.sigmoid = nn.Sigmoid()
    
    def forward(self, x):
        h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size)
        c0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size)
        
        out, _ = self.lstm(x, (h0, c0))
        out = self.dropout(out[:, -1, :])
        out = self.fc(out)
        return self.sigmoid(out)

def prepare_sequences(features, target, lookback=10):
    X, y = [], []
    for i in range(len(features) - lookback):
        X.append(features[i:i+lookback])
        y.append(target[i+lookback])
    return np.array(X), np.array(y)

回溯期是一个关键超参数。对于日频策略,从 10 到 20 个交易日的窗口开始。对于基于 5 分钟 K 线的日内策略,24 个周期的回溯期覆盖两个交易小时的上下文。最优回溯期取决于你要捕捉的模式的时序尺度,必须通过验证性能经验确定。

4、不自欺欺人的训练方法

构建模型是容易的部分。危险的部分是评估它是否真的学到了真实的东西,而不是记住了训练数据中的噪声。

过拟合是金融领域机器学习的核心失败模式。过拟合的模型在训练数据中找到了无法泛化的模式。在训练数据上表现优异,在新数据上表现为随机水平或更差。关键问题是,过拟合的样本内表现看起来与真实优势完全相同。你无法仅通过训练指标来区分它们。

解决方案是使用特定时序顺序的严格三数据集划分。

训练集是梯度下降运行的地方。模型在训练期间会反复看到这些数据并据此调整参数。永远不要在这个集合上评估泛化性能。

验证集是模型从未在其上训练过、但你在训练期间持续监控的数据。每个 epoch 后,计算验证损失。当验证损失停止改善并开始上升,而训练损失继续下降时,你就找到了过拟合阈值。立即停止训练,并保存验证损失最低的 epoch 的模型权重。这就是早停,它是你对抗过拟合的主要防御。

测试集是模型以任何形式都未曾影响过的数据。你只使用它一次:在使用训练集和验证集完成所有架构决策、所有超参数选择和所有特征工程决策之后。测试集结果是你对实时表现的诚实估计。使用测试集做额外的设计决策会使其失效。

对于金融时间序列,这些划分必须是时序的。训练期在时间上最早,验证期紧随其后,测试期是最新的。随机化划分会允许未来信息污染训练,这是一种前瞻偏差,会让你的回测表现好于实时表现。

带早停的完整训练循环:

def train_model(model, train_loader, val_loader, epochs=100, lr=0.001, patience=10):
    optimizer = torch.optim.Adam(model.parameters(), lr=lr)
    criterion = nn.BCELoss()
    scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(
        optimizer, patience=5, factor=0.5
    )
    
    best_val_loss = float('inf')
    best_weights = None
    patience_counter = 0
    
    for epoch in range(epochs):
        model.train()
        train_loss = 0
        for X_batch, y_batch in train_loader:
            optimizer.zero_grad()
            predictions = model(X_batch).squeeze()
            loss = criterion(predictions, y_batch)
            loss.backward()
            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
            optimizer.step()
            train_loss += loss.item()
        
        model.eval()
        val_loss = 0
        with torch.no_grad():
            for X_batch, y_batch in val_loader:
                predictions = model(X_batch).squeeze()
                loss = criterion(predictions, y_batch)
                val_loss += loss.item()
        
        val_loss /= len(val_loader)
        scheduler.step(val_loss)
        
        if val_loss < best_val_loss:
            best_val_loss = val_loss
            best_weights = model.state_dict().copy()
            patience_counter = 0
        else:
            patience_counter += 1
            if patience_counter >= patience:
                print(f"Early stopping at epoch {epoch}")
                break
    
    model.load_state_dict(best_weights)
    return model

此实现中包含梯度裁剪,原因在于:当学习的序列较长时,LSTM 梯度在训练期间可能爆炸。将梯度范数裁剪到最大值 1.0 可以防止这种情况,而不会显著影响收敛速度。

针对交易的最严谨评估框架是 Walk-Forward Validation。而不是单一的训练-验证-测试划分,你将训练窗口在时间上向前滚动,训练,然后在紧随其后的期间测试,然后推进窗口并重复。所有窗口中拼接的样本外预测给你一个现实的估计,即如果在历史上每个时间点部署该模型,它会如何表现。

Walk-forward validation:

def walk_forward_validation(features, target, train_size, test_size, step):
    all_predictions = []
    all_actuals = []
    
    for start in range(0, len(features) - train_size - test_size, step):
        train_end = start + train_size
        test_end = train_end + test_size
        
        X_train = features[start:train_end]
        y_train = target[start:train_end]
        X_test = features[train_end:test_end]
        y_test = target[train_end:test_end]
        
        scaler = StandardScaler()
        X_train_scaled = scaler.fit_transform(X_train.reshape(-1, X_train.shape[-1]))
        X_test_scaled = scaler.transform(X_test.reshape(-1, X_test.shape[-1]))
        
        model = TradingLSTM(input_size=X_train.shape[-1])
        # train model here
        
        with torch.no_grad():
            preds = model(torch.FloatTensor(X_test_scaled))
        
        all_predictions.extend(preds.numpy())
        all_actuals.extend(y_test)
    
    return np.array(all_predictions), np.array(all_actuals)

在一个良好构建的模型和优质特征上,预期的方向准确率通常在 52% 到 57%。这听起来平平无奇。但一个 54% 的方向信号,配合夏普比率高于 1.0,在数百笔交易中持续应用,并使用正确的 Kelly 衍生仓位管理,长期复合后能超越大多数自由裁量交易者。优势在于一致性和规模,而非单个信号的幅度。

5、完整实现流水线与部署

本节将前四部分的所有内容组装成生产就绪的信号流水线。这是完整的端到端实现。

步骤 1:数据获取与清洗

使用 Polygondotio 获取机构级别的 tick 和 OHLCV 数据。学习阶段 yfinance 足够。在任何特征工程之前处理缺失数据、公司行动和拆股。

import yfinance as yf
import pandas as pd

ticker = yf.Ticker("SPY")
df = ticker.history(period="5y", interval="1d")
df = df[['Open', 'High', 'Low', 'Close', 'Volume']].copy()
df.dropna(inplace=True)

步骤 2:特征工程

构建你完整的平稳特征集。每个特征都必须通过 Augmented Dickey-Fuller 检验。

def build_features(df):
    features = pd.DataFrame(index=df.index)
    
    close = df['Close']
    volume = df['Volume']
    returns = close.pct_change()
    
    features['return_1d'] = returns
    features['return_5d'] = close.pct_change(5)
    features['return_20d'] = close.pct_change(20)
    
    vol_5 = returns.rolling(5).std()
    vol_20 = returns.rolling(20).std()
    features['vol_ratio'] = vol_5 / vol_20
    features['momentum_norm'] = returns / vol_20
    
    features['volume_zscore'] = (
        (volume - volume.rolling(20).mean()) / volume.rolling(20).std()
    )
    
    high_low_range = (df['High'] - df['Low']) / close
    features['range_norm'] = high_low_range / high_low_range.rolling(20).mean()
    
    sma_5 = close.rolling(5).mean()
    sma_20 = close.rolling(20).mean()
    features['sma_ratio'] = sma_5 / sma_20 - 1
    
    target = (returns.shift(-1) > 0).astype(int)
    
    features = features.dropna()
    target = target.loc[features.index]
    
    return features, target

步骤 3:时序划分与缩放

def prepare_data(features, target, lookback=10, train_ratio=0.6, val_ratio=0.2):
    n = len(features)
    train_end = int(n * train_ratio)
    val_end = int(n * (train_ratio + val_ratio))
    
    scaler = StandardScaler()
    features_scaled = scaler.fit_transform(features[:train_end])
    
    features_array = np.vstack([
        features_scaled,
        scaler.transform(features[train_end:val_end]),
        scaler.transform(features[val_end:])
    ])
    
    X, y = prepare_sequences(features_array, target.values, lookback)
    
    X_train = torch.FloatTensor(X[:train_end - lookback])
    y_train = torch.FloatTensor(y[:train_end - lookback])
    X_val = torch.FloatTensor(X[train_end - lookback:val_end - lookback])
    y_val = torch.FloatTensor(y[train_end - lookback:val_end - lookback])
    X_test = torch.FloatTensor(X[val_end - lookback:])
    y_test = torch.FloatTensor(y[val_end - lookback:])
    
    return X_train, y_train, X_val, y_val, X_test, y_test, scaler

步骤 4:模型评估

除了方向准确率之外,使用对交易有意义的指标评估模型。

from sklearn.metrics import accuracy_score, roc_auc_score
import pandas as pd

def evaluate_trading_signal(predictions, actuals, returns):
    pred_direction = (predictions > 0.5).astype(int)
    
    accuracy = accuracy_score(actuals, pred_direction)
    auc = roc_auc_score(actuals, predictions)
    
    signal = np.where(pred_direction == 1, 1, -1)
    strategy_returns = signal * returns
    
    sharpe = strategy_returns.mean() / strategy_returns.std() * np.sqrt(252)
    
    cumulative = (1 + strategy_returns).cumprod()
    rolling_max = cumulative.cummax()
    drawdown = (cumulative - rolling_max) / rolling_max
    max_drawdown = drawdown.min()
    
    print(f"Directional Accuracy: {accuracy:.4f}")
    print(f"AUC-ROC: {auc:.4f}")
    print(f"Annualized Sharpe: {sharpe:.4f}")
    print(f"Maximum Drawdown: {max_drawdown:.4f}")
    
    return accuracy, sharpe, max_drawdown

步骤 5:信号到仓位大小

神经网络信号本身不是交易策略。你需要仓位管理。Kelly 分数将模型的优势转化为最优下注大小:

f* = (p × b - q) / b

其中 p 是你的方向准确率,q = 1 - p,b 是平均盈亏比。实践中使用半 Kelly 以考虑参数估计误差。无论 Kelly 建议如何,单一信号的风险都不要超过资本的 2%。

步骤 6:持续监控与重训

基于历史数据训练的模型在部署的那一刻就开始退化。市场 regime 发生变化。模型学到的条件期望会逐渐成为当前期望的更差近似。你必须实现监控系统,跟踪实时表现与验证基线的对比,并在分布漂移超出可接受阈值时触发重训。

标准指标是 Kolmogorov-Smirnov 统计量,比较你最近的实时预测分布与历史验证分布。当 KS 统计量超过 0.1 时,在最新数据上重训并重新评估后再继续部署。

对于预测市场和加密货币,regime 漂移比股票市场更频繁且更剧烈。在滚动 90 天窗口上重训,每 30 天进行一次 walk-forward 重新评估,是一个合理的基准。

6、结束语

神经网络不会给你水晶球。它给你的是一个系统性、数学严谨的框架,用于从历史数据中提取市场行为的条件期望。当特征平稳、架构适合序列数据、训练严谨且经过正确验证、信号配合恰当的仓位管理时,结果就是一致、可扩展、可复制的优势。

Two Sigma 同时运行 10,000 个信号通过这个框架。Renaissance 基于它建立了金融史上回报最高的基金。实现它的入门级研究员年薪 65 万美元。

完整的实现就在本文中。数学是可以学习的。代码可以在一个周末内构建。区分系统化基金与其他人的唯一区别,就是知道正确的框架并有不走捷径的纪律。

现在你知道了这个框架。

这里有一个问题,我想让你思考。

一个基于你当前指标集训练的神经网络将学习给定这些指标的市场回报的条件期望。但条件期望的好坏取决于你用来做条件的特征。如果你必须给模型增加一个你认为其他系统化交易者没在用的新特征,它会是什么?为什么?


原文链接:How to Use Neural Networks to Win Every Trade Before It Even Starts

汇智网翻译整理,转载请标明出处