2. 均值回归策略

任务目标

本案例将基于两只高度相关资产的价差序列，构建布林带与Z-Score均值回归策略。核心任务是：计算价差的Z-Score，设定入场与出场阈值，生成交易信号，并在历史数据上完成回测，输出策略的收益与风险指标。

实现思路

采用统计套利中经典的配对交易框架。首先计算两只资产价格的对数价差，然后通过滚动窗口计算价差的均值与标准差，进而得到标准化后的Z-Score序列。交易规则为：当Z-Score上穿入场阈值（如+1.5）时做空价差（卖高买低），下穿出场阈值（如+0.5）时平仓；对称地，当Z-Score下穿入场阈值（如-1.5）时做多价差，上穿出场阈值（如-0.5）时平仓。回测过程将逐日模拟持仓状态，并计算策略净值曲线及年化收益、夏普比率、最大回撤等核心风险指标。

完整代码

import numpy as np
import pandas as pd
from typing import Tuple

# ---------- 1. 生成模拟价差数据（用于演示，实际可替换为真实行情） ----------
def generate_spread_data(n: int = 500, seed: int = 42) -> pd.Series:
    """生成具有均值回归特性的价差序列（模拟数据）"""
    np.random.seed(seed)
    # 使用AR(1)过程模拟均值回归价差
    phi = 0.85  # 回归系数，越小回归越快
    noise_std = 0.5
    spread = np.zeros(n)
    for t in range(1, n):
        spread[t] = phi * spread[t-1] + np.random.normal(0, noise_std)
    return pd.Series(spread, index=pd.date_range('2022-01-01', periods=n, freq='D'))

# ---------- 2. 计算Z-Score ----------
def calculate_zscore(spread: pd.Series, window: int = 20) -> pd.Series:
    """计算滚动Z-Score： (当前值 - 滚动均值) / 滚动标准差 """
    rolling_mean = spread.rolling(window=window).mean()
    rolling_std = spread.rolling(window=window).std()
    zscore = (spread - rolling_mean) / rolling_std
    return zscore

# ---------- 3. 生成交易信号 ----------
def generate_signals(zscore: pd.Series, entry: float = 1.5, exit: float = 0.5) -> pd.Series:
    """
    生成持仓信号（1=做多价差，-1=做空价差，0=空仓）
    做多条件：Z-Score 下穿 -entry（价差过低，预期回升）
    做空条件：Z-Score 上穿 +entry（价差过高，预期回落）
    平仓条件：Z-Score 回归至 ±exit 以内
    """
    signals = pd.Series(0, index=zscore.index, dtype=int)
    position = 0  # 当前持仓状态

    for i in range(1, len(zscore)):
        z = zscore.iloc[i]
        z_prev = zscore.iloc[i-1]

        if position == 0:  # 空仓，寻找入场机会
            if z_prev < entry <= z:      # 上穿 +entry → 做空
                position = -1
            elif z_prev > -entry >= z:   # 下穿 -entry → 做多
                position = 1
        elif position == -1:  # 持空仓，寻找平仓或反转机会
            if z <= exit:                 # 回归至 +exit 以下 → 平空仓
                position = 0
            elif z <= -entry:             # 直接下穿 -entry → 反手做多
                position = 1
        elif position == 1:   # 持多仓
            if z >= -exit:                # 回归至 -exit 以上 → 平多仓
                position = 0
            elif z >= entry:              # 直接上穿 +entry → 反手做空
                position = -1

        signals.iloc[i] = position

    return signals

# ---------- 4. 回测引擎 ----------
def backtest(spread: pd.Series, signals: pd.Series) -> Tuple[pd.Series, dict]:
    """
    回测策略。假设每次交易1单位价差，不考虑交易成本。
    返回：策略净值序列 + 绩效指标字典
    """
    # 价差日收益率（价差本身可交易，收益率为差分）
    spread_ret = spread.diff().fillna(0)

    # 策略日收益 = 持仓信号 * 价差日变动（做多赚价差上涨，做空赚价差下跌）
    strategy_ret = signals.shift(1).fillna(0) * spread_ret

    # 净值曲线
    equity_curve = (1 + strategy_ret).cumprod()

    # ---------- 绩效指标 ----------
    total_return = equity_curve.iloc[-1] - 1
    years = len(spread) / 252
    annual_return = (1 + total_return) ** (1 / years) - 1 if years > 0 else 0

    daily_std = strategy_ret.std()
    sharpe = (strategy_ret.mean() / daily_std * np.sqrt(252)) if daily_std > 0 else 0

    # 最大回撤
    rolling_max = equity_curve.cummax()
    drawdown = equity_curve / rolling_max - 1
    max_drawdown = drawdown.min()

    metrics = {
        '总收益率': f'{total_return:.2%}',
        '年化收益率': f'{annual_return:.2%}',
        '夏普比率': f'{sharpe:.2f}',
        '最大回撤': f'{max_drawdown:.2%}',
        '交易次数': int((signals.diff() != 0).sum())
    }
    return equity_curve, metrics

# ---------- 5. 主程序：执行回测 ----------
if __name__ == '__main__':
    # 参数设置
    WINDOW = 20       # Z-Score 计算窗口
    ENTRY = 1.5       # 入场阈值
    EXIT = 0.5        # 出场阈值

    # 生成数据并计算
    spread = generate_spread_data()
    zscore = calculate_zscore(spread, WINDOW)
    signals = generate_signals(zscore, ENTRY, EXIT)

    # 执行回测
    equity_curve, metrics = backtest(spread, signals)

    # 输出结果
    print('=== 均值回归策略回测结果 ===')
    for key, value in metrics.items():
        print(f'{key}: {value}')

    # 展示最近10日信号与净值
    result_df = pd.DataFrame({
        '价差': spread,
        'Z-Score': zscore,
        '持仓信号': signals,
        '净值': equity_curve
    })
    print('\n=== 最近10日明细 ===')
    print(result_df.tail(10).round(4))

运行方式

1. 环境依赖：需要 numpy 和 pandas 库，可通过 pip install numpy pandas 安装。
2. 执行：将代码保存为 mean_reversion.py，在终端运行 python mean_reversion.py。

验证方法与通过标准

程序运行后应输出回测绩效指标及最近10日的交易明细。通过标准如下：

• 信号逻辑正确：检查明细表中持仓信号列，确认当Z-Score上穿+1.5时信号变为-1，下穿-1.5时变为+1，回归至±0.5以内时归零。
• 净值曲线合理：由于模拟数据具有均值回归特性，策略净值应呈现平稳上升趋势，总收益率为正。
• 指标完整性：输出应包含总收益率、年化收益率、夏普比率、最大回撤和交易次数五项指标，且数值格式正确。