跨资产统计套利：股票、债券、商品的均值回归机会

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

# 1. 获取跨资产数据
assets = {
    'stocks': ['SPY', 'QQQ', 'IWM'],          # 股票ETF
    'bonds': ['TLT', 'IEF', 'SHY'],           # 债券ETF
    'commodities': ['GLD', 'SLV', 'USO'],      # 商品ETF
    'credit': ['HYG', 'JNK', 'LQD']           # 信用债ETF
}

# 下载2015-2025年数据
data = pd.DataFrame()
for asset_class, tickers in assets.items():
    for ticker in tickers:
        data[ticker] = yf.download(ticker, start='2015-01-01', end='2025-12-31')['Adj Close']

# 2. 测试协整关系
def find_cointegrated_pairs(data, p_value_threshold=0.05):
    """寻找具有协整关系的跨资产对"""
    n = data.shape[1]
    score_matrix = np.zeros((n, n))
    pvalue_matrix = np.ones((n, n))
    keys = data.keys()
    pairs = []
    
    for i in range(n):
        for j in range(i+1, n):
            S1 = data[keys[i]]
            S2 = data[keys[j]]
            
            # 协整检验
            result = coint(S1, S2)
            score = result[0]
            pvalue = result[1]
            
            score_matrix[i, j] = score
            pvalue_matrix[i, j] = pvalue
            
            if pvalue < p_value_threshold:
                pairs.append((keys[i], keys[j], pvalue))
    
    return pairs, pvalue_matrix

# 3. 找出跨资产协整对
pairs, pvalues = find_cointegrated_pairs(data)
print(f"找到 {len(pairs)} 个协整对:")
for pair in pairs[:5]:
    print(f"{pair[0]} - {pair[1]}: p-value = {pair[2]:.4f}")

# 1. 计算价差（hedge ratio通过OLS回归得到）
def calculate_spread(series1, series2):
    """计算两个价格序列的价差"""
    # OLS回归获取对冲比例
    import statsmodels.api as sm
    X = sm.add_constant(series2)
    model = sm.OLS(series1, X).fit()
    hedge_ratio = model.params[1]
    
    # 计算价差
    spread = series1 - hedge_ratio * series2
    return spread, hedge_ratio

# 2. 计算价差的z-score
def calculate_zscore(spread, window=22):
    """计算价差的滚动z-score"""
    mean = spread.rolling(window=window).mean()
    std = spread.rolling(window=window).std()
    zscore = (spread - mean) / std
    return zscore

# 3. 生成交易信号
def generate_signals(zscore, entry_threshold=2.0, exit_threshold=0.5):
    """根据z-score生成交易信号"""
    signals = pd.Series(index=zscore.index, dtype=float)
    signals[:] = 0
    
    # 做多价差（买入被低估资产，卖出被高估资产）
    signals[zscore < -entry_threshold] = 1
    
    # 做空价差
    signals[zscore > entry_threshold] = -1
    
    # 平仓信号
    signals[(zscore > -exit_threshold) & (zscore < exit_threshold)] = 0
    
    return signals

class CrossAssetArbStrategy:
    def __init__(self, max_leverage=2.0, stop_loss=0.05):
        self.max_leverage = max_leverage
        self.stop_loss = stop_loss
        
    def calculate_position_size(self, zscore, volatility, account_equity):
        """基于波动率和z-score计算仓位大小"""
        # 基础仓位与z-score绝对值成正比
        base_position = abs(zscore) / 4.0  # z-score=4时满仓
        
        # 波动率调整（高波动时降低仓位）
        vol_adjustment = 0.15 / volatility  # 目标波动率15%
        
        # 杠杆限制
        position = min(base_position * vol_adjustment, self.max_leverage)
        
        # 转换为实际金额
        notional = position * account_equity
        return notional
    
    def risk_management(self, positions, pnl):
        """风险管理检查"""
        # 止损检查
        if pnl < -self.stop_loss * account_equity:
            return "CLOSE_ALL"
        
        # 集中度检查
        asset_class_exposure = {}
        for position in positions:
            asset_class = self.classify_asset(position.ticker)
            asset_class_exposure[asset_class] = asset_class_exposure.get(asset_class, 0) + position.notional
        
        # 单一资产类别暴露不超过50%
        for asset_class, exposure in asset_class_exposure.items():
            if abs(exposure) > 0.5 * account_equity:
                return f"REDUCE_{asset_class}_EXPOSURE"
        
        return "NORMAL"