PTrade 策略教程：基于行业动量的板块轮动交易

在PTrade中实现行业轮动策略主要包含以下几个步骤：

1. 获取行业成分股：使用 get_industry_stocks 获取指定行业的股票列表。
2. 获取历史行情：使用 get_history 获取这些股票的价格数据。
3. 计算板块涨幅：计算成分股的平均涨幅作为板块涨幅（也可以使用加权平均）。
4. 排序与轮动：根据涨幅对板块进行排序，买入排名靠前的板块成分股，卖出排名靠后的。

以下是详细的策略实现代码和说明。

策略逻辑说明

1. 行业池：我们在 initialize 中定义一组要轮动的行业代码（这里以聚源一级行业为例，如金融、工业、消费等）。
2. 轮动周期：设定每隔 N 天（例如20个交易日）进行一次调仓。
3. 动量计算：计算过去 M 天（例如20天）的板块平均涨幅。
4. 交易执行：
   • 选取涨幅最高的 Top N 个行业。
   • 清仓不在目标行业内的股票。
   • 等权买入目标行业内的所有股票（或为了资金利用率，买入目标行业的前X只龙头股，本示例为简化逻辑，买入行业内所有非停牌/ST股票）。

PTrade 策略代码

import numpy as np
import pandas as pd

def initialize(context):
    """
    策略初始化函数
    """
    # 1. 设定要轮动的行业代码列表 (这里使用聚源一级行业代码示例，后缀为.XBHS)
    # 110000: 农林牧渔, 220000: 基础化工, 340000: 食品饮料, 
    # 480000: 银行, 710000: 计算机, 280000: 汽车
    g.sector_list = [
        '110000.XBHS', '220000.XBHS', '340000.XBHS', 
        '480000.XBHS', '710000.XBHS', '280000.XBHS'
    ]
    
    # 2. 策略参数设置
    g.lookback_days = 20      # 回看天数，用于计算涨幅
    g.holding_days = 20       # 持仓周期，每20天轮动一次
    g.top_n_sectors = 1       # 持有涨幅最好的前N个板块
    g.days_counter = 0        # 计数器
    
    # 3. 设定基准
    set_benchmark('000300.SS')
    # 开启每日定时运行，时间为9:35
    run_daily(context, trade_logic, time='09:35')

def check_stock_valid(stock_list):
    """
    过滤停牌、ST、退市的股票
    """
    # 获取当前不可交易的股票（停牌、ST等）
    # filter_stock_by_status 默认过滤 ST, HALT, DELISTING
    valid_stocks = filter_stock_by_status(stock_list)
    return valid_stocks

def get_sector_momentum(context, sector_code, lookback):
    """
    计算指定行业的平均涨幅
    """
    # 1. 获取行业成分股
    stocks = get_industry_stocks(sector_code)
    if not stocks:
        return -999.0 # 如果行业为空，返回极小值
    
    # 2. 获取历史收盘价
    # 获取过去 lookback + 1 天的数据，以计算收益率
    # 注意：如果股票数量极大，get_history可能会较慢，建议分批或优化
    hist_data = get_history(lookback + 1, '1d', 'close', stocks, fq='pre', include=False)
    
    if hist_data is None or len(hist_data) == 0:
        return -999.0

    # 3. 计算收益率
    # PTrade get_history 返回 DataFrame，列为股票代码，行为时间 (Python 3.11环境)
    # 或者返回 Panel (Python 3.5环境)，这里做兼容处理
    
    # 计算每只股票的区间涨幅: (Current_Close - Start_Close) / Start_Close
    # 简单处理：取最后一天和第一天对比
    try:
        # 假设是DataFrame格式 (Time x Stocks) 或 (Stocks x Fields via query)
        # 这里简化逻辑，直接利用pandas计算
        df_close = hist_data['close'] if 'close' in hist_data else hist_data
        
        # 如果是多级索引或Panel，需要确保转为 DataFrame (Time x Stock)
        if not isinstance(df_close, pd.DataFrame):
             # 针对旧版环境的简单处理，实际环境建议打印type(hist_data)确认
             return -999.0
        
        # 计算个股涨幅
        # iloc[-1] 是最近一天，iloc[0] 是起始天
        stock_returns = (df_close.iloc[-1] - df_close.iloc[0]) / df_close.iloc[0]
        
        # 4. 计算板块平均涨幅 (剔除NaN值)
        avg_return = stock_returns.mean()
        
        # 如果全是NaN，返回极小值
        if np.isnan(avg_return):
            return -999.0
            
        return avg_return
        
    except Exception as e:
        log.error("计算板块 %s 涨幅出错: %s" % (sector_code, e))
        return -999.0

def trade_logic(context):
    """
    核心交易逻辑
    """
    # 1. 判断是否到达调仓日
    if g.days_counter % g.holding_days != 0:
        g.days_counter += 1
        return
    
    log.info("达到调仓周期，开始计算板块轮动...")
    
    # 2. 计算所有板块的涨幅
    sector_scores = []
    for sector in g.sector_list:
        score = get_sector_momentum(context, sector, g.lookback_days)
        sector_scores.append((sector, score))
        log.info("板块: %s, 过去%d日涨幅: %.2f%%" % (sector, g.lookback_days, score*100))
    
    # 3. 排序，取涨幅最大的 Top N
    # 按涨幅从大到小排序
    sector_scores.sort(key=lambda x: x[1], reverse=True)
    
    target_sectors = [x[0] for x in sector_scores[:g.top_n_sectors]]
    log.info("目标持有板块: %s" % target_sectors)
    
    # 4. 获取目标板块的所有成分股
    target_stocks = []
    for sec in target_sectors:
        components = get_industry_stocks(sec)
        # 过滤ST和停牌
        valid_components = check_stock_valid(components)
        target_stocks.extend(valid_components)
    
    # 如果目标股票池太大，可以限制数量，例如只买前20只，这里为了演示全买
    # 为防止资金过于分散，建议实际策略中对 target_stocks 再做一次筛选（如按市值排序取前10）
    if len(target_stocks) > 30:
        target_stocks = target_stocks[:30] # 简单截断，防止持仓过多
    
    # 5. 执行交易
    # 5.1 卖出不在目标池的股票
    current_holdings = list(context.portfolio.positions.keys())
    for stock in current_holdings:
        if stock not in target_stocks:
            order_target_value(stock, 0)
            log.info("卖出非目标板块股票: %s" % stock)
    
    # 5.2 买入目标池股票
    if len(target_stocks) > 0:
        # 均分资金
        cash = context.portfolio.portfolio_value # 使用总资产计算，或者使用 context.portfolio.cash
        position_value = cash / len(target_stocks)
        
        for stock in target_stocks:
            order_target_value(stock, position_value)
            log.info("买入目标板块股票: %s" % stock)
    
    # 更新计数器
    g.days_counter += 1

def handle_data(context, data):
    # 日级策略主要逻辑在 run_daily 中执行
    pass

关键API解析

1. get_industry_stocks(industry_code):

   • 功能: 获取指定行业的成分股列表。
   • 参数: industry_code 必须带后缀。例如聚源行业以 .XBHS 结尾（如 110000.XBHS），证监会行业也以 .XBHS 结尾（如 C27000.XBHS）。
   • 注意: 行业代码列表可以在 PTrade 帮助文档的“附录：行业列表”中找到。

2. get_history(count, frequency, field, security_list, ...):

   • 功能: 批量获取股票的历史行情。
   • 技巧: 直接传入 security_list 为一个列表，可以一次性获取多只股票的数据，返回的是 DataFrame 结构，便于使用 Pandas 进行向量化计算（如 mean()），这比循环获取单只股票效率高得多。

3. filter_stock_by_status(stocks):

   • 功能: 这是一个非常实用的辅助函数，用于剔除 ST、停牌、退市的股票，防止策略买入无法交易或高风险的标的。

常见问题 Q&A

Q: 如何查找行业代码？
A: 在 PTrade 帮助文档的“附录”章节有完整的行业代码表。常用的有：

• 聚源行业（推荐，分类较细）：如 270000.XBHS (电子), 340000.XBHS (食品饮料)。
• 证监会行业：如 C27000.XBHS (医药制造业)。

Q: 为什么代码中要限制持仓数量 (if len(target_stocks) > 30)？
A: 一个行业可能包含上百只股票。如果资金量不大，全额买入会导致每只股票分到的资金极少（甚至不足一手），导致无法成交（废单）。在实盘或回测中，通常会选取行业内的龙头股（如按市值排序取前5）来代表该行业进行交易。

Q: 如何计算加权涨幅而不是简单平均？
A: 你需要获取股票的流通市值。使用 get_fundamentals 获取 float_value (流通市值)，然后计算：
板块涨幅 = Sum(个股涨幅 * 个股流通市值) / Sum(个股流通市值)。

Q: get_history 返回的数据结构是什么？
A: 在 PTrade 的 Python 3 环境中，当传入多个股票代码时，通常返回一个 DataFrame，索引为时间，列为股票代码（针对单个字段查询）。如果查询多个字段，可能会返回 MultiIndex DataFrame。本示例中只查询 'close'，处理起来最简单。