第十一章：TWAP算法详解
TWAP，全称Time-Weighted Average Price，时间加权平均价格算法。说白了，就是要把一个大单子，均匀地拆成小单子，在指定时间段内慢慢喂给市场。
我刚开始做量化那会儿，觉得TWAP太简单了——不就是把单子切碎嘛。后来在实盘中被市场狠狠教育了一顿，才发现这里面门道不少。今天咱们就把TWAP从头到尾捋一遍。
11.1 算法原理
TWAP的核心思想很朴素：把总成交量均匀分配到每个时间切片上。假设你需要在T分钟内买入Q股，那就每分钟买入Q/T股。
为什么要这么做？
- 降低市场冲击：大单一次性砸进去，价格会被推高，你买得贵
- 隐藏交易意图：市场一看你猛买，就知道有大户进场，会反向操作
- 时间分散风险：价格波动是随机的，分散到不同时间点，平均成本更稳定
我个人习惯把TWAP看作一个时间上的等分器。它不关心价格高低，只关心时间到了没有。到点了，就发单，不管市场是涨是跌。
核心假设：在足够短的时间切片内，市场流动性是充足的，价格波动是随机的。TWAP赌的是——长期来看，你的成交均价会接近这段时间的VWAP（成交量加权均价）。
11.2 数学公式
TWAP的数学表达其实很简单。先看几个关键变量：
| 符号 | 含义 |
|---|---|
| T | 总执行时间（秒） |
| Δt | 时间切片长度（秒） |
| Q | 总成交量（股） |
| N | 切片数量，N = T / Δt |
| q_i | 第i个切片的目标成交量 |
每个切片的目标成交量：
q_i = Q / N就这么简单。每个切片都一样大。
但实际执行中，我们关心的是累计成交量和时间的关系。理想情况下，累计成交量应该是一条直线：
C(t) = (Q / T) × t其中C(t)是到时间t为止的累计成交量。这条线就是TWAP的基准线。
我曾在一次回测中发现，如果市场在某个切片突然放量，你的实际成交线会偏离基准线。这时候就需要偏差修正。修正后的公式：
q_i' = q_i + (C_target(t_i) - C_actual(t_i))说白了，就是前面少买了，后面就多买点，把进度追回来。
我的经验：偏差修正不能太激进。我曾经把修正系数设得太大，结果后半段疯狂追单，反而造成了更大的冲击。建议修正幅度不超过原始切片量的20%。
11.3 Python实现
下面给出一份完整的TWAP算法实现。我习惯用类来封装，方便扩展。
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
class TWAPExecutor:
"""
TWAP算法执行器
"""
def __init__(self, total_quantity, total_seconds, slice_seconds=60):
"""
参数：
total_quantity: 总成交量（股）
total_seconds: 总执行时间（秒）
slice_seconds: 每个时间切片的长度（秒），默认60秒
"""
self.Q = total_quantity
self.T = total_seconds
self.dt = slice_seconds
self.N = int(total_seconds / slice_seconds)
# 每个切片的目标量
self.slice_qty = self.Q / self.N
# 记录执行状态
self.current_slice = 0
self.executed_qty = 0
self.schedule = []
def generate_schedule(self):
"""
生成完整的发单计划
返回：[(时间戳, 目标量), ...]
"""
schedule = []
base_time = datetime.now()
for i in range(self.N):
slice_time = base_time + timedelta(seconds=i * self.dt)
schedule.append((slice_time, self.slice_qty))
self.schedule = schedule
return schedule
def get_next_order(self, current_time, executed_so_far):
"""
获取下一个切片应该发的单量
参数：
current_time: 当前时间
executed_so_far: 已经成交的量
返回：目标发单量
"""
# 计算当前应该在第几个切片
elapsed = (current_time - self.schedule[0][0]).total_seconds()
expected_slice = int(elapsed / self.dt)
# 边界检查
if expected_slice >= self.N:
return 0 # 时间到了，不再发单
# 计算理论累计成交量
expected_qty = self.slice_qty * (expected_slice + 1)
# 偏差修正
deviation = expected_qty - executed_so_far
# 当前切片的目标量
target = self.slice_qty + deviation
# 防止负数和过大
target = max(0, min(target, self.slice_qty * 2))
return target
def run_simulation(self, price_series, fill_rate=0.8):
"""
模拟执行
参数：
price_series: 价格序列，长度必须等于N
fill_rate: 成交率，默认80%
返回：成交记录DataFrame
"""
records = []
total_cost = 0
total_filled = 0
for i in range(self.N):
target = self.slice_qty
filled = target * fill_rate
price = price_series[i]
total_filled += filled
total_cost += filled * price
records.append({
'slice': i,
'target': target,
'filled': filled,
'price': price,
'cum_filled': total_filled,
'cum_cost': total_cost
})
df = pd.DataFrame(records)
df['avg_price'] = df['cum_cost'] / df['cum_filled']
return df
# 使用示例
if __name__ == "__main__":
# 创建执行器：买入10000股，用时300秒，每30秒发一次单
executor = TWAPExecutor(
total_quantity=10000,
total_seconds=300,
slice_seconds=30
)
# 生成发单计划
schedule = executor.generate_schedule()
print(f"共{len(schedule)}个切片，每个切片目标量{executor.slice_qty:.0f}股")
# 模拟价格序列（随机游走）
np.random.seed(42)
prices = 100 + np.cumsum(np.random.randn(10) * 0.5)
prices = np.maximum(prices, 95) # 防止价格过低
# 运行模拟
result = executor.run_simulation(prices, fill_rate=0.85)
print(f"成交均价: {result['avg_price'].iloc[-1]:.2f}")
print(f"总成交量: {result['cum_filled'].iloc[-1]:.0f}股")
避坑指南：我曾经在实盘中使用这个类，发现get_next_order方法里的偏差修正有个bug——如果连续几个切片都没成交，deviation会变得非常大，导致下一个切片发单量暴增。后来我加了个上限，就是代码里的min(target, self.slice_qty * 2)。这个经验是用真金白银换来的。
11.4 回测框架
光有算法不行，得回测。我搭了一个轻量级的回测框架，专门用来测试TWAP策略。
class TWAPBacktest:
"""
TWAP回测框架
"""
def __init__(self, tick_data, total_quantity, execution_window):
"""
参数：
tick_data: DataFrame，必须包含['time', 'price', 'volume']三列
total_quantity: 总成交量
execution_window: 执行时间窗口（秒）
"""
self.data = tick_data
self.Q = total_quantity
self.window = execution_window
def run(self, slice_seconds=60, fill_model='proportional'):
"""
运行回测
fill_model: 成交模型
'proportional' - 按市场成交量比例成交
'fixed' - 固定成交率
"""
# 将tick数据按时间切片聚合
self.data['time_bin'] = pd.cut(
self.data['time'],
bins=pd.date_range(
start=self.data['time'].min(),
end=self.data['time'].max(),
freq=f'{slice_seconds}S'
)
)
# 计算每个切片的成交量占比
slice_volumes = self.data.groupby('time_bin')['volume'].sum()
total_volume = slice_volumes.sum()
volume_ratio = slice_volumes / total_volume
# 按比例分配目标量
target_qty = self.Q * volume_ratio
# 模拟成交
results = []
cum_filled = 0
cum_cost = 0
for i, (idx, row) in enumerate(slice_volumes.items()):
target = target_qty.iloc[i]
if fill_model == 'proportional':
# 假设我们能拿到该切片市场成交量的20%
filled = row * 0.2
else:
filled = target * 0.8
# 取该切片内的平均价格
slice_data = self.data[self.data['time_bin'] == idx]
avg_price = slice_data['price'].mean()
cum_filled += filled
cum_cost += filled * avg_price
results.append({
'slice_start': idx.left,
'slice_end': idx.right,
'target': target,
'filled': filled,
'avg_price': avg_price,
'cum_filled': cum_filled,
'cum_cost': cum_cost
})
df = pd.DataFrame(results)
df['avg_price'] = df['cum_cost'] / df['cum_filled']
# 计算绩效指标
metrics = {
'total_filled': cum_filled,
'total_cost': cum_cost,
'vwap': cum_cost / cum_filled,
'slippage': (cum_cost / cum_filled) - self.data['price'].mean(),
'completion_rate': cum_filled / self.Q,
'execution_time': (self.data['time'].max() - self.data['time'].min()).total_seconds()
}
return df, metrics
# 回测示例
def run_twap_backtest():
# 生成模拟tick数据
np.random.seed(123)
n_ticks = 1000
base_time = datetime.now().replace(hour=9, minute=30, second=0)
tick_data = pd.DataFrame({
'time': [base_time + timedelta(seconds=i*3) for i in range(n_ticks)],
'price': 100 + np.cumsum(np.random.randn(n_ticks) * 0.1),
'volume': np.random.poisson(lam=100, size=n_ticks)
})
# 创建回测器
bt = TWAPBacktest(
tick_data=tick_data,
total_quantity=50000,
execution_window=1800 # 30分钟
)
# 运行回测
result_df, metrics = bt.run(slice_seconds=60)
print("=== TWAP回测结果 ===")
print(f"总成交: {metrics['total_filled']:.0f}股")
print(f"VWAP: {metrics['vwap']:.2f}")
print(f"滑点: {metrics['slippage']:.4f}")
print(f"完成率: {metrics['completion_rate']:.2%}")
return result_df, metrics
11.5 核心逻辑流程图
下面我用SVG画了一张TWAP算法的核心流程图，把整个执行过程串起来。
11.6 实战经验总结
做了这么多年TWAP，我总结了几条铁律：
- 切片长度别太短：我见过有人设1秒一个切片，结果高频摩擦成本比冲击成本还高。建议至少30秒以上。
- 偏差修正要温和：前面说过了，修正太猛会引发二次冲击。我一般控制在20%以内。
- 关注市场微观结构：TWAP不是万能的。在流动性枯竭的市场，TWAP就是灾难。我曾在某小盘股上跑TWAP，结果把价格从10块推到12块，滑点惨不忍睹。
- 结合VWAP使用：我个人习惯把TWAP和VWAP结合起来。在流动性好的时段用VWAP，流动性差的时段用TWAP。效果比单独用任何一个都好。
最后说一句：TWAP是入门级的算法，但入门不代表简单。真正把TWAP做好，需要对市场有深刻理解。我见过太多人把TWAP当成万能药，结果亏得底裤都不剩。记住，算法只是工具，对市场的敬畏才是根本。