ptrade 实盘中如何通过 on_trade_response 实时维护高精度本地持仓字典？

为什么需要利用 on_trade_response 维护本地持仓？

在 ptrade 实盘交易中，柜台返回的持仓数据（如通过 context.portfolio.positions 或 get_position 获取）通常存在一定的同步时滞（一般在 6 秒左右）。如果策略采用高频交易、Tick 级交易或需要在成交后立即进行反向操作（如网格交易、日内回转），直接依赖柜台持仓极易导致重复下单或可用资金/持仓不足而委托失败。

通过 on_trade_response（成交主推事件）回调，策略可以在成交发生的毫秒级时间内收到通知。利用该事件增量更新一个本地维护的持仓字典（g.local_positions），可以实现高精度、超低延迟的仓位管理。

核心设计思路

1. 初始化同步：在 initialize 或 before_trading_start 中，通过 get_positions() 获取柜台初始持仓，并同步到本地全局字典 g.local_positions 中。
2. 成交主推监听：实现 on_trade_response(context, trade_list) 函数。当有新成交时，ptrade 会自动回调此函数。
3. 增量更新算法：
   • 买入成交（entrust_bs == '1'）：本地持仓数量增加，持仓成本重新计算。
   • 卖出成交（entrust_bs == '2'）：本地持仓数量减少。若持仓减至 0，则从字典中移除该标的。
4. 异常与策略外成交兼容：主推数据中可能包含策略外手动交易产生的主推（此时 order_id 为空 ""），本地缓存同样需要对其进行增量更新，以保证与账户真实持仓绝对一致。

完整 Python 策略源码

以下是基于 ptrade 标准 API 实现的实时持仓管理策略示例：

def initialize(context):
    # 设置操作的股票池
    g.security = ['600570.SS', '000001.SZ']
    set_universe(g.security)
    
    # 开启接收非本策略产生的成交主推（可选，建议开启以保持全局一致）
    set_parameters(receive_other_response="1")
    
    # 初始化本地持仓缓存字典
    g.local_positions = {}
    
    # 首次同步柜台真实持仓
    sync_positions_from_counter()

def sync_positions_from_counter():
    """从柜台同步初始持仓"""
    counter_positions = get_positions()
    g.local_positions.clear()
    for code, pos in counter_positions.items():
        g.local_positions[code] = {
            'amount': pos.amount,
            'enable_amount': pos.enable_amount,
            'cost_basis': pos.cost_basis
        }
    log.info("【系统初始化】已从柜台同步初始持仓: %s" % str(g.local_positions))

def on_trade_response(context, trade_list):
    """
    成交主推回调函数
    trade_list 结构示例:
    [{'status': '8', 'business_id': '76', 'business_amount': 200, 'order_id': 'xxx', 
      'entrust_no': 700006, 'business_balance': 7390.0, 'business_price': 36.95, 
      'stock_code': '600570.SS', 'entrust_bs': '1', 'business_time': '2022-05-10 15:51:47'}]
    """
    for trade in trade_list:
        code = trade['stock_code']
        bs_direction = trade['entrust_bs'] # '1' 为买，'2' 为卖
        amount = int(trade['business_amount'])
        price = float(trade['business_price'])
        balance = float(trade['business_balance'])
        
        log.info("【成交主推】收到成交回报 -> 标的: %s, 方向: %s, 数量: %d, 价格: %s" % 
                 (code, '买入' if bs_direction == '1' else '卖出', amount, price))
        
        # 如果本地缓存中没有该股票，先初始化
        if code not in g.local_positions:
            g.local_positions[code] = {'amount': 0, 'enable_amount': 0, 'cost_basis': 0.0}
            
        pos = g.local_positions[code]
        
        if bs_direction == '1': # 买入成交
            old_amount = pos['amount']
            new_amount = old_amount + amount
            # 重新计算持仓均价
            if new_amount > 0:
                pos['cost_basis'] = round((pos['cost_basis'] * old_amount + balance) / new_amount, 3)
            pos['amount'] = new_amount
            # 注意：实盘中买入的股票 T+1 可用，此处不增加 enable_amount
            
        elif bs_direction == '2': # 卖出成交
            pos['amount'] = max(0, pos['amount'] - amount)
            pos['enable_amount'] = max(0, pos['enable_amount'] - amount)
            
            # 如果持仓清空，则移除缓存
            if pos['amount'] == 0:
                g.local_positions.pop(code, None)
                
        log.info("【本地持仓更新】当前本地缓存持仓: %s" % str(g.local_positions))

def before_trading_start(context, data):
    # 每日开盘前，强制与柜台做一次全量对账，纠正可能存在的偏差
    sync_positions_from_counter()

def handle_data(context, data):
    # 策略逻辑中，使用 g.local_positions 代替 context.portfolio.positions 进行仓位判断
    for code in g.security:
        local_pos = g.local_positions.get(code, None)
        if local_pos and local_pos['amount'] > 0:
            log.info("【策略盘中】%s 本地高精度持仓数量: %d" % (code, local_pos['amount']))

实盘注意事项

1. T+1 规则限制：A股市场实行 T+1 交易制度。买入成交时，本地缓存的 amount（总持仓）应立即增加，但 enable_amount（可用持仓）在当天不应增加，只有在隔日 before_trading_start 重新同步柜台时才会更新为可用。
2. 每日盘前对账：虽然主推更新非常精准，但为了防止网络丢包、柜台异常等极端情况导致的本地缓存与柜台数据不一致，必须在每日盘前（before_trading_start）调用柜台接口进行一次全量覆盖同步。
3. 多策略共享账户限制：如果同一个资金账户运行了多个不同的 ptrade 策略，建议在 initialize 中设置 set_parameters(receive_other_response="1")，以便本地缓存能够捕获其他策略或手动交易引起的仓位变化。