如何利用 PTrade API 获取板块涨跌幅排名，并构建行业轮动策略？

在PTrade中，获取实时热门板块排名并进行交易的核心API是 get_sort_msg。该函数属于交易模块专用接口（通常需要开通Level-2行情权限），能够返回板块（概念、行业、地域等）的实时涨跌幅排名，以及板块内的领涨个股。

策略设计思路

1. 获取排名：使用 get_sort_msg 获取当前涨幅最高的概念板块（XBHS.GN）。
2. 锁定标的：从排名第一的板块数据中，提取 rise_first_grp（领涨股票列表），选取前N只股票。
3. 代码转换：get_sort_msg 返回的股票代码通常没有后缀（如 '600570'），需要根据市场类型转换为PTrade标准格式（如 '600570.SS'）。
4. 交易逻辑：
   • 卖出：遍历当前持仓，如果持仓股票不在当天的目标股票列表中，则清仓卖出。
   • 买入：遍历目标股票列表，如果未持仓且资金充足，则买入。

PTrade 策略代码

# 策略名称：实时热门板块追涨策略
# 运行环境：PTrade实盘/仿真交易（回测模式下该API可能无数据，需在交易模块使用）

def initialize(context):
    """
    初始化函数，设置策略参数
    """
    # 设定要操作的板块类型：'XBHS.GN'为概念板块，'XBHS.HY'为行业板块
    g.sector_type = 'XBHS.GN'
    
    # 设定排序字段：'px_change_rate' 为涨跌幅
    g.sort_field = 'px_change_rate'
    
    # 设定买入排名前几的板块
    g.top_sector_count = 1
    
    # 设定每个板块买入前几名的龙头股
    g.top_stock_count = 2
    
    # 设置每只股票的买入金额（示例：每只买入20000元）
    g.trade_value = 20000
    
    # 设置运行频率：每60秒运行一次扫描和交易
    run_interval(context, trade_logic, seconds=60)

def trade_logic(context):
    """
    核心交易逻辑：获取排名 -> 筛选股票 -> 执行买卖
    """
    # 1. 获取板块涨幅排名
    # sort_type=1 表示降序（涨幅从大到小）
    # data_count=5 获取前5名，减少数据量
    sort_data = get_sort_msg(sort_type_grp=g.sector_type, 
                             sort_field_name=g.sort_field, 
                             sort_type=1, 
                             data_count=5)
    
    if not sort_data:
        log.info("未获取到板块排名数据，跳过本次执行")
        return

    # 2. 提取目标股票池
    target_stocks = []
    
    # 遍历排名前 N 的板块
    for i in range(min(len(sort_data), g.top_sector_count)):
        sector_info = sort_data[i]
        sector_name = sector_info.get('prod_name', '未知板块')
        change_rate = sector_info.get('px_change_rate', 0)
        
        log.info("当前排名第 %s 的板块: %s, 涨幅: %s%%" % (i+1, sector_name, change_rate))
        
        # 获取该板块内的领涨股票列表 (rise_first_grp)
        # 注意：rise_first_grp 是一个包含字典的列表
        leading_grp = sector_info.get('rise_first_grp', [])
        
        # 选取板块内领涨的前 N 只股票
        count = 0
        for stock_item in leading_grp:
            if count >= g.top_stock_count:
                break
            
            raw_code = stock_item.get('prod_code')
            hq_type = stock_item.get('hq_type_code') # 用于判断市场后缀
            
            # 转换代码格式 (例如 600570 -> 600570.SS)
            ptrade_code = format_code(raw_code, hq_type)
            
            if ptrade_code:
                target_stocks.append(ptrade_code)
                count += 1
    
    log.info("本轮目标股票池: %s" % target_stocks)
    
    if not target_stocks:
        return

    # 3. 交易执行：卖出逻辑
    # 获取当前所有持仓
    positions = context.portfolio.positions
    for stock in list(positions.keys()):
        # 如果持仓股票不在今日目标池中，且有持仓，则卖出
        if stock not in target_stocks and positions[stock].amount > 0:
            order_target(stock, 0)
            log.info("板块轮动，卖出非热门股: %s" % stock)

    # 4. 交易执行：买入逻辑
    for stock in target_stocks:
        # 过滤涨跌停状态（可选）：check_limit返回1或2表示涨停，-1或-2表示跌停
        # 这里简单处理：如果是涨停，可能买不进，但依然下单排队
        limit_status = check_limit(stock)
        
        # 如果没有持仓，则买入
        if context.portfolio.positions[stock].amount == 0:
            # 检查资金是否足够
            if context.portfolio.cash >= g.trade_value:
                log.info("买入热门板块龙头: %s" % stock)
                # 下单指定金额
                order_value(stock, g.trade_value)
            else:
                log.info("资金不足，无法买入: %s" % stock)

def format_code(raw_code, hq_type_code):
    """
    辅助函数：将原始代码转换为PTrade标准后缀代码
    raw_code: 股票代码，如 '600570'
    hq_type_code: 市场类型代码，如 'XSHG.ESA.M'
    """
    if not raw_code or not hq_type_code:
        return None
        
    suffix = ''
    if 'XSHG' in hq_type_code:
        suffix = '.SS'
    elif 'XSHE' in hq_type_code:
        suffix = '.SZ'
    else:
        # 如果无法从hq_type判断，尝试通过代码首位判断（备用逻辑）
        if raw_code.startswith('6'):
            suffix = '.SS'
        elif raw_code.startswith('0') or raw_code.startswith('3'):
            suffix = '.SZ'
        else:
            return None # 忽略北交所或其他品种
            
    return raw_code + suffix

def handle_data(context, data):
    """
    必须实现的函数，但在本策略中主要逻辑由 run_interval 驱动
    """
    pass

代码关键点解析

1. get_sort_msg 函数：

   • 这是获取排名的核心。
   • sort_type_grp='XBHS.GN'：指定获取概念板块排名。如果想追行业板块，可改为 'XBHS.HY'。
   • sort_field_name='px_change_rate'：按涨跌幅排序。
   • sort_type=1：降序排列，即涨幅最大的排在前面。

2. 数据结构解析：

   • get_sort_msg 返回的是一个列表，列表中的每个元素是一个字典，代表一个板块。
   • 字典中的 rise_first_grp 字段包含了该板块内涨幅靠前的股票信息（也是字典列表）。
   • 注意：返回的股票代码（prod_code）通常是纯数字（如 000001），而PTrade下单函数 order 需要带后缀的代码（如 000001.SZ）。因此代码中编写了 format_code 函数利用 hq_type_code 字段进行后缀补全。

3. 执行频率：

   • 使用了 run_interval(context, trade_logic, seconds=60)，这意味着策略会每分钟检查一次排名并调整仓位。这适合捕捉日内热点，但要注意交易成本和换手率。

4. 注意事项：

   • 环境限制：get_sort_msg 依赖实时行情推送，通常只能在实盘或仿真交易环境中使用。在普通的日线回测中，该函数可能返回空数据或不支持。
   • 未来函数风险：如果在回测中使用此逻辑，必须确保数据是当时时刻的快照。但在实盘中，这是获取实时数据的标准方法。

Q&A

Q: 为什么代码中要自己写 format_code 函数？
A: get_sort_msg 返回的原始数据中，股票代码是不带 .SS 或 .SZ 后缀的。PTrade 的交易函数（如 order）必须使用带后缀的标准代码才能识别。通过判断 hq_type_code 中是否包含 XSHG（上海）或 XSHE（深圳）是最准确的转换方式。

Q: 这个策略可以在回测中运行吗？
A: 通常不可以。get_sort_msg 是基于 Level-2 或实时行情推送的接口，大部分历史回测环境不支持该接口的调用。此策略主要用于实盘或盘中仿真。

Q: 如何避免买入已经涨停买不进去的股票？
A: 代码中调用了 check_limit(stock)。该函数返回 1 或 2 代表涨停。你可以在买入逻辑前加一个判断：if check_limit(stock) not in [1, 2]: order_value(...)。不过，对于追涨策略，有时即便涨停也需要挂单排队，因此示例代码中保留了下单动作。