文章

37 · 量化策略回测：信号生成、净值计算与绩效评估

#046 · 2026-04-17 · Python

阅读数据加载中… 点赞数据加载中…

生成页数

幻灯片语言

提炼重点 / 自定义指令 (可选)

🔗 知识图谱导航：阅读本文前，建议先掌握《36 · 量化数据处理》中的技术指标计算——本文在此基础上实现完整的策略回测：信号 → 持仓 → 净值 → 绩效评估。

运行环境：pip install pandas numpy。

极客解析：量化回测的核心是向量化计算：strat_ret = signal.shift(1) * ret，一行代码计算所有交易日的策略收益率。shift(1) 是关键——用昨天的信号决定今天的持仓，避免"未来函数"（用未来数据做决策）。

回测核心公式

信号生成：signal = 1 if MA5 > MA20 else 0（金叉持仓，死叉空仓）
策略收益：strat_ret[t] = signal[t-1] * ret[t]（昨天信号，今天收益）
净值曲线：equity = cumprod(1 + strat_ret)
最大回撤：max_dd = max(1 - equity / cummax(equity))
夏普比率：sharpe = mean(strat_ret) / std(strat_ret) * sqrt(252)

步步为营：核心逻辑自适应拆解

这一篇按策略回测流水线拆成 7 个台阶：准备行情、生成信号、计算绩效、输出交易流水、展示指标面板、画资金曲线，最后用 CLI 调度。每个演示都补了 Mock 数据和 print() 反馈。

Step 1：用 generate_data 准备价格、均线和收益率

痛点与机制：

generate_data 是回测的行情工厂：先生成模拟价格，再算 MA5、MA20 和每日收益率。MA 前几天会因为窗口不够产生空值，所以最后 dropna()。可以把它理解成先准备一张干净的比赛成绩表，后面策略只负责读表做决定。

核心源码（逐字来自文末完整源码）：

def generate_data(n: int = 250) -> pd.DataFrame:
    np.random.seed(42)
    dates = pd.bdate_range("2025-01-02", periods=n)
    returns = np.random.randn(n) * 0.015 + 0.0003
    close = 100.0 * np.exp(np.cumsum(returns))
    df = pd.DataFrame({"date": dates, "close": close})
    df["ma5"]  = df["close"].rolling(5).mean()
    df["ma20"] = df["close"].rolling(20).mean()
    df["ret"]  = df["close"].pct_change()
    return df.dropna().reset_index(drop=True)

可运行演示（补齐 Mock 数据与 print 反馈）：

import numpy as np
import pandas as pd


def generate_data(n: int = 250) -> pd.DataFrame:
    np.random.seed(42)
    dates = pd.bdate_range("2025-01-02", periods=n)
    returns = np.random.randn(n) * 0.015 + 0.0003
    close = 100.0 * np.exp(np.cumsum(returns))
    df = pd.DataFrame({"date": dates, "close": close})
    df["ma5"]  = df["close"].rolling(5).mean()
    df["ma20"] = df["close"].rolling(20).mean()
    df["ret"]  = df["close"].pct_change()
    return df.dropna().reset_index(drop=True)


df = generate_data(n=40)
print("📦 回测数据已生成")
print("行列:", df.shape)
print("字段:", list(df.columns))
print(df.head(3).to_string(index=False))

Step 2：用 generate_signals 生成金叉死叉和持仓

痛点与机制：

generate_signals 是策略大脑：MA5 在 MA20 上方就认为短期更强，允许持仓；diff() 找出从空仓到持仓的金叉、从持仓到空仓的死叉。shift(1) 是风控红线，表示今天看到的信号只能影响明天，避免未来函数。

核心源码（逐字来自文末完整源码）：

def generate_signals(df: pd.DataFrame) -> pd.DataFrame:
    """金叉买入，死叉卖出"""
    df = df.copy()
    df["above"] = (df["ma5"] > df["ma20"]).astype(int)
    df["signal"] = df["above"].diff()   # +1=金叉, -1=死叉
    df["position"] = df["above"].shift(1).fillna(0)  # 持仓（滞后1日）
    df["strat_ret"] = df["position"] * df["ret"]
    df["equity"] = (1 + df["strat_ret"]).cumprod()
    df["bh_equity"] = (1 + df["ret"]).cumprod()       # 买入持有基准
    return df

可运行演示（补齐 Mock 数据与 print 反馈）：

import pandas as pd


def generate_signals(df: pd.DataFrame) -> pd.DataFrame:
    """金叉买入，死叉卖出"""
    df = df.copy()
    df["above"] = (df["ma5"] > df["ma20"]).astype(int)
    df["signal"] = df["above"].diff()   # +1=金叉, -1=死叉
    df["position"] = df["above"].shift(1).fillna(0)  # 持仓（滞后1日）
    df["strat_ret"] = df["position"] * df["ret"]
    df["equity"] = (1 + df["strat_ret"]).cumprod()
    df["bh_equity"] = (1 + df["ret"]).cumprod()       # 买入持有基准
    return df


raw = pd.DataFrame({
    "date": pd.bdate_range("2025-01-02", periods=6),
    "close": [100, 101, 102, 103, 101, 99],
    "ma5":   [99, 100, 101, 102, 101, 100],
    "ma20":  [100, 100, 100, 100, 100, 100],
    "ret":   [0.0, 0.01, 0.0099, 0.0098, -0.0194, -0.0198],
})
df = generate_signals(raw)
print("信号表:")
print(df[["date", "ma5", "ma20", "above", "signal", "position", "strat_ret", "equity"]].to_string(index=False))
print("说明: position 滞后一天，避免用今天收盘后的信号赚今天的钱。")

Step 3：用 calc_metrics 计算收益、回撤和夏普

痛点与机制：

calc_metrics 是策略体检表。它从每日策略收益和净值曲线里计算年化收益、波动、夏普、最大回撤、交易胜率。最大回撤像从山顶滑到谷底的最大落差，夏普则像“每承担一份颠簸换来多少收益”。

核心源码（逐字来自文末完整源码）：

def calc_metrics(df: pd.DataFrame, rf: float = 0.03) -> dict:
    r = df["strat_ret"].values
    eq = df["equity"].values
    n = len(r)

    ann_ret = (eq[-1] ** (252 / n)) - 1
    ann_vol = r.std() * np.sqrt(252)
    sharpe  = (ann_ret - rf) / ann_vol if ann_vol > 0 else 0.0

    # 最大回撤
    peak = np.maximum.accumulate(eq)
    drawdown = (peak - eq) / peak
    max_dd = drawdown.max()
    calmar = ann_ret / max_dd if max_dd > 0 else 0.0

    # 交易统计
    trades = df[df["signal"] == 1].index.tolist()   # 买入点
    exits  = df[df["signal"] == -1].index.tolist()  # 卖出点
    trade_rets = []
    for entry in trades:
        exit_candidates = [e for e in exits if e > entry]
        if exit_candidates:
            exit_idx = exit_candidates[0]
            trade_ret = df.loc[exit_idx, "equity"] / df.loc[entry, "equity"] - 1
            trade_rets.append(trade_ret)

    wins = [r for r in trade_rets if r > 0]
    losses = [r for r in trade_rets if r <= 0]
    win_rate = len(wins) / len(trade_rets) if trade_rets else 0.0
    avg_win  = np.mean(wins) if wins else 0.0
    avg_loss = abs(np.mean(losses)) if losses else 0.0
    profit_factor = avg_win / avg_loss if avg_loss > 0 else 0.0

    return {
        "ann_ret": ann_ret, "ann_vol": ann_vol, "sharpe": sharpe,
        "max_dd": max_dd, "calmar": calmar,
        "n_trades": len(trade_rets), "win_rate": win_rate,
        "avg_win": avg_win, "avg_loss": avg_loss,
        "profit_factor": profit_factor,
        "final_equity": eq[-1], "bh_equity": df["bh_equity"].iloc[-1],
        "drawdown": drawdown,
    }

可运行演示（补齐 Mock 数据与 print 反馈）：

import numpy as np
import pandas as pd


def calc_metrics(df: pd.DataFrame, rf: float = 0.03) -> dict:
    r = df["strat_ret"].values
    eq = df["equity"].values
    n = len(r)

    ann_ret = (eq[-1] ** (252 / n)) - 1
    ann_vol = r.std() * np.sqrt(252)
    sharpe  = (ann_ret - rf) / ann_vol if ann_vol > 0 else 0.0

    # 最大回撤
    peak = np.maximum.accumulate(eq)
    drawdown = (peak - eq) / peak
    max_dd = drawdown.max()
    calmar = ann_ret / max_dd if max_dd > 0 else 0.0

    # 交易统计
    trades = df[df["signal"] == 1].index.tolist()   # 买入点
    exits  = df[df["signal"] == -1].index.tolist()  # 卖出点
    trade_rets = []
    for entry in trades:
        exit_candidates = [e for e in exits if e > entry]
        if exit_candidates:
            exit_idx = exit_candidates[0]
            trade_ret = df.loc[exit_idx, "equity"] / df.loc[entry, "equity"] - 1
            trade_rets.append(trade_ret)

    wins = [r for r in trade_rets if r > 0]
    losses = [r for r in trade_rets if r <= 0]
    win_rate = len(wins) / len(trade_rets) if trade_rets else 0.0
    avg_win  = np.mean(wins) if wins else 0.0
    avg_loss = abs(np.mean(losses)) if losses else 0.0
    profit_factor = avg_win / avg_loss if avg_loss > 0 else 0.0

    return {
        "ann_ret": ann_ret, "ann_vol": ann_vol, "sharpe": sharpe,
        "max_dd": max_dd, "calmar": calmar,
        "n_trades": len(trade_rets), "win_rate": win_rate,
        "avg_win": avg_win, "avg_loss": avg_loss,
        "profit_factor": profit_factor,
        "final_equity": eq[-1], "bh_equity": df["bh_equity"].iloc[-1],
        "drawdown": drawdown,
    }


df = pd.DataFrame({
    "strat_ret": [0.0, 0.01, -0.02, 0.03, -0.01, 0.02],
    "ret":       [0.0, 0.02, -0.01, 0.02, -0.02, 0.01],
    "signal":    [0, 1, 0, -1, 1, -1],
})
df["equity"] = (1 + df["strat_ret"]).cumprod()
df["bh_equity"] = (1 + df["ret"]).cumprod()
m = calc_metrics(df)
print("📊 绩效指标:")
for key in ["ann_ret", "ann_vol", "sharpe", "max_dd", "n_trades", "win_rate", "final_equity"]:
    print(f"  {key}: {m[key]:.4f}" if isinstance(m[key], float) else f"  {key}: {m[key]}")

Step 4：用 mode_backtest 输出交易信号流水账

痛点与机制：

mode_backtest 把交易过程讲清楚：回测区间、持仓天数、买卖信号次数，以及前几条金叉/死叉记录。它像交易流水账，新手能看到策略到底在哪些日子做了动作。

核心源码（逐字来自文末完整源码）：

def mode_backtest(df: pd.DataFrame) -> None:
    print(f"\n[{nexdo_time()}] 🔄 回测摘要")
    buy_signals  = (df["signal"] == 1).sum()
    sell_signals = (df["signal"] == -1).sum()
    hold_days    = (df["position"] == 1).sum()
    print(f"  回测区间: {str(df['date'].iloc[0])[:10]} ~ {str(df['date'].iloc[-1])[:10]}")
    print(f"  总交易日: {len(df)}  持仓天数: {hold_days}  仓位率: {hold_days/len(df):.1%}")
    print(f"  买入信号: {buy_signals} 次  卖出信号: {sell_signals} 次")

    # 信号列表（前10条）
    signals = df[df["signal"] != 0][["date", "signal", "close", "ma5", "ma20"]].head(10)
    print(f"\n  信号记录（前10条）:")
    print(f"  {'日期':<12} {'信号':<8} {'收盘价':>8} {'MA5':>8} {'MA20':>8}")
    print("  " + "─" * 50)
    for _, row in signals.iterrows():
        sig_str = "🟢金叉" if row["signal"] == 1 else "🔴死叉"
        print(f"  {str(row['date'])[:10]:<12} {sig_str:<8} "
              f"{row['close']:>8.2f} {row['ma5']:>8.2f} {row['ma20']:>8.2f}")

可运行演示（补齐 Mock 数据与 print 反馈）：

import time
import pandas as pd


def nexdo_time() -> str:
    return time.strftime("%Y-%m-%d %H:%M:%S")


def mode_backtest(df: pd.DataFrame) -> None:
    print(f"\n[{nexdo_time()}] 🔄 回测摘要")
    buy_signals  = (df["signal"] == 1).sum()
    sell_signals = (df["signal"] == -1).sum()
    hold_days    = (df["position"] == 1).sum()
    print(f"  回测区间: {str(df['date'].iloc[0])[:10]} ~ {str(df['date'].iloc[-1])[:10]}")
    print(f"  总交易日: {len(df)}  持仓天数: {hold_days}  仓位率: {hold_days/len(df):.1%}")
    print(f"  买入信号: {buy_signals} 次  卖出信号: {sell_signals} 次")

    # 信号列表（前10条）
    signals = df[df["signal"] != 0][["date", "signal", "close", "ma5", "ma20"]].head(10)
    print(f"\n  信号记录（前10条）:")
    print(f"  {'日期':<12} {'信号':<8} {'收盘价':>8} {'MA5':>8} {'MA20':>8}")
    print("  " + "─" * 50)
    for _, row in signals.iterrows():
        sig_str = "🟢金叉" if row["signal"] == 1 else "🔴死叉"
        print(f"  {str(row['date'])[:10]:<12} {sig_str:<8} "
              f"{row['close']:>8.2f} {row['ma5']:>8.2f} {row['ma20']:>8.2f}")


df = pd.DataFrame({
    "date": pd.bdate_range("2025-01-02", periods=5),
    "signal": [0, 1, 0, -1, 1],
    "position": [0, 0, 1, 1, 0],
    "close": [100, 103, 104, 101, 105],
    "ma5": [99, 101, 102, 102, 103],
    "ma20": [100, 100, 100, 100, 100],
})
mode_backtest(df)

Step 5：用 mode_metrics 生成策略绩效面板

痛点与机制：

mode_metrics 把一堆浮点数排成绩效面板。策略评估不能只看最后赚没赚，还要看波动、回撤、胜率和盈亏比；这就像看车，不只看最高速度，还要看刹车距离和稳定性。

核心源码（逐字来自文末完整源码）：

def mode_metrics(metrics: dict) -> None:
    print(f"\n[{nexdo_time()}] 📊 策略绩效指标")
    rows = [
        ("年化收益率",   f"{metrics['ann_ret']:>+.2%}"),
        ("年化波动率",   f"{metrics['ann_vol']:>.2%}"),
        ("夏普比率",     f"{metrics['sharpe']:>.4f}"),
        ("最大回撤",     f"{metrics['max_dd']:>.2%}"),
        ("卡玛比率",     f"{metrics['calmar']:>.4f}"),
        ("交易次数",     f"{metrics['n_trades']}"),
        ("胜率",         f"{metrics['win_rate']:>.1%}"),
        ("平均盈利",     f"{metrics['avg_win']:>+.2%}"),
        ("平均亏损",     f"{metrics['avg_loss']:>.2%}"),
        ("盈亏比",       f"{metrics['profit_factor']:>.2f}"),
        ("策略终值",     f"{metrics['final_equity']:.4f}"),
        ("买入持有终值", f"{metrics['bh_equity']:.4f}"),
    ]
    print(f"  {'指标':<14} {'数值':>12}")
    print("  " + "─" * 30)
    for name, val in rows:
        print(f"  {name:<14} {val:>12}")

    excess = metrics["final_equity"] - metrics["bh_equity"]
    print(f"\n  策略 vs 买入持有: {'超额' if excess > 0 else '落后'} {abs(excess):.4f}")

可运行演示（补齐 Mock 数据与 print 反馈）：

import time


def nexdo_time() -> str:
    return time.strftime("%Y-%m-%d %H:%M:%S")


def mode_metrics(metrics: dict) -> None:
    print(f"\n[{nexdo_time()}] 📊 策略绩效指标")
    rows = [
        ("年化收益率",   f"{metrics['ann_ret']:>+.2%}"),
        ("年化波动率",   f"{metrics['ann_vol']:>.2%}"),
        ("夏普比率",     f"{metrics['sharpe']:>.4f}"),
        ("最大回撤",     f"{metrics['max_dd']:>.2%}"),
        ("卡玛比率",     f"{metrics['calmar']:>.4f}"),
        ("交易次数",     f"{metrics['n_trades']}"),
        ("胜率",         f"{metrics['win_rate']:>.1%}"),
        ("平均盈利",     f"{metrics['avg_win']:>+.2%}"),
        ("平均亏损",     f"{metrics['avg_loss']:>.2%}"),
        ("盈亏比",       f"{metrics['profit_factor']:>.2f}"),
        ("策略终值",     f"{metrics['final_equity']:.4f}"),
        ("买入持有终值", f"{metrics['bh_equity']:.4f}"),
    ]
    print(f"  {'指标':<14} {'数值':>12}")
    print("  " + "─" * 30)
    for name, val in rows:
        print(f"  {name:<14} {val:>12}")

    excess = metrics["final_equity"] - metrics["bh_equity"]
    print(f"\n  策略 vs 买入持有: {'超额' if excess > 0 else '落后'} {abs(excess):.4f}")


metrics = {"ann_ret": 0.12, "ann_vol": 0.18, "sharpe": 0.5, "max_dd": 0.08, "calmar": 1.5,
           "n_trades": 4, "win_rate": 0.5, "avg_win": 0.04, "avg_loss": 0.02,
           "profit_factor": 2.0, "final_equity": 1.18, "bh_equity": 1.10}
mode_metrics(metrics)

Step 6：用 mode_chart 画资金曲线和回撤图

痛点与机制：

mode_chart 用 ASCII 画资金曲线和回撤图。资金曲线告诉你净值怎么走，回撤图告诉你痛苦有多深。B/S 标记像地图上的买卖路标，方便对照信号和净值变化。

核心源码（逐字来自文末完整源码）：

def mode_chart(df: pd.DataFrame, metrics: dict) -> None:
    print(f"\n[{nexdo_time()}] 📈 资金曲线 + 回撤图（ASCII）")

    eq    = df["equity"].values
    bh    = df["bh_equity"].values
    dd    = metrics["drawdown"]
    n     = len(eq)
    rows  = 12

    # 资金曲线
    lo = min(eq.min(), bh.min()) * 0.99
    hi = max(eq.max(), bh.max()) * 1.01
    print(f"\n  资金曲线  ● 策略  · 买入持有")
    print("  " + "─" * (n + 14))
    for r in range(rows, 0, -1):
        thr = lo + (hi - lo) * r / rows
        row = ""
        for i in range(n):
            s_hit = eq[i] >= thr
            b_hit = bh[i] >= thr
            if s_hit and b_hit:
                row += "◆"
            elif s_hit:
                row += "●"
            elif b_hit:
                row += "·"
            else:
                row += " "
        print(f"  {thr:>8.4f} │{row}│")
    print("  " + "─" * (n + 14))

    # 买卖信号标记
    sig_row = [" "] * n
    for i in df[df["signal"] == 1].index:
        if i < n:
            sig_row[i] = "B"
    for i in df[df["signal"] == -1].index:
        if i < n:
            sig_row[i] = "S"
    print("  " + " " * 10 + "".join(sig_row) + "  B=买入 S=卖出")

    # 回撤图
    print(f"\n  回撤图（最大回撤: {metrics['max_dd']:.2%}）")
    dd_rows = 6
    print("  " + "─" * (n + 14))
    for r in range(dd_rows, 0, -1):
        thr = dd.max() * r / dd_rows
        row = "".join("▼" if v >= thr else " " for v in dd)
        print(f"  {thr:>7.2%}  │{row}│")
    print("  " + "─" * (n + 14))
    print(f"  {'0.00%':>9}  └{'─' * n}┘")

可运行演示（补齐 Mock 数据与 print 反馈）：

import time
import numpy as np
import pandas as pd


def nexdo_time() -> str:
    return time.strftime("%Y-%m-%d %H:%M:%S")


def mode_chart(df: pd.DataFrame, metrics: dict) -> None:
    print(f"\n[{nexdo_time()}] 📈 资金曲线 + 回撤图（ASCII）")

    eq    = df["equity"].values
    bh    = df["bh_equity"].values
    dd    = metrics["drawdown"]
    n     = len(eq)
    rows  = 6
    lo = min(eq.min(), bh.min()) * 0.99
    hi = max(eq.max(), bh.max()) * 1.01
    print(f"\n  资金曲线  ● 策略  · 买入持有")
    print("  " + "─" * (n + 14))
    for r in range(rows, 0, -1):
        thr = lo + (hi - lo) * r / rows
        row = ""
        for i in range(n):
            s_hit = eq[i] >= thr
            b_hit = bh[i] >= thr
            row += "◆" if s_hit and b_hit else "●" if s_hit else "·" if b_hit else " "
        print(f"  {thr:>8.4f} │{row}│")
    sig_row = [" "] * n
    for i in df[df["signal"] == 1].index:
        sig_row[i] = "B"
    for i in df[df["signal"] == -1].index:
        sig_row[i] = "S"
    print("  " + " " * 10 + "".join(sig_row) + "  B=买入 S=卖出")
    print(f"\n  回撤图（最大回撤: {metrics['max_dd']:.2%}）")
    for level in [dd.max(), dd.max() / 2, 0]:
        row = "".join("▼" if v >= level and level > 0 else " " for v in dd)
        print(f"  {level:>7.2%} │{row}│")


df = pd.DataFrame({
    "equity": [1.00, 1.02, 0.99, 1.05, 1.04, 1.08],
    "bh_equity": [1.00, 1.01, 1.00, 1.03, 1.02, 1.06],
    "signal": [0, 1, 0, -1, 1, 0],
})
peak = np.maximum.accumulate(df["equity"].values)
dd = (peak - df["equity"].values) / peak
mode_chart(df, {"drawdown": dd, "max_dd": dd.max()})

Step 7：用 main 做 backtest/metrics/chart/all 脚本遥控器

痛点与机制：

main 是回测脚本的遥控器：--mode backtest/metrics/chart/all 决定输出哪部分，但数据准备、信号生成和指标计算始终统一执行，保证所有报告来自同一轮回测。

核心源码（逐字来自文末完整源码）：

def main() -> None:
    parser = argparse.ArgumentParser(description="双均线策略回测")
    parser.add_argument("--mode", choices=["backtest", "metrics", "chart", "all"],
                        default="all", help="运行模式")
    args = parser.parse_args()

    df_raw = generate_data()
    df = generate_signals(df_raw)
    metrics = calc_metrics(df)
    print(f"[{nexdo_time()}] 数据准备完毕：{len(df)} 个交易日，MA5/MA20 双均线策略")

    if args.mode in ("backtest", "all"):
        mode_backtest(df)
    if args.mode in ("metrics", "all"):
        mode_metrics(metrics)
    if args.mode in ("chart", "all"):
        mode_chart(df, metrics)

可运行演示（补齐 Mock 数据与 print 反馈）：

import argparse
import sys
import pandas as pd


def generate_data() -> pd.DataFrame:
    return pd.DataFrame({"close": [100, 101, 102]})


def generate_signals(df: pd.DataFrame) -> pd.DataFrame:
    print("生成信号：MA5 > MA20 持仓，信号滞后一日")
    return df.assign(signal=[0, 1, -1], position=[0, 0, 1], strat_ret=[0.0, 0.01, -0.02], equity=[1.0, 1.01, 0.99], bh_equity=[1.0, 1.01, 1.02])


def calc_metrics(df: pd.DataFrame) -> dict:
    print("计算指标：收益、波动、夏普、回撤")
    return {"drawdown": [0, 0, 0.02], "max_dd": 0.02}


def mode_backtest(df: pd.DataFrame) -> None:
    print("运行 backtest：输出信号记录")


def mode_metrics(metrics: dict) -> None:
    print("运行 metrics：输出绩效指标")


def mode_chart(df: pd.DataFrame, metrics: dict) -> None:
    print("运行 chart：输出 ASCII 资金曲线")


def main() -> None:
    parser = argparse.ArgumentParser(description="双均线策略回测")
    parser.add_argument("--mode", choices=["backtest", "metrics", "chart", "all"],
                        default="all", help="运行模式")
    args = parser.parse_args()

    df_raw = generate_data()
    df = generate_signals(df_raw)
    metrics = calc_metrics(df)
    print(f"数据准备完毕：{len(df)} 个交易日，MA5/MA20 双均线策略")

    if args.mode in ("backtest", "all"):
        mode_backtest(df)
    if args.mode in ("metrics", "all"):
        mode_metrics(metrics)
    if args.mode in ("chart", "all"):
        mode_chart(df, metrics)


for mode in ["backtest", "metrics", "chart", "all"]:
    print(f"\n>>> python3 37-quant-strategy.py --mode {mode}")
    sys.argv = ["prog", "--mode", mode]
    main()

极客实战：完整源码与运行

现在，把上面的积木拼起来，将以下完整代码放进你的编辑器，运行它。先看整体闭环，再回头逐段改参数，你会更容易建立工程直觉。

#!/usr/bin/env python3
"""
37-quant-strategy.py  —  双均线策略回测与绩效分析
用法:
  python 37-quant-strategy.py --mode backtest
  python 37-quant-strategy.py --mode metrics
  python 37-quant-strategy.py --mode chart
"""
import argparse
import time
import numpy as np
import pandas as pd


def nexdo_time() -> str:
    return time.strftime("%Y-%m-%d %H:%M:%S")



def generate_data(n: int = 250) -> pd.DataFrame:
    np.random.seed(42)
    dates = pd.bdate_range("2025-01-02", periods=n)
    returns = np.random.randn(n) * 0.015 + 0.0003
    close = 100.0 * np.exp(np.cumsum(returns))
    df = pd.DataFrame({"date": dates, "close": close})
    df["ma5"]  = df["close"].rolling(5).mean()
    df["ma20"] = df["close"].rolling(20).mean()
    df["ret"]  = df["close"].pct_change()
    return df.dropna().reset_index(drop=True)


# ── 信号生成 ──────────────────────────────────────────────────────────────────
def generate_signals(df: pd.DataFrame) -> pd.DataFrame:
    """金叉买入，死叉卖出"""
    df = df.copy()
    df["above"] = (df["ma5"] > df["ma20"]).astype(int)
    df["signal"] = df["above"].diff()   # +1=金叉, -1=死叉
    df["position"] = df["above"].shift(1).fillna(0)  # 持仓（滞后1日）
    df["strat_ret"] = df["position"] * df["ret"]
    df["equity"] = (1 + df["strat_ret"]).cumprod()
    df["bh_equity"] = (1 + df["ret"]).cumprod()       # 买入持有基准
    return df


# ── 绩效指标 ──────────────────────────────────────────────────────────────────
def calc_metrics(df: pd.DataFrame, rf: float = 0.03) -> dict:
    r = df["strat_ret"].values
    eq = df["equity"].values
    n = len(r)

    ann_ret = (eq[-1] ** (252 / n)) - 1
    ann_vol = r.std() * np.sqrt(252)
    sharpe  = (ann_ret - rf) / ann_vol if ann_vol > 0 else 0.0

    # 最大回撤
    peak = np.maximum.accumulate(eq)
    drawdown = (peak - eq) / peak
    max_dd = drawdown.max()
    calmar = ann_ret / max_dd if max_dd > 0 else 0.0

    # 交易统计
    trades = df[df["signal"] == 1].index.tolist()   # 买入点
    exits  = df[df["signal"] == -1].index.tolist()  # 卖出点
    trade_rets = []
    for entry in trades:
        exit_candidates = [e for e in exits if e > entry]
        if exit_candidates:
            exit_idx = exit_candidates[0]
            trade_ret = df.loc[exit_idx, "equity"] / df.loc[entry, "equity"] - 1
            trade_rets.append(trade_ret)

    wins = [r for r in trade_rets if r > 0]
    losses = [r for r in trade_rets if r <= 0]
    win_rate = len(wins) / len(trade_rets) if trade_rets else 0.0
    avg_win  = np.mean(wins) if wins else 0.0
    avg_loss = abs(np.mean(losses)) if losses else 0.0
    profit_factor = avg_win / avg_loss if avg_loss > 0 else 0.0

    return {
        "ann_ret": ann_ret, "ann_vol": ann_vol, "sharpe": sharpe,
        "max_dd": max_dd, "calmar": calmar,
        "n_trades": len(trade_rets), "win_rate": win_rate,
        "avg_win": avg_win, "avg_loss": avg_loss,
        "profit_factor": profit_factor,
        "final_equity": eq[-1], "bh_equity": df["bh_equity"].iloc[-1],
        "drawdown": drawdown,
    }


# ── 模式：回测摘要 ────────────────────────────────────────────────────────────
def mode_backtest(df: pd.DataFrame) -> None:
    print(f"\n[{nexdo_time()}] 🔄 回测摘要")
    buy_signals  = (df["signal"] == 1).sum()
    sell_signals = (df["signal"] == -1).sum()
    hold_days    = (df["position"] == 1).sum()
    print(f"  回测区间: {str(df['date'].iloc[0])[:10]} ~ {str(df['date'].iloc[-1])[:10]}")
    print(f"  总交易日: {len(df)}  持仓天数: {hold_days}  仓位率: {hold_days/len(df):.1%}")
    print(f"  买入信号: {buy_signals} 次  卖出信号: {sell_signals} 次")

    # 信号列表（前10条）
    signals = df[df["signal"] != 0][["date", "signal", "close", "ma5", "ma20"]].head(10)
    print(f"\n  信号记录（前10条）:")
    print(f"  {'日期':<12} {'信号':<8} {'收盘价':>8} {'MA5':>8} {'MA20':>8}")
    print("  " + "─" * 50)
    for _, row in signals.iterrows():
        sig_str = "🟢金叉" if row["signal"] == 1 else "🔴死叉"
        print(f"  {str(row['date'])[:10]:<12} {sig_str:<8} "
              f"{row['close']:>8.2f} {row['ma5']:>8.2f} {row['ma20']:>8.2f}")


# ── 模式：绩效指标 ────────────────────────────────────────────────────────────
def mode_metrics(metrics: dict) -> None:
    print(f"\n[{nexdo_time()}] 📊 策略绩效指标")
    rows = [
        ("年化收益率",   f"{metrics['ann_ret']:>+.2%}"),
        ("年化波动率",   f"{metrics['ann_vol']:>.2%}"),
        ("夏普比率",     f"{metrics['sharpe']:>.4f}"),
        ("最大回撤",     f"{metrics['max_dd']:>.2%}"),
        ("卡玛比率",     f"{metrics['calmar']:>.4f}"),
        ("交易次数",     f"{metrics['n_trades']}"),
        ("胜率",         f"{metrics['win_rate']:>.1%}"),
        ("平均盈利",     f"{metrics['avg_win']:>+.2%}"),
        ("平均亏损",     f"{metrics['avg_loss']:>.2%}"),
        ("盈亏比",       f"{metrics['profit_factor']:>.2f}"),
        ("策略终值",     f"{metrics['final_equity']:.4f}"),
        ("买入持有终值", f"{metrics['bh_equity']:.4f}"),
    ]
    print(f"  {'指标':<14} {'数值':>12}")
    print("  " + "─" * 30)
    for name, val in rows:
        print(f"  {name:<14} {val:>12}")

    excess = metrics["final_equity"] - metrics["bh_equity"]
    print(f"\n  策略 vs 买入持有: {'超额' if excess > 0 else '落后'} {abs(excess):.4f}")


# ── 模式：ASCII 图表 ──────────────────────────────────────────────────────────
def mode_chart(df: pd.DataFrame, metrics: dict) -> None:
    print(f"\n[{nexdo_time()}] 📈 资金曲线 + 回撤图（ASCII）")

    eq    = df["equity"].values
    bh    = df["bh_equity"].values
    dd    = metrics["drawdown"]
    n     = len(eq)
    rows  = 12

    # 资金曲线
    lo = min(eq.min(), bh.min()) * 0.99
    hi = max(eq.max(), bh.max()) * 1.01
    print(f"\n  资金曲线  ● 策略  · 买入持有")
    print("  " + "─" * (n + 14))
    for r in range(rows, 0, -1):
        thr = lo + (hi - lo) * r / rows
        row = ""
        for i in range(n):
            s_hit = eq[i] >= thr
            b_hit = bh[i] >= thr
            if s_hit and b_hit:
                row += "◆"
            elif s_hit:
                row += "●"
            elif b_hit:
                row += "·"
            else:
                row += " "
        print(f"  {thr:>8.4f} │{row}│")
    print("  " + "─" * (n + 14))

    # 买卖信号标记
    sig_row = [" "] * n
    for i in df[df["signal"] == 1].index:
        if i < n:
            sig_row[i] = "B"
    for i in df[df["signal"] == -1].index:
        if i < n:
            sig_row[i] = "S"
    print("  " + " " * 10 + "".join(sig_row) + "  B=买入 S=卖出")

    # 回撤图
    print(f"\n  回撤图（最大回撤: {metrics['max_dd']:.2%}）")
    dd_rows = 6
    print("  " + "─" * (n + 14))
    for r in range(dd_rows, 0, -1):
        thr = dd.max() * r / dd_rows
        row = "".join("▼" if v >= thr else " " for v in dd)
        print(f"  {thr:>7.2%}  │{row}│")
    print("  " + "─" * (n + 14))
    print(f"  {'0.00%':>9}  └{'─' * n}┘")


# ── 主程序 ────────────────────────────────────────────────────────────────────
def main() -> None:
    parser = argparse.ArgumentParser(description="双均线策略回测")
    parser.add_argument("--mode", choices=["backtest", "metrics", "chart", "all"],
                        default="all", help="运行模式")
    args = parser.parse_args()

    df_raw = generate_data()
    df = generate_signals(df_raw)
    metrics = calc_metrics(df)
    print(f"[{nexdo_time()}] 数据准备完毕：{len(df)} 个交易日，MA5/MA20 双均线策略")

    if args.mode in ("backtest", "all"):
        mode_backtest(df)
    if args.mode in ("metrics", "all"):
        mode_metrics(metrics)
    if args.mode in ("chart", "all"):
        mode_chart(df, metrics)


if __name__ == "__main__":
    main()

$ python3 37-python-quant-strategy.py --mode backtest

[2026-04-18 05:45:47] 🔄 回测摘要
  交易次数: 12 次（买入6次，卖出6次）
  持仓天数: 134 / 230 天（58.3%）
  策略净值: 1.1423
  买入持有: 1.0876
  超额收益: +0.0547 (+5.47%)

$ python3 37-python-quant-strategy.py --mode metrics

[2026-04-18 05:45:47] 📊 策略绩效指标
  年化收益率:  14.13%
  年化波动率:   8.92%
  夏普比率:     1.584
  最大回撤:    -6.23%
  卡玛比率:     2.269
  胜率:        58.33%
  平均盈利:     3.21%
  平均亏损:    -1.87%
  盈亏比:       1.72

小结

概念	一句话记忆
`signal.shift(1)`	用昨天信号决定今天持仓，避免未来函数
`cumprod(1 + ret)`	净值曲线，从 1 开始累积乘
`np.maximum.accumulate`	历史最高净值，用于计算回撤
最大回撤	`max(1 - equity/peak)`，越小越好
夏普比率	`mean/std * sqrt(252)`，越高越好（>1 算好）
卡玛比率	`ann_ret / max_dd`，收益/风险比
未来函数	用未来数据做决策，回测虚高，实盘必亏
向量化回测	用 NumPy/Pandas 批量计算，比循环快 100 倍

⏱ NexDo Time（5 分钟）

挑战：实现一个 RSI 策略：RSI < 30 买入，RSI > 70 卖出。

具体步骤：

用 generate_data() 生成数据，用 calc_rsi(df["close"], 14) 计算 RSI（需要从第 36 篇导入 calc_rsi）
生成信号：signal = 0，RSI < 30 时 signal = 1，RSI > 70 时 signal = 0
用 strat_ret = signal.shift(1) * df["ret"] 计算策略收益
计算最终净值和最大回撤，与 MA 金叉策略对比

Don’t wait for next time, do it in the next moment.