AI量化交易实战：从数据预处理到策略回测的全流程解析

1次阅读

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

在量化交易中，金融数据处理是构建稳健策略的基础。然而，金融数据存在几个关键痛点，直接影响策略的最终表现。

非平稳性 ：金融时间序列往往呈现趋势性、周期性和突变性，统计特性随时间变化，传统模型假设的平稳性难以满足。
低信噪比 ：市场噪音远大于有效信号，尤其在短周期交易中，噪声可能完全掩盖真实规律。
幸存者偏差 ：回测时仅使用历史存活到现在的标的，忽略了已退市或表现极差的数据，导致策略评估失真。

ARIMA/GARCH：适用于平稳或可差分平稳的线性关系建模，计算效率高但难以捕捉非线性特征。
卡尔曼滤波 ：对状态空间模型有效，适合处理带噪声的观测序列，但需要手动定义状态转移方程。

LSTM：擅长捕捉长期时间依赖关系，适合处理高频 tick 数据或跨周期特征，但训练成本较高。
XGBoost：对特征工程要求较低，能自动处理非线性关系，在中小规模数据集上表现优异。

from pykalman import KalmanFilter
import numpy as np

def kalman_smoothing(prices: np.ndarray) -> np.ndarray:
    kf = KalmanFilter(transition_matrices=[1],
        observation_matrices=[1],
        initial_state_mean=prices[0],
        initial_state_covariance=1,
        observation_covariance=1,
        transition_covariance=0.01
    )
    state_means, _ = kf.filter(prices)
    return state_means

import talib
import pandas as pd

def create_features(df: pd.DataFrame) -> pd.DataFrame:
    df['rsi'] = talib.RSI(df['close'].values, timeperiod=14)
    df['macd'], _, _ = talib.MACD(df['close'].values)
    df['atr'] = talib.ATR(df['high'].values, 
        df['low'].values, 
        df['close'].values, 
        timeperiod=14
    )
    return df

from abc import ABC, abstractmethod

class BaseStrategy(ABC):
    @abstractmethod
    def generate_signal(self, data: pd.DataFrame) -> pd.Series:
        pass

class MeanReversionStrategy(BaseStrategy):
    def __init__(self, lookback: int = 20, threshold: float = 1.0):
        self.lookback = lookback
        self.threshold = threshold

    def generate_signal(self, data: pd.DataFrame) -> pd.Series:
        zscore = (data['close'] - data['close'].rolling(self.lookback).mean()) \
               / data['close'].rolling(self.lookback).std()
        return np.where(zscore > self.threshold, -1, 
                       np.where(zscore < -self.threshold, 1, 0))

import backtrader as bt

class MLStrategy(bt.Strategy):
    params = (('lookback', 20), ('threshold', 1.0))

    def __init__(self):
        self.signal = self.data0.signal  # 预计算好的信号列

    def next(self):
        if not self.position:
            if self.signal[0] > 0:
                self.buy(size=self.p.stake)
            elif self.signal[0] < 0:
                self.sell(size=self.p.stake)
        else:
            if (self.position.size > 0 and self.signal[0] < 0) or \
               (self.position.size < 0 and self.signal[0] > 0):
                self.close()

# 多周期回测设置               
cerebro = bt.Cerebro()
data = bt.feeds.PandasData(dataname=df)
resampled_data = data.resample(timeframe=bt.TimeFrame.Weeks)
cerebro.adddata(data)
cerebro.adddata(resampled_data)
cerebro.addstrategy(MLStrategy)
results = cerebro.run()