PTrade 高频实战：监控订单生成到成交的系统延时

在PTrade中，计算订单从发送到成交的时间延迟（Latency），通常用于实盘交易中评估策略执行速度或网络状况。

核心思路

1. 记录发送时间：在调用下单函数（如 order）的前一刻或拿到 order_id 的瞬间，记录当前系统时间。
2. 获取成交时间：利用交易主推回调函数 on_trade_response 获取成交回报，其中包含成交时间（business_time）。
3. 计算差值：将成交时间转换为时间戳或 datetime 对象，与发送时间相减，得出延迟。

代码实现

以下是一个完整的策略示例，展示了如何记录下单时间并通过回调计算延迟。

import datetime
import time

def initialize(context):
    # 设置股票池
    g.security = '600570.SS'
    set_universe(g.security)
    
    # 用于存储订单发送时间的字典
    # Key: order_id, Value: datetime对象 (发送时间)
    g.order_send_times = {}
    
    # 限制下单次数，仅作演示用
    g.trade_count = 0

def handle_data(context, data):
    # 仅演示一次下单
    if g.trade_count > 0:
        return

    # 1. 记录发送前的时间 (使用系统当前时间，而非回测时间)
    send_time = datetime.datetime.now()
    
    # 2. 下单
    # 注意：这里使用市价单以确保尽快成交，方便观察延迟
    # 实盘中请根据实际情况调整价格和数量
    order_id = order(g.security, 100)
    
    # 3. 如果下单成功（返回了order_id），记录发送时间
    if order_id is not None:
        g.order_send_times[order_id] = send_time
        log.info("订单已发送，ID: %s，发送时间: %s" % (order_id, send_time))
        g.trade_count += 1

def on_trade_response(context, trade_list):
    """
    交易主推回调函数
    当有成交发生时，该函数会被触发
    """
    for trade in trade_list:
        # 获取成交单对应的 order_id
        order_id = trade.get('order_id')
        
        # 检查该订单是否是我们记录过的
        if order_id in g.order_send_times:
            # 获取当初的发送时间
            send_time = g.order_send_times[order_id]
            
            # 获取成交时间 (字符串格式，例如 '2023-10-27 10:30:05')
            business_time_str = trade.get('business_time')
            
            # 将成交时间字符串转换为 datetime 对象
            # 注意：实盘返回的时间格式通常为 "%Y-%m-%d %H:%M:%S"
            # 如果包含毫秒，需要根据实际情况调整格式
            try:
                trade_time = datetime.datetime.strptime(str(business_time_str), "%Y-%m-%d %H:%M:%S")
                
                # 计算延迟 (单位：秒)
                latency = (trade_time - send_time).total_seconds()
                
                log.info("订单 %s 成交回报收到。" % order_id)
                log.info("发送时间: %s" % send_time)
                log.info("成交时间: %s" % trade_time)
                log.info("成交延迟: %.4f 秒" % latency)
                
                # 计算完毕后，可以从字典中移除，防止内存泄漏
                del g.order_send_times[order_id]
                
            except Exception as e:
                log.error("时间解析或计算错误: %s" % e)

关键点说明

1. datetime.datetime.now():

   • 在实盘中，这是计算延迟的关键，它代表代码运行到这一行的物理时间。
   • 在回测中，datetime.now() 返回的是你运行回测时的服务器系统时间，而不是回测历史时刻。回测中通常不存在真实的网络延迟，成交通常在下一个bar或即时撮合（取决于回测设置），因此计算延迟主要用于实盘性能监控。

2. on_trade_response:

   • 这是PTrade提供的交易主推接口，比在 handle_data 中轮询 get_trades() 响应更快，能更准确地捕捉到成交回报到达策略终端的时刻。
   • trade_list 中的 business_time 是柜台或交易所记录的成交时间。

3. 时间同步:

   • 计算出的延迟包含了：策略逻辑处理时间 + 网络传输时间 + 柜台处理时间 + 交易所撮合时间 + 回报返回时间。
   • 如果 business_time 是交易所时间，而 send_time 是本地服务器时间，两者可能存在时钟偏差（Clock Skew），导致计算结果出现误差（甚至负数）。在极高要求的场景下，需要确保服务器时间与交易所时间高度同步。

4. Python 版本兼容性:

   • 代码使用了 % 格式化字符串，而非 f-string，以确保兼容 Python 3.5 及以上版本。

Q&A

Q: 为什么在回测中计算出的延迟看起来不合理？
A: 回测环境是模拟的。datetime.now() 获取的是当前现实世界的时间，而 business_time 是历史回测数据的时间（例如2020年）。两者相减没有意义。此逻辑仅适用于实盘交易（Trade）模式。

Q: on_trade_response 没有触发怎么办？
A: 请检查策略是否开启了主推接收功能。虽然默认通常开启，但某些券商环境可能需要显式设置。另外，只有订单真正成交（部分成交或全部成交）时才会触发此回调，仅委托但未成交是不会触发的。

Q: 如何计算从“委托”到“柜台确认”的延迟？
A: 可以使用 on_order_response 回调函数。逻辑与上面类似，记录发送时间，然后在 on_order_response 中收到 status 为 "已报" 或 "已确认" 的回报时记录时间并计算差值。