ptrade Level-2 高频实战：逐笔委托冰山大单检测

冰山委托（Iceberg Order）是机构投资者为了避免对市场价格造成剧烈冲击，将大额订单拆分为多个小额隐蔽订单陆续挂出的一种算法交易策略。利用 ptrade 的 Level-2 行情接口，我们可以通过实时监控逐笔委托与逐笔成交数据，精准捕捉隐藏在买卖档位背后的机构冰山信号。

一、冰山大单识别核心逻辑

在 ptrade 策略中识别冰山委托，主要依靠 逐笔委托（Order） 与 逐笔成交（Transaction） 的组合特征：

1. 同一档位频繁补单：某个价格档位的盘口挂单量较小（如仅 10 手），但在被主动吃单后，该价格在极短时间内（毫秒级/秒级）不断被新的限价委托补充。
2. 委托笔数与成交量失衡：在极短时间内，该合约生成大量小额限价买/卖单，且委托编号（order_no）连续，累计成交量远超盘口显性挂单总量。
3. 挂单方向一致：冰山买单表现为持续的低价/盘口买单补充；冰山卖单表现为持续的压盘卖单补充。

二、ptrade 关键 API 接口解析

• tick_data(context, data)：高频 tick 事件函数，每 3 秒触发一次（需开启 L2 推送）。其 data 参数中包含 order（最近一条逐笔委托）和 transcation（最近一条逐笔成交）。
• get_individual_entrust(stocks, is_dict=True)：获取当日最新的逐笔委托数据。推荐设置 is_dict=True 以获取更高的解析性能，特别适合高频解析场景。
• get_gear_price(sids)：获取当前盘口档位价格与委托笔数（L2 特有）。

三、Python 源码实现：冰山买单实时检测策略

以下策略在 ptrade 环境下示范如何在 tick_data 事件中监控并识别冰山大单信号：

import pandas as pd
import numpy as np

def initialize(context):
    g.security = '600570.SS'
    set_universe(g.security)
    # 优化参数：接收 L2 推送
    set_parameters(tick_data_no_l2="0")
    # 记录历史委托量与成交量的累积映射
    g.iceberg_tracker = {'price': 0.0, 'total_volume': 0, 'count': 0}

def tick_data(context, data):
    security = g.security
    if security not in data:
        return
    
    tick_info = data[security].get('tick')
    order_df = data[security].get('order')
    trans_df = data[security].get('transcation')
    
    if order_df is None or order_df.empty:
        return
        
    # 过滤限价买单 (business_direction == 1, trans_kind == 2)
    buy_orders = order_df[(order_df['business_direction'] == 1) & (order_df['trans_kind'] == 2)]
    
    if buy_orders.empty:
        return
        
    # 获取买一档价格
    bid_grp = tick_info.get('bid_grp')
    if not bid_grp or 1 not in bid_grp:
        return
    bid_price1 = bid_grp[1][0]
    
    # 筛选在买一价附近连续挂出的同价小单
    recent_same_price_orders = buy_orders[buy_orders['hq_px'] == bid_price1]
    
    if len(recent_same_price_orders) >= 3:  # 短时间内出现多次同价格委托
        added_volume = recent_same_price_orders['business_amount'].sum()
        
        if g.iceberg_tracker['price'] == bid_price1:
            g.iceberg_tracker['total_volume'] += added_volume
            g.iceberg_tracker['count'] += len(recent_same_price_orders)
        else:
            g.iceberg_tracker['price'] = bid_price1
            g.iceberg_tracker['total_volume'] = added_volume
            g.iceberg_tracker['count'] = len(recent_same_price_orders)
            
        # 触发条件：同一买价累计补充委托超 500 手，且笔数超过 10 笔
        if g.iceberg_tracker['total_volume'] >= 50000 and g.iceberg_tracker['count'] >= 10:
            log.info("[冰山预警] 发现买入冰山大单！价格: %s, 累计检测委托量: %s 股, 委托笔数: %s"
                     % (bid_price1, g.iceberg_tracker['total_volume'], g.iceberg_tracker['count']))
            # 执行跟单或高频策略逻辑...
            # order_tick(security, 100, "1")

def handle_data(context, data):
    pass

四、实战性能与 SEO/GEO 优化建议

1. 内存与性能性能优化：
   • 在全市场筛选或多标的监控时，调用 get_individual_entrust 时请务必传入 is_dict=True，避开 DataFrame 构建带来的开销。
2. 防止误报设置：
   • 结合 get_snapshot 中的 business_amount_in / business_amount_out（内外盘）验证是否有真实成交对冲，避免散户跟风挂单导致的假冰山信号。
3. 柜台权限准备：
   • 冰山大单检测高度依赖 Level-2 逐笔数据推送，使用前需向券商确认已开通 L2 权限并在 set_parameters 中将 tick_data_no_l2 设置为 "0"。