第四章：核心算法一：TWAP（时间加权平均价格）算法

各位同学，今天我们来聊聊TWAP。这个算法在量化交易圈子里，可以说是最基础、最经典的大单拆解策略之一。我刚开始做算法交易那会儿，第一个上生产环境的策略就是TWAP，所以对它感情挺深的。

TWAP的全称是Time-Weighted Average Price，时间加权平均价格。说白了，就是把一个大订单，按照时间均匀地拆成小单子，然后分批执行。目标很简单——让我们的成交均价，尽量接近这段时间的市场平均价。

一、数学公式：其实很简单

TWAP的数学表达，我个人觉得是算法交易里最直观的。你想想看，我们要把总量为Q的订单，在T个时间切片内执行完。每个切片分配的数量就是：

q = Q / T

其中：

• Q：总订单数量
• T：总时间切片数（比如240个1分钟切片）
• q：每个切片要执行的量

那最终的TWAP价格怎么算？其实就是每个切片成交价格的加权平均，权重就是每个切片成交的数量：

TWAP = Σ(P_i × q_i) / Σ(q_i)

这里P_i是第i个切片的成交价格，q_i是第i个切片实际成交的数量。如果每个切片都完全成交，那TWAP就等于所有切片价格的算术平均。

二、适用场景：低流动性、非紧急订单

TWAP最适合什么场景？我总结了两类：

1. 低流动性市场：比如某些小盘股、债券、或者流动性差的期货合约。你一个大单砸进去，直接就把价格打飞了。TWAP帮你把单子拆碎，慢慢喂给市场。
2. 非紧急订单：你不急着在几分钟内完成交易。比如你是做指数增强的，今天要调仓，但可以慢慢来，花一整天甚至几天完成。

我记得有一次做港股交易，某只小盘股一天成交量才几百万港币。客户要买两百万，如果用市价单直接冲，估计能把股价拉高5%。后来我们用TWAP，从早上9:30一直拆到下午3:55，最后成交均价只比开盘价高了0.3%。客户很满意。

三、局限性：TWAP不是万能的

任何算法都有短板，TWAP也不例外。我把它的问题归纳为三点：

局限性	具体表现	我的建议
无视市场微观结构	TWAP只关心时间，不关心订单簿的深度、买卖盘口的变化。如果某个时刻买单特别少，它还是会硬塞进去，导致滑点。	可以结合VWAP或者加入流动性检测逻辑
容易被对手盘识别	因为每个时间切片都均匀下单，对手盘如果监测到你的规律，可以在你下单前挂单等你。	加入随机化，比如每个切片的时间点偏移几秒
无法应对极端行情	如果市场突然暴跌，TWAP不会暂停或调整，依然按节奏下单。这可能导致你在下跌过程中不断买入。	设置价格保护阈值，超过一定幅度就暂停

嗯，这里要特别说一下第一条。我曾经在回测时发现TWAP表现很好，但实盘时滑点比预期大了一倍。后来排查发现，是因为回测时假设每个切片都能按中间价成交，但实盘时某些切片正好遇到买单稀薄，导致成交价被拉高。从那以后，我所有TWAP策略都会加上一个「流动性过滤器」——如果当前盘口的买单深度不够，就延迟下单。

四、代码实现：一个简单的TWAP引擎

下面我给一个Python实现。这个版本我做了简化，但核心逻辑都在。你可以在自己的回测框架里直接套用。

import time
import random
from datetime import datetime, timedelta

class TWAPEngine:
    """
    一个简单的TWAP执行引擎
    """
    def __init__(self, total_qty, total_seconds, slices=240):
        self.total_qty = total_qty          # 总订单量
        self.total_seconds = total_seconds  # 总执行时间（秒）
        self.slices = slices                # 切片数
        self.slice_qty = total_qty / slices # 每个切片的理论量
        self.slice_interval = total_seconds / slices  # 每个切片的间隔
        
    def generate_schedule(self):
        """
        生成执行计划
        """
        schedule = []
        for i in range(self.slices):
            # 这里可以加入随机偏移，避免被识别
            delay = random.uniform(-0.5, 0.5)  # 偏移±0.5秒
            exec_time = i * self.slice_interval + delay
            schedule.append({
                'time': exec_time,
                'qty': self.slice_qty
            })
        return schedule
    
    def execute(self, market_api):
        """
        执行TWAP策略
        market_api: 模拟的市场接口，需要实现 place_order() 方法
        """
        schedule = self.generate_schedule()
        executed_qty = 0
        total_cost = 0
        
        for task in schedule:
            # 等待到指定时间
            time.sleep(task['time'] - (datetime.now() - start_time).total_seconds())
            
            # 检查市场状态（简化版）
            if market_api.is_market_open():
                # 下单
                order = market_api.place_order(
                    qty=task['qty'],
                    order_type='limit',
                    price=market_api.get_current_price()
                )
                # 记录成交
                executed_qty += order['filled_qty']
                total_cost += order['filled_qty'] * order['avg_price']
        
        # 计算TWAP
        if executed_qty > 0:
            twap_price = total_cost / executed_qty
        else:
            twap_price = 0
            
        return {
            'executed_qty': executed_qty,
            'twap_price': twap_price,
            'total_cost': total_cost
        }

# 使用示例
if __name__ == '__main__':
    engine = TWAPEngine(total_qty=100000, total_seconds=3600, slices=60)
    # 假设 market_api 已经初始化
    # result = engine.execute(market_api)
    # print(f"成交数量: {result['executed_qty']}, TWAP价格: {result['twap_price']}")

五、知识体系图

下面我用一张SVG图，把TWAP的核心逻辑串起来。你可以把它当作一个快速参考。

这张图把TWAP的输入、核心逻辑、执行过程、输出，以及适用场景和局限性都串起来了。你可以把它打印出来贴在工位上，写代码时瞄一眼，思路会清晰很多。

---

好了，TWAP就讲到这里。说白了，它是一个「笨但可靠」的算法。在大多数常规场景下，它都能很好地完成任务。但如果你遇到的是高流动性、或者需要抢单的场景，那就要考虑下一章要讲的VWAP或者更激进的策略了。

记住一点：没有最好的算法，只有最合适的算法。TWAP是你的工具箱里最基础的那把扳手——不一定最漂亮，但一定最常用。