第 28 篇：均值回归策略：跌多了会涨吗

# 块 1：获取宽基指数日收盘价（含本地造数据 fallback）
# uv pip install akshare pandas numpy
import numpy as np
import pandas as pd

def load_price() -> pd.Series:
    """拉沪深300日收盘价；akshare 不可用时造一条带均值回归特征的数据。"""
    try:
        import akshare as ak
        df = ak.index_zh_a_hist(symbol="000300", period="daily",
                                start_date="20200101", end_date="20241231")
        s = pd.Series(df["收盘"].values, index=pd.to_datetime(df["日期"]))
        return s.astype(float)
    except Exception:
        # fallback：AR(1) 对数价格——围绕锚缠绕，天然震荡（教学专用，非真实行情）
        rng = np.random.default_rng(11)
        idx = pd.bdate_range("2020-01-01", "2024-12-31")
        logp = np.zeros(len(idx))
        for t in range(1, len(idx)):
            logp[t] = 0.96 * logp[t - 1] + rng.normal(0, 0.013)  # 偏离会被拉回
        return pd.Series(100 * np.exp(logp), index=idx)

price = load_price()
print(f"共 {len(price)} 个交易日，"
      f"起止：{price.index[0].date()} ~ {price.index[-1].date()}")
# 顺手预览最新一天的 z-score（块 2 会正式使用这套逻辑）
z_now = (price.iloc[-1] - price.rolling(20).mean().iloc[-1]) / price.rolling(20).std().iloc[-1]
print(f"最新 20 日窗口 z-score：{z_now:.2f}")

# 块 2：z-score 均值回归回测
import numpy as np
import pandas as pd

N = 20                        # 滚动窗口：20 个交易日的均值与标准差
Z_IN, Z_OUT = -1.5, 0.0       # 入场：z ≤ -1.5；出场：z 回到 0
STOP_Z, STOP_PCT = -3.0, 0.08 # 止损：z ≤ -3，或自入场亏损超 8%
FEE = 0.0005                  # 单边手续费 0.05%，双边合计 0.1%（万 10）

def zscore_backtest(price: pd.Series) -> tuple[pd.Series, pd.DataFrame]:
    """z-score 均值回归。约定与第 27 篇相同：信号用截至昨收的数据，
    按昨收价成交，当日损益归属昨收形成的仓位，无未来函数。"""
    ma = price.rolling(N).mean()                  # 回归锚：20 日滚动均值
    sd = price.rolling(N).std()
    z = (price - ma) / sd                         # 偏离几个标准差
    ret = price.pct_change().fillna(0)
    nav, trades = {}, []
    pos, entry, cooldown = 0, np.nan, False       # pos：0 空仓 / 1 持有
    for i in range(N, len(price)):
        date, z_prev = price.index[i], z.iloc[i - 1]   # 昨收可知的 z
        fee_today = 0.0
        if pos == 1 and (z_prev >= Z_OUT or z_prev <= STOP_Z
                         or price.iloc[i - 1] / entry - 1 <= -STOP_PCT):
            hit_stop = (z_prev <= STOP_Z
                        or price.iloc[i - 1] / entry - 1 <= -STOP_PCT)
            trades.append((price.index[i - 1],
                           "止损离场" if hit_stop else "回归离场",
                           round(price.iloc[i - 1], 2)))
            pos, fee_today, cooldown = 0, FEE, hit_stop   # 止损离场才进冷却
        elif pos == 0 and not cooldown and z_prev <= Z_IN:
            trades.append((price.index[i - 1], "入场",
                           round(price.iloc[i - 1], 2)))
            pos, entry, fee_today = 1, price.iloc[i - 1], FEE
        elif pos == 0 and cooldown and z_prev >= Z_OUT:
            cooldown = False                     # 冷却解除：z 修复回均值再观察
        nav[date] = (ret.iloc[i] * pos) - fee_today
    nav = pd.Series(nav, name="均值回归").add(1).cumprod()
    return nav, pd.DataFrame(trades, columns=["日期", "动作", "成交参考价"])

# 块 3：与买入持有基准对比，并查看交易明细
import pandas as pd

price = load_price()
nav, trades = zscore_backtest(price)
bench = price / price.iloc[0]                    # 买入持有基准

comp = pd.concat([nav, bench], axis=1).dropna()
comp = comp / comp.iloc[0]                       # 从首个共同交易日（窗口填满后）归一
comp.columns = ["均值回归", "买入持有"]
for name in comp.columns:
    s = comp[name]
    mdd = (s / s.cummax() - 1).min()
    ann = s.iloc[-1] ** (252 / len(s)) - 1
    print(f"{name}: 累计 {s.iloc[-1] - 1:.1%} | 年化 {ann:.1%} | 最大回撤 {mdd:.1%}")

print(f"共 {len(trades)} 笔交易，动作分布：")
print(trades["动作"].value_counts())
print("前 5 笔明细：")
print(trades.head().to_string(index=False))

共 1217 个交易日，起止：2020-01-02 ~ 2024-12-31
最新 20 日窗口 z-score：-0.87        # 数值仅为示意

均值回归: 累计 21.4% | 年化 4.0% | 最大回撤 -6.3%
买入持有: 累计 9.7% | 年化 1.9% | 最大回撤 -15.1%   # 数值均为示意
共 14 笔交易，动作分布：
入场      7
回归离场   4
止损离场   3
前 5 笔明细：
      日期 动作  成交参考价
2020-03-09 入场     3851.53
2020-03-23 止损离场  3518.20
2020-05-29 入场     3715.66
2020-07-06 回归离场  3892.10
2020-09-30 入场     3985.42