AI量化炒股实战：从数据清洗到策略回测的技术实现

1次阅读

没有评论

共计 2754 个字符，预计需要花费 7 分钟才能阅读完成。

量化交易的核心挑战在于金融数据的独特性质。与传统数据集不同，金融数据具有两个致命特点：

非平稳性 ：股票价格的时间序列统计特性会随时间变化，导致传统机器学习模型在训练集表现良好，但在测试集迅速失效。
低信噪比 ：市场噪声远大于真实信号，尤其在高频交易中，有效信号可能只占数据方差的 1 -2%。

此外，高频交易对延迟极度敏感。从信号生成到订单执行的全链路延迟必须控制在微秒级，这对模型复杂度提出了硬性约束。

传统量化策略主要依赖统计套利，但存在明显局限：

线性假设强：基于均值回归或协整关系的策略难以捕捉市场突变
规则脆弱：参数固化后无法适应市场机制变化（如涨跌停规则调整）

深度学习方案的优势在于：

LSTM 网络 ：天然适合处理时序数据，通过门控机制记忆长期依赖
Attention 机制 ：动态聚焦关键时间点，例如财报发布前后的价格异常

实验表明，LSTM+Attention 组合在沪深 300 成分股预测中，夏普比率比传统 ARIMA 模型高 0.3-0.5。

import yfinance as yf
import pandas as pd

# 获取苹果公司 2015-2020 年日线数据
data = yf.download('AAPL', start='2015-01-01', end='2020-12-31')

# 异常值处理：剔除 3 倍标准差以外的数据
close = data['Close']
mean, std = close.mean(), close.std()
data = data[(close > mean - 3*std) & (close < mean + 3*std)]

# 缺失值填充：前向填充
data.ffill(inplace=True)

使用 TA-Lib 计算技术指标：

import talib

# 计算 MACD
data['macd'], data['signal'], _ = talib.MACD(data['Close'])

# 计算布林带
upper, middle, lower = talib.BBANDS(data['Close'])
data['bb_upper'] = upper
data['bb_lower'] = lower

import numpy as np

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

import torch
import torch.nn as nn

class LSTMModel(nn.Module):
    def __init__(self, input_size, hidden_size, num_layers):
        super().__init__()
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
        self.attention = nn.Sequential(nn.Linear(hidden_size, 1),
            nn.Softmax(dim=1)
        )
        self.dropout = nn.Dropout(0.2)
        self.fc = nn.Linear(hidden_size, 1)

    def forward(self, x):
        out, _ = self.lstm(x)
        weights = self.attention(out)
        context = torch.sum(weights * out, dim=1)
        context = self.dropout(context)
        return self.fc(context)

import backtrader as bt

class MLStrategy(bt.Strategy):
    params = (('window_size', 60), ('threshold', 0.02))

    def __init__(self):
        self.model = load_pretrained_model()
        self.data_window = []

    def next(self):
        self.data_window.append(self.datas[0].close[0])
        if len(self.data_window) > self.p.window_size:
            self.data_window.pop(0)
            prediction = self.model.predict(self.data_window)
            if prediction > self.p.threshold:
                self.buy()
            elif prediction < -self.p.threshold:
                self.sell()

滑点建模 ：在回测中添加 0.1% 的随机滑点
手续费计算 ：按成交金额的 0.025% 双向收取

cerebro.broker.set_slippage_perc(0.001)  # 滑点
cerebro.broker.setcommission(commission=0.00025)  # 手续费

每周计算预测分布与历史分布的 KL 散度：

from scipy.stats import entropy

kl_divergence = entropy(p_current, q_historical)
if kl_divergence > threshold:
    trigger_retrain()

使用时间序列交叉验证（TimeSeriesSplit）：

from sklearn.model_selection import TimeSeriesSplit

tscv = TimeSeriesSplit(n_splits=5)
for train_idx, test_idx in tscv.split(X):
    X_train, X_test = X[train_idx], X[test_idx]
    y_train, y_test = y[train_idx], y[test_idx]

网格搜索时优先调整关键参数：