第2章 TWAP算法详解

TWAP算法详解：时间加权平均价格算法

各位同学，今天我们来聊聊TWAP算法。

TWAP，全称Time-Weighted Average Price，时间加权平均价格。说白了，就是把一个大订单拆成很多小单，在固定的时间段内均匀地发出去。我刚开始做量化交易那会儿，觉得这玩意儿太简单了，不就是定时定量嘛。后来在实盘中被市场狠狠教育过，才明白里面的门道有多深。

一、TWAP的核心原理

TWAP的数学表达其实很朴素：

TWAP = (P1 + P2 + ... + Pn) / n

其中P1到Pn是每个时间点的成交价格。我们的目标，就是让最终成交均价尽可能接近这个值。

具体怎么操作呢？假设你要在1小时内买入10万股的某只股票。TWAP的做法是：

把1小时切成N个时间片，比如60个（每分钟一个）
每个时间片只发 100000/60 ≈ 1667 股
每个时间片内，再随机选个时机发单

嗯，这里要注意。很多新手以为TWAP就是机械地每秒发单，那你就大错特错了。我见过有人这么干，结果被市场盯上了，每次发单都被吃，滑点惨不忍睹。

核心要点：TWAP追求的是时间上的均匀分布，而不是价格上的最优。它假设市场在统计意义上是有效的，长期来看价格会回归均值。

二、适用场景

TWAP不是什么万能药。我个人习惯把它用在以下几个场景：

低流动性品种：比如一些小盘股、冷门期货合约。你一次性砸进去，价格直接被打飞。TWAP慢慢磨，市场还能消化。
大额订单：超过日均成交量5%的订单，我建议优先考虑TWAP。我在做股指期货套利时，单笔经常上千手，不用TWAP根本没法做。
被动型策略：如果你只是做指数跟踪、ETF申赎，不需要抢跑，TWAP是最省心的选择。
市场波动平稳时：如果市场风平浪静，TWAP表现很好。但要是遇到突发消息，嗯...后面我会讲怎么处理。

我的经验：TWAP最适合用在开盘后30分钟到收盘前30分钟这个时间段。开盘和收盘那会儿，市场情绪太激烈，TWAP容易被带偏。

三、优缺点分析

优点	缺点
实现简单，代码量少	完全不考虑市场微观结构
可预测性强，风控容易做	遇到趋势行情，滑点可能很大
对市场冲击小	无法利用短期价格机会
适合监管合规要求高的场景	执行效率偏低，时间成本高

我曾经在一个趋势明显的行情里硬用TWAP，结果均价比VWAP高了0.3%。那笔交易亏了不少。所以你看，任何算法都有它的边界。

避坑指南：千万不要在单边行情中死用TWAP。如果市场在持续上涨，你还在均匀卖出，那不是傻吗？我曾经吃过这个亏，后来给TWAP加了个价格偏离检测，超过阈值就暂停。

四、TWAP的变体与改进

基础的TWAP太死板了。我在实战中一般会用它的几个变体：

随机化TWAP：每个时间片的量在均值附近随机波动，防止被识别
自适应TWAP：根据市场波动率动态调整时间片长度
带偏移的TWAP：在开盘和收盘时适当减少量，中间时段增加

你想想看，如果你每次都在整点零分零秒发单，对手盘一眼就能看出你的规律。随机化一下，至少能增加对手的识别成本。

五、Python实现一个简单的TWAP

好了，理论说完了，咱们直接上代码。这是我个人比较喜欢的一个实现方式，简洁但实用：

import time
import random
from datetime import datetime, timedelta

class SimpleTWAP:
    """
    一个简单的TWAP算法实现
    适合小规模订单，日均成交量在100万以下的品种
    """
    
    def __init__(self, total_quantity, start_time, end_time, 
                 num_slices=20, randomize=True):
        """
        :param total_quantity: 总交易量
        :param start_time: 开始时间 (datetime对象)
        :param end_time: 结束时间 (datetime对象)
        :param num_slices: 时间片数量
        :param randomize: 是否随机化每片数量
        """
        self.total_qty = total_quantity
        self.start_time = start_time
        self.end_time = end_time
        self.num_slices = num_slices
        self.randomize = randomize
        
        # 计算每个时间片的基础量
        self.base_qty = total_quantity / num_slices
        
        # 计算时间间隔
        total_seconds = (end_time - start_time).total_seconds()
        self.interval = total_seconds / num_slices
        
        print(f"[TWAP] 总时间: {total_seconds:.0f}秒, "
              f"每片间隔: {self.interval:.1f}秒")
    
    def generate_schedule(self):
        """生成交易计划"""
        schedule = []
        current_time = self.start_time
        
        for i in range(self.num_slices):
            # 确定本片交易量
            if self.randomize:
                # 在基础量的80%-120%之间随机
                factor = random.uniform(0.8, 1.2)
                slice_qty = self.base_qty * factor
            else:
                slice_qty = self.base_qty
            
            # 在本片内随机选一个执行时间
            offset = random.uniform(0, self.interval * 0.8)
            exec_time = current_time + timedelta(seconds=offset)
            
            schedule.append({
                'slice_id': i + 1,
                'time': exec_time,
                'quantity': round(slice_qty, 2),
                'status': 'pending'
            })
            
            current_time += timedelta(seconds=self.interval)
        
        return schedule
    
    def execute(self, market_data_func):
        """
        执行TWAP
        :param market_data_func: 获取市场数据的函数，返回当前价格
        """
        schedule = self.generate_schedule()
        executed_qty = 0
        total_cost = 0.0
        
        print(f"\n[TWAP] 开始执行，共{len(schedule)}个时间片")
        print("-" * 50)
        
        for slice_info in schedule:
            # 等待到执行时间
            now = datetime.now()
            wait_seconds = (slice_info['time'] - now).total_seconds()
            
            if wait_seconds > 0:
                time.sleep(wait_seconds)
            
            # 获取当前价格并执行
            current_price = market_data_func()
            slice_qty = slice_info['quantity']
            
            # 模拟成交
            # 这里可以接入真实的交易接口
            executed_qty += slice_qty
            total_cost += slice_qty * current_price
            
            avg_price = total_cost / executed_qty
            
            print(f"  片{slice_info['slice_id']:2d}: "
                  f"时间={slice_info['time'].strftime('%H:%M:%S')}, "
                  f"量={slice_qty:8.2f}, "
                  f"价格={current_price:8.2f}, "
                  f"均价={avg_price:8.2f}")
        
        final_avg = total_cost / executed_qty
        print("-" * 50)
        print(f"[TWAP] 执行完成")
        print(f"  总成交量: {executed_qty:.2f}")
        print(f"  总成本: {total_cost:.2f}")
        print(f"  加权均价: {final_avg:.4f}")
        
        return {
            'avg_price': final_avg,
            'total_qty': executed_qty,
            'total_cost': total_cost,
            'schedule': schedule
        }


# 使用示例
if __name__ == "__main__":
    # 模拟市场数据
    def mock_market_data():
        return random.uniform(10.0, 10.5)
    
    # 创建TWAP实例
    start = datetime.now()
    end = start + timedelta(minutes=5)
    
    twap = SimpleTWAP(
        total_quantity=10000,
        start_time=start,
        end_time=end,
        num_slices=10,
        randomize=True
    )
    
    # 执行
    result = twap.execute(mock_market_data)

代码说明：这个实现里我加了随机化功能。每个时间片的量在80%-120%之间浮动，执行时间也在片内随机偏移。这样能有效防止被对手盘识别出规律。实际生产中，我还会加上价格偏离检测和风控熔断。

六、TWAP的核心逻辑图

下面这张图展示了TWAP算法的完整流程。我画得比较简洁，但核心逻辑都在里面了：

这张图把TWAP的整个流程串起来了。从输入参数开始，到计算每片基础量，再到随机化处理，最后循环执行。嗯，看起来简单，但每一步都有坑。比如随机化因子怎么设，时间偏移取多少，这些都需要根据具体品种和流动性来调。

七、实战中的注意事项

最后，我总结几条实战经验：

时间片数量要合理：太少容易被识别，太多会增加交易成本。我个人习惯在20-60片之间。
一定要加随机化：这是防止被对手盘狙击的关键。我曾经见过有人用固定间隔的TWAP，结果被高频交易者盯上，每次发单都被吃掉。
监控市场状态：如果市场突然放量或者出现大单，TWAP要能暂停或调整。我一般会加个波动率检测，超过阈值就切换到VWAP或者直接暂停。
回测要真实：回测时别用1分钟K线，要用tick级数据。否则你根本看不到TWAP在微观层面的表现。

重要提醒：TWAP不是万能的。在趋势行情、重大消息发布前后、开盘收盘等特殊时段，TWAP的表现会很差。这时候要么不用，要么加保护机制。我吃过亏，所以特别强调这一点。

好了，TWAP的内容就讲到这里。代码我已经给了，你们可以拿去跑跑看。记住，纸上得来终觉浅，绝知此事要躬行。多跑几次回测，多观察实盘表现，你才能真正理解TWAP的精髓。

公众号:蓝海资料掘金营，微信deep3321

📚 课程目录

TWAP算法详解：时间加权平均价格算法

一、TWAP的核心原理

二、适用场景

三、优缺点分析

四、TWAP的变体与改进

五、Python实现一个简单的TWAP

六、TWAP的核心逻辑图

七、实战中的注意事项