ptrade 盘中异动捕获：跌停板开板抄底交易策略

在盘中交易中，股票从跌停板打开（即卖一档跌停价挂单被快速吃掉/撤单）往往意味着抄底资金或主力资金入场。利用 ptrade 平台的实时行情快照与高频触发机制，可以实现秒级捕捉开板信号并快速下单交易。

一、核心实现原理

1. 行情监控（get_snapshot）：

   • 通过 get_snapshot(security) 获取股票的最新行情快照。
   • 检查 last_px（最新成交价）与 down_px（跌停价）。当股票处于跌停状态时，last_px == down_px，且卖一档 offer_grp[1] 挂牌价格为跌停价且挂单量巨大。
   • 当最新价大于跌停价（last_px > down_px），或卖一档价格高于跌停价时，即判定为跌停板已打开。

2. 执行频率（run_interval / tick_data）：

   • 普通盘中策略可使用 run_interval(context, func, seconds=3) 实现最小 3 秒一次的高频监控。
   • 在实盘 tick 级别策略中，也可在 tick_data(context, data) 中利用 data[security]['tick'] 进行 3 秒级别的逐 tick 触发与秒级下单。

3. 下单执行（order / order_tick）：

   • 捕捉到开板信号后，使用 order(security, amount, limit_price) 发送限价单，或在 tick_data 中通过 order_tick 以指定买档/卖档快速报单。

二、ptrade 跌停开板抄底策略源码

以下为基于 run_interval 间隔 3 秒轮询监控监控池股票跌停开板的完整策略示例：

# -*- coding: utf-8 -*-

def initialize(context):
    # 1. 设置监控股票池
    g.security_list = ['600570.SS', '000001.SZ']
    set_universe(g.security_list)
    
    # 2. 标记股票当日是否已触发买入，避免重复下单
    g.traded_flags = {sec: False for sec in g.security_list}
    
    # 3. 设置定时轮询，每 3 秒执行一次监控函数
    run_interval(context, check_limit_open, seconds=3)

def before_trading_start(context, data):
    # 每日盘前重置交易标记
    for sec in g.security_list:
        g.traded_flags[sec] = False

def check_limit_open(context):
    """每 3 秒检查一次行情快照，捕获跌停开板"""
    for security in g.security_list:
        # 如果今日已下单，跳过
        if g.traded_flags[security]:
            continue
        
        # 获取实时行情快照
        snapshot = get_snapshot(security)
        if not snapshot or security not in snapshot:
            continue
        
        sec_data = snapshot[security]
        last_px = sec_data.get('last_px', 0)
        down_px = sec_data.get('down_px', 0)
        trade_status = sec_data.get('trade_status', '')
        
        # 过滤非连续竞价交易状态
        if trade_status != 'TRADE':
            continue
            
        # 确认该股历史/此前处于跌停状态（如低开或早盘触及跌停）
        # 当最新价高于跌停价，且存在有效跌停价时，判断为跌停打开
        if down_px > 0 and last_px > down_px:
            # 获取卖一档信息 offer_grp = {1: [价格, 委托量, 委托笔数], ...}
            offer_grp = sec_data.get('offer_grp', {})
            sell_1_price = offer_grp.get(1, [0])[0] if 1 in offer_grp else 0
            
            # 如果卖一价也高于跌停价，说明卖盘跌停压单已被消化/撤单
            if sell_1_price > down_px:
                log.info(f"[异动捕获] 股票 {security} 跌停板打开！最新价: {last_px}, 跌停价: {down_px}")
                
                # 计算买入数量（示例买入 1000 股）
                buy_amount = 1000
                cash = context.portfolio.cash
                
                # 资金校验
                if cash >= buy_amount * last_px:
                    # 发送限价单买入（以最新价或买一价下单）
                    order_id = order(security, buy_amount, limit_price=last_px)
                    if order_id:
                        log.info(f"已对 {security} 发送抄底买单，订单ID: {order_id}")
                        g.traded_flags[security] = True

def handle_data(context, data):
    # 基础逻辑已在 run_interval 中处理
    pass

三、注意事项与优化建议

1. 价格精度控制：
   • 股票买卖限价 limit_price 需保留 2 位小数（可转债/ETF 需保留 3 位小数），避免因精度问题导致报单被柜台拒绝。
2. 防止重复下单与时滞：
   • 柜台持仓与资金更新存在微秒至秒级时滞，务必使用全局变量（如 g.traded_flags）在本地记录下单状态，防止在 3 秒轮询中连续触发重复报单。
3. 挂单量（大单吃单）辅助验证：
   • 可结合 check_limit(security) 函数（返回 2 表示触板，1 表示封板，0 表示未涨跌停）或直接分析 offer_grp 卖一档挂单量的骤减趋势，进一步提高开板捕捉的准确率。