TWAP算法基础代码实现：Python实现一个简单的TWAP算法

各位同学，今天我们来聊聊TWAP算法的代码实现。说实话，很多人在学TWAP时，一上来就盯着复杂的数学公式看，结果把自己绕晕了。我个人习惯是——先跑通一个最简单的版本，再慢慢优化。

TWAP，全称是Time-Weighted Average Price，时间加权平均价格。说白了，就是把一个大订单，拆成很多小订单，在固定时间间隔里均匀地发出去。这样做的好处是啥？减少对市场的冲击，让成交价更接近这段时间的平均价。

1. 最简单的TWAP模型

我们先从最基础的逻辑开始。假设我要买入10万股的某只ETF，交易时间是上午9:30到11:30，一共120分钟。那我每分钟应该发多少？

嗯，很简单：100,000股 ÷ 120分钟 ≈ 833股/分钟。

这就是TWAP最朴素的想法。但实际交易中，你不能真的每分钟都发833股，因为A股最小交易单位是100股（1手），所以得取整。我在项目中遇到过这种情况，取整后剩下的股数，我习惯在最后一笔补上。

2. Python代码实现

下面我给出一个最基础的TWAP实现。这个版本没有考虑任何高级优化，就是让你看清楚核心逻辑。

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

class SimpleTWAP:
    """
    最简单的TWAP算法实现
    适合初学者理解核心逻辑
    """
    def __init__(self, total_shares, start_time, end_time, interval_minutes=1):
        """
        初始化TWAP参数
        
        :param total_shares: 总委托股数
        :param start_time: 开始时间 (字符串格式 'HH:MM')
        :param end_time: 结束时间 (字符串格式 'HH:MM')
        :param interval_minutes: 时间间隔，默认1分钟
        """
        self.total_shares = total_shares
        self.start_time = datetime.strptime(start_time, '%H:%M')
        self.end_time = datetime.strptime(end_time, '%H:%M')
        self.interval_minutes = interval_minutes
        
        # 计算总切片数
        total_minutes = (self.end_time - self.start_time).seconds // 60
        self.num_slices = total_minutes // interval_minutes
        
        print(f"总交易时间: {total_minutes} 分钟")
        print(f"切片数量: {self.num_slices} 个")
    
    def generate_schedule(self):
        """
        生成订单切片计划
        """
        # 计算每片的基础股数（取整到100股）
        base_shares = (self.total_shares // self.num_slices) // 100 * 100
        
        # 计算剩余股数
        remaining = self.total_shares - base_shares * self.num_slices
        
        schedule = []
        current_time = self.start_time
        
        for i in range(self.num_slices):
            # 前几笔用基础股数，最后一笔补上剩余
            if i == self.num_slices - 1:
                slice_shares = base_shares + remaining
            else:
                slice_shares = base_shares
            
            # 生成时间戳
            time_str = current_time.strftime('%H:%M')
            
            schedule.append({
                'slice_id': i + 1,
                'time': time_str,
                'shares': slice_shares,
                'cumulative': sum([s['shares'] for s in schedule]) + slice_shares
            })
            
            current_time += timedelta(minutes=self.interval_minutes)
        
        return pd.DataFrame(schedule)
    
    def print_summary(self, df):
        """
        打印执行计划摘要
        """
        print("\n=== TWAP执行计划 ===")
        print(f"总股数: {self.total_shares}")
        print(f"切片数: {self.num_slices}")
        print(f"每片基础股数: {df['shares'].iloc[0]}")
        print(f"最后一片调整: {df['shares'].iloc[-1]}")
        print(f"预计完成时间: {df['time'].iloc[-1]}")
        print("\n前5笔切片:")
        print(df.head().to_string(index=False))

3. 运行示例

我们来跑一个实际例子。假设我要在上午10:00到10:30之间，买入50,000股某ETF。

# 创建TWAP实例
twap = SimpleTWAP(
    total_shares=50000,
    start_time='10:00',
    end_time='10:30',
    interval_minutes=1
)

# 生成计划
schedule_df = twap.generate_schedule()

# 打印摘要
twap.print_summary(schedule_df)

# 输出完整计划
print("\n完整切片计划:")
print(schedule_df.to_string(index=False))

运行结果会显示：总交易时间30分钟，切片30个，每片基础股数1600股（因为50000÷30≈1666，取整到100股就是1600），最后一片补上剩余的2000股。

4. 订单切片与时间调度的核心逻辑

刚才的代码展示了最基础的切片逻辑。但实际交易中，时间调度没那么简单。我总结了几点需要注意的地方：

• 交易时间边界： 开盘前和收盘前，市场波动大，我一般会避开这些时段。比如9:30-9:45和14:45-15:00，我会减少切片数量。
• 间隔选择： 1分钟间隔适合流动性好的ETF。如果ETF流动性差，我建议用3分钟或5分钟间隔，给市场一些消化时间。
• 随机化处理： 固定时间发单容易被别人猜到你的意图。我习惯在每笔发单时间上加一个随机偏移，比如±5秒。

5. 知识体系结构图

下面这张图，是我自己梳理的TWAP算法知识结构。你可以看到，核心逻辑其实就三个部分：参数设置、切片计算、执行调度。

6. 代码优化方向

刚才的代码只是个起点。在实际项目中，我还会做以下几件事：

1. 加入实时行情判断： 如果当前价格偏离均价太多，我会暂停发单或者调整股数。
2. 支持动态调整： 市场波动大时，自动拉长时间间隔；市场平稳时，加快执行速度。
3. 日志记录： 每笔成交都记录下来，方便事后复盘。我习惯用CSV格式，字段包括时间、委托价、成交价、滑点等。

好了，这一章的内容就到这里。代码虽然简单，但核心逻辑都在里面了。你可以在自己的电脑上跑一跑，改改参数，看看切片计划怎么变化。下一章我们会聊TWAP的进阶优化——如何结合市场微观结构来调整切片策略。

---