PTrade 策略技巧：编写一个限价单发出后 N 秒未成交即自动撤单的函数

在PTrade中实现“限价单超过30秒未成交自动撤单”的功能，主要思路如下：

1. 记录订单时间：在下单时，将订单ID和当前时间戳存入一个全局字典（如 g.pending_orders）中。
2. 轮询检查：使用 run_interval 函数（仅实盘/模拟盘有效）设定一个较短的检查周期（例如每3秒检查一次）。
3. 判断与撤单：在检查函数中，遍历记录的订单，计算（当前时间 - 下单时间）。如果超过30秒且订单状态仍为未成交（或部分成交），则调用 cancel_order 进行撤单，并从记录中移除。

核心代码实现

以下是一个完整的策略示例代码。该代码在 initialize 中设置了轮询任务，在 handle_data 中模拟下单，并包含处理超时的逻辑。

import time

def initialize(context):
    # 设置股票池
    g.security = '600570.SS'
    set_universe(g.security)
    
    # 定义一个全局字典，用于存储待监测的订单
    # 格式: { 'order_id':下单时间戳 }
    g.pending_orders = {}
    
    # 设定一个标志位，仅用于演示只下一笔单
    g.order_placed = False
    
    # 开启一个定时任务，每隔3秒运行一次 check_timeout_orders 函数
    # 注意：run_interval 仅在交易模式（实盘/模拟盘）下生效
    run_interval(context, check_timeout_orders, seconds=3)

def handle_data(context, data):
    # --- 演示下单逻辑 ---
    if not g.order_placed:
        # 获取当前价格
        last_price = data[g.security]['close']
        # 下一个限价单（故意挂低一点，模拟未成交的情况）
        limit_price = last_price * 0.98 
        
        # 下单并获取订单ID
        order_id = order(g.security, 100, limit_price=limit_price)
        
        if order_id:
            # 记录订单ID和当前时间
            g.pending_orders[order_id] = time.time()
            log.info("已下单 ID: %s, 价格: %s, 等待成交..." % (order_id, limit_price))
            g.order_placed = True

def check_timeout_orders(context):
    """
    每3秒执行一次，检查是否有订单超时
    """
    # 获取当前时间
    current_time = time.time()
    # 待移除的订单列表（避免在遍历字典时修改字典）
    orders_to_remove = []
    
    for order_id, create_time in g.pending_orders.items():
        # 获取订单对象
        # 注意：get_order 返回的是一个列表，包含一个Order对象
        order_list = get_order(order_id)
        
        if not order_list:
            continue
            
        order_obj = order_list[0]
        status = order_obj.status
        
        # 订单状态说明：
        # '2': 已报 (Open)
        # '7': 部成 (Partially Filled)
        # '8': 已成 (Filled)
        # '6': 已撤 (Cancelled)
        # '9': 废单 (Rejected)
        
        # 如果订单已经完全成交、已撤单或废单，则不再监控
        if status in ['8', '6', '9']:
            orders_to_remove.append(order_id)
            continue
            
        # 计算已过去的时间
        elapsed_time = current_time - create_time
        
        # 如果超过30秒，且订单处于"已报"或"部成"状态，则撤单
        if elapsed_time > 30:
            log.info("订单 %s 已挂单 %.1f 秒，超过30秒未完全成交，正在撤单..." % (order_id, elapsed_time))
            cancel_order(order_id)
            # 标记为移除（撤单指令发出后，下一轮检查会因为状态变为已撤而移除，这里直接移除也可以）
            orders_to_remove.append(order_id)
    
    # 从监控字典中清理掉已处理的订单
    for oid in orders_to_remove:
        if oid in g.pending_orders:
            del g.pending_orders[oid]

关键点说明

1. run_interval 函数：

   • 这是实现“秒级”监控的关键。普通的 handle_data 在日线或分钟线回测中频率不够高。
   • run_interval 只能在交易模块（实盘或仿真交易）中使用，回测模式下通常无法精确模拟30秒的撤单逻辑（回测最小单位通常是分钟）。

2. get_order(order_id)：

   • PTrade API 中 get_order 返回的是一个 list，所以代码中使用了 order_list[0] 来获取具体的 Order 对象。

3. 状态判断：

   • 我们只对状态为 '2' (已报) 或 '7' (部成) 的订单进行撤单操作。
   • 如果订单已经是 '8' (已成)，则不需要撤单，直接从监控列表中移除即可。

4. 时间管理：

   • 使用了 Python 标准库 time.time() 来获取Unix时间戳进行秒级计算，这是最简单有效的方法。

Q&A

Q: 这个策略可以在回测中使用吗？
A: 不可以精确模拟。run_interval 仅在交易模式下有效。在回测中，handle_data 最快是每分钟运行一次。如果在回测中需要模拟，只能在 handle_data 中判断“如果订单在上一分钟产生且未成交，则撤单”，但这代表的是60秒超时，而非30秒。

Q: 如果撤单失败怎么办？
A: cancel_order 发出指令后，订单状态不会立即改变。代码中的逻辑会在下一次 run_interval 触发时再次检查。如果状态变成了 '6' (已撤)，它会被移出监控列表；如果状态还是 '2' (已报) 且时间仍超时，它会再次尝试撤单。这形成了一种简单的重试机制。

Q: 为什么使用 g.pending_orders 而不是直接遍历 get_orders()?
A: get_orders() 会返回当日所有订单。随着交易进行，订单数量会越来越多，遍历所有订单效率较低。使用 g.pending_orders 只记录我们需要关注的“活跃”订单，效率更高。