SMA + 杠杆 ETF 轮动策略回测框架

以 SPY 200 日移动平均线作为宏观状态过滤器，在杠杆 ETF（Risk-on）与防御性资产（Risk-off）之间轮动，防止长期熊市中波动率损耗摧毁杠杆复利。

核心思路

SPY 收盘价 > 200 SMA × 1.04  →  买入杠杆 ETF（如 TQQQ / UPRO）
SPY 收盘价 < 200 SMA × 0.97  →  切换至防御资产（如 QQQ / SPY / SGOV）
中间地带                       →  维持当前仓位（减少 whipsaw）

快速开始

# 安装
pip install -e ".[dev]"

# 用默认配置运行 5 个策略的完整回测
python3 -m smaetf configs/default.yaml

# 指定输出路径
python3 -m smaetf configs/default.yaml -o my_report.html

# 只生成报告，不输出终端表格
python3 -m smaetf configs/default.yaml --no-terminal

# 详细日志
python3 -m smaetf configs/default.yaml -v

运行后会输出：

终端：策略摘要表、多策略对比表、危机分析表、Top-5 回撤事件
文件：自包含 HTML 报告（净值曲线 + 指标对比 + 水下图 + 年度回报）

YAML 配置

所有策略通过 YAML 定义。configs/default.yaml 包含 5 个预置策略：

backtest:
  annual_trading_days: 252
  commission_bps: 0.0
  slippage_bps: 0.0

benchmarks:
  - SPY
  - QQQ

strategies:
  - name: "TQQQ/QQQ Buffered Binary (SPY signal)"
    signal:
      ma: { method: sma, window: 200 }
      signal_source: SPY
      method: buffered
      buy_buffer_pct: 0.04
      sell_buffer_pct: 0.03
    position:
      method: binary
    risk_on_asset: TQQQ
    risk_off: { asset: QQQ, transition: immediate }

可配置维度

维度	选项
均线算法	`sma` / `ema` / `wma`，窗口可调
信号方法	`bare`（裸均线）/ `buffered`（非对称缓冲带）
仓位管理	`binary`（二元切换）/ `tiered`（阶梯式）/ `linear`（线性渐进）
Risk-on	`TQQQ` / `QLD` / `UPRO` / `SSO`
Risk-off	`QQQ` / `SPY` / `cash` / `SGOV`
过渡模式	`immediate` / `dca_6mo` / `dca_12mo`
信号源	`SPY`（跨资产）/ `QQQ`（同资产）

输出指标

CAGR、最大回撤、Sharpe / Sortino / Calmar / Omega Ratio、VaR / CVaR、Alpha / Beta / Information Ratio、交易次数、胜率、盈亏比、回撤持续期。

危机分析窗口

危机	时间段	类型
互联网泡沫	2000-03 ~ 2002-10	长期阴跌
金融危机	2007-10 ~ 2009-03	系统性崩盘
COVID V反	2020-02 ~ 2020-08	闪崩反转
通胀熊市	2022-01 ~ 2022-10	股债正相关

测试

python3 -m pytest tests/ -m unit -v                    # 单元测试
python3 -m pytest tests/ -m integration -v             # 集成测试
python3 -m pytest tests/ -m golden -v                  # Golden File 回归
python3 -m pytest tests/ -m "unit or integration or property or extreme or golden" -v  # 全量

275 个测试通过（248 unit + 20 integration + 6 property + 7 extreme + 3 golden）。

数据

data/full/ 包含 1993~2026 年的完整日线数据（6 个资产的 Parquet 文件）。杠杆 ETF 发行前的数据由底层指数日收益率合成重建（含融资成本建模）。

AI Agent 使用指南

本系统设计为 CLI 驱动、YAML 配置、结构化输出，天然适合 AI Agent 进行多轮策略探索。以下是 AI 可执行的完整操作手册。

环境准备

cd /path/to/SMAETF
pip install -e ".[dev]"
# 确认数据就绪
ls data/full/  # 应有 SPY.parquet, QQQ.parquet, TQQQ.parquet, QLD.parquet, SSO.parquet, UPRO.parquet

操作 1：运行预置策略

python3 -m smaetf configs/default.yaml -o report.html --no-terminal

输出文件 report.html 包含完整图表和指标对比。终端输出（去掉 --no-terminal）包含 Rich 表格。

操作 2：创建自定义 YAML 并回测

写入一个新 YAML 文件，然后运行。示例——测试 EMA 150 + 线性仓位 + DCA 过渡：

# configs/experiment_01.yaml
backtest:
  annual_trading_days: 252
  commission_bps: 5.0
  slippage_bps: 5.0

benchmarks:
  - SPY

strategies:
  - name: "EMA150 Linear DCA TQQQ"
    signal:
      ma: { method: ema, window: 150 }
      signal_source: SPY
      method: buffered
      buy_buffer_pct: 0.03
      sell_buffer_pct: 0.02
    position:
      method: linear
      linear_start_threshold: -0.05
      linear_full_threshold: 0.08
    risk_on_asset: TQQQ
    risk_off: { asset: QQQ, transition: dca_6mo }

python3 -m smaetf configs/experiment_01.yaml -o experiment_01.html --no-terminal

操作 3：通过 Python API 获取结构化数据

当需要精确数值而非 HTML 报告时，用 Python 脚本：

from pathlib import Path
from smaetf.cli import load_config, load_market_data, _collect_tickers
from smaetf.engine.backtest import run_backtest
from smaetf.analytics.metrics import compute_metrics
from smaetf.analytics.crisis import analyze_all_crises
from smaetf.analytics.drawdown import find_drawdown_events

# 加载
cfg_path = Path("configs/default.yaml")
backtest_cfg, strategies, benchmarks = load_config(cfg_path)
tickers = _collect_tickers(strategies, benchmarks)
market_data = load_market_data(tickers, Path("data/full"))

# 对每个策略取指标
for strategy in strategies:
    result = run_backtest(strategy, backtest_cfg, market_data)
    metrics = compute_metrics(result)
    crises = analyze_all_crises(result)
    drawdowns = find_drawdown_events(result.nav)

    print(f"\n=== {strategy.name} ===")
    print(f"  CAGR:    {metrics['cagr']:.4f}")
    print(f"  MaxDD:   {metrics['max_drawdown']:.4f}")
    print(f"  Sharpe:  {metrics['sharpe_ratio']:.4f}")
    print(f"  Sortino: {metrics['sortino_ratio']:.4f}")
    print(f"  Calmar:  {metrics['calmar_ratio']:.4f}")
    print(f"  Trades:  {metrics['total_trades']}")
    print(f"  WinRate: {metrics['win_rate']:.4f}")

    for c in crises:
        if c.get("data_available"):
            print(f"  {c['crisis_name']}: return={c['strategy_return']:.4f}, maxdd={c['strategy_max_dd']:.4f}")

    for d in drawdowns[:3]:
        print(f"  Drawdown: {d.peak_date}→{d.trough_date}, depth={d.depth:.4f}, days={d.underwater_days}")

操作 4：参数扫描（AI 多轮实验模板）

AI 可以批量生成 YAML、运行、收集结果。以下是一个扫描 SMA 窗口的模板：

import yaml, json
from pathlib import Path
from smaetf.cli import load_market_data
from smaetf.config import *
from smaetf.engine.backtest import run_backtest
from smaetf.analytics.metrics import compute_metrics

market_data = load_market_data(
    {"SPY", "QQQ", "TQQQ"}, Path("data/full")
)

results_log = []
for window in [100, 150, 200, 250, 300]:
    for buy_buf in [0.02, 0.04, 0.06]:
        for sell_buf in [0.02, 0.03, 0.04]:
            strategy = StrategyConfig(
                name=f"SMA{window}_buy{buy_buf}_sell{sell_buf}",
                signal=SignalConfig(
                    ma=MAConfig(method="sma", window=window),
                    method="buffered",
                    buy_buffer_pct=buy_buf,
                    sell_buffer_pct=sell_buf,
                ),
                position=PositionConfig(method="binary"),
                risk_on_asset="TQQQ",
                risk_off=RiskOffConfig(asset="QQQ", transition="immediate"),
            )
            result = run_backtest(strategy, BacktestConfig(), market_data)
            m = compute_metrics(result)
            row = {
                "window": window,
                "buy_buf": buy_buf,
                "sell_buf": sell_buf,
                "cagr": round(m["cagr"], 4),
                "max_drawdown": round(m["max_drawdown"], 4),
                "sharpe": round(m["sharpe_ratio"], 4),
                "trades": m["total_trades"],
            }
            results_log.append(row)
            print(json.dumps(row))

# 保存为 JSON 供后续分析
Path("scan_results.json").write_text(json.dumps(results_log, indent=2))

操作 5：对比两种仓位模式

configs = [
    ("binary", PositionConfig(method="binary")),
    ("tiered", PositionConfig(method="tiered", tiers=(
        PositionTier(-0.08, 0.0),
        PositionTier(-0.03, 0.3),
        PositionTier(0.0, 0.5),
        PositionTier(0.05, 0.8),
        PositionTier(0.10, 1.0),
    ))),
    ("linear", PositionConfig(method="linear",
        linear_start_threshold=-0.03, linear_full_threshold=0.10)),
]

signal = SignalConfig(
    ma=MAConfig(method="sma", window=200),
    method="buffered",
    buy_buffer_pct=0.04,
    sell_buffer_pct=0.03,
)

for name, pos_cfg in configs:
    strategy = StrategyConfig(
        name=name,
        signal=signal,
        position=pos_cfg,
        risk_on_asset="TQQQ",
        risk_off=RiskOffConfig(asset="QQQ", transition="immediate"),
    )
    result = run_backtest(strategy, BacktestConfig(), market_data)
    m = compute_metrics(result)
    print(f"{name:8s}  CAGR={m['cagr']:.4f}  MaxDD={m['max_drawdown']:.4f}  Sharpe={m['sharpe_ratio']:.4f}")

AI 实验设计建议

以下是有价值的多轮探索方向，每个都可以通过修改 YAML 或 Python 脚本实现：

实验	变量	目标
均线窗口扫描	`window`: 50~400	找到 Sharpe 最大化的窗口
缓冲带宽度扫描	`buy_buffer_pct` / `sell_buffer_pct`: 0~0.10	最小化 whipsaw 同时保持响应性
2x vs 3x	`TQQQ` vs `QLD`, `UPRO` vs `SSO`	风险调整后哪个更优
过渡模式	`immediate` vs `dca_6mo` vs `dca_12mo`	DCA 是否降低 whipsaw 成本
信号源对比	`signal_source: SPY` vs `QQQ`	跨资产信号是否优于同资产
交易成本敏感性	`commission_bps` + `slippage_bps`: 0~20	策略在真实成本下是否仍有优势
危机类型分析	用 `analyze_all_crises()`	不同配置在四种危机中的表现差异
仓位模式对比	`binary` vs `tiered` vs `linear`	渐进入场是否优于全仓切换

关键数据结构参考

BacktestResult:
  .nav              → pd.Series (每日净值)
  .returns          → pd.Series (每日收益率)
  .regime_series    → pd.Series (每日 Regime 枚举)
  .trades           → list[Trade] (买卖循环记录)
  .strategy_name    → str
  .total_cost       → float

compute_metrics() 返回 dict:
  cagr, total_return, max_drawdown, max_drawdown_duration,
  annual_volatility, downside_volatility, var_95, cvar_95,
  sharpe_ratio, sortino_ratio, calmar_ratio, omega_ratio,
  total_trades, avg_holding_days, win_rate, profit_loss_ratio,
  max_win, max_loss, alpha, beta, information_ratio,
  annual_returns (dict[int, float])

analyze_all_crises() 返回 list[dict]:
  每个 dict: crisis_name, start, end, data_available,
  strategy_return, strategy_max_dd, trading_days, regime_distribution

find_drawdown_events() 返回 list[DrawdownEvent]:
  peak_date, trough_date, recovery_date, depth,
  duration_to_trough, duration_to_recovery, underwater_days

可用资产

Ticker	类型	杠杆	族
SPY	标普 500	1x	sp500
SSO	标普 500	2x	sp500
UPRO	标普 500	3x	sp500
QQQ	纳斯达克 100	1x	nasdaq100
QLD	纳斯达克 100	2x	nasdaq100
TQQQ	纳斯达克 100	3x	nasdaq100

限制

vix_killswitch、macd_confirm、bodyguard 三个高级信号尚未实现（Phase 7）
数据需要预先存在于 data/full/，CLI 没有自动下载命令
信号方法仅支持 bare 和 buffered

项目结构

configs/
  default.yaml               — 默认 5 策略配置
src/smaetf/
  __main__.py                — python -m smaetf 入口
  cli.py                     — CLI 主逻辑 (YAML 解析 / 回测编排 / 报告输出)
  config.py                  — 核心 dataclass (Regime / BacktestConfig / StrategyConfig 等)
  data/
    registry.py              — ETF 元数据注册表
    downloader.py            — yfinance 下载封装
    cache.py                 — Parquet 本地缓存
    synthetic.py             — 合成 LETF 价格重建
  signals/
    base.py                  — SignalGenerator Protocol
    ma.py                    — SMA / EMA / WMA 统一计算
    sma_bare.py              — 裸均线信号
    sma_buffered.py          — 非对称缓冲带信号
    composite.py             — 多信号组合器
  positioning/               — 仓位管理 (binary / tiered / linear)
  engine/
    portfolio.py             — BacktestResult / Trade dataclass
    backtest.py              — 回测主循环 run_backtest()
    transition.py            — 过渡引擎 (即时切换 / DCA 状态机)
  analytics/
    metrics.py               — 核心绩效指标 compute_metrics()
    drawdown.py              — 回撤深度分析
    rolling.py               — 滚动窗口分析
    crisis.py                — 危机分析 (4 个预定义窗口)
    comparison.py            — 多策略对比表
  reporting/
    terminal.py              — Rich 终端表格
    charts.py                — Plotly 交互图表
    html_report.py           — 自包含 HTML 报告
data/full/                   — 合成+真实合并的完整序列 (1993~2026)
tests/                       — 275 个测试 (unit / integration / property / extreme / golden)
  golden/                    — Golden File 基线 (JSON)

实现进度

Phase	内容	状态
1	基础设施 (配置 / 注册表 / 下载 / 缓存 / 合成)	✅
2	均线与信号 (SMA/EMA/WMA / 裸均线 / 缓冲带 / 组合器)	✅
3	仓位管理 (binary / tiered / linear)	✅
4	回测引擎 (主循环 / DCA 状态机)	✅
5	分析与报告 (指标 / 回撤 / 滚动 / 危机 / 图表)	✅
6	集成 (CLI / YAML 配置 / Golden File)	✅
7	高级信号 (VIX 熔断 / MACD 确认 / 泡沫偏离保镖)	待开发

技术栈

文档

需求文档 — 策略需求与回测框架规格
开发文档 — 技术架构、7 阶段 34 步实现计划
进度跟踪 — 各步骤完成状态
问题记录 — 实现过程中的技术决策

Name		Name	Last commit message	Last commit date
configs		configs
data		data
docs		docs
scripts		scripts
src/smaetf		src/smaetf
tests		tests
.gitignore		.gitignore
CLAUDE.md		CLAUDE.md
ISSUES.md		ISSUES.md
PROGRESS.md		PROGRESS.md
README.md		README.md
pyproject.toml		pyproject.toml
report.html		report.html

Navigation Menu

Repository files navigation

SMA + 杠杆 ETF 轮动策略回测框架

核心思路

快速开始

YAML 配置

可配置维度

输出指标

危机分析窗口

测试

数据

AI Agent 使用指南

环境准备

操作 1：运行预置策略

操作 2：创建自定义 YAML 并回测

操作 3：通过 Python API 获取结构化数据

操作 4：参数扫描（AI 多轮实验模板）

操作 5：对比两种仓位模式

AI 实验设计建议

关键数据结构参考

可用资产

限制

项目结构

实现进度

技术栈

文档

About

Releases

Packages

Contributors

Languages

Latest commit

History

Folders and files

Repository files navigation

SMA + 杠杆 ETF 轮动策略回测框架

核心思路

快速开始

YAML 配置

可配置维度

输出指标

危机分析窗口

测试

数据

AI Agent 使用指南

环境准备

操作 1：运行预置策略

操作 2：创建自定义 YAML 并回测

操作 3：通过 Python API 获取结构化数据

操作 4：参数扫描（AI 多轮实验模板）

操作 5：对比两种仓位模式

AI 实验设计建议

关键数据结构参考

可用资产

限制

项目结构

实现进度

技术栈

文档

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages