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

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

套利交易虽然波动较小，但一旦遭遇基本面逻辑破裂（如产业链断裂或退市风险），价差将永久发散导致巨大穿仓。本节深入剖析套利策略的动态跟踪出场、价差极端发散止损以及跨市场执行腿滑点控制。本讲义原课以 TB/MC 演示，此处为 Python 等价实现。

💡 核心导读

• 套利策略出场与硬风控三大铁律：
  • 均值回归平仓（Mean Reversion Exit）：当 $Z$-Score 回归至 $[-0.5, 0.5]$ 正常区间时，双腿同步平仓获利；
  • 极端发散硬止损（Divergence Hard Stop）：当 $Z$-Score 突破 $\pm 3.5$ 时，判定协整关系破裂，无条件止损离场；
  • 时间止损保护（Arbitrage Time Stop）：超过预期均值回归半衰期 3 倍时间仍未回归，主动退出以防资金被套（本节代码未实现，作为课后练习）。

┌─────────────────────────────────────────────────────────────┐
│                    套利策略完整风控与出场状态机拓扑图       │
└──────────────────────────────┬──────────────────────────────┘
                               │
 ┌─────────────────────────────┼─────────────────────────────┐
 ▼                             ▼                             ▼
┌─────────────────────────┐   ┌─────────────────────────┐   ┌─────────────────────────┐
│ 正常均值回归获利平仓    │   │ 协整破裂极端硬止损      │   │ 超时强制平仓            │
├─────────────────────────┤   ├─────────────────────────┤   ├─────────────────────────┤
│ • $|Z| < 0.5$ 获利兑现  │   │ • $|Z| > 3.5$ 关系破裂  │   │ • 持仓时间 > $3 \times$ 半衰期 │
│ • 释放双腿保证金占用    │   │ • 坚决砍仓，防穿仓风险  │   │ • 提高资金流转周转效率  │
└─────────────────────────┘   └─────────────────────────┘   └─────────────────────────┘

一、Ornstein-Uhlenbeck 过程与半衰期（Half-Life）测算

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

价差序列服从 OU 均值回归随机微分方程 $d S_t = \theta (\mu - S_t) dt + \sigma dW_t$。半衰期 $T_{\text{half}}$ 计算公式为：

$$\Delta S_t = -\lambda S_{t-1} + \epsilon, \quad T_{\text{half}} = \frac{\ln 2}{\lambda}$$

import pandas as pd
import numpy as np

def calculate_spread_half_life(spread: pd.Series) -> float:
    """
    利用 AR(1) 模型测算价差序列的均值回归半衰期（合成样例，历史参数不代表未来）
    """
    spread_lag = spread.shift(1).dropna()
    spread_diff = spread.diff().dropna()
    
    # 回归 ΔS_t on S_{t-1}
    valid = pd.DataFrame({'lag': spread_lag, 'diff': spread_diff}).dropna()
    gamma = np.cov(valid['lag'], valid['diff'])[0, 1] / np.var(valid['lag'])
    
    if gamma >= 0:
        return np.inf # 无均值回归特性
    half_life = -np.log(2) / gamma
    return float(half_life)

# 合成一个具有均值回归特性的价差序列
rng = np.random.default_rng(42)
n_samples = 100
t = np.linspace(0, 10, n_samples)
# 合成价差 = 均值回归项 + 噪声
spread_series = pd.Series(3.0 * np.sin(t) + rng.normal(0, 0.3, n_samples))

half_life_result = calculate_spread_half_life(spread_series)
print("===== 价差均值回归半衰期测算 =====")
print(f"合成价差序列长度: {len(spread_series)}")
print(f"测算半衰期: {half_life_result:.2f} Bars")
if np.isinf(half_life_result):
    print("结论: 序列未检测到均值回归特性")
else:
    print(f"结论: 序列表现出均值回归，半衰期约 {half_life_result:.1f} 期")

===== 价差均值回归半衰期测算 =====
合成价差序列长度: 100
测算半衰期: 61.33 Bars
结论: 序列表现出均值回归，半衰期约 61.3 期

二、套利系统完整生命周期状态机 Python 实现

包含开仓、均值回归止盈与发散止损的套利闭环控制；超时平仓（时间止损）作为课后练习扩展。

import pandas as pd
import numpy as np

class PairsArbitrageStateMachine:
    """
    配对套利完整风控执行状态机（合成教学样例，实盘需考虑滑点、手续费与强制平仓风险）
    """
    def __init__(self, entry_z: float = 2.0, exit_z: float = 0.5, stop_z: float = 3.5):
        self.entry_z = entry_z
        self.exit_z = exit_z
        self.stop_z = stop_z
        self.position = 0 # 0: 空仓, 1: 多价差, -1: 空价差
        
    def step(self, z: float) -> str:
        if self.position == 0:
            if z > self.entry_z:
                self.position = -1
                return "OPEN_SHORT_SPREAD"
            elif z < -self.entry_z:
                self.position = 1
                return "OPEN_LONG_SPREAD"
        elif self.position == 1:
            if z >= -self.exit_z:
                self.position = 0
                return "TAKE_PROFIT_CLOSE"
            elif z <= -self.stop_z:
                self.position = 0
                return "STOP_LOSS_CLOSE"
        elif self.position == -1:
            if z <= self.exit_z:
                self.position = 0
                return "TAKE_PROFIT_CLOSE"
            elif z >= self.stop_z:
                self.position = 0
                return "STOP_LOSS_CLOSE"
        return "HOLD"

# 合成一个Z值序列，模拟策略执行过程
rng = np.random.default_rng(123)
n_steps = 20
# 生成一个包含趋势和震荡的序列，以触发不同状态
z_values = np.cumsum(rng.normal(0, 0.5, n_steps))
z_values = pd.Series(z_values)

sm = PairsArbitrageStateMachine(entry_z=1.0, exit_z=0.3, stop_z=2.0)
actions = []
positions = []

for z in z_values:
    action = sm.step(z)
    actions.append(action)
    positions.append(sm.position)

print("===== 配对套利状态机模拟执行 =====")
print(f"参数: entry_z={sm.entry_z}, exit_z={sm.exit_z}, stop_z={sm.stop_z}")
print("步骤 |    Z值   | 仓位 | 动作")
print("-" * 40)
for i in range(min(10, len(z_values))): # 打印前10步
    print(f"{i+1:4d} | {z_values.iloc[i]:7.3f} | {positions[i]:3d} | {actions[i]}")
print("...")
print(f"总步数: {n_steps}")
print(f"开仓次数: {actions.count('OPEN_LONG_SPREAD') + actions.count('OPEN_SHORT_SPREAD')}")
print(f"止盈次数: {actions.count('TAKE_PROFIT_CLOSE')}")
print(f"止损次数: {actions.count('STOP_LOSS_CLOSE')}")

===== 配对套利状态机模拟执行 =====
参数: entry_z=1.0, exit_z=0.3, stop_z=2.0
步骤 |    Z值   | 仓位 | 动作
----------------------------------------
   1 |  -0.495 |   0 | HOLD
   2 |  -0.678 |   0 | HOLD
   3 |  -0.034 |   0 | HOLD
   4 |   0.062 |   0 | HOLD
   5 |   0.523 |   0 | HOLD
   6 |   0.811 |   0 | HOLD
   7 |   0.493 |   0 | HOLD
   8 |   0.764 |   0 | HOLD
   9 |   0.606 |   0 | HOLD
  10 |   0.444 |   0 | HOLD
...
总步数: 20
开仓次数: 1
止盈次数: 0
止损次数: 0

📝 动手练一练

编写 Python 仿真测试，在模拟发散与回归的价差走势中运行套利状态机：

import pandas as pd
import numpy as np

def calculate_spread_half_life(spread: pd.Series) -> float:
    """
    利用 AR(1) 模型测算价差序列的均值回归半衰期
    """
    spread_lag = spread.shift(1).dropna()
    spread_diff = spread.diff().dropna()
    
    valid = pd.DataFrame({'lag': spread_lag, 'diff': spread_diff}).dropna()
    gamma = np.cov(valid['lag'], valid['diff'])[0, 1] / np.var(valid['lag'])
    
    if gamma >= 0:
        return np.inf
    half_life = -np.log(2) / gamma
    return float(half_life)

class PairsArbitrageStateMachine:
    """
    配对套利完整风控执行状态机
    """
    def __init__(self, entry_z: float = 2.0, exit_z: float = 0.5, stop_z: float = 3.5):
        self.entry_z = entry_z
        self.exit_z = exit_z
        self.stop_z = stop_z
        self.position = 0
        
    def step(self, z: float) -> str:
        if self.position == 0:
            if z > self.entry_z:
                self.position = -1
                return "OPEN_SHORT_SPREAD"
            elif z < -self.entry_z:
                self.position = 1
                return "OPEN_LONG_SPREAD"
        elif self.position == 1:
            if z >= -self.exit_z:
                self.position = 0
                return "TAKE_PROFIT_CLOSE"
            elif z <= -self.stop_z:
                self.position = 0
                return "STOP_LOSS_CLOSE"
        elif self.position == -1:
            if z <= self.exit_z:
                self.position = 0
                return "TAKE_PROFIT_CLOSE"
            elif z >= self.stop_z:
                self.position = 0
                return "STOP_LOSS_CLOSE"
        return "HOLD"

rng = np.random.default_rng(42)
t = np.linspace(0, 12, 120)
spread = pd.Series(5.0 * np.sin(t) + rng.normal(0, 0.5, 120))

half_life = calculate_spread_half_life(spread)
z_series = (spread - spread.mean()) / spread.std()

sm = PairsArbitrageStateMachine(entry_z=1.5, exit_z=0.3, stop_z=3.0)
actions = [sm.step(z) for z in z_series]

tp_count = actions.count("TAKE_PROFIT_CLOSE")
sl_count = actions.count("STOP_LOSS_CLOSE")
open_count = actions.count("OPEN_LONG_SPREAD") + actions.count("OPEN_SHORT_SPREAD")

print("================ MC 套利策略 Python 等价实现（下）================")
print(f"价差均值回归半衰期: {half_life:.2f} Bars")
print(f"策略动作统计:")
print(f"  开仓信号次数 : {open_count}")
print(f"  止盈离场次数 : {tp_count}")
print(f"  止损离场次数 : {sl_count}")
print(f"  持仓不动次数 : {actions.count('HOLD')}")
if half_life > 0 and not np.isinf(half_life):
    print("✅ 半衰期测算与状态机基础逻辑验证 PASS")
else:
    print("⚠️  半衰期测算异常，请检查价差序列")
print("================================================================")

运行结果输出：

================ MC 套利策略 Python 等价实现（下）================
价差均值回归半衰期: 50.85 Bars
策略动作统计:
  开仓信号次数 : 1
  止盈离场次数 : 0
  止损离场次数 : 0
  持仓不动次数 : 119
✅ 半衰期测算与状态机基础逻辑验证 PASS
================================================================

本章小结

1. 科学测算均值回归半衰期：通过 OU 过程预先评估套利标的的回归速度，合理规划资金占用周期。注意本例实测半衰期高达 50.85 Bars（已占 120 Bars 样本窗口的 42%），说明该配对在该窗口内回归缓慢——这正是反面教材：半衰期相对交易窗口过长时应放弃该配对，而非重仓硬扛等待回归。
2. 坚决执行极端发散止损：警惕基本面脱钩引发的价差无限扩大，当 $Z$-Score 突破极限时果断止损。
3. 重视单腿成交滑点与单边裸奔风险：在实盘下单中采用挂单等候与超限市价追单组合策略，确保双腿完全对冲。

📋 行动清单 (Action Checklist)

• 运行套利风控状态机代码，核验在极端突发价差下的止损触发行为；
• 编写价差回测收益率与单腿滑点敏感性分析模块；
• 预习下一节，掌握套利策略的优化体系与多品种组合绩效评估。

—— 小象教研组