如何在 PTrade 中实现 VWAP 算法以降低大额订单的交易成本？

编写一个 VWAP（成交量加权平均价）算法的核心逻辑在于：利用历史数据的成交量分布规律，预测当日的成交量分布，并据此安排下单节奏。

简单来说，如果历史上某只股票在上午 10:00 到 10:05 这段时间成交活跃（占全天成交量的 5%），那么你的算法也应该在这段时间完成总委托量的 5%。

以下是在 PTrade 平台上实现 VWAP 策略的完整代码和详细说明。

策略逻辑步骤

1. 初始化 (initialize)：设置要交易的股票、总目标数量、以及计算历史分布的回溯天数。
2. 盘前准备 (before_trading_start)：
   • 获取过去 N 天的分钟级成交量数据。
   • 按时间（如 09:31, 09:32...）对成交量进行平均。
   • 计算每一分钟的累计成交量占比（Cumulative Volume Percentage）。
   • 生成当日的“目标持仓曲线”。
3. 盘中执行 (handle_data)：
   • 每分钟运行一次。
   • 获取当前时间对应的“目标累计占比”。
   • 计算“当前应该完成的总数量” = 总目标单量 * 目标占比。
   • 计算“还需要买入的数量” = 应该完成的数量 - 当前已持仓数量。
   • 下单补足差额。

PTrade 策略代码

import pandas as pd
import numpy as np
import datetime

def initialize(context):
    """
    策略初始化函数
    """
    # 1. 设置要操作的股票
    g.security = '600570.SS'  # 示例：恒生电子
    set_universe(g.security)
    
    # 2. 设置VWAP执行参数
    g.target_volume = 10000   # 计划今天总共买入多少股
    g.lookback_days = 10      # 回溯过去多少天来计算成交量分布
    
    # 3. 全局变量，用于存储当日的时间-比例计划表
    g.vwap_schedule = None    # 格式：Series, index为时间字符串(HH:MM), value为累计比例
    
    # 4. 记录是否已经完成全部交易
    g.finished = False

def before_trading_start(context, data):
    """
    盘前处理：计算历史成交量分布，生成当日执行计划
    """
    g.finished = False
    log.info("开始计算历史成交量分布...")
    
    # 获取历史分钟线数据
    # 获取天数 * 240分钟的数据量
    count = g.lookback_days * 240
    
    # 获取分钟级成交量数据
    # 注意：get_history 返回的 DataFrame index 是 datetime 对象
    hist_data = get_history(count, frequency='1m', field='volume', security_list=g.security)
    
    if hist_data is None or len(hist_data) == 0:
        log.error("未获取到历史数据，无法执行VWAP策略")
        return

    # --- 数据处理逻辑 ---
    # 1. 将数据转换为 DataFrame (如果是单只股票，get_history通常返回DataFrame，列名为volume)
    # 如果返回的是字典(多股情况)，需要取对应股票的数据
    if isinstance(hist_data, dict):
        df = pd.DataFrame({'volume': hist_data[g.security]['volume']})
        df.index = hist_data[g.security]['datetime'] # 假设字典里有datetime字段，具体视版本而定
        # PTrade get_history 单股通常直接返回DataFrame，索引为时间
    else:
        df = hist_data
    
    # 2. 提取时间（小时:分钟），忽略日期
    # 创建一个新的列 'time_str' 用于分组
    # Python 3.5 兼容写法
    df['time_str'] = [x.strftime('%H:%M') for x in df.index]
    
    # 3. 按时间分组，计算平均成交量
    # 这里计算的是过去N天，每一分钟的平均成交量
    volume_profile = df.groupby('time_str')['volume'].mean()
    
    # 4. 计算累计成交量占比
    # cumsum() 计算累计值，然后除以总和，得到 0 到 1 之间的曲线
    total_avg_volume = volume_profile.sum()
    if total_avg_volume == 0:
        log.error("历史平均成交量为0，无法计算分布")
        return
        
    cumulative_ratio = volume_profile.cumsum() / total_avg_volume
    
    # 5. 存储到全局变量 g.vwap_schedule
    g.vwap_schedule = cumulative_ratio
    
    log.info("VWAP 计划生成完毕。")
    # 打印前5个和后5个时间点用于检查
    log.info("开盘前5分钟计划占比: %s" % str(g.vwap_schedule.head(5).to_dict()))
    log.info("收盘前5分钟计划占比: %s" % str(g.vwap_schedule.tail(5).to_dict()))

def handle_data(context, data):
    """
    盘中运行：每分钟检查进度并下单
    """
    # 如果已经完成或计划表为空，则跳过
    if g.finished or g.vwap_schedule is None:
        return

    # 1. 获取当前时间
    current_dt = context.blotter.current_dt
    current_time_str = current_dt.strftime('%H:%M')
    
    # 2. 检查当前时间是否在交易计划中
    # 注意：PTrade的分钟线时间通常是 K线结束时间，如 09:31 代表 09:30-09:31 的数据
    if current_time_str not in g.vwap_schedule.index:
        return
        
    # 3. 获取当前时间点应该达到的累计进度 (0.0 - 1.0)
    target_ratio = g.vwap_schedule[current_time_str]
    
    # 4. 计算当前理论上应该持有的总数量
    target_position = int(g.target_volume * target_ratio)
    
    # 5. 获取当前实际持仓数量
    # get_position 返回的是 Position 对象
    current_position = get_position(g.security).amount
    
    # 6. 计算差额（需要买入的数量）
    # 目标持仓 - 当前持仓
    order_qty = target_position - current_position
    
    # 7. 下单逻辑
    # 设置一个最小下单阈值，避免碎股频繁下单（例如小于100股不下单，除非是最后收盘）
    if order_qty >= 100:
        # 使用 order_target 直接调整持仓到目标数量，这比 order 更稳健，能自动处理之前的废单或未成交情况
        # 也可以使用 limit_price 挂单，这里为了演示简单使用市价或默认逻辑
        order_target(g.security, target_position)
        
        log.info("时间: %s, 目标进度: %.2f%%, 目标持仓: %d, 当前持仓: %d, 下单: %d" % (
            current_time_str, 
            target_ratio * 100, 
            target_position, 
            current_position, 
            order_qty
        ))
    
    # 8. 检查是否接近收盘，强制完成剩余部分
    if current_time_str == '14:57':
        remaining = g.target_volume - current_position
        if remaining > 0:
            log.info("收盘前强制补足剩余数量: %d" % remaining)
            order_target(g.security, g.target_volume)
        g.finished = True

代码关键点解析

1. 数据获取 (get_history)：

   • 我们获取了 g.lookback_days * 240 个分钟数据。A股每天交易时间为 4 小时，即 240 分钟。
   • frequency='1m' 确保我们有足够细的颗粒度来拟合曲线。

2. 分布计算 (Pandas Groupby)：

   • df['time_str'] 将日期时间索引转换成了纯时间字符串（如 "09:35"）。
   • groupby('time_str')['volume'].mean() 是算法的核心。它把过去 10 天里，每天 09:35 的成交量加在一起求平均。这样就消除了某一天的异常波动，得到了该股票典型的日内成交特征（通常是早盘和尾盘量大，午盘量小）。
   • cumsum() 计算累计占比，生成一条从 0 增加到 1 的 S 型曲线。

3. 执行逻辑 (order_target)：

   • 我们没有使用 order()（按数量买入），而是使用了 order_target()（调整至目标持仓）。
   • 原因：如果使用 order()，假设你在 10:00 下单 1000 股，但因为价格波动只成交了 500 股。到了 10:01，算法计算发现你还缺货，如果用 order_target，它会自动识别你只成交了 500，并计算新的差额。如果用 order 简单的累加，容易导致计算偏差。

4. 时间处理：

   • PTrade 的 context.blotter.current_dt 获取的是当前回测/交易时间。
   • 代码中加入了收盘前（14:57）的强制检查，防止因为计算误差或最小下单单位限制导致最终没有买够数量。

如何优化这个策略？

如果你想让这个算法更接近实盘专业级水平，可以考虑以下改进：

1. 限价单保护：目前的 order_target 可能会以市价成交。你可以结合 get_snapshot 获取买一卖一价，使用 limit_price 进行限价委托，防止在大额买入时吃掉太多流动性导致成本过高。
2. 随机化拆单：目前的算法每分钟都会检查并下单，这在市场上可能会留下明显的算法痕迹。可以引入随机机制，例如每 5 分钟检查一次，或者在计算出的 order_qty 上增加一个小的随机浮动。
3. 动态 VWAP：目前的算法是基于历史分布（Static VWAP）。更高级的算法会结合当日已经发生的成交量来动态调整后续的预测曲线（Dynamic VWAP）。

Q&A

Q: 为什么我的 get_history 获取的数据在盘前是空的？
A: before_trading_start 运行时（通常是 08:30 或 09:00），当天的行情还没开始。get_history 默认获取的是截止到前一个交易日的数据，这是正常的。我们正是利用过去 N 天的数据来预测今天。

Q: 这个策略可以直接做空（卖出）吗？
A: 可以。只需将 g.target_volume 设为 0（如果是清仓）或者负数（如果是期货/融券），并将 order_qty 的计算逻辑稍作调整（卖出时 current > target）。对于简单的清仓 VWAP，只需将 target_position 计算逻辑改为 int(g.initial_position * (1 - target_ratio)) 即可。

Q: 如何在回测中查看效果？
A: 运行回测后，查看“交易详情”或日志。你应该能看到密集的下单记录，且成交价格的均价应该非常接近当天的市场 VWAP 均价。