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

TWAP，全称 Time-Weighted Average Price。说白了，就是把一个大单子，均匀地拆成很多小单子，然后按时间一个一个扔出去。

我刚开始做算法交易的时候，第一个接触的就是 TWAP。为什么？因为它简单、直观，而且大部分场景下都够用。你想想看，如果你要买 100 万股某只股票，直接挂个市价单，那价格肯定被打飞。TWAP 就是帮你把这一百万股，拆成比如每 5 分钟扔 1 万股，这样市场就感觉不到你的存在了。

2.1 TWAP 的核心原理

TWAP 的数学表达其实很简单：

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

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

具体执行时，算法会做两件事：

1. 时间切片：把整个交易时段，比如 9:30 到 15:00，切成 N 个等长的时间片。
2. 均匀分配：把总订单量，平均分配到每个时间片里。

举个例子：

参数	值
总订单量	100,000 股
交易时段	9:30 - 15:00 (330 分钟)
时间片长度	5 分钟
时间片数量	66 个
每片下单量	约 1,515 股

嗯，这里要注意：实际执行时，每个时间片内的下单量，还要考虑该时间片内的历史成交量占比。不过这是进阶玩法了，后面再聊。

2.2 适用场景

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

• 流动性好的股票：比如沪深 300 成分股，每天成交几十亿，TWAP 效果很好。
• 非紧急订单：如果你不急着在 1 分钟内买完，TWAP 是最省心的选择。
• 大单拆分：订单量超过日均成交量的 5% 时，必须用算法拆分。
• 指数跟踪：做指数增强或者 ETF 套利时，TWAP 能帮你平滑建仓成本。

2.3 Python 代码实现

下面是我自己写的一个 TWAP 实现。代码不复杂，但包含了实际交易中需要考虑的细节：

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class TWAPExecutor:
    """
    TWAP 算法执行器
    """
    def __init__(self, total_quantity, start_time, end_time, slice_minutes=5):
        self.total_qty = total_quantity
        self.start = start_time
        self.end = end_time
        self.slice_min = slice_minutes
        
        # 计算时间片数量
        total_minutes = (end_time - start_time).seconds / 60
        self.num_slices = int(total_minutes / slice_minutes)
        
        # 每片基础下单量
        self.base_qty_per_slice = total_quantity / self.num_slices
        
    def generate_schedule(self):
        """
        生成交易时间表
        """
        schedule = []
        current_time = self.start
        
        for i in range(self.num_slices):
            slice_end = current_time + timedelta(minutes=self.slice_min)
            
            # 这里可以加入随机偏移，避免被识别
            actual_qty = self.base_qty_per_slice * (1 + np.random.uniform(-0.05, 0.05))
            
            schedule.append({
                'time': current_time,
                'quantity': round(actual_qty),
                'slice_end': slice_end
            })
            
            current_time = slice_end
            
        return pd.DataFrame(schedule)
    
    def execute(self, market_data):
        """
        模拟执行
        market_data: DataFrame 包含时间、价格、成交量
        """
        schedule = self.generate_schedule()
        executed_trades = []
        
        for _, slice_info in schedule.iterrows():
            # 获取该时间片内的市场数据
            mask = (market_data['time'] >= slice_info['time']) & \
                   (market_data['time'] < slice_info['slice_end'])
            slice_data = market_data[mask]
            
            if len(slice_data) == 0:
                continue
                
            # 模拟成交：用该时间片的 VWAP 作为成交价
            avg_price = (slice_data['price'] * slice_data['volume']).sum() / \
                        slice_data['volume'].sum()
            
            executed_trades.append({
                'time': slice_info['time'],
                'quantity': slice_info['quantity'],
                'price': avg_price,
                'slice_end': slice_info['slice_end']
            })
            
        return pd.DataFrame(executed_trades)

# 使用示例
if __name__ == "__main__":
    # 参数设置
    total_qty = 100000  # 10万股
    start = datetime(2024, 1, 15, 9, 30)
    end = datetime(2024, 1, 15, 15, 0)
    
    # 创建执行器
    twap = TWAPExecutor(total_qty, start, end, slice_minutes=5)
    
    # 生成交易计划
    schedule = twap.generate_schedule()
    print("交易计划：")
    print(schedule.head())
    
    # 模拟执行（假设有市场数据）
    # executed = twap.execute(market_data)

2.4 TWAP 的优缺点

优点	缺点
实现简单，逻辑清晰	不考虑成交量分布，可能在高成交量时段下单不足
对市场影响小，尤其是大单	在低流动性时段可能无法完成
执行成本可预测	无法利用市场波动获得更好的价格
适合指数跟踪和被动策略	在趋势行情中表现较差

2.5 核心逻辑流程图

下面这张图，是我自己画的 TWAP 核心流程。你看一眼就能明白整个算法是怎么跑的：

你看，整个流程其实就四步。但实际跑起来，要考虑的东西就多了。比如：

• 开盘和收盘时段，成交量通常比较大，要不要调整分配比例？
• 如果某个时间片没成交完，剩下的量怎么处理？
• 市场突然出现大单，要不要暂停一下？

这些都是实战中会遇到的问题。我建议你先从最基础的 TWAP 开始跑，跑熟了再慢慢加这些优化。毕竟，能跑通的算法，才是好算法。

好了，TWAP 就讲到这里。代码我已经贴出来了，你可以直接复制到自己的回测框架里跑跑看。有什么问题，欢迎交流。

---

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