策略案例 · 大单拆分（TWAP / VWAP）

本文给出两种常见的大单拆分算法：TWAP（按固定时间间隔平均开仓）和 VWAP（在 TWAP 基础上加入成交量权重动态调整手数），用于降低冲击成本、减小滑点损失。

场景

大交易量用户一次性下单会冲击市场、造成较大滑点。将母单按时间或成交量拆分成多笔子单逐步执行，可以平滑冲击成本。期魔方支持在策略中自定义拆单逻辑，下面给出 TWAP 与 VWAP 两种实现片段。

---

案例一：TWAP 拆单算法

核心逻辑

• 将总开仓手数 volume 拆分为 num_orders 笔子单，每笔固定 order_size 手
• 以 twap_start_time_bull 为起始时间，每隔 1 分钟发一笔
• 每笔取当前收盘价 close_array[-1] 作为限价单价格
• 已执行量达到 volume 即清理状态并退出

代码片段

# 循环执行，循环次数为总订单数减去当前已执行的 TWAP 多头订单数
for _ in range(num_orders - context.twap_order_count_bull):
    # 检查当前已执行的 TWAP 多头交易量是否达到了目标交易量
    if context.twap_executed_volume_bull >= context.volume:
        # 若达到目标交易量，打印提示信息，表示 TWAP 多头交易已完成
        print("TWAP 多头交易完成，总下单量已达到目标")
        # 删除 TWAP 多头交易的开始时间记录
        del context.twap_start_time_bull
        # 删除 TWAP 多头交易的订单计数记录
        del context.twap_order_count_bull
        # 删除 TWAP 多头交易的已执行交易量记录
        del context.twap_executed_volume_bull
        # 函数返回，结束当前操作
        return
    # 计算下一次交易的时间，即 TWAP 多头交易开始时间加上当前已执行的订单数对应的分钟数
    next_trade_time = context.twap_start_time_bull + timedelta(minutes=context.twap_order_count_bull)
    # 检查当前时间是否已经达到或超过下一次交易的时间
    if current_time >= next_trade_time:
        # 获取当前的收盘价作为下单价格
        current_price = context.close_array[-1]
        # 计算实际的下单量，取预设订单量和剩余目标交易量中的较小值
        actual_order_size = min(order_size, context.volume - context.twap_executed_volume_bull)
        # 打印当前下单信息，包括订单序号、下单价格和下单手数
        print(f"TWAP 多单 {context.twap_order_count_bull+1}/{num_orders}: 价格 {current_price}, 手数 {actual_order_size}")
        # 执行买入开仓操作，传入实际下单量、基础合约 ID 和下单价格
        buy_open(actual_order_size, context.base_instrument_id, price=current_price)
        # 累加已执行的 TWAP 多头订单数
        context.twap_order_count_bull += 1
        # 累加已执行的 TWAP 多头交易量
        context.twap_executed_volume_bull += actual_order_size

关键变量

变量	含义
context.twap_start_time_bull	TWAP 多头开始时间，用于计算下一次下单时点
context.twap_order_count_bull	已执行的 TWAP 多头订单数
context.twap_executed_volume_bull	已执行的 TWAP 多头累计手数
context.volume	目标总开仓手数
num_orders	总拆单次数
order_size	每笔基础下单手数

---

案例二：VWAP 拆单算法

核心逻辑

VWAP 在 TWAP 的固定时间间隔基础上，加入成交量权重动态调整每次开仓手数：

1. 以 total_time / interval 计算总拆单次数 times
2. 每笔基础手数 volume_per_order = int(volume / times)
3. 每次循环获取最新 K 线的成交量，滚动更新成交量序列 volume_data
4. 计算实时均量 avg_volume，调整系数 volume_factor = max(0.5, min(1, current_volume / avg_volume))
5. 按调整系数下单，并确保：① 不超剩余量；② 最少 1 手；③ 不超总手数

代码片段

# 异步 VWAP 拆单算法：根据成交量调整开仓，该算法会在指定的总时间内，按照一定的时间间隔进行拆单操作，并根据实时成交量动态调整每次下单的手数
async def async_vwap_order(context, direction, volume, total_time, interval):
    # 计算总拆单次数，通过总时间除以时间间隔得到，并将结果转换为整数
    times = int(total_time / interval)
    # 计算每次下单的基础手数，通过总开仓手数除以拆单次数得到，并将结果转换为整数
    volume_per_order = int(volume / times)
    # 初始化一个变量，用于记录已经开仓的手数，初始值为 0
    total_opened = 0
    # 初始化成交量数组，调用 get_kline 函数获取指定交易标的、周期和时间段的 K 线数据
    klines = get_kline(context.base_instrument_id, context.base_period, context.vwap_period)
    # 从 K 线数据中提取成交量数据，并将其转换为numpy数组
    volume_data = np.array(klines.get("volume"))
    # 开始循环，循环次数为拆单次数
    for i in range(times):
        # 获取最新的 K 线数据，只获取最近的一根 K 线
        latest_klines = get_kline(context.base_instrument_id, context.base_period, 1)
        # 从最新的 K 线数据中提取最新的成交量
        current_volume = latest_klines.get("volume")[-1]
        # 更新成交量数组，将数组元素向后滚动一位，去掉最早的成交量数据
        volume_data = np.roll(volume_data, -1)
        # 将最新的成交量数据添加到数组的最后一位
        volume_data[-1] = current_volume
        # 实时计算成交量数组的平均值，作为当前的均量
        avg_volume = np.mean(volume_data)
        # 计算剩余需要开仓的手数，通过总开仓手数减去已经开仓的手数得到
        remaining_volume = volume - total_opened
        # 判断剩余手数是否小于等于 0，如果是，则说明所有仓位已开完，停止开仓
        if remaining_volume <= 0:
            # 记录日志，提示所有仓位已开完，停止开仓
            put_log("所有仓位已开完，停止开仓", "INFO")
            # 跳出循环
            break
        # 动态调整下单量的系数，将其限制在 0.5 到 1 倍之间
        volume_factor = max(0.5, min(1, current_volume / avg_volume))
        # 根据调整系数计算调整后的下单手数
        adjusted_volume = int(volume_per_order * volume_factor)
        # 确保调整后的下单手数不超过剩余需要开仓的手数
        adjusted_volume = min(adjusted_volume, remaining_volume)
        # 确保调整后的下单手数最小为 1 手
        adjusted_volume = max(1, adjusted_volume)
        # 防止超开总手数，如果调整后的下单手数加上已经开仓的手数超过总开仓手数，则进行调整
        if total_opened + adjusted_volume > volume:
            adjusted_volume = volume - total_opened
        # 根据交易方向进行开仓操作，如果方向为 1，则进行买入开仓
        if direction == 1:
            buy_open(adjusted_volume, context.base_instrument_id)
        # 如果方向为 -1，则进行卖出开仓
        elif direction == -1:
            sell_open(adjusted_volume, context.base_instrument_id)
            # 更新已经开仓的手数，将本次开仓的手数累加到总开仓手数中
        total_opened += adjusted_volume

参数说明

参数	含义
direction	交易方向：1 买入开仓，-1 卖出开仓
volume	目标总开仓手数
total_time	拆单总时长
interval	拆单时间间隔
context.vwap_period	用于取历史成交量序列的 K 线周期

---

关键要点

1. TWAP vs VWAP：TWAP 按等时间间隔、等手数拆分，实现简单；VWAP 引入实时成交量权重，成交量放大时多下、萎缩时少下，更贴合市场节奏
2. 限价 vs 市价：TWAP 示例使用 price=current_price 限价单，避免市价单冲击；VWAP 示例使用默认市价单，实际可按需改造
3. 状态清理：TWAP 达到目标后用 del 清理 context 上的临时状态，避免下次循环误判；VWAP 通过 remaining_volume <= 0 提前 break 退出
4. 成交量滚动窗口：VWAP 用 np.roll(volume_data, -1) 滚动更新成交量序列，保持窗口长度恒定，均量计算更平稳
5. 手数上下界保护：VWAP 对 adjusted_volume 做了三重保护——不超过剩余量、不少于 1 手、不超过总手数，防止超开或空单
6. 异步执行：VWAP 用 async def 声明，适合在事件驱动的策略主循环中非阻塞调用；TWAP 片段为同步循环，需配合 on_tick / on_bar 的时钟判断触发

---

相关阅读

• [核心API 历史K线数据]
• [核心API 账户持仓查询]
• [核心API 衍生数据]
• [案例：双均线金叉死叉]
• [案例：网格策略]
• [案例：日内清仓与套利]