均值回归 量化经典策略
假设资产价格会围绕其均值波动，当价格偏离均值超过一定阈值时，会向均值回归。
本案例以 A 股股票（沪深 300 成分股）为标的，
完整覆盖「数据采集→数据清洗→特征工程→策略生成→策略评估」全流程，
并补充模拟交易 / 实盘验证的关键要点。


一、环境准备
先安装所需依赖库：
pip install tushare pandas numpy talib backtrader pyfolio matplotlib
tushare：免费获取 A 股行情数据；
talib：计算技术指标（如均线、标准差）；
backtrader：策略回测框架；
pyfolio：策略绩效分析。


二、Step1：数据采集
目标
获取沪深 300 成分股的日度行情数据（OHLCV、成交量），
时间范围 2020-01-01 至 2024-12-31。


import tushare as ts
import pandas as pd
import numpy as np

# 设置Tushare token（需注册Tushare获取：https://tushare.pro/）
ts.set_token("你的Tushare Token")
pro = ts.pro_api()

# 1. 获取沪深300成分股列表
hs300 = pro.index_weight(index_code='000300.SH', start_date='20200101', end_date='20241231')
stock_codes = hs300['con_code'].unique()[:10]  # 取前10只股票（减少计算量）

# 2. 批量获取股票日度行情数据
def get_stock_data(code, start, end):
    df = pro.daily(ts_code=code, start_date=start, end_date=end)
    # 整理字段：时间升序、复权处理（前复权）
    df['trade_date'] = pd.to_datetime(df['trade_date'])
    df = df.sort_values('trade_date').reset_index(drop=True)
    # 补充前复权因子（Tushare需单独获取）
    adj = pro.adj_factor(ts_code=code, start_date=start, end_date=end)
    adj['trade_date'] = pd.to_datetime(adj['trade_date'])
    df = pd.merge(df, adj[['trade_date', 'adj_factor']], on='trade_date', how='left')
    # 前复权计算
    for col in ['open', 'high', 'low', 'close']:
        df[col] = df[col] * df['adj_factor']
    return df

# 采集数据并合并
data_list = []
for code in stock_codes:
    try:
        df = get_stock_data(code, '20200101', '20241231')
        df['stock_code'] = code
        data_list.append(df)
    except Exception as e:
        print(f"获取{code}数据失败：{e}")
raw_data = pd.concat(data_list, ignore_index=True)

# 保存原始数据（可选）
raw_data.to_csv('hs300_stock_data.csv', index=False)
print("数据采集完成，数据量：", len(raw_data))



三、Step2：数据清洗
目标
消除缺失值、异常值，统一数据格式，保证时序一致性。


# 加载数据（若已保存）
# raw_data = pd.read_csv('hs300_stock_data.csv', parse_dates=['trade_date'])

# 1. 缺失值处理
clean_data = raw_data.copy()
# 填充缺失的复权因子（线性插值）
clean_data['adj_factor'] = clean_data.groupby('stock_code')['adj_factor'].interpolate()
# 删除关键字段（收盘价、成交量）缺失的行
clean_data = clean_data.dropna(subset=['close', 'vol'])

# 2. 异常值处理
# 3σ原则剔除收盘价异常值
def remove_outliers(group):
    mean = group['close'].mean()
    std = group['close'].std()
    return group[(group['close'] >= mean - 3*std) & (group['close'] <=  mean + 3*std)]
clean_data = clean_data.groupby('stock_code').apply(remove_outliers).reset_index(drop=True)

# 3. 剔除停牌/涨跌停数据（业务规则）
# 涨跌停判断：A股涨跌幅限制±10%（ST股±5%，此处简化）
clean_data['pct_chg'] = clean_data.groupby('stock_code')['close'].pct_change()
clean_data = clean_data[
    (clean_data['pct_chg'] > -0.1) & (clean_data['pct_chg'] < 0.1)  # 剔除涨跌停
    & (clean_data['vol'] > 0)  # 剔除停牌（成交量为0）
]

# 4. 数据标准化：时间戳格式统一、字段筛选
clean_data = clean_data[['trade_date', 'stock_code', 'open', 'high', 'low', 'close', 'vol', 'pct_chg']]
clean_data = clean_data.sort_values(['stock_code', 'trade_date']).reset_index(drop=True)

print("数据清洗完成，清洗后数据量：", len(clean_data))
print("缺失值检查：\n", clean_data.isnull().sum())




四、Step3：特征工程
目标
提取均值回归核心特征：滚动均值（均值中枢）、滚动标准差（偏离度）、Z-score（标准化偏离程度）。

import talib

# 按股票分组计算特征
def add_features(group):
    # 1. 滚动均值（20日均线，均值中枢）
    group['ma20'] = talib.MA(group['close'], timeperiod=20)
    # 2. 滚动标准差（20日，衡量波动）
    group['std20'] = talib.STDDEV(group['close'], timeperiod=20)
    # 3. Z-score（偏离均值的标准化程度）
    group['z_score'] = (group['close'] - group['ma20']) / group['std20']
    return group

# 特征工程
feature_data = clean_data.groupby('stock_code').apply(add_features).reset_index(drop=True)
# 删除特征计算导致的缺失值（前20行无均线）
feature_data = feature_data.dropna(subset=['ma20', 'std20', 'z_score'])

print("特征工程完成，特征字段：", feature_data.columns.tolist())
print("特征数据示例：\n", feature_data[['trade_date', 'stock_code', 'close', 'ma20', 'z_score']].head())



五、Step4：策略生成
策略逻辑（均值回归核心规则）
买入条件：Z-score < -1.5（价格低于均值 1.5 个标准差，超跌）；
卖出条件：Z-score > 0（价格回归至均值附近）；
止损条件：买入后跌幅超过 5%（避免极端下跌）；
仓位管理：单只股票仓位不超过总资金的 10%，最大持仓 10 只股票。

Backtrader 策略代码

import backtrader as bt
import warnings
warnings.filterwarnings('ignore')

# 定义均值回归策略
class MeanReversionStrategy(bt.Strategy):
    # 策略参数（可优化）
    params = (
        ('z_buy', -1.5),    # 买入Z-score阈值
        ('z_sell', 0),      # 卖出Z-score阈值
        ('stop_loss', 0.05),# 止损比例
        ('max_pos', 10),    # 最大持仓数
        ('single_pos', 0.1) # 单只股票最大仓位
    )

    def __init__(self):
        # 存储每只股票的Z-score和买入价
        self.z_score = {}
        self.buy_price = {}
        for data in self.datas:
            self.z_score[data] = data.z_score
            self.buy_price[data] = 0

    def next(self):
        # 遍历所有股票
        for data in self.datas:
            if not self.getposition(data):  # 无持仓
                # 买入条件：Z-score < 买入阈值，且有足够资金
                if self.z_score[data][0] < self.p.z_buy:
                    # 计算可买入仓位
                    cash = self.broker.getcash()
                    position_size = cash * self.p.single_pos // (data.close[0] * 100) * 100  # A股100股为1手
                    if position_size > 0 and len(self.positions) < self.p.max_pos:
                        self.buy(data=data, size=position_size)
                        self.buy_price[data] = data.close[0]  # 记录买入价
            else:  # 有持仓
                # 卖出条件1：Z-score回归至0以上
                if self.z_score[data][0] > self.p.z_sell:
                    self.sell(data=data, size=self.getposition(data).size)
                # 卖出条件2：止损（跌幅超过5%）
                elif (self.buy_price[data] - data.close[0]) / self.buy_price[data] > self.p.stop_loss:
                    self.sell(data=data, size=self.getposition(data).size)
                    print(f"止损：{data._name}，买入价{self.buy_price[data]:.2f}，当前价{data.close[0]:.2f}")

# 准备Backtrader数据馈送
def prepare_bt_data(df, stock_code):
    # 筛选单只股票数据（Backtrader需按股票单独馈送）
    stock_df = df[df['stock_code'] == stock_code].copy()
    stock_df = stock_df.set_index('trade_date')
    # 转换为Backtrader数据格式
    data = bt.feeds.PandasData(
        dataname=stock_df,
        datetime=None,
        open='open',
        high='high',
        low='low',
        close='close',
        volume='vol',
        openinterest=-1,
        # 新增Z-score字段
        addplot=True,
        plotcolumns=['z_score']
    )
    data._name = stock_code  # 标记股票代码
    return data

# 初始化回测引擎
cerebro = bt.Cerebro()
# 添加多只股票数据
for code in stock_codes[:5]:  # 取前5只股票（减少计算量）
    stock_df = feature_data[feature_data['stock_code'] == code]
    if len(stock_df) > 0:
        data = prepare_bt_data(stock_df, code)
        cerebro.adddata(data)
# 添加策略
cerebro.addstrategy(MeanReversionStrategy)
# 设置初始资金
cerebro.broker.setcash(1000000.0)  # 初始100万
# 设置交易成本（A股：佣金0.03% + 印花税0.1%）
cerebro.broker.setcommission(commission=0.0003, stampduty=0.001)
# 设置滑点（0.05%，模拟真实交易）
cerebro.broker.set_slippage_perc(perc=0.0005)

print("初始资金：", cerebro.broker.getvalue())




六、Step5：策略评估
目标
计算核心绩效指标，验证策略盈利能力和稳定性。

# 运行回测
results = cerebro.run()
print("回测完成，最终资金：", cerebro.broker.getvalue())

# 1. 提取回测结果，计算核心指标
strategy = results[0]
# 计算年化收益率
total_return = (cerebro.broker.getvalue() - 1000000) / 1000000
years = (feature_data['trade_date'].max() - feature_data['trade_date'].min()).days / 365
annual_return = (1 + total_return) ** (1/years) - 1

# 计算最大回撤（Backtrader内置分析器）
cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe', riskfreerate=0.02)  # 无风险利率2%
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trade_analyzer')

# 重新运行回测（添加分析器）
cerebro = bt.Cerebro()
for code in stock_codes[:5]:
    stock_df = feature_data[feature_data['stock_code'] == code]
    if len(stock_df) > 0:
        data = prepare_bt_data(stock_df, code)
        cerebro.adddata(data)
cerebro.addstrategy(MeanReversionStrategy)
cerebro.broker.setcash(1000000.0)
cerebro.broker.setcommission(commission=0.0003, stampduty=0.001)
cerebro.broker.set_slippage_perc(perc=0.0005)
results = cerebro.run()

# 提取分析结果
drawdown = results[0].analyzers.drawdown.get_analysis()
sharpe = results[0].analyzers.sharpe.get_analysis()
trade_analyzer = results[0].analyzers.trade_analyzer.get_analysis()

# 整理评估指标
eval_metrics = {
    '累计收益率': f"{total_return*100:.2f}%",
    '年化收益率': f"{annual_return*100:.2f}%",
    '最大回撤': f"{drawdown['max']['drawdown']:.2f}%",
    '夏普比率': f"{sharpe['sharperatio']:.2f}",
    '总交易次数': trade_analyzer.total.closed if 'closed' in trade_analyzer.total else 0,
    '胜率': f"{(trade_analyzer.won.total / trade_analyzer.total.closed)*100:.2f}%" if trade_analyzer.total.closed >0 else "0%",
    '盈亏比': f"{trade_analyzer.won.pnl.total / abs(trade_analyzer.lost.pnl.total):.2f}" if 'lost' in trade_analyzer and trade_analyzer.lost.pnl.total !=0 else "0"
}

# 打印评估结果
print("\n=== 策略评估指标 ===")
for k, v in eval_metrics.items():
    print(f"{k}: {v}")

# 2. 绘制回测曲线（净值+最大回撤）
cerebro.plot(style='candlestick', iplot=False, volume=False)




七、Step6：模拟交易
关键操作（无代码，核心步骤）

对接模拟交易平台：
选择券商模拟盘（如华泰涨乐财富通、中信证券信 e 投）或量化平台模拟盘（聚宽、米筐）；
将策略代码适配平台 API（如聚宽使用set_universe设置股票池，order_target_percent下单）

模拟交易设置：
同步实盘行情（使用实时行情接口，如 Tushare pro 的实时行情）；
严格模拟交易成本：佣金 0.03%、印花税 0.1%、滑点 0.05%；
监控订单成交率（避免模拟盘 “秒成交”，实盘可能因流动性无法成交）

模拟交易监控：
每日记录策略信号与成交结果，对比模拟收益与回测收益的差异；
重点关注：Z-score 阈值是否合理、止损是否触发过多、仓位是否超限制


八、Step7：实盘验证
核心步骤（风险控制优先）

小资金试盘：
初始资金：总资金的 10%-20%（如 100 万资金仅投入 10-20 万）；
执行逻辑：严格按策略信号下单，禁止人工干预；
监控维度：订单成交时间、滑点实际值、冲击成本（大额订单对价格的影响）

实盘优化：
若实盘收益显著低于回测 / 模拟盘：
检查滑点设置是否过低（实盘滑点可能更高）；
优化 Z-score 阈值（如将买入阈值从 - 1.5 调整为 - 1.8，减少频繁交易）；
增加行业分散度（避免持仓集中在单一行业）

逐步加仓：
试盘周期：至少 1-3 个月，确保策略在不同市场环境（震荡 / 单边）下稳定；
加仓规则：每月收益率稳定在 1%-2%，最大回撤 < 5%，可加仓 10%-20%


九、策略优化与风险提示

1. 策略优化方向
参数优化：
用网格搜索优化 Z-score 买入 / 卖出阈值（如 - 1.2~-2.0）、均线周期（15/20/30 日）；
因子增强：
加入成交量因子（超跌 + 缩量更可靠）、行业因子（剔除强趋势行业）；
多周期验证：
分样本内（2020-2022）、样本外（2023-2024）验证，避免过拟合


2. 核心风险提示
均值回归失效：
若市场出现单边趋势（如 2021 年新能源单边上涨），策略会持续亏损；
流动性风险：
沪深 300 成分股流动性较好，但小市值股票可能出现订单无法成交

参数过拟合：
若过度优化 Z-score 阈值，策略在样本外会失效；

极端行情：
黑天鹅事件（如 2020 年疫情暴跌）会导致 Z-score 偏离极值，止损触发过多



总结
本案例完整实现了均值回归策略的研发流程，核心逻辑简单且可复现，但实际应用中需重点关注：
避免过拟合（样本外验证是关键）；
严格控制交易成本（滑点和手续费会显著影响收益）；
策略迭代（市场风格切换时，需调整 Z-score 阈值或加入新因子）

若需适配高频均值回归策略（如分钟级数据），
可将代码中的日度数据改为分钟级，
并用 C++/Cython 优化计算速度，减少回测耗时