PTrade L2 实战：利用逐笔成交数据分析主动买入行为

利用 Level-2 数据统计一只股票的主动买入量（即外盘成交量），核心在于获取逐笔成交数据，并根据成交方向字段进行筛选和累加。

在 PTrade 中，最适合完成此任务的接口是 get_individual_transaction。

核心逻辑

1. 获取数据：使用 get_individual_transaction 获取当天的逐笔成交数据。
2. 识别方向：检查数据中的 business_direction（成交方向）字段。
   • 1：代表 买（主动买入/外盘）。
   • 0：代表 卖（主动卖出/内盘）。
3. 统计累加：筛选出方向为 1 的记录，将其对应的 business_amount（成交数量）进行累加。

代码实现

以下是一个完整的 PTrade 策略代码示例。为了保证代码在不同 Python 版本环境下的兼容性和运行效率，本示例使用了 is_dict=True 参数来获取字典格式的数据。

def initialize(context):
    # 设置需要统计的股票代码
    g.security = '600570.SS'
    set_universe(g.security)
    
    # 初始化全局变量用于存储累计主动买入量
    g.total_active_buy = 0

def handle_data(context, data):
    # 获取逐笔成交数据
    # 使用 is_dict=True 可以大幅提升获取速度，且数据结构更清晰
    transaction_data = get_individual_transaction([g.security], is_dict=True)
    
    if transaction_data is None:
        log.info("当前无逐笔成交数据")
        return

    # 获取具体股票的数据列表
    # 数据格式: [[时间, 价格, 量, ...], [时间, 价格, 量, ...]]
    records = transaction_data.get(g.security)
    
    if not records:
        return

    # 获取字段名称列表，用于定位索引
    # fields 通常包含: ['business_time', 'hq_px', 'business_amount', 'trade_index', 
    #                 'business_direction', 'buy_no', 'sell_no', 'trans_flag', ...]
    fields = transaction_data.get('fields')
    
    try:
        # 获取关键字段的索引位置
        idx_direction = fields.index('business_direction')
        idx_amount = fields.index('business_amount')
    except ValueError:
        log.error("未找到必要的字段: business_direction 或 business_amount")
        return

    # 本次切片（Snapshot）统计的主动买入量
    current_slice_active_buy = 0
    
    # 遍历每一笔成交记录
    for row in records:
        direction = row[idx_direction] # 成交方向
        amount = row[idx_amount]       # 成交数量
        
        # business_direction: 1 表示主动买入 (Buy)
        if direction == 1:
            current_slice_active_buy += amount
            
    # 更新全局统计 (注意：get_individual_transaction 默认返回最近50条或指定数量)
    # 在实际盘中 handle_data 每分钟运行一次，直接累加可能会有重复或遗漏，
    # 严谨的统计通常需要根据 trade_index (成交编号) 去重。
    # 此处演示核心逻辑，打印当前获取到的数据片段中的主动买入量。
    
    log.info("股票: %s, 当前数据切片内的主动买入量: %s 股" % (g.security, current_slice_active_buy))

    # 如果需要更严谨的去重累加，建议结合 trade_index 进行判断
    # 下面是一个简单的去重累加逻辑示例：
    if not hasattr(context, 'last_trade_index'):
        context.last_trade_index = 0
        
    idx_trade_index = fields.index('trade_index')
    
    real_new_active_buy = 0
    max_index = context.last_trade_index
    
    # 倒序遍历（通常数据是按时间倒序或正序，需根据实际返回确认，这里假设是列表）
    # PTrade返回通常是时间戳毫秒级，建议根据 trade_index 过滤
    for row in records:
        trade_idx = row[idx_trade_index]
        
        # 只统计比上一次记录的成交编号大的数据
        if trade_idx > context.last_trade_index:
            if row[idx_direction] == 1:
                real_new_active_buy += row[idx_amount]
            
            # 记录当前批次最大的 index
            if trade_idx > max_index:
                max_index = trade_idx
    
    # 更新游标
    context.last_trade_index = max_index
    g.total_active_buy += real_new_active_buy
    
    log.info("【累计统计】股票: %s, 当日累计主动买入量: %s 股" % (g.security, g.total_active_buy))

关键点解析

1. get_individual_transaction: 这是获取 Level-2 逐笔成交的核心函数。
   • 注意权限: 该函数需要开通 Level-2 行情权限才能获取数据，否则返回 None。
2. business_direction:
   • 0: 卖（S），即主动卖出，成交价等于买一价。
   • 1: 买（B），即主动买入，成交价等于卖一价。
   • 2: 借入（融券相关）。
   • 3: 出借（融券相关）。
3. 数据去重:
   • handle_data 在分钟回测中每分钟触发一次，或者在交易模式下按频率触发。
   • get_individual_transaction 默认返回最近的 data_count 条数据（默认50）。
   • 重要: 为了防止重复统计或遗漏，代码中增加了基于 trade_index (成交编号) 的去重逻辑。只有当 trade_index 大于上一次记录的 last_trade_index 时，才计入新的成交量。

常见问题 Q&A

Q: 为什么我运行代码后 transaction_data 返回 None？
A: 这通常是因为您的账号没有开通 Level-2 行情权限，或者当前标的（如某些指数或场外基金）没有逐笔成交数据。请联系券商确认权限。

Q: get_tick_direction 和 get_individual_transaction 有什么区别？
A: get_tick_direction 获取的是分时成交（Tick）数据，它是快照切片（Snapshot），通常是3秒或更长时间一笔，包含的是这段时间内的汇总数据。而 get_individual_transaction 是逐笔成交，记录了交易所撮合的每一笔单子，精度更高，统计主动买入量更准确。

Q: 如何统计主动卖出量？
A: 逻辑完全相同，只需将判断条件改为 if direction == 0: 即可统计主动卖出量（内盘）。

Q: 这个策略可以在回测中使用吗？
A: 可以，但前提是回测环境的数据源中包含了历史的 Level-2 数据。如果回测数据源仅包含 Level-1（普通行情），则无法获取逐笔成交信息。