AI量化交易选股入门：从数据清洗到策略回测的全流程实战

1次阅读

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

量化交易听起来高大上，但实际操作中会遇到不少坑。新手最容易忽略的三个问题是：市场噪声大、数据维度高、策略易过拟合。市场噪声大意味着价格波动常常没有规律可循；数据维度高则可能导致模型难以训练；策略过拟合则是回测表现很好，实盘却一塌糊涂。

优点：逻辑简单，易于理解，比如 MACD、RSI 等
缺点：适应性差，无法捕捉复杂市场模式

LSTM：适合处理时间序列数据，能捕捉长期依赖关系
XGBoost：处理结构化数据效果好，训练速度快

这里以 Tushare 为例，获取股票历史数据：

import tushare as ts

def get_stock_data(stock_code: str, start_date: str, end_date: str):
    """
    获取股票历史数据
    :param stock_code: 股票代码
    :param start_date: 开始日期
    :param end_date: 结束日期
    """pro = ts.pro_api(' 你的 token')
    df = pro.daily(ts_code=stock_code, start_date=start_date, end_date=end_date)
    return df

计算技术指标并标准化：

import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler

def calculate_technical_indicators(df: pd.DataFrame):
    """计算技术指标"""
    # 计算简单移动平均
    df['MA5'] = df['close'].rolling(5).mean()
    df['MA10'] = df['close'].rolling(10).mean()

    # 计算 RSI
    delta = df['close'].diff()
    gain = delta.where(delta > 0, 0)
    loss = -delta.where(delta < 0, 0)
    avg_gain = gain.rolling(14).mean()
    avg_loss = loss.rolling(14).mean()
    rs = avg_gain / avg_loss
    df['RSI'] = 100 - (100 / (1 + rs))

    return df

def normalize_features(df: pd.DataFrame):
    """特征标准化"""
    scaler = StandardScaler()
    features = df[['MA5', 'MA10', 'RSI']]
    df[['MA5', 'MA10', 'RSI']] = scaler.fit_transform(features)
    return df

使用 XGBoost 进行训练：

import xgboost as xgb
from sklearn.model_selection import TimeSeriesSplit

def train_model(X: pd.DataFrame, y: pd.Series):
    """训练 XGBoost 模型"""
    # 使用时序交叉验证
    tscv = TimeSeriesSplit(n_splits=5)

    model = xgb.XGBClassifier(
        objective='binary:logistic',
        n_estimators=100,
        max_depth=3,
        learning_rate=0.1
    )

    for train_index, test_index in tscv.split(X):
        X_train, X_test = X.iloc[train_index], X.iloc[test_index]
        y_train, y_test = y.iloc[train_index], y.iloc[test_index]

        model.fit(X_train, y_train)
        # 这里可以添加评估代码

    return model

使用 backtrader 进行回测：

import backtrader as bt

class MyStrategy(bt.Strategy):
    params = (('maperiod', 15),)

    def __init__(self):
        self.dataclose = self.datas[0].close
        self.sma = bt.indicators.SimpleMovingAverage(self.datas[0], period=self.params.maperiod
        )

    def next(self):
        if self.dataclose[0] > self.sma[0]:
            self.buy()
        elif self.dataclose[0] < self.sma[0]:
            self.sell()

# 创建回测引擎
cerebro = bt.Cerebro()
# 添加数据
data = bt.feeds.PandasData(dataname=df)
cerebro.adddata(data)
# 添加策略
cerebro.addstrategy(MyStrategy)
# 设置初始资金
cerebro.broker.setcash(100000.0)
# 运行回测
cerebro.run()
# 绘制结果
cerebro.plot()