如何利用 PTrade API 构建基于相对强弱的行业轮动策略？

这是一个基于PTrade平台的行业轮动策略。

策略逻辑

1. 核心思想：动量效应（Momentum）。强者恒强，资金倾向于流入近期表现强势的行业。
2. 行业池：选取证监会（CSRC）一级行业指数作为轮动标的（如制造业、金融业、采矿业等）。
3. 轮动周期：每20个交易日（约一个月）调仓一次。
4. 选行业标准：计算各行业指数过去20日的涨跌幅，选取涨幅最大的前3个行业。
5. 选股标准：在选中的每个行业中，选取总市值最大的前3只股票作为龙头代表进行买入。
6. 资金分配：资金平均分配到选中的股票中。

策略代码

import pandas as pd
import numpy as np

def initialize(context):
    """
    初始化函数，设置策略参数和全局变量
    """
    # 设置回测频率
    set_commission(commission_ratio=0.0003, min_commission=5.0, type="STOCK")
    set_slippage(slippage=0.002)
    
    # 设定行业池（证监会一级行业代码，后缀为.XBHS）
    # 参考文档附录中的行业列表
    g.industry_list = [
        'A00000.XBHS', # 农、林、牧、渔业
        'B00000.XBHS', # 采矿业
        'C00000.XBHS', # 制造业
        'D00000.XBHS', # 电力、热力、燃气及水生产和供应
        'E00000.XBHS', # 建筑业
        'F00000.XBHS', # 批发和零售业
        'G00000.XBHS', # 交通运输、仓储和邮政业
        'H00000.XBHS', # 住宿和餐饮业
        'I00000.XBHS', # 信息传输、软件和信息技术服务业
        'J00000.XBHS', # 金融业
        'K00000.XBHS', # 房地产业
        'L00000.XBHS', # 租赁和商务服务业
        'M00000.XBHS', # 科学研究和技术服务业
        'N00000.XBHS', # 水利、环境和公共设施管理业
        'Q00000.XBHS', # 卫生和社会工作
        'R00000.XBHS'  # 文化、体育和娱乐业
    ]
    
    # 策略参数
    g.momentum_days = 20       # 计算动量的时间窗口
    g.rebalance_days = 20      # 调仓周期
    g.top_industry_count = 3   # 选取排名前N的行业
    g.top_stock_count = 3      # 每个行业选取的股票数量
    
    # 计数器，用于控制调仓频率
    g.days_counter = 0
    
    # 当前持仓的目标股票列表
    g.target_stocks = []

def before_trading_start(context, data):
    """
    盘前处理：判断是否调仓，如果调仓则进行选股
    """
    # 增加计数器
    g.days_counter += 1
    
    # 仅在调仓日执行选股逻辑
    if g.days_counter % g.rebalance_days != 1:
        return
    
    log.info("开始进行行业轮动选股...")
    
    # 1. 计算行业动量
    industry_momentum = []
    
    for ind_code in g.industry_list:
        # 获取行业指数过去N天的收盘价
        # count设置为 momentum_days + 1 是为了计算收益率 (今天收盘 / N天前收盘 - 1)
        hist = get_history(g.momentum_days + 1, '1d', 'close', security_list=ind_code, fq=None, include=True)
        
        if hist is not None and len(hist) > 1:
            # 计算涨跌幅：(最新收盘价 - N天前收盘价) / N天前收盘价
            # 注意：get_history返回的DataFrame索引是时间，列是code（如果是多股）或字段（如果是单股）
            # 这里单只获取，返回的是DataFrame，列名为close
            close_prices = hist['close'].values
            if len(close_prices) >= 2:
                ret = (close_prices[-1] - close_prices[0]) / close_prices[0]
                industry_momentum.append((ind_code, ret))
    
    # 2. 对行业按涨幅降序排序
    industry_momentum.sort(key=lambda x: x[1], reverse=True)
    
    # 选取排名前N的行业
    top_industries = [item[0] for item in industry_momentum[:g.top_industry_count]]
    log.info("选中的强势行业: %s" % str(top_industries))
    
    # 3. 在选中行业中选股（选市值最大的龙头）
    selected_stocks = []
    
    for ind_code in top_industries:
        # 获取该行业下的所有成分股
        stocks_in_industry = get_industry_stocks(ind_code)
        
        if not stocks_in_industry:
            continue
            
        # 过滤ST、停牌、退市股票
        # 使用 filter_stock_by_status 接口 (PTrade特有)
        valid_stocks = filter_stock_by_status(stocks_in_industry, filter_type=["ST", "HALT", "DELISTING"])
        
        if not valid_stocks:
            continue
            
        # 获取基本面数据：总市值 (total_value)
        # 注意：get_fundamentals 在回测中 date 默认取回测日期
        q = get_fundamentals(valid_stocks, 'valuation', fields='total_value')
        
        if q is None or q.empty:
            continue
            
        # 对市值进行降序排序
        # q 的索引是股票代码
        q = q.sort_values(by='total_value', ascending=False)
        
        # 选取该行业市值最大的前M只股票
        top_stocks = q.index[:g.top_stock_count].tolist()
        selected_stocks.extend(top_stocks)
        
    g.target_stocks = selected_stocks
    log.info("本期目标持仓股票: %s" % str(g.target_stocks))

def handle_data(context, data):
    """
    盘中处理：执行交易
    """
    # 仅在调仓日执行交易逻辑
    if g.days_counter % g.rebalance_days != 1:
        return
        
    # 获取当前持仓
    current_positions = list(context.portfolio.positions.keys())
    
    # 1. 卖出不在目标列表中的股票
    for stock in current_positions:
        if stock not in g.target_stocks:
            # 检查停牌状态，避免报错
            if data[stock].is_open == 0:
                continue
            order_target_value(stock, 0)
            log.info("卖出非目标股票: %s" % stock)
            
    # 2. 买入目标股票
    if len(g.target_stocks) > 0:
        # 等权重分配资金
        # 预留一点现金防止手续费不够
        available_cash = context.portfolio.portfolio_value * 0.98
        target_value_per_stock = available_cash / len(g.target_stocks)
        
        for stock in g.target_stocks:
            # 检查停牌状态
            if stock in data and data[stock].is_open == 0:
                log.info("股票 %s 停牌，跳过买入" % stock)
                continue
                
            order_target_value(stock, target_value_per_stock)
            # log.info("调整股票 %s 持仓至市值 %.2f" % (stock, target_value_per_stock))

def after_trading_end(context, data):
    """
    盘后处理
    """
    pass

策略要点解析

1. 行业数据获取：

   • 使用了 get_history 获取行业指数（如 A00000.XBHS）的历史行情。PTrade支持直接获取这些指数的K线数据，这是计算行业涨跌幅的基础。
   • 代码中列出了证监会一级行业代码作为基础池。

2. 动量计算：

   • 公式：(当前收盘价 - N天前收盘价) / N天前收盘价。
   • 这里选取了20天（g.momentum_days）作为观察窗口，这是一个经典的中短期动量参数。

3. 股票筛选：

   • 成分股获取：使用 get_industry_stocks(ind_code) 获取行业下的所有股票。
   • 状态过滤：使用 filter_stock_by_status 剔除了ST股、停牌股和退市股，防止买入无法交易或风险过大的标的。
   • 龙头优选：使用 get_fundamentals 获取市值数据（total_value），并按降序排列，选取行业内市值最大的股票。大市值股票通常流动性好，更能代表该行业的整体走势。

4. 交易执行：

   • 使用 order_target_value 进行调仓，确保每个选中的股票资金分配均匀（等权重）。
   • 加入了停牌检查 data[stock].is_open，避免对停牌股票下单导致废单或报错。

常见问题 (Q&A)

Q: 为什么选择证监会行业而不是申万行业？
A: PTrade内置了证监会行业指数（.XBHS后缀）的行情数据，调用更加直接稳定。如果需要使用申万行业，需要确认券商环境是否购买了相关数据包，并替换对应的行业代码。

Q: 如何调整调仓频率？
A: 修改 initialize 函数中的 g.rebalance_days 参数。例如，设置为 5 即为周频调仓，设置为 60 即为季频调仓。

Q: 为什么使用总市值选股？
A: 行业轮动策略通常配置的是该行业的Beta收益。市值大的股票通常是行业龙头，与行业指数的相关性最高，且流动性好，适合作为行业的代理标的。如果想追求Alpha，可以改为按ROE或成长性选股。