本文介绍通过多进程并行化显著提升协整对筛选效率的方法，针对滑动窗口场景下的高频计算需求，将原串行for循环重构为multiprocessing.Pool并行执行，实测在500只股票组合中将运行时间缩短至8秒级。
本文介绍通过多进程并行化显著提升协整对筛选效率的方法，针对滑动窗口场景下的高频计算需求，将原串行`for`循环重构为`multiprocessing.pool`并行执行，实测在500只股票组合中将运行时间缩短至8秒级。
在量化配对交易策略中，协整检验（如Johansen检验）是识别统计套利机会的核心步骤。然而，对S&P 500成分股等大规模资产池进行两两协整检验时，原始串行实现因O(n²)组合复杂度与单核瓶颈，极易成为性能瓶颈——尤其在滑动窗口回测中需反复调用，延迟直接影响策略时效性。
核心优化思路：从串行到并行
原代码中嵌套循环遍历所有股票对，并对每对执行coint_johansen检验、协方差建模与信号计算。该流程天然具备任务独立性：任意两支股票的协整检验互不影响，完全可并行化。我们采用multiprocessing.Pool替代for循环，在多核CPU上并发执行，理论加速比接近核心数（实践中受I/O与内存带宽限制，通常达3–4倍）。
关键重构要点
- 函数解耦：将单对检验逻辑抽离为独立函数test_cointegration(pair)，接收(stock1, stock2)元组，返回None或三元组结果；
- 预处理前置：在主函数find_cointegrated_pairs中一次性完成数据清洗（replace(0, np.nan).dropna(axis=1)），避免子进程重复操作；
- 并行调度：使用Pool(4)（可根据CPU核心数调整，如os.cpu_count()）并行映射所有组合；
- 结果聚合：过滤None值后统一构建DataFrame，结构清晰且便于后续分析。
以下是优化后的完整实现：
import numpy as np
import pandas as pd
import yfinance as yf
from statsmodels.tsa.vector_ar.vecm import coint_johansen
from itertools import combinations
from multiprocessing import Pool
import time
# 数据获取（注意：实际使用需处理缺失ticker）
df1 = pd.read_html('https://en.wikipedia.org/wiki/List_of_S%26P_500_companies')[0]
tickers = df1.Symbol.tolist()
df = yf.download(tickers, period="2y")["Close"].dropna(axis=1) # 自动剔除无数据股票
def test_cointegration(pair):
"""单对股票协整检验函数，返回(Stock1, Stock2, Signal)或None"""
stock1, stock2 = pair
try:
# 构造一阶差分序列（协整检验输入要求平稳性）
diff_df = pd.concat([
df[stock1].diff().dropna(),
df[stock2].diff().dropna()
], axis=1).dropna().astype(float)
if len(diff_df) < 20: # 最小样本量保护
return None
# 执行Johansen协整检验（det_order=1: 含截距项）
result = coint_johansen(diff_df, det_order=1, k_ar_diff=1)
trace_stat, crit_val = result.lr1[0], result.cvt[0, 1] # 使用第一特征根检验
if trace_stat > crit_val:
# 简化协整向量估计：用协方差比近似β（替代OLS，更鲁棒）
cov_mat = df[[stock1, stock2]].cov()
beta = cov_mat.iloc[0, 1] / cov_mat.iloc[0, 0] if cov_mat.iloc[0, 0] != 0 else np.nan
if np.isnan(beta):
return None
spread = df[stock2] - beta * df[stock1]
signal = (spread - spread.mean()) / spread.std()
return (stock1, stock2, signal.iloc[-1])
except Exception as e:
pass # 忽略单对计算异常，保障整体流程
return None
def find_cointegrated_pairs(data):
start_time = time.time()
# 预处理：剔除全零列与缺失列
data_clean = data.replace(0, np.nan).dropna(axis=1)
pairs = list(combinations(data_clean.columns, 2))
# 并行执行（建议根据CPU核心数调整processes参数）
with Pool(processes=4) as pool:
results = pool.map(test_cointegration, pairs)
# 过滤有效结果并构造DataFrame
valid_results = [r for r in results if r is not None]
results_df = pd.DataFrame(valid_results, columns=['Stock1', 'Stock2', 'Signal'])
end_time = time.time()
print(f"✅ 协整对筛选完成 | 总耗时: {end_time - start_time:.3f}s | 有效配对数: {len(results_df)}")
return results_df
# 执行示例
cointegrated_pairs = find_cointegrated_pairs(df)
print(cointegrated_pairs.head())注意事项与进阶建议
- 内存与稳定性：yfinance批量下载易触发限流或缺失数据（如BRK.B时区错误），建议添加重试机制或预缓存本地数据；
- 检验参数调优：coint_johansen的det_order（趋势项）、k_ar_diff（滞后阶数）需根据数据频率校准，日频数据通常设det_order=1；
- 替代方案：若仅需Engle-Granger二步法（双变量），可用statsmodels.tsa.stattools.coint替代，速度更快但不支持多变量扩展；
- 滑动窗口集成：将find_cointegrated_pairs封装为窗口函数，配合pandas.DataFrame.rolling或自定义时间切片器，实现动态配对更新。
通过上述重构，协整检验从分钟级降至秒级，为实时化、高频化的配对交易策略提供了坚实的技术基础。
立即学习“Python免费学习笔记（深入）”；
