不使用内置算法，如何用 Python 手动实现 TWAP 下单？

在 QMT 中实现 TWAP（时间加权平均价格）策略，核心逻辑是将总委托量均匀分配到指定的时间段内，按固定的时间间隔分批下单。

虽然 QMT 提供了内置的 smart_algo_passorder（智能算法交易）可以直接调用 TWAP 算法，但自己编写代码实现可以让你对下单逻辑、间隔控制和委托价格有更精细的控制。

以下是手动实现 TWAP 分批下单的完整策略代码及详细说明。

TWAP 策略实现逻辑

1. 初始化 (init)：设定目标股票、总买入数量、开始时间、结束时间、分批次数（或时间间隔）。
2. 计算单笔委托量：单笔量 = 总数量 / 分批次数。
3. 事件循环 (handlebar)：
   • 获取当前最新行情时间。
   • 判断是否在交易时间窗口内。
   • 判断是否达到下一次下单的时间点。
   • 如果满足条件，执行下单（passorder），并更新下一次触发时间。
   • 处理最后一笔交易，确保剩余零股全部买入。

完整代码实现

# -*- coding: gbk -*-
import time
import math

def init(ContextInfo):
    # ================= 策略参数设置 =================
    ContextInfo.stock = '600000.SH'  # 目标股票：浦发银行
    ContextInfo.total_volume = 10000 # 总计划买入股数
    ContextInfo.account_id = '600000248' # 资金账号 (请修改为您的实际账号)
    
    # TWAP 时间设置 (格式 HHMMSS)
    ContextInfo.start_time = "093500" # 开始时间
    ContextInfo.end_time = "110000"   # 结束时间
    ContextInfo.interval = 5 * 60     # 下单间隔 (秒)，这里设为5分钟
    
    # ================= 内部变量初始化 =================
    ContextInfo.traded_volume = 0     # 已成交/已委托数量
    ContextInfo.last_order_time = 0   # 上次下单的时间戳
    
    # 计算总时长(秒)和预计分批次数
    # 注意：这里简单计算，未剔除中午休市时间，如跨中午需额外处理
    t_start = time.mktime(time.strptime('2023-01-01 ' + ContextInfo.start_time, "%Y-%m-%d %H%M%S"))
    t_end = time.mktime(time.strptime('2023-01-01 ' + ContextInfo.end_time, "%Y-%m-%d %H%M%S"))
    total_seconds = t_end - t_start
    
    if total_seconds <= 0:
        print("错误：结束时间必须晚于开始时间")
        ContextInfo.is_active = False
        return
    
    # 计算分批次数 (向上取整)
    ContextInfo.batch_count = math.ceil(total_seconds / ContextInfo.interval)
    
    # 计算单次下单量 (向下取整到100的倍数，避免碎股，最后一次补齐)
    raw_per_vol = ContextInfo.total_volume / ContextInfo.batch_count
    ContextInfo.vol_per_order = int(raw_per_vol // 100 * 100)
    
    if ContextInfo.vol_per_order < 100:
        ContextInfo.vol_per_order = 100 # 最小一手
        
    ContextInfo.is_active = True # 策略激活状态
    
    print(f"TWAP策略启动: 总量{ContextInfo.total_volume}, 预计分{ContextInfo.batch_count}批, 每批{ContextInfo.vol_per_order}股")

def handlebar(ContextInfo):
    # 如果策略未激活或已完成，则退出
    if not ContextInfo.is_active:
        return

    # 获取当前K线或Tick的时间戳 (毫秒)
    current_timetag = ContextInfo.get_tick_timetag()
    
    # 转换为 datetime 字符串以便比较 HHMMSS
    current_dt_str = timetag_to_datetime(current_timetag, '%H%M%S')
    current_timestamp = time.time() # 获取系统当前时间戳用于计算间隔
    
    # 1. 判断是否在交易区间内
    if current_dt_str < ContextInfo.start_time:
        return # 还没到开始时间
        
    if current_dt_str > ContextInfo.end_time:
        print("已过结束时间，策略停止")
        ContextInfo.is_active = False
        return

    # 2. 判断是否满足剩余量
    remaining_vol = ContextInfo.total_volume - ContextInfo.traded_volume
    if remaining_vol <= 0:
        print("计划数量已全部执行完毕")
        ContextInfo.is_active = False
        return

    # 3. 判断时间间隔是否到达
    # 如果是第一次下单，或者距离上次下单超过了设定的间隔
    if ContextInfo.last_order_time == 0 or (current_timestamp - ContextInfo.last_order_time >= ContextInfo.interval):
        
        # 计算本次下单量
        order_vol = ContextInfo.vol_per_order
        
        # 如果是最后一次或者剩余量不足单次量，则全买
        # 或者是时间快到了(接近end_time)，强制买入剩余
        time_left = time.mktime(time.strptime('2023-01-01 ' + ContextInfo.end_time, "%Y-%m-%d %H%M%S")) - time.mktime(time.strptime('2023-01-01 ' + current_dt_str, "%Y-%m-%d %H%M%S"))
        
        if remaining_vol < order_vol or time_left < ContextInfo.interval:
            order_vol = remaining_vol
            print("触发最后一次补齐交易")
        
        # 执行下单：这里使用最新价 (prType=5)
        # opType=23(买入), orderType=1101(单股单账号)
        passorder(23, 1101, ContextInfo.account_id, ContextInfo.stock, 5, -1, order_vol, ContextInfo)
        
        # 更新状态
        ContextInfo.traded_volume += order_vol
        ContextInfo.last_order_time = current_timestamp
        
        print(f"TWAP下单触发: 时间{current_dt_str}, 数量{order_vol}, 累计已发单{ContextInfo.traded_volume}/{ContextInfo.total_volume}")

代码关键点解析

1. 时间处理：

   • QMT 的 get_tick_timetag() 返回的是毫秒级时间戳。
   • 使用 timetag_to_datetime 将其转换为 HHMMSS 字符串，方便与设定的 start_time 和 end_time 进行比较。
   • 使用 time.time() 获取系统时间戳来计算 interval（间隔秒数），这样比解析 K 线时间更准确，适合实盘。

2. 手数整取：

   • int(raw_per_vol // 100 * 100)：A股交易通常要求买入数量为 100 的整数倍。这段代码确保了分批的单子是整手的。
   • 尾单处理：代码中包含逻辑 if remaining_vol < order_vol，用于处理最后一次交易，确保将剩余的零头（或不足一手的量）全部买入。

3. 下单函数 (passorder)：

   • opType=23：股票买入。
   • orderType=1101：单股、单账号、普通、股/手方式下单。
   • prType=5：使用最新价下单。如果你希望更激进成交，可以改为 14 (对手价) 或 12 (涨跌停价，即市价)。

4. 运行模式：

   • 此策略需要在实盘模式下运行，并且需要在行情界面或独立运行界面勾选目标股票。
   • handlebar 会随行情 Tick 推送触发，因此时间判断非常精准。

进阶优化建议

如果需要更接近专业级的 TWAP，可以考虑以下改进：

• 跨中午处理：目前的逻辑是简单的 end - start。如果时间段跨越了 11:30 - 13:00，需要手动扣除这 1.5 小时，否则分批数量计算会偏小。
• 价格限制：可以在下单前增加价格判断，例如：if current_price < limit_price: passorder(...)，实现限价 TWAP。
• 随机化：在 interval 和 vol_per_order 上增加小的随机扰动（例如 ±10%），以防止被市场识别出算法单痕迹。

Q&A

Q: 为什么不直接使用 smart_algo_passorder?
A: smart_algo_passorder 是 QMT 封装好的算法，使用非常简单（只需指定算法名称 "TWAP"）。但如果你需要自定义逻辑（例如：只有当大盘上涨时才执行 TWAP，或者每笔间隔根据成交量动态调整），则必须使用上述的手动代码实现方式。

Q: 如何在回测中测试这段代码？
A: 这段代码主要针对实盘设计（使用了 time.time()）。如果在回测中使用，需要将 current_timestamp 的获取方式改为解析 ContextInfo.get_bar_timetag(ContextInfo.barpos)，因为回测是基于历史时间轴的，不能使用系统当前时间。

Q: passorder 下单后如何确认成交？
A: passorder 只是发送委托。要确认成交，需要使用 get_trade_detail_data 查询成交明细，或者在 deal_callback 回调函数中接收成交推送。上述代码仅负责“定时发单”，不负责“确认成交后才发下一笔”。