恭喜你走到了最后一篇。前面四十九篇，每篇都在拼图上添一块：数据、指标、回测、策略、统计、机器学习、工程、风控、实盘准备。本篇是结业项目——把所有拼图拼成一张完整的图：一个「沪深 300 + 国债 ETF 动量轮动 + ATR 止损 + 月度调仓」的完整策略，从目标设定到纸面交易计划，八步走完一个真实策略的全生命周期。每一步都会回指对应篇章——如果你在某一步卡住，说明那一篇值得重做练习。
本篇目标
- 能独立走完策略全生命周期：目标 → 数据 → 信号 → 回测 → 评价 → 风控 → 敏感性 → 纸面计划
- 能把第 27（动量）、30（轮动）、35（ATR 止损）、48（风控）篇的模块组装成一个整体
- 能对成品做参数敏感性与 Walk-Forward 检验，确认它不是过拟合产物
- 能输出一份「切换纸面交易」的行动计划
- 能对着结业自检清单，确认自己从合集带走了真本事
核心概念
结业项目为什么选「轮动 + ATR 止损」
三个理由：难度适配——它综合了动量信号（第 27 篇）、多资产轮动（第 30 篇）、波动率风控（第 14、35 篇），但每个模块都不超出合集范围；逻辑自洽——轮动提供「站在强的那边」，ATR 止损提供「错了就撤」，月度调仓提供「低换手、成本可控」，三者互补不重叠；工程完整——它有双资产数据管道、月度调度、风控状态机，正好把第 41–43 篇的工程件用上。
八步流程总览
① 目标设定 → ② 数据获取 → ③ 信号生成 → ④ 回测实现
↓ ↓
⑧ 纸面计划 ← ⑦ 敏感性检验 ← ⑥ 风控设计 ← ⑤ 绩效评价
流程是环形的：第 ⑦ 步如果发现参数敏感（悬崖而非平原），要退回 ③④ 重选参数；第 ⑥ 步如果发现熔断线与策略回撤特征不匹配，要回到 ① 修正目标。真实的研究从来不是直线，而是带着检查点的循环——这本身就是本篇要教的东西。
完整代码
本篇依赖：uv pip install pandas numpy matplotlib。全部模块化呈现，拼起来就是完整项目。
# ---------- ① 目标设定 + ② 数据获取 ----------
import numpy as np
import pandas as pd
# 目标（写下来，防止做着做着跑偏）：
# 年化 > 6%（跑赢债券），最大回撤 < 15%（拿得住），月均换手 < 30%（成本可控）
def get_pair_data():
"""沪深300ETF与国债ETF日线。akshare优先，失败用模拟数据"""
try:
import akshare as ak
a = ak.fund_etf_hist_em(symbol="510300", period="daily",
start_date="20190101", end_date="20241231")
b = ak.fund_etf_hist_em(symbol="511010", period="daily",
start_date="20190101", end_date="20241231")
px = pd.DataFrame({
"stock": a.set_index(pd.to_datetime(a["日期"]))["收盘"],
"bond": b.set_index(pd.to_datetime(b["日期"]))["收盘"],
})
return px.dropna()
except Exception:
rng = np.random.default_rng(50)
n = 1450
idx = pd.bdate_range("2019-01-02", periods=n)
# 股票：三段结构（涨-跌-涨）；债券：低波缓涨
drift = np.concatenate([np.full(500, 0.0008), np.full(450, -0.0012),
np.full(500, 0.0006)])
stock = 4.0 * np.exp(np.cumsum(rng.normal(0, 0.015, n) + drift))
bond = 100 * np.exp(np.cumsum(rng.normal(0.0002, 0.002, n)))
return pd.DataFrame({"stock": stock, "bond": bond}, index=idx)
px = get_pair_data()
# ---------- ③ 信号生成：动量轮动 + ATR（模块：strategies/rotation.py） ----------
def build_features(px, mom_window=20, atr_window=14):
feat = px.copy()
feat["mom_stock"] = px["stock"].pct_change(mom_window)
feat["atr_stock"] = (px["stock"].rolling(atr_window).max()
- px["stock"].rolling(atr_window).min())
# 简化ATR：14日高低幅均值（教学口径，严格版见第14篇三情形TR）
feat["atr_pct"] = feat["atr_stock"] / px["stock"]
return feat.dropna()
def monthly_signal(feat):
"""月末：股票动量>0 → 持股；否则持债。返回目标持仓序列"""
month_end = feat.groupby([feat.index.year, feat.index.month]).tail(1).index
target = pd.Series(index=feat.index, dtype="object")
cur = "bond"
planned = {}
for d in month_end:
cur = "stock" if feat.loc[d, "mom_stock"] > 0 else "bond"
nxt = feat.index[feat.index > d]
if len(nxt):
planned[nxt[0]] = cur
for d in feat.index:
cur = planned.get(d, cur if d == feat.index[0] else target.iloc[
feat.index.get_loc(d) - 1])
target[d] = cur
return target
# ---------- ④⑤ 回测 + 评价（模块：backtest/engine.py + metrics.py） ----------
def run_backtest(feat, target, fee=0.001, stop_atr_k=2.5):
"""月度轮动 + 持仓期ATR止损（跌破入场价-k×ATR即切债）"""
rets = feat[["stock", "bond"]].pct_change().fillna(0)
daily = pd.Series(0.0, index=feat.index)
holding, entry_px, entry_atr = None, None, None
for i, d in enumerate(feat.index):
if i == 0:
holding = target[d]
entry_px, entry_atr = feat.loc[d, "stock"], feat.loc[d, "atr_stock"]
continue
tgt = target[d]
# ATR止损：持股期间跌破入场价-k×ATR → 提前切债
if holding == "stock" and tgt == "stock":
stop_line = entry_px - stop_atr_k * entry_atr
if feat.loc[d, "stock"] < stop_line:
tgt = "bond"
if tgt != holding: # 换仓：扣双边成本
daily[d] -= fee
if tgt == "stock":
entry_px = feat.loc[d, "stock"]
entry_atr = feat.loc[d, "atr_stock"]
holding = tgt
daily[d] += rets.loc[d, holding]
return daily
def evaluate(daily, rf=0.02):
nav = (1 + daily).cumprod()
years = len(daily) / 252
ann = nav.iloc[-1] ** (1 / years) - 1
vol = daily.std() * np.sqrt(252)
sharpe = (ann - rf) / vol
dd = (nav / nav.cummax() - 1).min()
return {"年化": ann, "夏普": sharpe, "最大回撤": dd,
"Calmar": ann / abs(dd)}, nav
feat = build_features(px)
target = monthly_signal(feat)
daily = run_backtest(feat, target)
stats, nav = evaluate(daily)
print("策略(示意):", {k: f"{v:.2f}" if k == "夏普" else f"{v:.1%}"
for k, v in stats.items()})
# ---------- ⑥ 风控设计（承接第48篇，对齐回撤特征） ----------
# 策略回撤(示意)约-12% → 熔断线设 15%（历史回撤×1.25 缓冲）：
# 半仓线 12%、空仓线 18%、冷却 10 个交易日、单笔风险预算 1%
# （若历史回撤改变，按 历史最大回撤×1.25 重算，不拍脑袋）
# ---------- ⑦ 敏感性检验：动量窗口与ATR倍数的参数平原 ----------
def sensitivity(px, mom_list=(10, 20, 40, 60), k_list=(2.0, 2.5, 3.0)):
grid = []
for mw in mom_list:
for k in k_list:
f = build_features(px, mom_window=mw)
t = monthly_signal(f)
s, _ = evaluate(run_backtest(f, t, stop_atr_k=k))
grid.append({"mom": mw, "k": k, "夏普": s["夏普"],
"回撤": s["最大回撤"]})
return pd.DataFrame(grid)
grid = sensitivity(px)
print(grid.pivot(index="mom", columns="k", values="夏普").round(2).to_string())
# ---------- ⑧ 纸面交易计划输出 ----------
def paper_plan(stats, grid, out="paper_plan.md"):
best = grid.loc[grid["夏普"].idxmax()]
lines = [
"# 纸面交易计划（第50篇结业）",
"",
"## 策略参数（冻结候选）",
f"- 动量窗口: {int(best['mom'])} 日（敏感性检验最优，见参数平原表）",
f"- ATR止损倍数: {best['k']}（入场价-k×ATR 切债）",
"- 调仓频率: 月度（月末信号、次月首日执行）",
"- 手续费口径: 双边 0.1%",
"",
"## 风控参数（第48篇口径）",
"- 熔断: 半仓12% / 空仓18% / 冷却10个交易日",
"- 资金计划: 纸面10万，切实盘时 ≤ 总资产30%",
"",
"## 纸面流程（第46篇）",
"1. 每交易日15:05 数据管道更新（第42篇）",
"2. 每月末信号计算与委托记录（第46篇PaperBroker）",
"3. 每周五净值复盘 + 心跳检查（第43篇）",
"4. 连续纸面3个月后对照切换清单（第46篇）评审",
"",
f"## 回测基准（示意，commit待绑定）",
f"- 年化 {stats['年化']:.1%} / 夏普 {stats['夏普']:.2f} / "
f"回撤 {stats['最大回撤']:.1%}",
"",
"> 本文策略仅用于教学，不构成投资建议，实盘前请充分测试。",
]
from pathlib import Path
Path(out).write_text("\n".join(lines), encoding="utf-8")
print(f"已生成: {out}")
paper_plan(stats, grid)
代码逐段解释
- 目标与数据：三个量化目标（收益、回撤、换手）先于代码存在——没有目标的回测是漫游，有目标的回测是实验。数据层 akshare 优先、模拟兜底，双资产对齐
dropna()。 - 信号模块：
build_features算 20 日动量与简化 ATR（严格三情形版见第 14 篇，此处教学简化已注明）；monthly_signal月末判动量、登记到次月首日——信号与执行的时序铁律（第 30 篇）原样复用。 - 回测引擎：在月度轮动骨架上加 ATR 止损——持股期间跌破「入场价 − k×ATR」立即切债，不等月末。换仓扣双边 0.1%；止损触发的换仓同样计费。这就是第 35 篇「退出规则属于策略定义」的落地。
- 评价：第 19 篇的四件套（年化/夏普/回撤/Calmar）。对照第 ① 步的目标检查：三项全达标才继续，不达标回炉。
- 风控对齐：熔断线不拍脑袋，按「历史最大回撤 × 1.25」反推（第 48 篇坑 3 的正面示范）。策略回撤变了，风控参数跟着重算——风控是策略的从变量。
- 敏感性：动量窗口 × ATR 倍数的夏普矩阵。看的是形状不是峰值：如果 20/2.5 是孤峰，参数不可信；连片的高地才敢用（第 25 篇）。
- 纸面计划：把参数、风控、流程、回测基准写成 markdown 落盘——这就是第 46 篇切换清单的前置文档。git commit 后，纸面期的一切对照都以此为准。
示例输出（示意，模拟行情结果）:
策略(示意): {'年化': '8.4%', '夏普': '0.79', '最大回撤': '-11.7%', 'Calmar': '0.72'}
k 2.0 2.5 3.0
mom
10 0.71 0.74 0.70
20 0.78 0.79 0.76
40 0.75 0.77 0.74
60 0.69 0.71 0.68
已生成: paper_plan.md
敏感性矩阵的形状比数字重要：整个 10~60 窗口、2.0~3.0 倍数的区域夏普都在 0.68~0.79 之间——参数平原清晰可见，20/2.5 只是平原中的一点而非孤峰。这样的参数才有资格进入纸面期。
结业自检清单
□ 我能不看教程，从空文件写出本篇的回测引擎（④）
□ 我能解释为什么信号要月末算、次月执行（③ → 第17/30篇）
□ 我能解释 ATR 止损的 k 为什么取 2.5、以及为什么不深究精确值（⑦ → 第25篇）
□ 我的熔断参数能说出推导过程，而不是抄来的（⑥ → 第48篇）
□ 我知道这份回测的三个局限（模拟数据/无幸存者偏差处理/单一市场状态）
□ 我有明确的纸面计划与切换标准（⑧ → 第46篇）
□ 全部代码在 git 里，参数与报告版本绑定（第44篇）
七条里能踏实打勾六条以上，这套合集的使命就完成了。
常见坑与小练习（结业版）
常见坑（结业项目本身的坑）
- 把本篇代码当模板照抄：能跑通不等于能盲写。合上教程从空文件重写一遍④⑤步，写不出来的地方就是你要回炉的篇章。
- 敏感性矩阵只看峰值：盯着 20/2.5 那一格的高分，而忽略整片平原的存在意义。选参数选的是区域不是点（第 25 篇）。
- 纸面计划生成后就束之高阁：paper_plan.md 是给未来三个月的自己立的约，不接上第 42 篇的定时任务、不每周复盘，它只是一段文本。
小练习
- 把本篇的回测引擎换成事件驱动版（第 20 篇的框架），对比两者净值曲线的差异并解释来源。
思路提示：参考第 20 篇末尾的差异三来源（整数股、现金基数、边界处理），轮动策略还会多一个「换仓时点」差异。
- 给策略加上第 33 篇的市场状态过滤：趋势下状态直接持债不动，观察回撤与换手的变化。
思路提示：把 monthly_signal 里的判断改成「动量>0 且非趋势下状态」，状态序列同样要月末取、次月生效。
- 用真实数据（akshare 拉 510300 与 511010）重跑全流程，把模拟数据与真实数据的结论差异写成半页报告。
思路提示：重点看三处：动量窗口的最优区域是否漂移、ATR 止损的触发频率、熔断线推导是否需要重算。
写在合集最后
五十篇，从「为什么要学 Python 量化」到一份能落地的纸面交易计划。最后送你三句话：
第一，工具会过时，方法不会。 pandas 的 API 会改、akshare 的接口会变、Backtrader 终有一天停止维护，但「先想亏多少再想买多少」「样本外才算数」「信号在前执行在后」这些方法层的纪律，再过十年依然成立。
第二，慢就是快。 三个月纸面期看起来是浪费时间，但它拦下的那一次「参数没冻结就上实盘」，可能就是你账户的生死线。量化的复利不只来自收益率，更来自「活得久」。
第三，市场永远比你对。 回测年化 30% 也不代表你掌握了市场，只代表历史数据里恰好有这个模式。保持敬畏，保持小仓位验证，把每一个「我以为」都当成假设去检验——这既是量化的方法，也是对待市场的态度。
结业不是终点。纸面交易跑起来、第一个真实委托下出去、第一份实盘偏差报告写出来——那才是这套教程真正的最后一课。祝你在市场里活得久、学得深。
本文策略仅用于教学，不构成投资建议，实盘前请充分测试。