4. VWAP的实战代码实现：从零搭建一个简单的VWAP执行引擎

好了，前面我们把VWAP的理论讲了个透。今天咱们来点实在的——动手写代码。

我个人习惯是，学任何策略都得先跑通一个最小可用的版本。说白了，就是先让轮子转起来，再考虑怎么让它转得更快、更稳。今天我们就从零开始，搭一个最简单的VWAP执行引擎。

4.1 核心思路：引擎到底在干什么？

先想清楚一个问题：VWAP执行引擎的本质是什么？

嗯，其实就两件事：

• 切单：把一个大订单，按时间切成若干小片
• 控速：让每一片的成交量，尽量匹配市场的成交量分布

你想想看，如果市场在上午10点成交量最大，那我们的订单也应该在那个时段多下一点。这就是VWAP的核心逻辑。

下面这张图，是我画的一个简易引擎流程图，帮你理清整体脉络：

4.2 第一步：准备历史数据

做VWAP，首先得有数据。我们需要的是分钟级的成交量数据。

我在项目中遇到过一个问题：有些股票在开盘和收盘时成交量特别大，中午反而很冷清。如果你只用一天的成交量分布，很容易被噪音带偏。所以我建议至少取过去5到10天的数据做平均。

来看代码：

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

def load_volume_profile(ticker, start_date, end_date):
    """
    加载历史分钟级成交量数据，计算平均成交量分布
    """
    # 假设我们已经从数据库或API拿到了分钟数据
    # 这里用模拟数据演示
    np.random.seed(42)
    
    # 生成一天的分钟时间戳（假设交易时间为9:30-16:00，共390分钟）
    minutes = 390
    dates = pd.date_range(start_date, end_date, freq='D')
    
    all_profiles = []
    
    for date in dates:
        # 模拟一天的分钟成交量（加入U型特征）
        base_vol = np.random.randint(1000, 5000, minutes)
        # 开盘和收盘加量
        u_shape = np.concatenate([
            np.linspace(1.5, 1.0, 30),   # 开盘30分钟
            np.ones(minutes - 60),        # 中间时段
            np.linspace(1.0, 1.5, 30)     # 收盘30分钟
        ])
        daily_vol = (base_vol * u_shape).astype(int)
        all_profiles.append(daily_vol)
    
    # 取平均
    avg_profile = np.mean(all_profiles, axis=0)
    return avg_profile

# 测试
profile = load_volume_profile('000001.SZ', '2024-01-01', '2024-01-10')
print(f"成交量分布形状: {profile.shape}")
print(f"前5分钟成交量: {profile[:5]}")

4.3 第二步：计算时间片权重

有了成交量分布，下一步就是算权重。这个其实很简单——每个时间片的成交量除以总成交量就行。

def compute_weights(volume_profile):
    """
    根据成交量分布计算每个时间片的权重
    """
    total_vol = np.sum(volume_profile)
    weights = volume_profile / total_vol
    return weights

weights = compute_weights(profile)
print(f"权重总和: {np.sum(weights):.4f}")  # 应该等于1.0
print(f"前5个时间片权重: {weights[:5]}")

嗯，这里要注意一点：权重总和必须严格等于1。如果因为浮点精度问题差了一点点，最后分配订单时可能会多出或缺少几股。我一般会做一次归一化处理。

4.4 第三步：生成执行计划

现在到了最核心的部分——把大订单切到每个时间片里。

def generate_schedule(total_shares, weights, start_time, end_time):
    """
    生成VWAP执行计划
    """
    # 计算每个时间片应发的股数
    raw_slices = total_shares * weights
    
    # 取整处理（股数必须是整数）
    schedule = np.floor(raw_slices).astype(int)
    
    # 处理余数：把剩余股数加到最后一个时间片
    remainder = total_shares - np.sum(schedule)
    if remainder > 0:
        schedule[-1] += remainder
    
    # 生成时间戳
    time_points = pd.date_range(start_time, end_time, periods=len(schedule) + 1)[:-1]
    
    # 构建执行计划表
    plan = pd.DataFrame({
        'time': time_points,
        'target_shares': schedule,
        'cumulative_shares': np.cumsum(schedule),
        'weight': weights
    })
    
    return plan

# 测试：执行10万股，从9:30到16:00
plan = generate_schedule(
    total_shares=100000,
    weights=weights,
    start_time='2024-01-15 09:30:00',
    end_time='2024-01-15 16:00:00'
)

print(plan.head(10))

4.5 完整的引擎类

把上面这些功能封装成一个类，用起来就方便多了。

class VWAPEngine:
    """
    简单的VWAP执行引擎
    """
    def __init__(self, ticker, lookback_days=10):
        self.ticker = ticker
        self.lookback_days = lookback_days
        self.volume_profile = None
        self.weights = None
    
    def train(self, start_date, end_date):
        """训练：加载历史数据并计算权重"""
        print(f"正在加载 {self.ticker} 的历史数据...")
        self.volume_profile = load_volume_profile(
            self.ticker, start_date, end_date
        )
        self.weights = compute_weights(self.volume_profile)
        print("训练完成！")
    
    def execute(self, total_shares, trade_date):
        """生成某一天的执行计划"""
        if self.weights is None:
            raise ValueError("请先调用 train() 方法进行训练")
        
        start_time = f"{trade_date} 09:30:00"
        end_time = f"{trade_date} 16:00:00"
        
        plan = generate_schedule(
            total_shares=total_shares,
            weights=self.weights,
            start_time=start_time,
            end_time=end_time
        )
        
        return plan

# 使用示例
engine = VWAPEngine('000001.SZ')
engine.train('2024-01-01', '2024-01-10')

plan = engine.execute(100000, '2024-01-15')
print(f"执行计划共 {len(plan)} 个时间片")
print(f"总股数: {plan['target_shares'].sum()}")

4.6 验证一下效果

代码写完了，总得看看效果吧？我们来模拟一下实际执行情况。

def simulate_execution(plan, actual_volume_profile):
    """
    模拟执行，计算实际VWAP
    """
    # 假设我们完全按照计划执行
    executed_shares = plan['target_shares'].values
    
    # 模拟成交价格（这里简单用随机数，实际应该用市场数据）
    np.random.seed(123)
    prices = np.random.uniform(10.0, 10.5, len(executed_shares))
    
    # 计算实际VWAP
    total_cost = np.sum(executed_shares * prices)
    total_volume = np.sum(executed_shares)
    actual_vwap = total_cost / total_volume
    
    # 计算市场VWAP（用实际成交量加权）
    market_vwap = np.sum(actual_volume_profile * prices) / np.sum(actual_volume_profile)
    
    # 计算偏离度
    deviation = abs(actual_vwap - market_vwap) / market_vwap * 100
    
    return {
        'actual_vwap': actual_vwap,
        'market_vwap': market_vwap,
        'deviation_pct': deviation
    }

# 模拟验证
result = simulate_execution(plan, profile)
print(f"实际执行VWAP: {result['actual_vwap']:.4f}")
print(f"市场VWAP: {result['market_vwap']:.4f}")
print(f"偏离度: {result['deviation_pct']:.2f}%")

4.7 避坑指南

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

• 数据对齐问题：历史数据的分钟区间一定要和交易时间对齐。我遇到过数据源把集合竞价也算进去了，导致权重算错。
• 除权除息日：遇到分红送股的日子，成交量会异常。我建议把这些日子从训练数据里剔除。
• 停牌处理：股票停牌复牌后，成交量分布会变。最好在复牌后重新训练模型。

好了，一个最简单的VWAP执行引擎就搭好了。虽然功能还比较基础，但核心逻辑已经跑通了。你可以拿真实数据试试看，感受一下VWAP的执行效果。

记住，任何复杂的系统都是从这个小东西长出来的。先把基础打牢，后面我们再慢慢加功能。

---

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