如何利用 PTrade 的 tick_data 逐笔委托与成交数据发现主力大单？

通过分析逐笔成交（Transaction）和逐笔委托（Entrust）数据，可以捕捉市场上主力资金的动向，例如大额买入、卖出或撤单行为。在PTrade中，这需要开通Level-2行情权限，并使用特定的API接口获取高频数据。

以下是具体的分析逻辑、API说明以及完整的策略代码示例。

核心分析逻辑

1. 定义“大单”：

   • 按金额：例如单笔成交额超过100万元。
   • 按手数：例如单笔成交量超过1000手。
   • 通常结合价格（hq_px）和数量（business_amount）计算成交金额（Turnover = Price * Volume）。

2. 逐笔成交分析 (get_individual_transaction)：

   • 关注 成交金额：筛选出金额巨大的成交记录。
   • 关注 成交方向 (business_direction)：判断是主动买入（主力扫货）还是主动卖出（主力出货）。
   • 关注 成交类型 (trans_flag)：区分是普通成交还是撤单成交。

3. 逐笔委托分析 (get_individual_entrust)：

   • 关注 委托金额：发现挂在盘口的大额挂单（支撑或压盘）。
   • 关注 委托类型 (trans_kind)：识别大额撤单行为（可能是虚假挂单诱多或诱空）。

关键API说明

• get_individual_transaction(stocks, is_dict=True): 获取逐笔成交数据。建议设置 is_dict=True 以提高回测和实盘的运行效率。
• get_individual_entrust(stocks, is_dict=True): 获取逐笔委托数据。

策略代码实现

以下代码展示了如何在PTrade中实时监控指定股票的逐笔数据，并打印出满足“大单”条件的交易详情。

def initialize(context):
    """
    策略初始化函数
    """
    # 设定要监控的股票
    g.security = '600570.SS'
    set_universe(g.security)
    
    # 定义大单阈值 (单位：元)
    # 这里设定为 100万元，可根据实际标的流动性调整
    g.big_order_threshold = 1000000 
    
    # 定义大单手数阈值 (单位：股)
    g.big_vol_threshold = 100000  # 1000手 (1手=100股)
    
    # 设置运行频率，这里演示使用 run_interval 每10秒检查一次L2数据
    # 注意：实盘中L2数据量巨大，需合理控制频率和处理逻辑
    run_interval(context, check_l2_data, seconds=10)

def check_l2_data(context):
    """
    定时检查L2数据的函数
    """
    if not is_trade():
        return

    # --- 1. 分析逐笔成交数据 ---
    # is_dict=True 返回字典格式，处理速度更快
    trans_data = get_individual_transaction([g.security], is_dict=True)
    
    if trans_data is not None and g.security in trans_data:
        # 获取字段索引，避免硬编码索引
        fields = trans_data['fields']
        try:
            idx_time = fields.index('business_time')
            idx_px = fields.index('hq_px')
            idx_amt = fields.index('business_amount')
            idx_dir = fields.index('business_direction') # 1-买, 2-卖
        except ValueError:
            log.info("逐笔成交数据字段解析失败")
            return

        # 获取该股票的数据列表
        records = trans_data[g.security]
        
        # 遍历每一笔成交
        for record in records:
            price = record[idx_px]
            volume = record[idx_amt]
            amount = price * volume # 计算成交额
            
            # 判断是否为大单 (金额达标 或 数量达标)
            if amount >= g.big_order_threshold or volume >= g.big_vol_threshold:
                direction_str = "买入" if record[idx_dir] == 1 else "卖出"
                if record[idx_dir] == 0: direction_str = "中性" # 竞价阶段等
                
                log.info("【大单成交监测】股票: %s, 时间: %s, 价格: %.2f, 数量: %d, 金额: %.2f, 方向: %s" % (
                    g.security, 
                    record[idx_time], 
                    price, 
                    volume, 
                    amount, 
                    direction_str
                ))

    # --- 2. 分析逐笔委托数据 ---
    entrust_data = get_individual_entrust([g.security], is_dict=True)
    
    if entrust_data is not None and g.security in entrust_data:
        fields = entrust_data['fields']
        try:
            idx_time = fields.index('business_time')
            idx_px = fields.index('hq_px')
            idx_amt = fields.index('business_amount')
            idx_dir = fields.index('business_direction') # 0-卖, 1-买
            idx_kind = fields.index('trans_kind') # 委托类别
        except ValueError:
            return

        records = entrust_data[g.security]
        
        for record in records:
            price = record[idx_px]
            volume = record[idx_amt]
            amount = price * volume
            
            # 简单的过滤逻辑
            if amount >= g.big_order_threshold:
                direction_str = "买单" if record[idx_dir] == 1 else "卖单"
                
                # 委托类别判断 (不同市场定义不同，参考文档)
                # 深圳: 1-市价, 2-限价, 3-本方最优
                # 上海: 4-增加订单, 5-删除订单(撤单)
                kind = record[idx_kind]
                kind_str = "委托"
                if kind == 5: # 上海撤单
                    kind_str = "撤单"
                
                log.info("【大单委托监测】股票: %s, 时间: %s, 价格: %.2f, 数量: %d, 金额: %.2f, 方向: %s, 类型: %s" % (
                    g.security, 
                    record[idx_time], 
                    price, 
                    volume, 
                    amount, 
                    direction_str,
                    kind_str
                ))

def handle_data(context, data):
    """
    必须实现的函数，此处留空，逻辑在 run_interval 中执行
    """
    pass

代码详解

1. 初始化 (initialize):

   • 设置了监控标的 g.security。
   • 定义了判断大单的阈值 g.big_order_threshold (金额) 和 g.big_vol_threshold (数量)。
   • 使用 run_interval 每10秒执行一次检查。逐笔数据量很大，如果每3秒（tick级别）处理一次，需要确保策略计算效率足够高。

2. 数据获取:

   • 使用了 get_individual_transaction 和 get_individual_entrust。
   • 关键点：设置了 is_dict=True。这是PTrade官方推荐的做法，返回字典格式比DataFrame格式在处理速度上快很多，特别是在处理大量L2数据时。

3. 字段解析:

   • 返回的数据是 list 结构，为了代码的健壮性，我们通过 fields.index('字段名') 动态获取数据在列表中的位置，而不是写死下标（例如 record[1]），这样即使API字段顺序调整，代码依然有效。

4. 逻辑判断:

   • 计算 amount = price * volume。
   • 对比阈值，如果满足条件，则通过 log.info 打印出来。在实际交易策略中，这里可以替换为下单逻辑（例如：发现主力大举买入时跟随买入）。

注意事项

1. 权限要求：逐笔成交和逐笔委托属于 Level-2 行情 数据。如果您的PTrade账号没有开通L2权限，调用这些接口将返回 None 或空数据。
2. 数据量与性能：L2数据非常密集，全市场监控会消耗极大的计算资源。建议只针对股票池中的少量股票进行监控。
3. 撤单分析：在逐笔委托中，上海市场的撤单通常标记为 trans_kind=5（删除订单）。识别大额撤单对于判断主力虚晃一枪非常有用。
4. Python版本：代码遵循Python 3.5语法规范（不使用f-string），确保在PTrade所有环境中均可运行。

Q&A

Q: 为什么代码中使用 is_dict=True？
A: 逐笔数据量极大，使用默认的 DataFrame 格式会消耗大量内存并降低处理速度。is_dict=True 返回原生的字典和列表结构，解析速度更快，适合高频策略。

Q: 如何区分主动买入和主动卖出？
A: 在逐笔成交数据中，business_direction 字段通常表示成交方向。一般约定：1 代表买入（主动买），2 代表卖出（主动卖）。这有助于判断资金流向。

Q: 逐笔委托中的 trans_kind 在沪深两市有什么区别？
A: 是的，有区别。

• 深圳市场：1 市价委托，2 限价委托，3 本方最优。
• 上海市场：4 增加订单（新挂单），5 删除订单（撤单）。
  在编写跨市场策略时，需要根据股票代码后缀（.SS 或 .SZ）分别处理。