PTrade 经典策略复现：如何实现基于协整的配对交易策略？

这是一个基于 Ptrade 量化交易平台的**股票配对交易（Pairs Trading）**策略。

策略原理

配对交易是一种市场中性策略。它选择两只历史价格走势高度相关（协整）的股票。

1. 计算价差/比率：计算两只股票价格的比率（Price A / Price B）。
2. 计算统计指标：基于过去 N 天的数据，计算该比率的均值（Mean）和标准差（Std）。
3. 计算 Z-Score：$Z = (当前比率 - 均值) / 标准差$。
4. 交易信号：
   • 当 Z-Score > 阈值（如 2.0）：比率过高，说明 A 贵 B 便宜。做空 A，做多 B。
   • 当 Z-Score < -阈值（如 -2.0）：比率过低，说明 A 便宜 B 贵。做多 A，做空 B。
   • 当 abs(Z-Score) < 平仓阈值（如 0.5）：回归均值，平掉所有仓位。

PTrade 策略代码

import numpy as np
import pandas as pd

def initialize(context):
    """
    初始化函数，设置策略参数和股票池
    """
    # 1. 设定两只配对股票
    # 这里以两只银行股为例：浦发银行(600000.SS) 和 招商银行(600036.SS)
    # 在实盘中，应选择经过协整检验的相关性极高的标的
    g.stock1 = '600000.SS'
    g.stock2 = '600036.SS'
    g.security_list = [g.stock1, g.stock2]
    
    # 2. 设定策略参数
    g.window = 20           # 计算均值和标准差的移动窗口（天）
    g.entry_threshold = 2.0 # 开仓阈值（Z-Score绝对值）
    g.exit_threshold = 0.5  # 平仓阈值（Z-Score绝对值）
    
    # 3. 设定资金分配比例 (每只股票占用总资金的比例)
    g.position_ratio = 0.4  # 留一些现金作为缓冲
    
    # 4. 设置股票池 (必须步骤)
    set_universe(g.security_list)
    
    # 5. 设置手续费 (可选，回测时建议设置)
    set_commission(commission_ratio=0.0003, min_commission=5.0, type="STOCK")

def before_trading_start(context, data):
    """
    盘前处理
    """
    # 确保股票池生效
    set_universe(g.security_list)

def handle_data(context, data):
    """
    盘中逻辑，每个周期（日线或分钟）运行一次
    """
    # -----------------------------------------------------------------
    # 1. 数据获取
    # -----------------------------------------------------------------
    # 获取过去 N+2 天的收盘价数据，确保数据量足够计算 rolling 指标
    # 注意：get_history 返回的数据结构可能因版本而异，这里处理为 DataFrame
    hist_data = get_history(g.window + 2, frequency='1d', field='close', security_list=g.security_list)
    
    # 检查数据是否获取成功
    if hist_data is None or len(hist_data) < g.window:
        log.info("历史数据不足，跳过本次计算")
        return

    # 提取两只股票的价格序列
    # PTrade get_history 多股返回时，列名为股票代码
    try:
        # 兼容不同版本的返回格式，确保获取到 DataFrame
        if not isinstance(hist_data, pd.DataFrame):
            # 如果是 Panel 或其他格式，需根据实际环境调整，通常 PTrade 新版直接返回 DataFrame
            log.error("数据格式非 DataFrame，需检查 API 返回")
            return
            
        s1_prices = hist_data[g.stock1].values
        s2_prices = hist_data[g.stock2].values
    except Exception as e:
        log.error("数据提取失败: %s" % e)
        return

    # 检查是否有停牌导致的数据缺失 (NaN)
    if np.isnan(s1_prices).any() or np.isnan(s2_prices).any():
        log.info("存在缺失数据(可能停牌)，跳过")
        return

    # -----------------------------------------------------------------
    # 2. 统计计算 (Z-Score)
    # -----------------------------------------------------------------
    # 计算价格比率序列 (Stock1 / Stock2)
    ratio_series = s1_prices / s2_prices
    
    # 获取最新的比率
    current_ratio = ratio_series[-1]
    
    # 计算移动窗口内的均值和标准差 (不包含最新一个点，避免未来函数，或者包含均可，这里取最后N个点)
    # 使用切片取最近 g.window 个数据进行计算
    calc_series = ratio_series[-g.window:]
    mean = np.mean(calc_series)
    std = np.std(calc_series)
    
    # 防止除以0
    if std == 0:
        return

    # 计算 Z-Score
    z_score = (current_ratio - mean) / std
    
    # 打印日志方便调试
    log.info("Stock1: %.2f, Stock2: %.2f, Ratio: %.4f, Z-Score: %.4f" % (
        s1_prices[-1], s2_prices[-1], current_ratio, z_score))

    # -----------------------------------------------------------------
    # 3. 交易逻辑
    # -----------------------------------------------------------------
    # 获取当前账户总资产
    total_value = context.portfolio.portfolio_value
    # 单只股票的目标市值
    target_value = total_value * g.position_ratio
    
    # 获取当前持仓
    pos_s1 = context.portfolio.positions[g.stock1].amount
    pos_s2 = context.portfolio.positions[g.stock2].amount

    # --- 信号判断 ---
    
    # 情况 A: Z-Score > 上轨 (比率过高，S1贵，S2便宜) -> 做空 S1，做多 S2
    if z_score > g.entry_threshold:
        # 注意：A股普通账户无法直接做空。
        # 如果是回测融券环境，amount 传负数即可。
        # 如果是普通账户，只能卖出已有持仓。
        # 下面的逻辑假设支持做空（融券）或仅作为策略逻辑展示。
        
        log.info("信号触发：做空 %s，做多 %s" % (g.stock1, g.stock2))
        order_target_value(g.stock1, -target_value) # 卖出/做空 S1
        order_target_value(g.stock2, target_value)  # 买入 S2

    # 情况 B: Z-Score < 下轨 (比率过低，S1便宜，S2贵) -> 做多 S1，做空 S2
    elif z_score < -g.entry_threshold:
        
        log.info("信号触发：做多 %s，做空 %s" % (g.stock1, g.stock2))
        order_target_value(g.stock1, target_value)  # 买入 S1
        order_target_value(g.stock2, -target_value) # 卖出/做空 S2

    # 情况 C: Z-Score 回归均值附近 -> 平仓
    elif abs(z_score) < g.exit_threshold:
        
        # 如果有持仓（无论是多还是空），则平仓
        if pos_s1 != 0 or pos_s2 != 0:
            log.info("均值回归，平掉所有仓位")
            order_target_value(g.stock1, 0)
            order_target_value(g.stock2, 0)

代码关键点解析

1. 数据获取 (get_history)：

   • 我们获取了 g.window + 2 天的数据，这是为了确保计算移动平均和标准差时有足够的数据点。
   • security_list 传入列表，返回的是一个 DataFrame，索引是时间，列名是股票代码。

2. Z-Score 计算：

   • 这是配对交易的核心。公式为：$Z = (Current - Mean) / Std$。
   • 它衡量当前价差偏离历史均值的程度（以标准差为单位）。

3. 交易执行 (order_target_value)：

   • 使用 order_target_value 可以自动计算需要买卖的数量，使持仓市值达到目标值。
   • 关于做空：代码中使用了负数 -target_value。
     • 在 回测 环境中，如果开启了融券功能，这会产生空头仓位。
     • 在 实盘 环境中，A股普通账户不支持直接做空。如果你使用的是两融账户（信用账户），需要使用 marginsec_open (融券卖出) 和 marginsec_close (买券还券) 等专用接口，或者确认券商柜台是否支持通过 order 接口自动路由到两融柜台。

4. Python 3.5 兼容性：

   • 代码中使用了 % 格式化字符串（如 "Ratio: %.4f" % current_ratio），而不是 f-string，以确保在 PTrade 旧版本 Python 环境中也能运行。

常见问题 (Q&A)

Q: 这个策略可以直接用于A股实盘吗？
A: A股普通账户不支持做空（卖出你没有持有的股票）。如果你只有普通账户，这个策略只能做“轮动”，即两只股票都长期持有底仓，根据信号调整仓位权重。如果要实现真正的多空配对，必须开通融资融券账户。

Q: 如何选择配对的股票？
A: 代码中硬编码了 600000.SS 和 600036.SS。在实际研究中，你需要先在研究环境（Notebook）中进行协整性检验（Cointegration Test），找出走势长期绑定的一对股票（通常是同行业的龙头，如两桶油、两大行、航空股等）。

Q: 为什么使用 order_target_value 而不是 order？
A: order_target_value 更适合这种策略，因为它可以自动处理调仓。例如，如果当前已经持有 10万元 S1，信号触发要求持有 10万元 S1，函数会自动判断不需要操作；如果要求持有 0，它会自动卖出。这样简化了逻辑判断。