PTrade 机器学习实战：构建随机森林多因子选股策略

使用随机森林（Random Forest）算法进行多因子选股是一种经典的机器学习量化策略。其核心思想是将选股问题转化为一个监督学习问题：

1. 特征（X）：选取股票的各类因子（如市盈率、市净率、ROE、动量、波动率等）。
2. 标签（y）：股票未来的收益率（回归问题）或是否跑赢基准（分类问题）。
3. 预测：利用历史数据训练随机森林模型，预测当前股票池中各股票未来的预期收益，买入预期收益最高的股票。

以下是在 PTrade 平台上实现该策略的详细步骤和代码。

策略逻辑概述

1. 股票池：沪深300成分股。
2. 调仓频率：按月调仓（每20个交易日）。
3. 因子选择：
   • 基本面因子：PE（市盈率）、PB（市净率）、ROE（净资产收益率）。
   • 技术面因子：20日动量（过去20日涨跌幅）。
4. 模型训练：
   • 在调仓日，获取过去 N 天的历史数据作为训练集。
   • 训练集特征：历史因子数据。
   • 训练集标签：历史因子对应的下一期收益率。
   • 使用 RandomForestRegressor 进行训练。
5. 交易执行：
   • 输入当前因子数据到模型，预测未来收益率。
   • 选取预测得分最高的 10 只股票等权重买入。

PTrade 策略代码实现

import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestRegressor

def initialize(context):
    """
    初始化函数
    """
    # 1. 设置股票池为沪深300
    g.index = '000300.SS'
    set_universe(get_index_stocks(g.index))
    
    # 2. 策略参数设置
    g.hold_num = 10          # 持仓股票数量
    g.train_days = 60        # 训练数据的时间窗口长度（过去多少天的数据用于训练）
    g.rebalance_days = 20    # 调仓周期（每20个交易日调仓一次）
    g.days_counter = 0       # 计数器
    
    # 3. 初始化随机森林模型
    # n_estimators: 树的数量, max_depth: 树的深度 (防止过拟合)
    g.model = RandomForestRegressor(n_estimators=50, max_depth=5, random_state=42, n_jobs=1)

def before_trading_start(context, data):
    """
    盘前处理：判断是否调仓，如果调仓则训练模型并生成信号
    """
    # 获取当前日期
    current_date = context.blotter.current_dt.strftime("%Y%m%d")
    
    # 每日更新股票池（剔除ST、停牌等）
    all_stocks = get_index_stocks(g.index)
    # 过滤掉停牌和ST股 (PTrade API: filter_stock_by_status)
    g.target_stocks = filter_stock_by_status(all_stocks, filter_type=["ST", "HALT", "DELISTING"])
    set_universe(g.target_stocks)
    
    # 判断是否到达调仓日
    if g.days_counter % g.rebalance_days == 0:
        g.rebalance = True
        log.info("=== 开始进行随机森林模型训练与选股 ===")
        
        # 1. 准备训练数据
        # 为了简化演示，我们取过去 g.train_days 的数据，
        # 实际上应该构建 (T-1期的因子) 对应 (T期的收益) 作为训练对
        
        # 获取历史收盘价计算收益率作为 Label (y)
        # 获取多一天的历史数据以便计算收益率
        hist_prices = get_history(g.train_days + g.rebalance_days + 1, '1d', 'close', g.target_stocks, fq='pre')
        
        # 构建训练集
        X_train = []
        y_train = []
        
        # 这里我们采用一个简化的滚动训练逻辑：
        # 使用 (当前日期 - 调仓周期) 的因子数据 预测 (当前日期 - 调仓周期 到 当前日期) 的收益率
        # 注意：实际工程中需要更严谨的时间对齐，防止未来函数
        
        # 获取上一期的因子数据 (作为特征 X)
        # 计算上一期的日期索引
        prev_idx = -1 - g.rebalance_days 
        
        # 获取上一期的基本面数据
        # 注意：get_history返回的是DataFrame，行是时间，列是股票
        # 我们需要构造一个 [n_samples, n_features] 的矩阵
        
        # 为了演示方便，我们循环获取每只股票的特征和标签
        # 在实际生产中，建议使用向量化操作以提高速度
        
        # 获取当天的基本面数据用于预测
        current_factors = get_factor_data(g.target_stocks, context)
        
        # 获取历史（上一期调仓日）的基本面数据用于训练
        # PTrade回测中 get_fundamentals 默认取回测时间的数据
        # 我们很难直接取到准确的历史某一天基本面快照用于训练(除非存了数据)，
        # 因此这里使用一种近似方法：利用当前因子预测未来(模拟盘)，
        # 或者在回测中，我们假设因子短期内变化不大，使用近期平均值。
        
        # --- 修正后的训练逻辑 ---
        # 由于在线获取历史任意时刻的因子比较困难，我们采用“截面回归”思路：
        # 使用最近一段历史的收益率作为y，使用最近的因子作为X（假设因子具有动量或持续性）
        # 或者更严谨的做法是：每天记录因子，存下来，但这在简单策略中难以实现。
        # 此处演示：使用当前因子预测未来20日收益（仅在回测中有意义，实盘需用历史因子预测当前）
        
        # 为了代码可运行且逻辑通顺，我们采用：
        # X: 过去20天的平均因子值 (PE, PB, ROE) + 过去20天动量
        # y: 过去20天的个股收益率
        # 逻辑假设：过去因子表现好的股票，未来也会好（动量效应）
        
        df_train = pd.DataFrame(index=g.target_stocks)
        
        # 计算标签 y: 过去20天的收益率
        close_df = hist_prices['close']
        # (今天收盘 - 20天前收盘) / 20天前收盘
        past_returns = (close_df.iloc[-1] - close_df.iloc[-1 - g.rebalance_days]) / close_df.iloc[-1 - g.rebalance_days]
        df_train['label'] = past_returns
        
        # 计算特征 X
        # 1. 动量因子 (再往前推20天的收益率)
        mom_returns = (close_df.iloc[-1 - g.rebalance_days] - close_df.iloc[-1 - g.rebalance_days*2]) / close_df.iloc[-1 - g.rebalance_days*2]
        df_train['momentum'] = mom_returns
        
        # 2. 基本面因子 (PE, PB, ROE)
        # 获取最近的财务数据
        q = get_fundamentals(g.target_stocks, 'valuation', ['pe_ttm', 'pb', 'turnover_rate'])
        q2 = get_fundamentals(g.target_stocks, 'profit_ability', ['roe_ttm'])
        
        # 合并数据
        if q is not None and q2 is not None:
            # PTrade返回的fundamentals索引是股票代码
            df_train['pe'] = q['pe_ttm']
            df_train['pb'] = q['pb']
            df_train['turnover'] = q['turnover_rate'] # 换手率
            df_train['roe'] = q2['roe_ttm']
        
        # 数据清洗：去除包含NaN的行
        df_train.dropna(inplace=True)
        
        if len(df_train) > 50: # 确保有足够的数据训练
            # 训练模型
            features = ['momentum', 'pe', 'pb', 'turnover', 'roe']
            X = df_train[features].values
            y = df_train['label'].values
            
            g.model.fit(X, y)
            
            # --- 预测阶段 ---
            # 使用当前最新的因子数据进行预测
            # 构造预测集 X_pred
            df_pred = pd.DataFrame(index=g.target_stocks)
            
            # 当前动量 (过去20天)
            df_pred['momentum'] = past_returns # 刚才计算的past_returns即为当前的过去20天动量
            
            # 当前基本面
            if q is not None and q2 is not None:
                df_pred['pe'] = q['pe_ttm']
                df_pred['pb'] = q['pb']
                df_pred['turnover'] = q['turnover_rate']
                df_pred['roe'] = q2['roe_ttm']
            
            df_pred.dropna(inplace=True)
            
            if len(df_pred) > 0:
                X_pred = df_pred[features].values
                # 预测得分
                df_pred['score'] = g.model.predict(X_pred)
                
                # 选股：取分数最高的 N 只
                df_pred.sort_values(by='score', ascending=False, inplace=True)
                g.buy_list = df_pred.index[:g.hold_num].tolist()
                log.info("选股完成，买入列表: %s" % g.buy_list)
            else:
                g.buy_list = []
        else:
            log.warning("有效训练数据不足，跳过本次调仓")
            g.buy_list = []
            
    else:
        g.rebalance = False

    g.days_counter += 1

def handle_data(context, data):
    """
    盘中交易：执行买卖操作
    """
    if not g.rebalance:
        return

    # 获取当前持仓
    current_positions = list(context.portfolio.positions.keys())
    
    # 1. 卖出不在买入列表中的股票
    for stock in current_positions:
        if stock not in g.buy_list:
            order_target_value(stock, 0)
            # log.info("卖出: %s" % stock)
    
    # 2. 买入在列表中的股票
    if len(g.buy_list) > 0:
        # 等权重分配资金
        position_value = context.portfolio.portfolio_value / len(g.buy_list)
        
        for stock in g.buy_list:
            order_target_value(stock, position_value)
            # log.info("买入: %s" % stock)
    
    # 重置调仓标志，防止重复交易
    g.rebalance = False

def get_factor_data(stocks, context):
    """
    辅助函数：获取因子数据 (此处仅为占位，逻辑已整合进 before_trading_start)
    """
    pass

代码关键点解析

1. 数据获取与对齐 (before_trading_start)：

   • 特征 (X)：我们选取了 momentum (动量), pe (市盈率), pb (市净率), turnover (换手率), roe (净资产收益率)。
   • 标签 (y)：在训练阶段，我们使用过去一段时间的收益率作为标签。
   • 注意：在 PTrade 中，get_fundamentals 获取的是当前回测时间点的数据。为了避免“未来函数”（即用未来的数据预测过去），训练逻辑通常需要严格的时间错位。上述代码采用了一种简化的截面回归逻辑：假设因子与收益率之间的关系在短期内是稳定的，利用最近已知的数据训练模型，然后应用到当前数据上。

2. 数据清洗：

   • 使用 dropna() 去除含有空值的股票。机器学习模型通常无法处理 NaN 值，必须先清洗。
   • 使用 filter_stock_by_status 剔除 ST 和停牌股票，防止买入无法交易的标的。

3. 模型训练：

   • 使用 sklearn.ensemble.RandomForestRegressor。
   • n_estimators=50：树的数量，数量越多越稳定但计算越慢。
   • max_depth=5：限制树的深度，防止过拟合（Overfitting）。

4. 交易执行 (handle_data)：

   • 先卖出：遍历当前持仓，如果不在新的 buy_list 中，则清仓。
   • 后买入：对 buy_list 中的股票进行等权重买入 (order_target_value)。

常见问题与优化建议

1. 回测速度：

   • get_history 和 get_fundamentals 涉及网络或数据库IO，尽量批量获取（如代码中直接传入股票列表 g.target_stocks），不要在 for 循环中单个获取。
   • 随机森林训练比较耗时，建议调仓频率不要太高（如月频或周频）。

2. 因子处理：

   • 去极值与标准化：虽然随机森林对异常值和量纲不敏感，但在量化中，通常还是建议对因子进行去极值（Winsorize）和标准化（Z-Score）处理，以提高模型稳定性。
   • 行业中性化：PE、PB 等因子在不同行业间差异巨大，建议进行行业中性化处理（减去行业均值）。

3. Python 版本兼容性：

   • PTrade 的 Python 环境可能是 3.5 或 3.8+。代码中避免使用了 f-string (如 f"{var}")，而是使用了兼容性更好的 "%s" % var 格式。

Q&A

Q: 为什么使用回归模型而不是分类模型？
A: 选股通常需要对股票进行排名（Ranking），选出最好的 N 只。回归模型可以预测具体的预期收益率，方便排序。分类模型（如预测涨/跌）输出的是概率或类别，虽然也可以用概率排序，但在多因子选股中回归更直观。

Q: 如何避免未来函数？
A: 确保训练集的数据（X和y）在预测时间点之前是完全已知的。例如，预测下个月收益时，训练用的 X 应该是上个月的因子，y 是上个月到现在的收益。绝不能用当天的收盘价去预测当天的收益。

Q: 可以在 PTrade 中保存训练好的模型吗？
A: 可以使用 Python 的 pickle 模块将 g.model 保存到研究目录中，下次直接读取，从而节省训练时间。但在回测中，由于市场环境变化，通常建议滚动训练（Rolling Training）以适应市场变化。