配对交易与协整分析：统计套利的核心技术

import numpy as np
import pandas as pd
from statsmodels.tsa.stattools import coint, adfuller
import yfinance as yf

# 下载股票数据
def download_data(ticker1, ticker2, start='2020-01-01'):
    stock1 = yf.download(ticker1, start=start)['Adj Close']
    stock2 = yf.download(ticker2, start=start)['Adj Close']
    return pd.DataFrame({ticker1: stock1, ticker2: stock2}).dropna()

# ADF检验
def adf_test(series, title=''):
    result = adfuller(series, autolag='AIC')
    print(f'\nADF Test: {title}')
    print(f'ADF Statistic: {result[0]:.4f}')
    print(f'p-value: {result[1]:.4f}')
    return result[1] <= 0.05

# 协整检验
def cointegration_test(s1, s2):
    score, pvalue, _ = coint(s1, s2)
    print(f'\nCointegration Test')
    print(f'Score: {score:.4f}')
    print(f'p-value: {pvalue:.4f}')
    return pvalue <= 0.05

# 示例：检验可口可乐和百事可乐
data = download_data('KO', 'PEP')
s1 = np.log(data['KO'])
s2 = np.log(data['PEP'])

print("=== 单位根检验 ===")
is_s1_stationary = adf_test(s1, 'KO')
is_s2_stationary = adf_test(s2, 'PEP')

if not is_s1_stationary and not is_s2_stationary:
    print("\n两只股票价格都是非平稳的，可以进行协整检验")
    is_cointegrated = cointegration_test(s1, s2)
    if is_cointegrated:
        print("✓ 两只股票存在协整关系！")

from sklearn.linear_model import LinearRegression

# 计算对冲比率
def calculate_hedge_ratio(s1, s2):
    X = s2.values.reshape(-1, 1)
    y = s1.values
    model = LinearRegression()
    model.fit(X, y)
    return model.coef_[0]

# 计算残差和Z-Score
hedge_ratio = calculate_hedge_ratio(s1, s2)
spread = s1 - hedge_ratio * s2
z_score = (spread - spread.mean()) / spread.std()

# 交易信号
def generate_signals(z_score, entry_threshold=2.0, exit_threshold=0.5):
    signals = pd.DataFrame(index=z_score.index)
    signals['z_score'] = z_score
    
    # 1: 做多价差, -1: 做空价差, 0: 平仓
    signals['position'] = 0
    signals.loc[z_score < -entry_threshold, 'position'] = 1  # 做多
    signals.loc[z_score > entry_threshold, 'position'] = -1   # 做空
    signals.loc[abs(z_score) < exit_threshold, 'position'] = 0  # 平仓
    
    return signals