如何在 PTrade 中捕捉开盘机会？集合竞价追涨停策略编写实例

这是一个基于PTrade平台的集合竞价追涨停策略。

策略逻辑说明

1. 运行时间：策略利用 run_daily 函数设定在 09:24 运行。A股集合竞价时间为 09:15-09:25，其中 09:20-09:25 为不可撤单阶段，09:24 分的价格通常非常接近最终开盘价，适合进行判断。
2. 核心判断：通过 get_snapshot 获取实时行情快照，比较当前价格（last_px）与涨停价（up_px）。
3. 交易执行：如果当前价格等于或高于涨停价，则以涨停价挂单买入。
4. 资金管理：示例中将可用资金平均分配给符合条件的股票。

PTrade 策略代码

def initialize(context):
    """
    初始化函数，设置股票池和定时任务
    """
    # 1. 设置要操作的股票池 (此处以恒生电子和平安银行为例)
    g.security_list = ['600570.SS', '000001.SZ']
    set_universe(g.security_list)
    
    # 2. 设置定时任务
    # 在交易日的 09:24 分运行集合竞价处理函数
    # 此时处于集合竞价不可撤单阶段，价格参考意义较大
    run_daily(context, aggregate_auction_trade, time='09:24')

def aggregate_auction_trade(context):
    """
    集合竞价交易逻辑函数
    """
    # 获取当前账户可用现金
    cash = context.portfolio.cash
    
    # 如果没有设置股票池或没钱，直接返回
    if not g.security_list or cash <= 0:
        return

    # 计算每只股票拟分配的资金 (简单均分)
    # 实际场景中建议根据策略逻辑调整仓位管理
    cash_per_stock = cash / len(g.security_list)

    # 批量获取行情快照，减少网络IO请求
    snapshots = get_snapshot(g.security_list)
    
    for stock in g.security_list:
        # 获取该股票的快照数据
        stock_data = snapshots.get(stock)
        
        if not stock_data:
            continue
            
        # 获取关键字段
        # last_px: 最新成交价 (在竞价阶段即为虚拟撮合价)
        # up_px: 涨停价
        # trade_status: 交易状态
        current_price = stock_data.get('last_px')
        limit_up_price = stock_data.get('up_px')
        status = stock_data.get('trade_status')
        
        # 数据有效性检查
        if current_price is None or limit_up_price is None:
            continue
            
        # 过滤非交易状态 (HALT: 暂停交易, SUSP: 停盘)
        if status in ['HALT', 'SUSP', 'STOPT']:
            log.info("股票 %s 处于停牌或不可交易状态，跳过" % stock)
            continue

        # --- 核心策略逻辑 ---
        # 判断当前价格是否触及涨停价
        # 注意：浮点数比较建议留一点微小的误差空间，或者直接比较是否 >=
        if float(current_price) >= float(limit_up_price):
            log.info("股票 %s 集合竞价触及涨停，准备买入。当前价: %s, 涨停价: %s" % (stock, current_price, limit_up_price))
            
            # 下单买入
            # 使用 order_value 按金额下单
            # limit_price 设置为涨停价，确保在涨停板上排队
            order_id = order_value(stock, cash_per_stock, limit_price=limit_up_price)
            
            if order_id:
                log.info("股票 %s 下单成功，订单ID: %s" % (stock, order_id))
            else:
                log.error("股票 %s 下单失败" % stock)

def handle_data(context, data):
    """
    盘中运行函数 (本策略主要在集合竞价运行，此处可留空或处理止盈止损)
    """
    pass

注意事项

1. 回测限制：get_snapshot 函数在普通的回测模式下可能无法获取到精准的集合竞价tick数据（取决于回测数据的精度）。该策略主要用于实盘或仿真交易环境。
2. 风险提示：追涨停属于高风险策略。在集合竞价阶段以涨停价买入，如果开盘后炸板（打开涨停），可能会面临当日大幅亏损。
3. API调用：代码中使用了 get_snapshot，这是获取盘口实时数据（包括涨跌停价格、交易状态）最高效的方式。
4. Python版本：代码已兼容 Python 3.5 语法（避免使用了 f-string）。

Q&A

Q: 为什么选择在 09:24 运行而不是 09:15 或 09:30？
A: 09:15-09:20 是可撤单阶段，价格变动剧烈且虚假申报较多。09:20-09:25 是不可撤单阶段，此时的撮合价格更接近真实的开盘价。选择 09:24 运行可以在最终撮合（09:25）之前完成下单，尽早排队。

Q: get_snapshot 返回的数据在回测中准确吗？
A: 在 PTrade 的回测环境中，get_snapshot 的支持有限，通常只能获取到当天的开盘、收盘等静态数据，无法模拟真实的逐笔竞价过程。此策略代码主要是为实盘设计的。

Q: 如何避免买入停牌的股票？
A: 代码中增加了 trade_status 的判断，如果状态是 HALT（暂停交易）或 SUSP（停盘），则会跳过该股票，防止无效下单或报错。