协整配对交易实战：A股统计套利10个经典案例

from statsmodels.tsa.stattools import adfuller

def adf_test(series, title=''):
    """Augmented Dickey-Fuller检验"""
    result = adfuller(series, autolag='AIC')
    
    print(f'ADF Statistic: {result[0]:.4f}')
    print(f'p-value: {result[1]:.4f}')
    
    if result[1] <= 0.05:
        print("→ 平稳（拒绝原假设）")
        return True
    else:
        print("→ 非平稳（接受原假设）")
        return False

# 检验两只股票的价差
spread = np.log(stock_a) - np.log(stock_b) * hedge_ratio
is_stationary = adf_test(spread, 'Spread')

from statsmodels.tsa.vector_ar.vecm import coint_johansen

def johansen_test(stock_a, stock_b):
    """Johansen协整检验"""
    data = pd.DataFrame({
        'stock_a': np.log(stock_a),
        'stock_b': np.log(stock_b)
    })
    
    result = coint_johansen(data, det_order=0, k_ar_diff=1)
    
    # 读取迹统计量
    trace_stat = result.lr1
    critical_value = result.cvt[:, 1]  # 95%置信度
    
    if trace_stat[0] > critical_value[0]:
        print("→ 存在协整关系（拒绝原假设）")
        return True
    else:
        print("→ 不存在协整关系")
        return False

import dash
import plotly.graph_objs as go

app = dash.Dash(__name__)

app.layout = html.Div([
    dcc.Graph(id='spread-zscore'),
    dcc.Graph(id='cumulative-pnl'),
    html.Table(id='open-positions')
])

# 实时更新价差Z-score
@app.callback(
    Output('spread-zscore', 'figure'),
    [Input('interval-component', 'n_intervals')]
)
def update_spread():
    # 从数据库读取最新价差
    spread = fetch_latest_spread()
    zscore = (spread - spread.mean()) / spread.std()
    
    fig = go.Figure()
    fig.add_trace(go.Scatter(x=spread.index, y=zscore))
    fig.add_hline(y=2, line_dash='dash', line_color='red')
    fig.add_hline(y=-2, line_dash='dash', line_color='green')
    
    return fig

from qclaw_trade_api import TradeAPI  # 假设的量化交易API

class PairsTradingBot:
    def __init__(self, pair, zscore_threshold=2.0):
        self.stock_a, self.stock_b = pair
        self.threshold = zscore_threshold
        self.api = TradeAPI()
        
    def check_signal(self):
        zscore = self.calculate_zscore()
        
        if zscore < -self.threshold:
            # 做多A，做空B
            self.api.place_order(self.stock_a, 'BUY', size=10000)
            self.api.place_order(self.stock_b, 'SELL', size=10000*self.hedge_ratio)
            
        elif zscore > self.threshold:
            # 做空A，做多B
            self.api.place_order(self.stock_a, 'SELL', size=10000)
            self.api.place_order(self.stock_b, 'BUY', size=10000*self.hedge_ratio)
            
        elif abs(zscore) < 0.5:
            # 平仓
            self.api.close_all_positions()