问题描述
请帮我写一个配对交易策略，需要包含如何用协整检验来找股票对
解决方案
这是一个基于 聚宽 (JoinQuant) 平台的配对交易（Pair Trading）策略示例。
策略原理
- 协整性 (Cointegration)：不同于简单的相关性（Correlation），协整性衡量的是两个价格序列的线性组合是否具有平稳性（Stationarity）。如果两个股票协整，它们的价差（Spread）会围绕一个均值上下波动（均值回归）。
- 选股：使用 Engle-Granger 两步法（
statsmodels.tsa.stattools.coint）对股票池内的股票两两进行检验，找出 P值最小（协整关系最强）的一对股票。 - 交易信号：
- 计算价差：$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)
代码关键点解析
-
协整检验 (
coint):- 使用了
statsmodels.tsa.stattools.coint。 - 它返回三个值，第二个值是 p-value。
- 如果 p-value < 0.05，通常认为两个序列具有显著的协整关系（即它们的价差是平稳的，会均值回归）。
- 使用了
-
对冲比例 ($\beta$):
- 我们不能简单地做多1股A，做空1股B，因为股价不同且波动率不同。
- 通过 OLS 线性回归 ($Price_Y = \beta \times Price_X + \alpha$) 计算出 $\beta$ (
hedge_ratio)。 - 价差公式定义为 $Spread = Y - \beta X$。
-
融资融券 (
stock_margin):- 配对交易的核心是多空对冲。在A股市场，做空需要通过融券实现。
- 代码中使用了
set_subportfolios(..., type='stock_margin')来初始化融资融券账户。 - 使用了
margincash_open(融资买入) 和marginsec_open(融券卖出) 函数。
-
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相近）进行初筛，缩小范围后再进行协整检验，这样既能提高计算速度，也能增加逻辑的可靠性。