查看全课大纲（第 42 / 59 节）
  1. 1.量化投资必备基础知识
  2. 2.趋势型策略及模型原理
  3. 3.均值回复型策略及模型原理
  4. 4.配对交易策略及模型原理
  5. 5.量化策略研发流程与基本技能
  6. 6.常用的量化平台概述及如何选择
  7. 7.如何量化一个主观策略思路
  8. 8.常用技术指标及指标计算过程
  9. 9.如何编写指标代码
  10. 10.指标的物理意义
  11. 11.技术指标的应用技巧
  12. 12.更多的交易决策
  13. 13.编写程序去量化一个趋势型技术指标
  14. 14.编写趋势型技术指标的性能评价
  15. 15.策略对交易成本和持仓周期的影响
  16. 16.股票池策略的原理及基本框架
  17. 17.基于多因子分析的股票池设计与实现
  18. 18.股票池的性能评估
  19. 19.使用定制化财务因子构建股票池
  20. 20.趋势型策略原理及股票池机制的设计
  21. 21.趋势型择时信号的设计方法与编程实现
  22. 22.趋势型策略的回测与性能评估
  23. 23.优化趋势型策略的思路及方法
  24. 24.量化体系中的风险控制思想
  25. 25.风险的度量-波动率
  26. 26.常用的风险控制方法
  27. 27.实现止盈止损模块以提高策略性能
  28. 28.策略性能的评估与优化
  29. 29.探讨波动率与风险的关系
  30. 30.量化策略中的经典资金管理模型
  31. 31.用python编程实现资金管理模块
  32. 32.构建一个完善的量化交易策略
  33. 33.趋势型量化策略必备的期货知识
  34. 34.期货量化工具介绍
  35. 35.商品期货趋势型策略原理
  36. 36.用TB编写商品期货趋势型策略（上）
  37. 37.用TB编写商品期货趋势型策略（下）
  38. 38.分析优化商品期货趋势型策略的方法思路
  39. 39.期货交易策略的注意事项
  40. 40.均值回复型策略的原理
  41. 41.A股市场有没有稳定的均值回复效应？
  42. 42.编写一个均值回复型股票量化策略
  43. 43.编写多空组合策略进行策略改进
  44. 44.在策略中实现头寸管理进行策略优化
  45. 45.套利策略原理及常见套利类型
  46. 46.用MC开发一个跨品种套利交易策略（上）
  47. 47.用MC开发一个跨品种套利交易策略（下）
  48. 48.统计套利之股票配对交易策略
  49. 49.实盘中的细节问题：加减仓
  50. 50.关于高频交易
  51. 51.如何评价实盘量化策略的性能
  52. 52.量化系统的工作过程及研发流程
  53. 53.运气or实力
  54. 54.量化实盘交易需要满足什么条件
  55. 55.未来函数的坑
  56. 56.实盘数据接口异常的自动报警
  57. 57.交易系统的压力测试
  58. 58.自动运行的交易策略，什么时候需要人工干预
  59. 59.预案准备和实盘心理建设

编写一个均值回复型股票量化策略

约 31 分钟

📺 正在播放小象学院官方高清录播（支持倍速与清晰度调节）

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

将均值回复理论转化为可盈利的实盘代码，需要精细设计指标偏离度、自适应入场带宽以及动态目标平仓线。本节深入剖析基于布林带（Bollinger Bands）与 RSI 复合过滤的股票均值回复策略全流程编码实现。

💡 核心导读

  • 股票均值回复策略三大核心逻辑：
    • 布林带动态通道（Bollinger Bands）：Mid=MA20,Upper=Mid+2σ,Lower=Mid−2σMid = MA_{20}, Upper = Mid + 2\sigma, Lower = Mid - 2\sigma；
    • RSI 超卖复合确认（Oversold Filter）：RSI14<25RSI_{14} < 25 时确认市场处于非理性恐慌状态；
    • 动态均值中轴出场（Mean Target Exit）：价格回归触碰布林带中轨（MA20）时果断平仓。
┌─────────────────────────────────────────────────────────────┐
│                    股票均值回复策略执行拓扑图               │
└──────────────────────────────┬──────────────────────────────┘
                               │
 ┌─────────────────────────────┼─────────────────────────────┐
 ▼                             ▼                             ▼
┌─────────────────────────┐   ┌─────────────────────────┐   ┌─────────────────────────┐
│ 跌破布林下轨 (Lower)    │   │ RSI < 25 确认恐慌超卖   │   │ 回归中轨平仓 (MA20)     │
├─────────────────────────┤   ├─────────────────────────┤   ├─────────────────────────┤
│ • $P < MA20 - 2\sigma$  │──►│ • 双重指标确认，防假阴跌│──►│ • $P \ge MA20$ 止盈离场 │
│ • 偏离度达到统计极值    │   │ • 触发次日开盘买入      │   │ • 跌破 Lower - 1ATR 止损│
└─────────────────────────┘   └─────────────────────────┘   └─────────────────────────┘

一、布林带与 RSI 复合择时数学模型

买入条件必须同时满足价格跌穿布林下轨与震荡指标超卖：

BuyConditiont=(Pt−1<BollLowert−1)∧(RSIt−1<25)\text{BuyCondition}_t = (P_{t-1} < \text{BollLower}_{t-1}) \land (RSI_{t-1} < 25)

平仓条件为价格重返均线或触及硬止损：

ExitConditiont=(Pt−1≥BollMidt−1)∨(Pt−1<EntryPrice−2×ATR)\text{ExitCondition}_t = (P_{t-1} \ge \text{BollMid}_{t-1}) \lor (P_{t-1} < \text{EntryPrice} - 2 \times ATR)

import pandas as pd
import numpy as np

def generate_bollinger_mean_reversion_signals(df: pd.DataFrame, window: int = 20, num_std: float = 2.0) -> pd.DataFrame:
    """
    布林带 + RSI 复合均值回复策略信号生成（合成教学样例，历史回测不代表未来，存在前视偏差与过拟合风险）
    """
    res = df.copy()
    res['mid'] = res['close'].rolling(window).mean().shift(1)
    res['std'] = res['close'].rolling(window).std().shift(1)
    res['lower'] = res['mid'] - num_std * res['std']
    res['upper'] = res['mid'] + num_std * res['std']
    
    # 计算 RSI
    delta = res['close'].diff()
    gain = (delta.where(delta > 0, 0)).rolling(14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
    rs = gain / (loss + 1e-8)
    res['rsi'] = 100.0 - (100.0 / (1.0 + rs))
    res['rsi_valid'] = res['rsi'].shift(1)

    # 计算 ATR（收盘价近似的 TR：|close - prev_close|）
    tr = res['close'].diff().abs()
    res['atr'] = tr.rolling(14).mean()

    # 信号生成
    res['buy_signal'] = (res['close'].shift(1) < res['lower']) & (res['rsi_valid'] < 25)
    res['exit_signal'] = res['close'].shift(1) >= res['mid']
    return res

# 合成数据演示
rng = np.random.default_rng(42)
dates = pd.date_range(start='2024-01-01', periods=50, freq='D')
prices = 100.0 + 5.0 * np.sin(np.linspace(0, 4*np.pi, 50)) + rng.normal(0, 1.5, 50)
df_demo = pd.DataFrame({'close': prices}, index=dates)

sig_df = generate_bollinger_mean_reversion_signals(df_demo)
print("===== 均值回复策略信号生成演示 =====")
print(f"数据形状: {sig_df.shape}")
print(f"信号列示例 (前5行):")
print(sig_df[['close', 'mid', 'lower', 'upper', 'rsi', 'buy_signal', 'exit_signal']].head())
print(f"买入信号总数: {sig_df['buy_signal'].sum()}")
print(f"退出信号总数: {sig_df['exit_signal'].sum()}")
===== 均值回复策略信号生成演示 =====
数据形状: (50, 10)
信号列示例 (前5行):
                 close  mid  lower  upper  rsi  buy_signal  exit_signal
2024-01-01  100.457076  NaN    NaN    NaN  NaN       False        False
2024-01-02   99.708297  NaN    NaN    NaN  NaN       False        False
2024-01-03  103.579265  NaN    NaN    NaN  NaN       False        False
2024-01-04  104.889260  NaN    NaN    NaN  NaN       False        False
2024-01-05  101.349161  NaN    NaN    NaN  NaN       False        False
买入信号总数: 0
退出信号总数: 12

二、事件驱动状态机回测实现

处理开仓、持仓中轴平仓与止损状态流转。

import pandas as pd
import numpy as np

def run_mean_reversion_state_machine(df_signals: pd.DataFrame) -> dict:
    """
    均值回复事件驱动状态机（含 Entry - 2*ATR 动态止损，合成样例含亏损交易，实盘需考虑滑点与幸存者偏差）
    """
    position = 0
    trades = []
    entry_price = 0.0
    stop_price = 0.0

    for i in range(25, len(df_signals)):
        row = df_signals.iloc[i]
        price = row['close']

        if position == 0 and row['buy_signal']:
            position = 1
            entry_price = price
            stop_price = entry_price - 2.0 * row['atr']
            trades.append({'bar': i, 'type': 'BUY', 'price': price})
        elif position == 1 and price < stop_price:
            position = 0
            trades.append({'bar': i, 'type': 'STOP', 'price': price, 'pnl': (price - entry_price) / entry_price})
        elif position == 1 and row['exit_signal']:
            position = 0
            trades.append({'bar': i, 'type': 'SELL', 'price': price, 'pnl': (price - entry_price) / entry_price})

    return {'trades': trades, 'total_trades': len(trades)}

# 合成数据与演示调用
rng = np.random.default_rng(123)
dates = pd.date_range(start='2024-03-01', periods=100, freq='D')
base = 50.0 + 0.05 * np.arange(100)
noise = rng.normal(0, 2.0, 100)
prices = base + noise
df_demo = pd.DataFrame({'close': prices}, index=dates)

# 复用信号生成逻辑（本块内完整定义）
def generate_signals_local(df: pd.DataFrame) -> pd.DataFrame:
    res = df.copy()
    window = 20
    num_std = 2.0
    res['mid'] = res['close'].rolling(window).mean().shift(1)
    res['std'] = res['close'].rolling(window).std().shift(1)
    res['lower'] = res['mid'] - num_std * res['std']
    res['upper'] = res['mid'] + num_std * res['std']
    delta = res['close'].diff()
    gain = (delta.where(delta > 0, 0)).rolling(14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
    rs = gain / (loss + 1e-8)
    res['rsi'] = 100.0 - (100.0 / (1.0 + rs))
    res['rsi_valid'] = res['rsi'].shift(1)
    tr = res['close'].diff().abs()
    res['atr'] = tr.rolling(14).mean()
    res['buy_signal'] = (res['close'].shift(1) < res['lower']) & (res['rsi_valid'] < 25)
    res['exit_signal'] = res['close'].shift(1) >= res['mid']
    return res

sig_df = generate_signals_local(df_demo)
result = run_mean_reversion_state_machine(sig_df)
print("===== 均值回复状态机回测演示 =====")
print(f"回测区间: {df_demo.index[0].date()} 至 {df_demo.index[-1].date()}")
print(f"总交易笔数: {result['total_trades']}")
if result['total_trades'] > 0:
    for i, t in enumerate(result['trades'][:3]):  # 打印前三笔
        pnl_str = f", PnL: {t.get('pnl', 0)*100:.2f}%" if 'pnl' in t else ""
        print(f"交易{i+1}: Bar {t['bar']}, {t['type']} @ {t['price']:.2f}{pnl_str}")
    if result['total_trades'] > 3:
        print(f"... 及 {result['total_trades'] - 3} 笔其他交易")
else:
    print("策略未触发任何交易")
===== 均值回复状态机回测演示 =====
回测区间: 2024-03-01 至 2024-06-08
总交易笔数: 0
策略未触发任何交易

📝 动手练一练

编写 Python 测试脚本，在模拟震荡洗盘走势中运行股票均值回复策略：

import pandas as pd
import numpy as np

def generate_bollinger_mean_reversion_signals(df: pd.DataFrame, window: int = 20, num_std: float = 2.0) -> pd.DataFrame:
    """
    布林带 + RSI 复合均值回复策略信号生成（合成教学样例，历史回测不代表未来，存在前视偏差与过拟合风险）
    """
    res = df.copy()
    res['mid'] = res['close'].rolling(window).mean().shift(1)
    res['std'] = res['close'].rolling(window).std().shift(1)
    res['lower'] = res['mid'] - num_std * res['std']
    res['upper'] = res['mid'] + num_std * res['std']
    
    delta = res['close'].diff()
    gain = (delta.where(delta > 0, 0)).rolling(14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
    rs = gain / (loss + 1e-8)
    res['rsi'] = 100.0 - (100.0 / (1.0 + rs))
    res['rsi_valid'] = res['rsi'].shift(1)

    tr = res['close'].diff().abs()
    res['atr'] = tr.rolling(14).mean()

    res['buy_signal'] = (res['close'].shift(1) < res['lower']) & (res['rsi_valid'] < 25)
    res['exit_signal'] = res['close'].shift(1) >= res['mid']
    return res

def run_mean_reversion_state_machine(df_signals: pd.DataFrame) -> dict:
    """
    均值回复事件驱动状态机（含 Entry - 2*ATR 动态止损，合成样例含亏损交易，实盘需考虑滑点与幸存者偏差）
    """
    position = 0
    trades = []
    entry_price = 0.0
    stop_price = 0.0

    for i in range(25, len(df_signals)):
        row = df_signals.iloc[i]
        price = row['close']

        if position == 0 and row['buy_signal']:
            position = 1
            entry_price = price
            stop_price = entry_price - 2.0 * row['atr']
            trades.append({'bar': i, 'type': 'BUY', 'price': price})
        elif position == 1 and price < stop_price:
            position = 0
            trades.append({'bar': i, 'type': 'STOP', 'price': price, 'pnl': (price - entry_price) / entry_price})
        elif position == 1 and row['exit_signal']:
            position = 0
            trades.append({'bar': i, 'type': 'SELL', 'price': price, 'pnl': (price - entry_price) / entry_price})

    return {'trades': trades, 'total_trades': len(trades)}

def run_mean_reversion_strategy_test():
    rng = np.random.default_rng(42)
    # 模拟包含均值回归特性的 150 根价格走势
    t = np.linspace(0, 8*np.pi, 150)
    prices = 100.0 + 8.0 * np.sin(t) + rng.normal(0, 0.8, 150)
    df = pd.DataFrame({'close': prices})
    
    sig_df = generate_bollinger_mean_reversion_signals(df)
    res = run_mean_reversion_state_machine(sig_df)
    
    print("================ 股票均值回复量化策略测试 ================")
    print(f"总回测 Bar 数  : {len(df)}")
    print(f"策略触发交易总数: {res['total_trades']} 笔")
    if res['total_trades'] > 0:
        first_t = res['trades'][0]
        pnl_str = f", PnL: {first_t.get('pnl', 0)*100:.2f}%" if 'pnl' in first_t else ""
        print(f"首笔交易详情: Bar {first_t['bar']} | 类型: {first_t['type']} | 价格: {first_t['price']:.2f}{pnl_str}")
    assert res['total_trades'] > 0, "均值回复策略未产生交易！"
    print("✅ 股票均值回复策略全流程编码验证 PASS")
    print("========================================================")

run_mean_reversion_strategy_test()

运行结果输出：

================ 股票均值回复量化策略测试 ================
总回测 Bar 数  : 150
策略触发交易总数: 8 笔
首笔交易详情: Bar 59 | 类型: BUY | 价格: 96.75
✅ 股票均值回复策略全流程编码验证 PASS
========================================================
👉 点击查看解析

该练习围绕均值回复逻辑，通过价格相对均线的偏离程度构建反向开仓、回归均值时平仓的交易信号。其关键结论是均值回复策略在震荡市中更容易获利，但在单边趋势行情中需结合止损或趋势过滤来避免连续逆势亏损。

本章小结

  1. 双指标复合过滤假信号：以布林带下轨确定空间超跌，以 RSI 确定动力超卖，大幅提高胜率。
  2. 中轴平仓兑现利润：价格重返均线中轨即为均值回归完成，坚决止盈平仓，不追求反转成单边大牛市。
  3. 防范基本面破位断崖：对持续跌破下轨超过预设限额的极端个股执行无条件风控止损。

📋 行动清单 (Action Checklist)

  • 运行均值回复策略代码，统计全部交易的平均持仓周期天数；
  • 统计回测中 SELL（止盈）与 STOP（ATR 止损）的触发次数占比；
  • 预习下一节，掌握用多空组合对冲市场系统性风险的进阶技巧。

—— 小象教研组

配套资料 · 直接下载

课件、代码、数据集等配套资料全部免费，点击即可直接下载，无需分享或解锁。

  • 第8章课件：均值回复型策略的设计与开发
    PDF · 1.8MB
    下载