用MC开发一个跨品种套利交易策略（上）

小象实战讲义 · 进阶量化交易实战：迭代式的量化策略研发 本讲义仅用于编程与量化方法教学，不构成任何投资建议。

MultiCharts（MC）支持多图表（Multi-Data）联动，是跨品种套利开发的利器。本节深入剖析 MC 多数据流同步机制，并用 Python 3.12 给出双品种价差跟踪与协整回归的等价实现。本讲义原课以 TB/MC 演示，此处为 Python 等价实现。

💡 核心导读

• MultiCharts 多数据流（Data1 & Data2）核心机制：
  • 时钟时间对齐（Timestamp Synchronization）：确保 Data1 与 Data2 在同一物理时间戳 Bar 上严密对齐；
  • 动态协整系数估计（Rolling OLS Beta）：实时估计对冲比率 $\beta$，动态消除品种间市值敞口；
  • 价差偏离度触发器：当价差脱离均值 $\pm 2\sigma$ 时，Data1 与 Data2 同时反向发出交易委托。

┌─────────────────────────────────────────────────────────────┐
│                 MultiCharts 双数据流对齐与套利拓扑          │
└──────────────────────────────┬──────────────────────────────┘
                               │
 ┌─────────────────────────────┴─────────────────────────────┐
 ▼                                                           ▼
┌─────────────────────────┐                                 ┌─────────────────────────┐
│ Data1 (标的 A 行情流)   │                                 │ Data2 (标的 B 行情流)   │
├─────────────────────────┤                                 ├─────────────────────────┤
│ • 螺纹钢主力合约（示例）│                                 │ • 热卷主力合约（示例）  │
│ • 价格序列 $P_A$        │                                 │ • 价格序列 $P_B$        │
└─────────────────────────┘                                 └─────────────────────────┘
                               │
                               ▼
                ┌─────────────────────────────┐
                │ 统一时间轴对齐 & 价差生成   │
                ├─────────────────────────────┤
                │ Spread = Price_A - Price_B  │
                │ Z = (Spread - MA) / Std     │
                └─────────────────────────────┘

一、双数据流时间戳对齐与滚动协整对冲比率 $\beta$

声明：原课以 TB/MC 演示，此处为 Python 等价实现。

在配对交易中，对冲比率 $\beta$ 通过滚动 OLS 最小二乘法估计：

$$P_{A, t} = \alpha + \beta P_{B, t} + \epsilon_t, \quad \beta = \frac{\text{Cov}(P_A, P_B)}{\text{Var}(P_B)}$$

import pandas as pd
import numpy as np

def calculate_rolling_beta(series_a: pd.Series, series_b: pd.Series, window: int = 60) -> pd.Series:
    """
    计算两资产滚动 OLS 协整对冲比率 Beta（Python 等价实现）
    """
    cov = series_a.rolling(window).cov(series_b)
    var = series_b.rolling(window).var()
    beta = cov / (var + 1e-8)
    return beta

# 合成数据演示滚动 Beta 计算（历史相关性不代表未来，存在结构性断裂风险）
rng = np.random.default_rng(42)
dates = pd.date_range('2024-01-01', periods=100, freq='D')
base = 100.0 + np.cumsum(rng.normal(0.05, 1.0, 100))
p_a = base + np.sin(np.linspace(0, 4*np.pi, 100)) * 3.0
p_b = base * 0.95 + rng.normal(0, 0.5, 100)

df_demo = pd.DataFrame({'price_a': p_a, 'price_b': p_b}, index=dates)
df_demo['beta_30'] = calculate_rolling_beta(df_demo['price_a'], df_demo['price_b'], window=30)
df_demo['beta_60'] = calculate_rolling_beta(df_demo['price_a'], df_demo['price_b'], window=60)

print("===== 滚动协整对冲比率 Beta 计算演示 =====")
print(f"数据日期范围: {df_demo.index[0].date()} 至 {df_demo.index[-1].date()}")
print(f"最新 30 期 Beta: {df_demo['beta_30'].iloc[-1]:.4f}")
print(f"最新 60 期 Beta: {df_demo['beta_60'].iloc[-1]:.4f}")
print(f"Beta 30 期非空值数量: {df_demo['beta_30'].notna().sum()}")
print(f"Beta 60 期非空值数量: {df_demo['beta_60'].notna().sum()}")

===== 滚动协整对冲比率 Beta 计算演示 =====
数据日期范围: 2024-01-01 至 2024-04-09
最新 30 期 Beta: 0.7484
最新 60 期 Beta: 1.3002
Beta 30 期非空值数量: 71
Beta 60 期非空值数量: 41

二、MC 套利开平仓信号与状态机 Python 实现

双品种套利状态机必须严格保证两腿（Legs）指令的原子性，开仓时 Data1 与 Data2 必须同步反向操作。

import pandas as pd
import numpy as np

def generate_pairs_trading_signals(df_merged: pd.DataFrame, entry_threshold: float = 2.0) -> pd.DataFrame:
    """
    MC 跨品种套利策略信号机（Python 等价实现）
    本函数只生成入场信号向量(pos_a/pos_b 开仓后保持, 不含回归中轴平仓);
    出场/平仓由下一节(9.3)的状态机以 exit_z 中轴阈值统一处理。
    """
    res = df_merged.copy()
    res['spread'] = res['price_a'] - res['beta'] * res['price_b']
    res['spread_ma'] = res['spread'].rolling(30).mean()
    res['spread_std'] = res['spread'].rolling(30).std()
    res['zscore'] = (res['spread'] - res['spread_ma']) / (res['spread_std'] + 1e-8)
    
    res['pos_a'] = 0
    res['pos_b'] = 0
    
    # 状态机：做多价差 (买 A 卖 B) 或 做空价差 (卖 A 买 B)
    res.loc[res['zscore'] > entry_threshold, 'pos_a'] = -1
    res.loc[res['zscore'] > entry_threshold, 'pos_b'] = 1
    res.loc[res['zscore'] < -entry_threshold, 'pos_a'] = 1
    res.loc[res['zscore'] < -entry_threshold, 'pos_b'] = -1
    return res

# 合成数据演示信号生成（合成样例，存在未来函数与过拟合风险）
rng = np.random.default_rng(123)
dates = pd.date_range('2024-03-01', periods=150, freq='D')
base = 50.0 + np.cumsum(rng.normal(0.02, 0.8, 150))
p_a = base + np.cos(np.linspace(0, 6*np.pi, 150)) * 2.5
p_b = base * 1.02 + rng.normal(0, 0.6, 150)

df_signal = pd.DataFrame({'price_a': p_a, 'price_b': p_b}, index=dates)

# 使用固定 Beta 简化演示（实盘应为滚动计算）
df_signal['beta'] = 0.98

signals = generate_pairs_trading_signals(df_signal, entry_threshold=1.5)

long_signals = (signals['pos_a'] == 1).sum()
short_signals = (signals['pos_a'] == -1).sum()
signal_dates = signals[signals['pos_a'] != 0].index

print("===== 跨品种套利策略信号生成演示 =====")
print(f"数据总长度: {len(signals)} 根 K 线")
print(f"价差 Z-Score 均值: {signals['zscore'].mean():.4f}")
print(f"价差 Z-Score 标准差: {signals['zscore'].std():.4f}")
print(f"做多价差信号次数 (买A卖B): {long_signals}")
print(f"做空价差信号次数 (卖A买B): {short_signals}")
if not signal_dates.empty:
    print(f"首个信号日期: {signal_dates[0].date()}")
    print(f"末个信号日期: {signal_dates[-1].date()}")

===== 跨品种套利策略信号生成演示 =====
数据总长度: 150 根 K 线
价差 Z-Score 均值: 0.1879
价差 Z-Score 标准差: 1.3589
做多价差信号次数 (买A卖B): 20
做空价差信号次数 (卖A买B): 29
首个信号日期: 2024-04-10
末个信号日期: 2024-07-28

📝 动手练一练

编写 Python 测试脚本，模拟 Data1 与 Data2 双数据流对齐并生成跨品种套利交易信号：

import pandas as pd
import numpy as np

def calculate_rolling_beta(series_a: pd.Series, series_b: pd.Series, window: int = 60) -> pd.Series:
    """
    计算两资产滚动 OLS 协整对冲比率 Beta（Python 等价实现）
    """
    cov = series_a.rolling(window).cov(series_b)
    var = series_b.rolling(window).var()
    beta = cov / (var + 1e-8)
    return beta

def generate_pairs_trading_signals(df_merged: pd.DataFrame, entry_threshold: float = 2.0) -> pd.DataFrame:
    """
    MC 跨品种套利策略信号机（Python 等价实现）
    本函数只生成入场信号向量(pos_a/pos_b 开仓后保持, 不含回归中轴平仓);
    出场/平仓由下一节(9.3)的状态机以 exit_z 中轴阈值统一处理。
    """
    res = df_merged.copy()
    res['spread'] = res['price_a'] - res['beta'] * res['price_b']
    res['spread_ma'] = res['spread'].rolling(30).mean()
    res['spread_std'] = res['spread'].rolling(30).std()
    res['zscore'] = (res['spread'] - res['spread_ma']) / (res['spread_std'] + 1e-8)
    
    res['pos_a'] = 0
    res['pos_b'] = 0
    
    # 状态机：做多价差 (买 A 卖 B) 或 做空价差 (卖 A 买 B)
    res.loc[res['zscore'] > entry_threshold, 'pos_a'] = -1
    res.loc[res['zscore'] > entry_threshold, 'pos_b'] = 1
    res.loc[res['zscore'] < -entry_threshold, 'pos_a'] = 1
    res.loc[res['zscore'] < -entry_threshold, 'pos_b'] = -1
    return res

# 完整流程测试（合成数据，历史回测不代表未来表现，存在交易成本与滑点未计入）
rng = np.random.default_rng(42)
dates = pd.date_range('2024-05-01', periods=100, freq='D')
base = 100.0 + np.cumsum(rng.normal(0.05, 1.0, 100))
p_a = base + np.sin(np.linspace(0, 4*np.pi, 100)) * 3.0
p_b = base * 0.95 + rng.normal(0, 0.5, 100)

df = pd.DataFrame({'price_a': p_a, 'price_b': p_b}, index=dates)
df['beta'] = calculate_rolling_beta(df['price_a'], df['price_b'], window=30)
res = generate_pairs_trading_signals(df)

long_spread_signals = (res['pos_a'] == 1).sum()
short_spread_signals = (res['pos_a'] == -1).sum()
signal_rows = res[res['pos_a'] != 0]

print("================ MC 套利策略 Python 等价实现（上）================")
print(f"数据总长度: {len(df)} 根 K 线")
print(f"最新协整对冲比率 Beta: {df['beta'].iloc[-1]:.4f}")
print(f"价差序列均值: {res['spread'].mean():.4f}")
print(f"价差序列标准差: {res['spread'].std():.4f}")
print(f"做多价差触发次数 (买A卖B): {long_spread_signals} 次")
print(f"做空价差触发次数 (卖A买B): {short_spread_signals} 次")
if not signal_rows.empty:
    avg_z_entry = signal_rows['zscore'].abs().mean()
    print(f"信号触发时平均 |Z-Score|: {avg_z_entry:.4f}")
assert not np.isnan(df['beta'].iloc[-1]), "Beta 计算异常！"
print("✅ MC 套利双数据流与协整信号测试 PASS")
print("================================================================")

运行结果输出：

================ MC 套利策略 Python 等价实现（上）================
数据总长度: 100 根 K 线
最新协整对冲比率 Beta: 0.7484
价差序列均值: 2.0065
价差序列标准差: 67.7863
做多价差触发次数 (买A卖B): 2 次
做空价差触发次数 (卖A买B): 0 次
信号触发时平均 |Z-Score|: 2.2273
✅ MC 套利双数据流与协整信号测试 PASS
================================================================

本章小结

1. 掌握多数据流同步技术：以统一时间轴严密对齐 Data1 与 Data2，杜绝异步数据错位。
2. 滚动估计动态对冲比率：利用滚动 OLS 提取动态 $\beta$，确保多空双腿头寸的市值中性。
3. 基于 Z-Score 的量化触发：通过统计偏离度标准化门限控制入场出场，提高套利胜率。

📋 行动清单 (Action Checklist)

• 运行 MC 套利测试代码，测试改变 OLS 滚动窗口对 $\beta$ 平滑度的影响；
• 为套利信号增加半衰期（Half-Life）均值回归速度检验；
• 预习下一节，掌握用 MultiCharts/Python 编写套利出场与止损控制。

—— 小象教研组