PTrade 高频策略揭秘：如何使用 get_individual_entrust 和 get_individual_transaction 分析 Level-2 数据？

在PTrade量化交易平台中，获取一只股票的实时逐笔成交数据（L2数据）主要通过调用 get_individual_transaction 接口来实现。

该接口仅在交易模式（实盘或模拟盘）下可用，且需要账号开通Level-2行情权限。回测模式下无法获取逐笔成交数据。

以下是具体的实现方法和代码示例。

核心接口说明

get_individual_transaction(stocks=None, data_count=50, start_pos=0, search_direction=1, is_dict=False)

• stocks: 股票代码列表，如 ['600570.SS']。
• data_count: 获取的数据条数，默认为50，最大200。
• is_dict: 返回数据类型。建议设置为 True，返回字典格式，速度更快且易于处理。

策略代码示例

以下策略展示了如何在盘中实时获取指定股票的逐笔成交数据，并分析主动买入（主买）和主动卖出（主卖）的成交量。

def initialize(context):
    # 设置要分析的股票代码
    g.security = '600570.SS'
    # 设置股票池
    set_universe(g.security)
    
    # 设置定时任务，例如每3秒运行一次（模拟tick级别监控）
    # 注意：run_interval 仅在交易模式有效
    run_interval(context, analyze_transaction, seconds=3)

def analyze_transaction(context):
    """
    自定义函数：获取并分析逐笔成交数据
    """
    # 获取最新的50条逐笔成交数据，使用字典格式返回以提高性能
    # 注意：此函数仅在交易模块可用，回测中调用会报错或返回None
    transactions = get_individual_transaction(stocks=[g.security], data_count=50, is_dict=True)
    
    if transactions is None:
        log.info("未获取到逐笔成交数据，请检查是否为交易时间或是否有L2权限。")
        return

    # 获取特定股票的数据
    # 返回格式示例: {'600570.SS': [[时间, 价格, 量, ...], ...], 'fields': [...]}
    stock_data = transactions.get(g.security)
    fields = transactions.get('fields')
    
    if not stock_data:
        return

    # 字段索引映射 (根据文档)
    # fields通常为: ["business_time", "hq_px", "business_amount", "trade_index", 
    #              "business_direction", "buy_no", "sell_no", "trans_flag", ...]
    # business_direction: 1-买(主买), 2-卖(主卖)
    
    try:
        idx_price = fields.index("hq_px")
        idx_amount = fields.index("business_amount")
        idx_direction = fields.index("business_direction")
        idx_time = fields.index("business_time")
    except ValueError:
        log.error("字段解析错误")
        return

    buy_vol = 0  # 主买量
    sell_vol = 0 # 主卖量
    latest_price = 0
    
    # 遍历逐笔数据进行统计
    for item in stock_data:
        direction = item[idx_direction] # 成交方向
        amount = item[idx_amount]       # 成交数量
        price = item[idx_price]         # 成交价格
        
        # 记录最新一笔的价格
        latest_price = price
        
        # 统计主买主卖
        # 注意：返回的数据类型可能为字符串或数字，建议转换
        if str(direction) == '1':
            buy_vol += int(amount)
        elif str(direction) == '2':
            sell_vol += int(amount)
            
    # 打印分析结果
    log.info("股票: %s, 最新价: %s" % (g.security, latest_price))
    log.info("最近50笔统计 - 主买量: %s, 主卖量: %s" % (buy_vol, sell_vol))
    
    # 简单的策略逻辑示例：如果主买量远大于主卖量，且价格上涨，则执行买入
    # (此处仅为逻辑演示，实际交易需更严谨的判断)
    if buy_vol > sell_vol * 1.5:
        log.info("主买力量强劲，触发信号监测...")

def handle_data(context, data):
    # 日线/分钟线级别的逻辑可在此处编写
    pass

关键字段解析

在 get_individual_transaction 返回的数据中，以下字段对于分析最为关键：

1. business_direction (成交方向):
   • 1: 买 (主动买入，内盘被吃，股价倾向于上涨)。
   • 2: 卖 (主动卖出，外盘被吃，股价倾向于下跌)。
   • 0: 未知/中性。
2. business_amount (成交数量): 该笔成交的股数。
3. hq_px (成交价格): 该笔成交的价格。
4. business_time (时间戳): 毫秒级时间戳，用于判断数据的时效性。

注意事项

1. 环境限制: 该代码不能在回测中运行。如果在回测环境中调用 get_individual_transaction，通常会返回 None 或报错。请在“模拟交易”或“实盘交易”中调试。
2. 权限要求: 必须确认您的PTrade账号开通了Level-2行情权限。如果没有L2权限，该接口将返回空数据。
3. 数据频率: 逐笔数据量非常大，建议使用 is_dict=True 以字典形式获取，处理速度比DataFrame快得多。
4. 调用频率: 虽然示例使用了 run_interval 每3秒调用一次，但在高频策略中，也可以使用 tick_data(context, data) 事件驱动函数，该函数会在有新的Tick推送时自动触发（注意 tick_data 推送的是快照Tick，内部也可以调用逐笔接口）。

Q&A

Q: 为什么我在回测中运行这段代码没有数据？
A: get_individual_transaction 是交易专用接口，依赖于实时的Level-2行情流。回测环境通常只提供K线数据（分钟/日线），不支持逐笔成交的历史回放。

Q: 如何判断一笔成交是大单还是小单？
A: 可以通过 business_amount (成交量) 乘以 hq_px (成交价) 计算成交金额。通常策略会设定一个阈值（例如50万元或100万元），超过该阈值的被视为大单。

Q: tick_data 函数和 get_individual_transaction 有什么区别？
A: tick_data(context, data) 是一个事件回调函数，每3秒左右触发一次，推送的是行情快照（Tick）。get_individual_transaction 是一个主动查询函数，用于获取详细的逐笔成交流水。通常在 tick_data 或定时任务中调用 get_individual_transaction 来进行更细致的盘口分析。