PTrade 配对交易策略 (包含协整检验)

这是一个基于统计套利（Statistical Arbitrage）的配对交易策略。该策略的核心思想是寻找两只价格走势具有长期稳定关系（协整关系）的股票。当它们的价格价差（Spread）偏离历史均值时进行反向操作，期待价差回归。

策略逻辑概述

1. 选股：在 initialize 中设定两只属于同一行业或具有强相关性的股票（例如：浦发银行与华夏银行）。
2. 协整检验 (Cointegration Test)：
   • 使用 Engle-Granger 两步法。
   • 第一步：对两只股票的历史价格进行 OLS 线性回归，得到残差序列。
   • 第二步：对残差序列进行 ADF (Augmented Dickey-Fuller) 检验。如果残差是平稳的（p-value < 0.05），则认为两只股票具有协整关系。
3. 信号生成：
   • 计算当前的价差（Spread）= Stock A - HedgeRatio * Stock B。
   • 计算价差的 Z-Score = (当前价差 - 价差均值) / 价差标准差。
4. 交易执行：
   • 做空价差 (Z-Score > 阈值)：卖出 Stock A，买入 Stock B。
   • 做多价差 (Z-Score < -阈值)：买入 Stock A，卖出 Stock B。
   • 平仓 (abs(Z-Score) < 平仓阈值)：平掉所有仓位。

策略代码

import numpy as np
import pandas as pd
import statsmodels.api as sm
from statsmodels.tsa.stattools import adfuller

def initialize(context):
    """
    初始化函数
    """
    # 1. 设定要进行配对交易的两只股票
    # 这里以银行股为例：浦发银行(600000.SS) 和 华夏银行(600015.SS)
    g.stock_y = '600000.SS' # 标的Y (因变量)
    g.stock_x = '600015.SS' # 标的X (自变量)
    g.security = [g.stock_y, g.stock_x]
    set_universe(g.security)
    
    # 2. 策略参数设置
    g.window = 20           # 计算均值和标准差的滚动窗口
    g.entry_threshold = 2.0 # 开仓阈值 (Z-Score)
    g.exit_threshold = 0.5  # 平仓阈值 (Z-Score)
    
    # 3. 协整检验参数
    g.coint_check_days = 120 # 用于协整检验的历史数据长度
    g.is_cointegrated = False # 标记是否通过协整检验
    g.hedge_ratio = 0.0       # 对冲比例 (Beta)
    
    # 设定基准
    set_benchmark('000300.SS')
    # 开启每日定时运行，用于盘前检查
    run_daily(context, check_cointegration_and_calc_params, time='09:30')

def check_cointegration_and_calc_params(context):
    """
    盘前/每日处理：
    1. 获取历史数据
    2. 进行协整检验
    3. 计算对冲比例(Hedge Ratio)
    """
    # 获取过去 N 天的收盘价数据
    hist = get_history(g.coint_check_days, '1d', 'close', g.security, fq='pre')
    
    # 数据预处理：转为 DataFrame 格式，列为股票代码
    # 注意：get_history 返回格式可能因版本不同而异，这里做通用处理
    if isinstance(hist, pd.DataFrame) and 'code' in hist.columns:
        # Python 3.11+ PTrade 常见格式
        df_close = hist.pivot(index='datetime', columns='code', values='close')
    else:
        # 旧版本或字典格式处理
        df_close = pd.DataFrame()
        if isinstance(hist, dict):
            for code in g.security:
                # 假设返回的是 numpy array 结构
                df_close[code] = hist[code]['close']
        else:
            # 假设是 Panel 或直接 DataFrame
            df_close = hist
            
    # 确保数据完整
    if df_close.isnull().values.any() or len(df_close) < g.coint_check_days:
        log.warning("历史数据不足或包含空值，跳过今日计算")
        g.is_cointegrated = False
        return

    # 提取价格序列 (取对数价格通常更稳定，但这里为了直观使用原始价格)
    Y = df_close[g.stock_y]
    X = df_close[g.stock_x]
    
    # --- 步骤1: OLS 回归计算对冲比例 (Hedge Ratio) ---
    # Y = beta * X + alpha
    X_add_const = sm.add_constant(X)
    model = sm.OLS(Y, X_add_const).fit()
    g.hedge_ratio = model.params[g.stock_x] # 斜率即为对冲比例
    alpha = model.params['const']
    
    # --- 步骤2: 协整检验 (ADF Test on Residuals) ---
    residuals = Y - (g.hedge_ratio * X + alpha)
    
    # 使用 ADF 检验残差的平稳性
    # adfuller 返回值: (adf_stat, pvalue, usedlag, nobs, crit_values, icbest)
    adf_result = adfuller(residuals)
    p_value = adf_result[1]
    
    # 判断是否协整 (通常 p-value < 0.05 认为显著)
    if p_value < 0.05:
        g.is_cointegrated = True
        log.info("协整检验通过: p-value=%.4f, Hedge Ratio=%.4f" % (p_value, g.hedge_ratio))
    else:
        g.is_cointegrated = False
        log.info("协整检验失败: p-value=%.4f. 关系不稳定，今日不交易。" % p_value)

def handle_data(context, data):
    """
    盘中交易逻辑
    """
    # 如果没有通过协整检验，或者数据不全，则清仓并暂停
    if not g.is_cointegrated:
        if len(context.portfolio.positions) > 0:
            for stock in context.portfolio.positions:
                order_target(stock, 0)
        return

    # 获取当前价格
    try:
        price_y = data[g.stock_y]['close']
        price_x = data[g.stock_x]['close']
    except:
        return # 数据缺失则跳过

    # 获取用于计算 Z-Score 的近期历史数据
    hist = get_history(g.window, '1d', 'close', g.security, fq='pre')
    
    # 数据格式化 (同上)
    if isinstance(hist, pd.DataFrame) and 'code' in hist.columns:
        df_recent = hist.pivot(index='datetime', columns='code', values='close')
    else:
        df_recent = pd.DataFrame()
        if isinstance(hist, dict):
            for code in g.security:
                df_recent[code] = hist[code]['close']
        else:
            df_recent = hist

    # 计算价差序列 Spread = Y - ratio * X
    # 注意：这里简化处理，忽略了 alpha，主要关注价差的波动
    spread_series = df_recent[g.stock_y] - g.hedge_ratio * df_recent[g.stock_x]
    
    # 计算统计量
    mean_spread = spread_series.mean()
    std_spread = spread_series.std()
    
    # 计算当前价差
    current_spread = price_y - g.hedge_ratio * price_x
    
    # 计算 Z-Score
    if std_spread > 0:
        z_score = (current_spread - mean_spread) / std_spread
    else:
        z_score = 0
        
    # 获取当前持仓
    pos_y = get_position(g.stock_y).amount
    pos_x = get_position(g.stock_x).amount
    
    # --- 交易信号逻辑 ---
    
    # 1. 做空价差 (Spread 过大，预计回归)：卖出 Y，买入 X
    if z_score > g.entry_threshold:
        # 简单的资金分配：各使用一半资金（实际应根据对冲比例调整）
        target_value = context.portfolio.portfolio_value * 0.5
        
        # 卖出 Y (PTrade 股票不能直接做空，这里假设是融券或平掉之前的多头)
        # 注意：如果是普通A股账户，只能先平仓 Y。如果是两融账户，可以使用 marginsec_open
        # 这里演示逻辑：假设我们持有底仓或者使用两融接口(需开通权限)
        # 为保证代码通用性，这里写成：如果持有Y则卖出，买入X
        
        # 实际配对交易通常需要融券。以下逻辑适配普通账户(只能做多)和两融逻辑的混合示意：
        # 逻辑：看空 Spread -> 减少 Y 持仓，增加 X 持仓
        
        log.info("Z-Score: %.2f > 阈值. 触发做空价差: 卖 %s, 买 %s" % (z_score, g.stock_y, g.stock_x))
        
        # 卖出 Y
        order_target_value(g.stock_y, 0) 
        # 买入 X
        order_target_value(g.stock_x, target_value)

    # 2. 做多价差 (Spread 过小，预计回归)：买入 Y，卖出 X
    elif z_score < -g.entry_threshold:
        target_value = context.portfolio.portfolio_value * 0.5
        
        log.info("Z-Score: %.2f < -阈值. 触发做多价差: 买 %s, 卖 %s" % (z_score, g.stock_y, g.stock_x))
        
        # 卖出 X
        order_target_value(g.stock_x, 0)
        # 买入 Y
        order_target_value(g.stock_y, target_value)

    # 3. 平仓 (Spread 回归均值)
    elif abs(z_score) < g.exit_threshold:
        if pos_y != 0 or pos_x != 0:
            log.info("Z-Score: %.2f 回归均值. 平仓所有头寸." % z_score)
            order_target(g.stock_y, 0)
            order_target(g.stock_x, 0)

关键点解析

1. 协整检验 (Cointegration Check)：

   • 代码中使用了 statsmodels 库。这是 Python 中最标准的统计学库。
   • sm.OLS 用于计算两个价格序列之间的线性关系，从而确定对冲比例（Hedge Ratio）。
   • adfuller (ADF检验) 用于验证残差是否平稳。如果残差平稳，说明两者之间存在长期均衡关系，价差会均值回归。
   • 注意：协整关系不是永久的，所以我们在 check_cointegration_and_calc_params 中每天（或定期）重新检验。

2. Z-Score 计算：

   • Z-Score 标准化了价差的波动，使得我们可以用统一的阈值（如 2.0 和 0.5）来衡量偏离程度，而不需要关心股价的绝对数值。

3. PTrade 特性适配：

   • 使用了 get_history 获取数据。
   • 使用了 run_daily 在盘前进行耗时的统计计算，避免在 handle_data 中每分钟重复计算，提高效率。
   • 使用了 order_target_value 来管理仓位。

4. 做空限制说明：

   • 上述代码使用的是标准的 order_target 接口。在中国 A 股普通账户中，不能直接裸卖空。
   • 如果策略信号提示“卖出 Y，买入 X”，在普通账户下，只能是“卖出持有的 Y（如果有），买入 X”。
   • 如果您使用的是信用账户（两融账户），请将下单函数替换为 marginsec_open (融券卖出) 和 margincash_open (融资买入) 等专用接口。

Q&A

Q: 为什么使用协整而不是相关性(Correlation)?
A: 相关性高只代表两只股票涨跌方向一致，但它们的价差可能会无限扩大。协整保证了价差是平稳的，即价差会围绕一个均值波动，这才是配对交易均值回归获利的基础。

Q: 如何选择股票对?
A: 通常选择同行业的龙头股（如两桶油、两大行、航空股等），或者产业链上下游关系紧密的企业。

Q: 代码报错 ImportError: No module named statsmodels 怎么办?
A: PTrade 的研究和回测环境通常内置了 statsmodels。如果实盘环境缺失该库，您需要手动实现简化的均值回归逻辑（仅计算价差均值和标准差），或者联系券商在服务器安装该库。