4. 时间加权平均价格算法（TWAP）：原理、适用场景、Python实现

4.1 什么是TWAP？说白了就是“时间切片”

TWAP，全称Time-Weighted Average Price。我习惯叫它“时间加权平均价格算法”。

它的核心思想很简单：把一个大单拆成很多小单，然后均匀地撒在时间轴上。

举个例子。你想买100万股某只股票。如果一口气全买进去，市场直接被你拉飞了。你想想看，这得多亏？

TWAP的做法是：把交易时间分成N个等长的时间段，比如每5分钟下一笔单。每笔单的量都一样。这样市场冲击就被分散了。

嗯，这里要注意：TWAP不关心价格波动。它只关心时间。说白了，它就是个“定时定量”的机器人。

4.2 原理：数学上怎么算？

TWAP的目标是让成交均价尽量接近这段时间内的市场平均价。数学表达式是这样的：

TWAP = (P₁ + P₂ + ... + Pₙ) / N

其中Pᵢ是每个时间点的价格，N是时间点总数。

但实际执行时，我们更关心的是如何拆单。拆单公式很简单：

每单数量 = 总数量 / 总时间段数

举个例子：

参数	值
总交易量	1,000,000 股
交易时长	2小时（120分钟）
切片间隔	5分钟
总切片数	24个
每单数量	1,000,000 / 24 ≈ 41,667 股

我在项目中遇到过一个问题：如果最后一笔没成交完怎么办？我的做法是——把剩余量在最后一个切片里全部打出去。当然，这需要配合风控逻辑。

4.3 适用场景：什么时候用TWAP？

TWAP不是万能的。我总结了几种最适合的场景：

• 流动性好的品种：比如大盘蓝筹股、主流ETF。这些品种你拆成小单，市场能消化。
• 非紧急交易：你不急着在某个价格买到，只是想把单子“润物细无声”地执行掉。
• 被动型策略：比如指数跟踪、定投。这些策略本身就不追求择时。
• 测试市场深度：有时候我想摸摸市场的底，TWAP是个好工具。

4.4 Python实现：从零写一个TWAP引擎

下面我给出一个完整的TWAP实现。代码不长，但该有的都有了。

import time
import datetime
import pandas as pd
import numpy as np

class TWAPExecutor:
    """
    TWAP执行引擎
    """
    def __init__(self, total_quantity, total_seconds, slice_interval=60):
        """
        :param total_quantity: 总交易量
        :param total_seconds: 总交易时长（秒）
        :param slice_interval: 切片间隔（秒），默认60秒
        """
        self.total_qty = total_quantity
        self.total_sec = total_seconds
        self.interval = slice_interval
        
        # 计算切片数量
        self.num_slices = total_seconds // slice_interval
        # 每片数量
        self.qty_per_slice = total_quantity / self.num_slices
        
        print(f"[TWAP] 总切片数: {self.num_slices}")
        print(f"[TWAP] 每片数量: {self.qty_per_slice:.2f}")
    
    def generate_schedule(self):
        """
        生成交易时间表
        """
        schedule = []
        current_time = datetime.datetime.now()
        
        for i in range(self.num_slices):
            exec_time = current_time + datetime.timedelta(seconds=i * self.interval)
            schedule.append({
                'slice_id': i + 1,
                'exec_time': exec_time,
                'quantity': self.qty_per_slice
            })
        
        return pd.DataFrame(schedule)
    
    def execute(self, market_data_feed):
        """
        执行TWAP策略
        :param market_data_feed: 模拟行情数据，实际中替换为API
        """
        schedule = self.generate_schedule()
        executed_qty = 0
        total_cost = 0.0
        
        print("\n[TWAP] 开始执行...")
        
        for _, row in schedule.iterrows():
            # 模拟等待到指定时间
            # 实际中这里会sleep到指定时间点
            time.sleep(0.1)  # 演示用
            
            # 获取当前市场价格（模拟）
            current_price = market_data_feed.get_price()
            
            # 执行交易
            trade_qty = row['quantity']
            trade_cost = trade_qty * current_price
            
            executed_qty += trade_qty
            total_cost += trade_cost
            
            # 打印执行日志
            print(f"[执行] 切片{row['slice_id']:2d} | "
                  f"时间{row['exec_time'].strftime('%H:%M:%S')} | "
                  f"价格{current_price:.2f} | "
                  f"数量{trade_qty:.0f} | "
                  f"累计成交{executed_qty:.0f}")
        
        # 处理剩余量
        remaining = self.total_qty - executed_qty
        if remaining > 0:
            print(f"\n[警告] 剩余{remaining:.0f}股未成交，执行尾单处理")
            # 这里可以加尾单逻辑
        
        avg_price = total_cost / executed_qty if executed_qty > 0 else 0
        print(f"\n[TWAP] 执行完成")
        print(f"[TWAP] 总成交: {executed_qty:.0f} 股")
        print(f"[TWAP] 均价: {avg_price:.2f}")
        
        return executed_qty, avg_price


# 模拟行情数据
class MockMarketData:
    def __init__(self, base_price=100.0, volatility=0.01):
        self.base = base_price
        self.vol = volatility
    
    def get_price(self):
        # 模拟价格随机波动
        return self.base * (1 + np.random.randn() * self.vol)


# 使用示例
if __name__ == "__main__":
    # 参数设置
    total_qty = 1000000  # 100万股
    total_sec = 7200     # 2小时
    interval = 300       # 5分钟切片
    
    # 创建执行器
    executor = TWAPExecutor(total_qty, total_sec, interval)
    
    # 创建模拟行情
    market = MockMarketData(base_price=100.0, volatility=0.005)
    
    # 执行
    executor.execute(market)

4.5 TWAP的优缺点

我做了个表格，方便你对比：

优点	缺点
实现简单，逻辑清晰	不关注价格，可能买在最高点
市场冲击小	在趋势行情中表现差
适合大单拆单	对流动性敏感
可预测性强	无法利用短期价格优势

4.6 核心逻辑流程图

下面我用SVG画了一张TWAP的核心流程图，帮你理清整个执行链路：

4.7 几个实战中的坑

最后，我分享几个实战中踩过的坑：

• 时间精度问题：我曾经用time.sleep()做定时，结果系统时钟漂移导致切片错位。后来改用NTP同步+绝对时间戳。
• 滑点吞噬利润：TWAP不控制滑点。在快速波动的市场中，滑点可能吃掉所有收益。我建议配合限价单使用。
• 不要死板执行：如果市场突然出现大单砸盘，TWAP还在傻傻地买，那就亏大了。我习惯加一个“暂停/恢复”机制。