【量化投资从入门到精通 #02】你的第一个量化策略双均线金叉历史上一共赚了多少

import os, datetime, statistics
from mairui import Client

CODE = "000001"
FAST, SLOW = 5, 20

mc = Client("<你的证书>")   # 证书从环境变量读取，不写死在代码里


def load_closes(code, n_days=250):
    """取最近 n_days 根日 K 线（前复权），按日期升序返回收盘价列表。"""
    end = datetime.date.today().strftime("%Y%m%d")
    st = (datetime.date.today() - datetime.timedelta(days=int(n_days * 1.6) + 30)).strftime("%Y%m%d")
    rows = mc.stock_history(code, "d", "n", st=st, et=end)      # "n" = 前复权
    rows = [r for r in rows if r.get("c") is not None and r.get("t")]
    rows.sort(key=lambda r: r["t"])
    return [float(r["c"]) for r in rows[-n_days:]]


def sma(vals, n):
    """简单移动平均，前 n-1 位为 None。"""
    out = [None] * len(vals)
    for i in range(n - 1, len(vals)):
        out[i] = sum(vals[i - n + 1:i + 1]) / n
    return out


def backtest_ma_cross(closes, fast, slow):
    """金叉持有 / 死叉空仓。信号只用当日及之前的均线——没有未来函数。"""
    ma_f, ma_s = sma(closes, fast), sma(closes, slow)
    pos = [0] * len(closes)
    for i in range(slow, len(closes)):
        prev, cur = ma_f[i - 1] - ma_s[i - 1], ma_f[i] - ma_s[i]
        if prev <= 0 < cur:
            pos[i] = 1                      # 金叉
        elif prev >= 0 > cur:
            pos[i] = 0                      # 死叉
        else:
            pos[i] = pos[i - 1]             # 维持原仓位
    equity, trades = [1.0], 0
    for i in range(1, len(closes)):
        day_ret = closes[i] / closes[i - 1] - 1
        equity.append(equity[-1] * (1 + day_ret * pos[i]))
        if pos[i] != pos[i - 1]:
            trades += 1
    return equity, trades


def calc_metrics(eq, periods=252):
    rets = [eq[i] / eq[i - 1] - 1 for i in range(1, len(eq))]
    peak, max_dd = eq[0], 0.0
    for v in eq:
        peak = max(peak, v)
        max_dd = min(max_dd, v / peak - 1)
    ann_vol = statistics.pstdev(rets) * (periods ** 0.5)
    ann_ret = eq[-1] ** (periods / (len(eq) - 1)) - 1
    return {"cum_ret": eq[-1] / eq[0] - 1, "max_dd": max_dd,
            "ann_vol": ann_vol, "sharpe": (ann_ret / ann_vol) if ann_vol else 0.0}


closes = load_closes(CODE, 250)
equity, n_trades = backtest_ma_cross(closes, FAST, SLOW)
m = calc_metrics(equity)

print(f"策略累计收益：{m['cum_ret'] * 100:.2f}%")
print(f"最大回撤：{m['max_dd'] * 100:.2f}%")
print(f"年化波动：{m['ann_vol'] * 100:.2f}%")
print(f"夏普比率：{m['sharpe']:.2f}")
print(f"买卖次数：{n_trades}")