第12章：VWAP算法详解

VWAP，全称Volume-Weighted Average Price，成交量加权平均价格。这玩意儿在量化交易圈子里，几乎是无人不知。说白了，它就是衡量你交易执行质量的一把尺子。

我刚开始做算法交易那会儿，对VWAP的理解特别肤浅，以为就是算个均价。后来在实盘中吃过亏，才真正搞懂它的门道。今天咱们就把VWAP算法掰开揉碎了讲清楚。

12.1 什么是VWAP？

VWAP的计算公式其实很简单：

VWAP = Σ(价格 × 成交量) / Σ(成交量)

嗯，就是每个时间点的成交价格，乘以该时间点的成交量，然后累加，再除以总成交量。这个值代表了一段时间内，市场参与者的平均成交成本。

举个例子：

VWAP = 452,500 / 4,500 = 100.56

你想想看，如果你在这段时间内买入，平均成本低于100.56，说明你执行得不错。高于这个数，那就得反思一下了。

12.2 VWAP算法的核心逻辑

VWAP算法交易的目标，就是让我们的订单执行价格，尽可能贴近市场的VWAP。说白了，就是跟着市场的节奏走。

我个人习惯把VWAP算法拆成三个核心步骤：

1. 预测成交量分布：先预估一天内每个时间段的成交量占比
2. 切片下单：按照预测的成交量曲线，把大单切成小单
3. 动态调整：根据实际成交情况，实时修正后续的下单节奏

这里有个关键点——成交量预测。我见过不少新手直接拿历史平均数据来用，结果在突发事件面前被打得措手不及。我曾经在一次财报发布日，就因为用了静态的成交量曲线，导致执行成本飙升了十几个基点。

12.3 成交量预测模型

预测成交量分布，常用的方法有这么几种：

• 历史平均法：取过去N天的同一时段成交量均值。简单，但不够灵活。
• 指数加权法：给近期的数据更高的权重。我个人比较喜欢这个，因为它能更快地适应市场变化。
• 机器学习法：用LSTM、XGBoost等模型来预测。效果可能更好，但复杂度也上去了。

这里我贴一段简单的Python实现，用的是指数加权法：

import numpy as np
import pandas as pd

def predict_volume_profile(hist_volumes, alpha=0.3):
    """
    指数加权预测成交量分布
    hist_volumes: 历史成交量矩阵，每行是一天，每列是一个时间段
    """
    weights = np.array([(1-alpha)**i for i in range(len(hist_volumes))])
    weights = weights[::-1]  # 最近的一天权重最高
    weights = weights / weights.sum()
    
    # 加权平均
    predicted = np.average(hist_volumes, axis=0, weights=weights)
    # 归一化
    return predicted / predicted.sum()

# 示例：假设有5天的数据，每天分13个30分钟时段
hist_data = np.random.rand(5, 13) * 10000
profile = predict_volume_profile(hist_data)
print("预测的成交量分布：", profile)

12.4 VWAP算法的完整实现

下面是一个完整的VWAP算法实现，包含了订单切片和动态调整的逻辑：

class VWAPAlgo:
    def __init__(self, total_qty, start_time, end_time, 
                 volume_profile, freq='1min'):
        self.total_qty = total_qty
        self.start_time = start_time
        self.end_time = end_time
        self.volume_profile = volume_profile  # 预测的成交量占比
        self.freq = freq
        
        self.executed_qty = 0
        self.executed_amount = 0
        self.current_vwap = 0
        
    def get_target_qty(self, current_time):
        """计算当前时间应该完成的累计成交量"""
        elapsed = (current_time - self.start_time).total_seconds()
        total = (self.end_time - self.start_time).total_seconds()
        progress = elapsed / total
        
        # 找到对应的成交量占比
        idx = int(progress * len(self.volume_profile))
        idx = min(idx, len(self.volume_profile) - 1)
        
        target_ratio = self.volume_profile[:idx+1].sum()
        return self.total_qty * target_ratio
    
    def get_order_qty(self, current_time, last_price, last_volume):
        """计算当前需要下单的量"""
        target_qty = self.get_target_qty(current_time)
        remaining = target_qty - self.executed_qty
        
        # 更新VWAP
        self.executed_amount += last_price * last_volume
        self.executed_qty += last_volume
        if self.executed_qty > 0:
            self.current_vwap = self.executed_amount / self.executed_qty
        
        # 动态调整：如果落后了，就多下一点
        if remaining > 0:
            # 加上一个调整项，防止过度滞后
            adjustment = max(0, remaining * 0.1)
            return min(remaining + adjustment, self.total_qty - self.executed_qty)
        return 0

# 使用示例
profile = np.array([0.05, 0.08, 0.12, 0.15, 0.18, 
                    0.15, 0.12, 0.08, 0.05, 0.02])
algo = VWAPAlgo(total_qty=100000, 
                start_time=pd.Timestamp('09:30'),
                end_time=pd.Timestamp('16:00'),
                volume_profile=profile)

12.5 VWAP回测框架

回测是检验算法有效性的唯一标准。我搭建VWAP回测框架时，最看重三个指标：

• VWAP偏离度：执行价格与市场VWAP的差值，越小越好
• 完成率：实际成交量占目标成交量的比例
• 冲击成本：因为我们的订单导致价格变动的成本

下面是一个简单的回测框架：

def backtest_vwap(algo, market_data):
    """
    market_data: DataFrame，包含时间、价格、成交量
    """
    results = []
    for idx, row in market_data.iterrows():
        current_time = row['time']
        last_price = row['price']
        last_volume = row['volume']
        
        # 计算下单量
        order_qty = algo.get_order_qty(current_time, last_price, last_volume)
        
        # 模拟成交（假设全部成交）
        if order_qty > 0:
            executed_price = last_price * (1 + 0.0001)  # 假设0.01%的滑点
            results.append({
                'time': current_time,
                'price': executed_price,
                'qty': order_qty,
                'vwap': algo.current_vwap
            })
    
    return pd.DataFrame(results)

# 运行回测
market_data = pd.DataFrame({
    'time': pd.date_range('09:30', '16:00', freq='1min'),
    'price': np.random.randn(390) * 0.5 + 100,
    'volume': np.random.randint(1000, 5000, 390)
})

profile = np.ones(390) / 390  # 均匀分布
algo = VWAPAlgo(100000, market_data['time'].iloc[0], 
                market_data['time'].iloc[-1], profile)
result_df = backtest_vwap(algo, market_data)

12.6 VWAP算法的知识体系

为了让你更直观地理解VWAP算法的全貌，我画了一张图：

12.7 实战中的避坑指南

做VWAP算法交易这几年，我踩过的坑不少。这里挑几个典型的说说：

• 不要迷信历史数据：市场结构会变，去年的成交量分布不一定适用于今天。我建议每周重新训练一次模型。
• 注意开盘和收盘：这两个时间段的成交量波动最大，算法容易出问题。我曾经在收盘前5分钟，因为算法没处理好，导致大量订单没成交。
• 滑点比你想象的大：回测时假设0.01%的滑点，实盘中可能翻倍。我一般会在回测中加一个安全系数，比如1.5倍滑点。
• 大单要拆得更细：如果你的订单量占日成交量的5%以上，建议把切片粒度降到秒级。不然冲击成本会让你怀疑人生。