PTrade 机器学习：基于 XGBoost 的股价趋势预测策略

在 PTrade 中使用 XGBoost 进行股价预测是一个典型的量化机器学习流程。由于 PTrade 内置了 xgboost（版本 0.6a2）、pandas 和 scikit-learn 等库，我们可以直接在策略中调用这些库进行模型训练和预测。

核心思路

1. 数据获取：使用 get_history 获取历史 K 线数据。
2. 特征工程：构造技术指标（如均线、收益率、波动率）作为模型输入（Features, X）。
3. 数据标注：将“未来一天的涨跌”作为标签（Label, Y）。如果明日收盘价 > 今日收盘价，标记为 1，否则为 0。
4. 模型训练：在 before_trading_start 中，使用历史数据训练 XGBClassifier。
5. 预测与交易：在 handle_data 中，提取当前特征输入模型，根据预测结果（涨/跌）执行买卖操作。

策略代码实现

以下是一个完整的 PTrade 策略示例。该策略选取单只股票（如贵州茅台），每日盘前利用过去 200 天的数据训练 XGBoost 模型，预测当日涨跌并进行交易。

import pandas as pd
import numpy as np
from xgboost import XGBClassifier

def initialize(context):
    """
    初始化函数
    """
    # 设定要交易的股票，这里以贵州茅台为例
    g.security = '600519.SS'
    set_universe(g.security)
    
    # 设定训练数据窗口长度（过去多少天的数据用于训练）
    g.train_window = 200
    
    # 全局模型变量
    g.model = None
    
    # 特征列表
    g.features = ['close_pct', 'ma5_ratio', 'ma10_ratio', 'volatility']

def get_features(df):
    """
    特征工程函数：计算技术指标
    输入：包含基础行情的DataFrame
    输出：包含特征列的DataFrame
    """
    df = df.copy()
    
    # 1. 每日收益率
    df['close_pct'] = df['close'].pct_change()
    
    # 2. 均线偏离度 (收盘价 / N日均线 - 1)
    df['ma5'] = df['close'].rolling(window=5).mean()
    df['ma10'] = df['close'].rolling(window=10).mean()
    df['ma5_ratio'] = df['close'] / df['ma5'] - 1
    df['ma10_ratio'] = df['close'] / df['ma10'] - 1
    
    # 3. 波动率 (5日收益率标准差)
    df['volatility'] = df['close_pct'].rolling(window=5).std()
    
    return df

def before_trading_start(context, data):
    """
    盘前处理：获取数据、特征工程、训练模型
    """
    # 获取历史数据：训练窗口 + 额外数据用于计算指标(如MA10需要前10天)
    lookback = g.train_window + 20
    
    # 获取历史行情 (不包含今天，防止未来函数)
    history_df = get_history(lookback, '1d', ['close', 'high', 'low', 'volume'], g.security, fq='pre', include=False)
    
    if len(history_df) < lookback:
        log.info("历史数据不足，跳过训练")
        g.model = None
        return

    # --- 1. 特征工程 ---
    data_df = get_features(history_df)
    
    # --- 2. 构造标签 (Label) ---
    # 我们要预测的是“下一天”的涨跌，所以 Label 是将收益率 shift(-1)
    # 如果下一天收益率 > 0，Label 为 1，否则为 0
    data_df['target'] = np.where(data_df['close_pct'].shift(-1) > 0, 1, 0)
    
    # 去除包含 NaN 的行 (因为计算 MA 和 shift 会产生 NaN)
    data_df.dropna(inplace=True)
    
    # 准备训练集
    X = data_df[g.features]
    y = data_df['target']
    
    # --- 3. 模型训练 ---
    # 初始化 XGBoost 分类器
    # 注意：PTrade 中的 xgboost 版本较老，参数尽量保持基础
    model = XGBClassifier(
        n_estimators=30,     # 树的数量
        max_depth=3,         # 树的深度，防止过拟合
        learning_rate=0.1,   # 学习率
        objective='binary:logistic'
    )
    
    try:
        model.fit(X, y)
        g.model = model
        log.info("XGBoost 模型训练完成")
    except Exception as e:
        log.error("模型训练失败: %s" % str(e))
        g.model = None

def handle_data(context, data):
    """
    盘中处理：获取当前特征、预测、交易
    """
    if g.model is None:
        return

    security = g.security
    
    # 获取最近的数据用于构建当天的特征输入
    # 我们需要足够的数据来计算 MA10 等指标
    recent_data = get_history(20, '1d', ['close', 'high', 'low', 'volume'], security, fq='pre', include=True)
    
    if len(recent_data) < 20:
        return

    # 计算特征
    current_df = get_features(recent_data)
    
    # 取最后一行（即当前最新的状态）作为预测输入
    # 注意：这里不需要 shift，因为我们是用“当前”的特征去预测“未来”
    current_X = current_df.iloc[[-1]][g.features]
    
    # 检查是否有 NaN (例如刚开盘可能导致某些指标无法计算)
    if current_X.isnull().values.any():
        log.info("特征数据包含空值，跳过预测")
        return

    # --- 预测 ---
    prediction = g.model.predict(current_X)[0]
    
    # 获取当前持仓和资金
    position = get_position(security).amount
    cash = context.portfolio.cash
    current_price = data[security]['close']
    
    # --- 交易逻辑 ---
    # 预测为 1 (上涨) 且空仓 -> 买入
    if prediction == 1 and position == 0:
        log.info("模型预测上涨，买入")
        order_value(security, cash)
        
    # 预测为 0 (下跌) 且持仓 -> 卖出
    elif prediction == 0 and position > 0:
        log.info("模型预测下跌，卖出")
        order_target(security, 0)

关键点解析

1. 特征与标签的错位（Shift）：

   • 在训练阶段（before_trading_start），我们需要用 $T$ 日的指标去预测 $T+1$ 日的涨跌。因此，在构造 target（标签）时，使用了 .shift(-1)。这意味着我们将“明天的结果”对齐到了“今天的数据”上作为训练目标。
   • 在预测阶段（handle_data），我们使用 $T$ 日（即当下）的实时指标，输入模型，模型会输出它认为的 $T+1$ 日（或当日剩余时间）的方向。

2. 防止未来函数：

   • 在 before_trading_start 中获取训练数据时，get_history 的 include 参数设为 False。这确保我们只使用昨天收盘及之前的数据来训练模型，模拟真实的盘前决策环境。

3. Python 版本兼容性：

   • PTrade 的 Python 环境通常基于 Python 3.5 或类似版本。代码中避免了 f-string（如 f"{var}"）等新语法，使用了 %s 格式化字符串，并使用了标准的 Pandas/Numpy 操作以确保兼容性。

4. XGBoost 参数：

   • n_estimators（树的数量）和 max_depth（树深）设置得较小（30 和 3），这是为了防止过拟合，同时也为了保证在回测和实盘中的运行速度。

5. 数据清洗：

   • 计算移动平均线（MA）和差分（pct_change）会产生 NaN 值。在送入模型训练前，必须使用 dropna() 清洗数据，否则 XGBoost 可能会报错或产生不可预期的结果。

常见问题 (Q&A)

Q: 为什么不在 initialize 里训练模型？
A: 在回测中，initialize 只在开始时运行一次。如果只在那时训练，模型将永远无法学习到最新的市场变化（即“概念漂移”）。在 before_trading_start 中每日（或定期）滚动训练，可以让模型适应最新的市场状态。

Q: 可以增加更多特征吗？
A: 可以。你可以利用 get_history 获取 high, low, volume 等字段，计算 RSI, MACD, KDJ, ATR 等指标加入 g.features。但要注意，特征越多，需要的训练数据量通常也越大，且更容易过拟合。

Q: 如何提高回测速度？
A: 每日训练模型非常消耗计算资源。如果回测速度过慢，可以考虑：

1. 减少 g.train_window（训练数据量）。
2. 减少 n_estimators。
3. 改为每周或每月训练一次（使用 run_daily 或日期判断逻辑控制训练频率）。