如何用协整检验筛选配对交易的股票对并实现策略？

这是一个基于 聚宽 (JoinQuant) 平台的配对交易（Pair Trading）策略示例。

策略原理

1. 协整性 (Cointegration)：不同于简单的相关性（Correlation），协整性衡量的是两个价格序列的线性组合是否具有平稳性（Stationarity）。如果两个股票协整，它们的价差（Spread）会围绕一个均值上下波动（均值回归）。
2. 选股：使用 Engle-Granger 两步法（statsmodels.tsa.stattools.coint）对股票池内的股票两两进行检验，找出 P值最小（协整关系最强）的一对股票。
3. 交易信号：
   • 计算价差：$Spread = Price_A - \beta \times Price_B$
   • 计算 Z-Score：$Z = \frac{Spread - Mean}{Std}$
   • 开仓：当 Z-Score 突破阈值（如 > 2 或 < -2）时，说明价差偏离正常水平，进行套利（做空高估的，做多低估的）。
   • 平仓：当 Z-Score 回归均值附近（如绝对值 < 0.5）时，平掉仓位。

策略代码

# -*- coding: utf-8 -*-
import numpy as np
import pandas as pd
import statsmodels.api as sm
from statsmodels.tsa.stattools import coint
from jqdata import *

def initialize(context):
    """
    初始化函数
    """
    # 1. 设定基准
    set_benchmark('000300.XSHG')
    # 2. 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 3. 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')
    
    # 4. 设定融资融券账户 (配对交易通常需要做空，必须开启融资融券)
    # 注意：回测资金量要足够，否则可能无法开仓
    set_subportfolios([SubPortfolioConfig(cash=context.portfolio.starting_cash, type='stock_margin')])

    # --- 策略参数设置 ---
    # 备选股票池：这里选取几只银行股作为示例，银行股同质化强，容易出现协整
    g.stock_pool = ['000001.XSHE', '600000.XSHG', '600036.XSHG', '601166.XSHG', '601398.XSHG']
    
    # 协整检验的时间窗口 (用于选对)
    g.check_period = 250 
    # 计算Z-Score的移动窗口 (用于交易信号)
    g.window = 20
    # 开仓阈值 (标准差倍数)
    g.entry_threshold = 2.0
    # 平仓阈值
    g.exit_threshold = 0.5
    
    # 全局变量记录当前交易的配对信息
    g.pair = []          # 存放 [Stock_Y, Stock_X]
    g.hedge_ratio = 0    # 对冲比例 beta
    g.spread_mean = 0    # 价差均值
    g.spread_std = 0     # 价差标准差
    
    # 设置定时运行：每月第一个交易日进行协整检验和选股
    run_monthly(select_pairs, 1, time='before_open')
    # 设置定时运行：每个交易日盘中进行交易判断
    run_daily(trade, '14:50')

def select_pairs(context):
    """
    选股逻辑：遍历股票池，寻找协整性最好的一对股票
    """
    # 获取历史收盘价数据
    df_price = history(g.check_period, '1d', 'close', g.stock_pool)
    
    best_pvalue = 1.0
    best_pair = []
    best_hedge_ratio = 0
    
    n = len(g.stock_pool)
    
    # 遍历所有组合进行协整检验
    for i in range(n):
        for j in range(i + 1, n):
            s1 = g.stock_pool[i]
            s2 = g.stock_pool[j]
            
            # 获取价格序列
            y = df_price[s1]
            x = df_price[s2]
            
            # 1. 协整检验 (Engle-Granger Test)
            # coint返回: t-statistic, p-value, crit_value
            _, pvalue, _ = coint(y, x)
            
            # 2. 如果P值更低，说明协整关系更强
            if pvalue < best_pvalue:
                best_pvalue = pvalue
                best_pair = [s1, s2]
                
                # 3. 计算对冲比例 (Hedge Ratio)
                # 使用 OLS 回归计算: Y = beta * X + alpha
                x_const = sm.add_constant(x)
                model = sm.OLS(y, x_const).fit()
                best_hedge_ratio = model.params[1] # 斜率即为 beta
    
    # 如果找到了显著的协整对 (通常要求 pvalue < 0.05，这里为了演示简单，只要有最优的就选)
    if best_pair and best_pvalue < 0.05:
        # 如果换了新的配对，先平掉旧仓位
        if g.pair and g.pair != best_pair:
            log.info(f"配对更换，平掉旧仓位: {g.pair}")
            close_all_positions(context)
            
        g.pair = best_pair
        g.hedge_ratio = best_hedge_ratio
        log.info(f"选中配对: {g.pair}, P-Value: {best_pvalue:.4f}, Hedge Ratio: {g.hedge_ratio:.4f}")
    else:
        log.info("未找到显著协整的股票对，本月空仓。")
        g.pair = []
        close_all_positions(context)

def trade(context):
    """
    交易逻辑：计算Z-Score并执行均值回归交易
    """
    if not g.pair:
        return
    
    s1 = g.pair[0] # Y
    s2 = g.pair[1] # X
    
    # 获取过去 g.window 天的数据来计算当前的 Z-Score
    prices = history(g.window, '1d', 'close', [s1, s2])
    y = prices[s1]
    x = prices[s2]
    
    # 计算价差序列 Spread = Y - beta * X
    spread_series = y - g.hedge_ratio * x
    
    # 计算统计量
    mean = spread_series.mean()
    std = spread_series.std()
    
    # 获取当前最新价格
    current_y = y.iloc[-1]
    current_x = x.iloc[-1]
    current_spread = current_y - g.hedge_ratio * current_x
    
    # 计算 Z-Score
    if std == 0:
        return
    z_score = (current_spread - mean) / std
    
    # 获取当前仓位
    pos_y = context.portfolio.positions[s1].total_amount
    pos_x = context.portfolio.positions[s2].total_amount
    
    # --- 交易信号判断 ---
    
    # 情况1: Z-Score > 阈值 (价差过大，做空价差 -> 卖Y 买X)
    if z_score > g.entry_threshold:
        # 如果当前没有持有"做空价差"的仓位
        # 注意：这里简化逻辑，通过判断持有Y的数量来确定方向。
        # 实际上做空Y意味着融券卖出，做多X意味着融资买入
        if pos_y >= 0: 
            log.info(f"Z-Score: {z_score:.2f} > {g.entry_threshold}, 触发做空价差 (卖{s1}, 买{s2})")
            # 资金分配：假设各用一半资金
            cash = context.portfolio.total_value * 0.5
            
            # 融券卖出 Y (s1)
            marginsec_open(s1, int(cash / current_y / 100) * 100)
            # 融资买入 X (s2)
            margincash_open(s2, int(cash / current_x / 100) * 100)

    # 情况2: Z-Score < -阈值 (价差过小，做多价差 -> 买Y 卖X)
    elif z_score < -g.entry_threshold:
        if pos_x >= 0: # 如果当前没有持有"做多价差"的仓位
            log.info(f"Z-Score: {z_score:.2f} < -{g.entry_threshold}, 触发做多价差 (买{s1}, 卖{s2})")
            cash = context.portfolio.total_value * 0.5
            
            # 融资买入 Y (s1)
            margincash_open(s1, int(cash / current_y / 100) * 100)
            # 融券卖出 X (s2)
            marginsec_open(s2, int(cash / current_x / 100) * 100)

    # 情况3: Z-Score 回归均值 (平仓)
    elif abs(z_score) < g.exit_threshold:
        if pos_y != 0 or pos_x != 0:
            log.info(f"Z-Score: {z_score:.2f} 回归均值，平仓")
            close_all_positions(context)

def close_all_positions(context):
    """
    平掉所有仓位（包括普通仓位和融资融券仓位）
    """
    for security in context.portfolio.positions.keys():
        pos = context.portfolio.positions[security]
        
        # 平多单 (卖出还款)
        if pos.total_amount > 0:
            order_target(security, 0) # 或者 margincash_close
            
        # 平空单 (买入还券)
        # 注意：聚宽的position对象中，空单数量通常在 short_positions 或通过 side 判断
        # 这里使用通用的 order_target(0) 在融资融券账户下通常能处理净仓位
        # 但为了严谨，针对融券仓位应使用 marginsec_close
        
    # 遍历融券卖出的仓位进行平仓
    for security in context.portfolio.short_positions.keys():
        amount = context.portfolio.short_positions[security].total_amount
        if amount > 0:
            marginsec_close(security, amount)

代码关键点解析

1. 协整检验 (coint):

   • 使用了 statsmodels.tsa.stattools.coint。
   • 它返回三个值，第二个值是 p-value。
   • 如果 p-value < 0.05，通常认为两个序列具有显著的协整关系（即它们的价差是平稳的，会均值回归）。

2. 对冲比例 ($\beta$):

   • 我们不能简单地做多1股A，做空1股B，因为股价不同且波动率不同。
   • 通过 OLS 线性回归 ($Price_Y = \beta \times Price_X + \alpha$) 计算出 $\beta$ (hedge_ratio)。
   • 价差公式定义为 $Spread = Y - \beta X$。

3. 融资融券 (stock_margin):

   • 配对交易的核心是多空对冲。在A股市场，做空需要通过融券实现。
   • 代码中使用了 set_subportfolios(..., type='stock_margin') 来初始化融资融券账户。
   • 使用了 margincash_open (融资买入) 和 marginsec_open (融券卖出) 函数。

4. Z-Score 信号:

   • Z-Score 标准化了价差，使得我们可以用统一的标准（如2倍标准差）来衡量价差是否偏离过大。
   • 当 Z-Score 回归到 0 附近时，意味着价差回到了历史平均水平，此时平仓获利。

注意事项

• 回测速度：协整检验计算量较大，如果股票池很大（如全市场），回测会非常慢甚至超时。建议将 g.stock_pool 限制在同一个行业（如银行、券商、航空）的几十只股票内。
• 做空限制：在实盘或模拟盘中，融券往往很难借到（券源不足）。回测环境中通常假设券源充足。
• 参数敏感性：check_period（检验窗口）和 window（交易窗口）对策略表现影响很大，需要根据具体品种进行调优。

Q&A

Q: 为什么使用协整而不是相关系数？
A: 相关系数高只代表两只股票走势方向一致，但它们的价差可能会无限扩大（不收敛）。协整保证了价差是平稳的，即价差无论跑多远，最终大概率会拉回来，这是均值回归策略盈利的数学基础。

Q: 为什么代码中要用 stock_margin 账户？
A: 配对交易必须同时买入一只股票并卖出另一只股票以对冲市场风险。在A股，卖出自己没有持有的股票（做空）必须通过信用账户（融资融券）进行。

Q: 如何优化选股池？
A: 可以先用基本面因子（如行业分类、PE/PB相近）进行初筛，缩小范围后再进行协整检验，这样既能提高计算速度，也能增加逻辑的可靠性。