Appearance
动量 + 风险平价复合策略（消融）· notebook 版
复刻聚宽帖（joinquant.com/post/61894）：主动动量选择 + 被动风险平价配置，带 2%–15% 权重约束。
- 资产四大类：商品(黄金)、美股(标普500)、国内权益(行业/风格/大小盘)、国内债券
- 动量得分 = 近 63 日日均收益 − 近 252 日日均收益；每月月末调仓
- 选择：商品/美股/债券常驻锚定；权益子类按动量取 top-k（行业3/风格2/大小盘2）
- 配置：选中标的做风险平价，单权重∈[2%,15%]，和为 1
- 消融：动量选择 ON vs OFF（OFF=对所有合格标的做同样的 capped 风险平价），检验“动量选择”主动层是否加分
仅供研究学习，不构成投资建议。脚本版（含全市场动态池/PIT 流动性筛选）见
scripts/run_momentum_riskparity.py。
python
%matplotlib inline
import os
while not os.path.exists("pyproject.toml") and os.getcwd() != "/":
os.chdir("..")
import pandas as pd, matplotlib, matplotlib.pyplot as plt
from matplotlib import font_manager
for _n in ["PingFang SC", "Hiragino Sans GB", "Arial Unicode MS", "Songti SC"]:
if _n in {f.name for f in font_manager.fontManager.ttflist}:
matplotlib.rcParams["font.sans-serif"] = [_n]; break
matplotlib.rcParams["axes.unicode_minus"] = False
from quant.strategy.risk_parity import risk_parity_weights # 内核：风险平价定权
from quant.backtest.portfolio import run_rebalance_backtest
from quant.reports.metrics import (annualized_return, annualized_volatility,
sharpe_ratio, max_drawdown, turnover)
# 候选池：手选 26 只（对齐原帖资产清单）
UNIVERSE = {
"518880.SH": ("commodity", "黄金"), "513500.SH": ("us", "标普500"),
"512000.SH": ("industry", "券商"), "512800.SH": ("industry", "银行"),
"512010.SH": ("industry", "医药"), "512480.SH": ("industry", "半导体"),
"515000.SH": ("industry", "科技"), "512660.SH": ("industry", "军工"),
"512690.SH": ("industry", "酒"), "159928.SZ": ("industry", "消费"),
"512400.SH": ("industry", "有色"), "512880.SH": ("industry", "证券"),
"512890.SH": ("style", "红利低波"), "512190.SH": ("style", "成长"),
"512900.SH": ("style", "价值"), "512950.SH": ("style", "质量"),
"510500.SH": ("size", "中证500"), "510300.SH": ("size", "沪深300"),
"159915.SZ": ("size", "创业板"), "588000.SH": ("size", "科创50"),
"511060.SH": ("bond", "5年地方债"), "511270.SH": ("bond", "10年地方债"),
"511090.SH": ("bond", "30年国债"),
}
ANCHORS = ("commodity", "us", "bond") # 常驻类别
SELECT_K = {"industry": 3, "style": 2, "size": 2} # 权益子类 top-k
START, END = "20210101", "20260613"
cat_map = {c: cat for c, (cat, n) in UNIVERSE.items()}
name_map = {c: n for c, (cat, n) in UNIVERSE.items()}
df = pd.read_parquet("data/raw/momentum_rp_etf.parquet") # 读缓存,离线
close = df.pivot(index="datetime", columns="instrument", values="close").sort_index()
cal = close.index[close["510300.SH"].notna()].tolist() # 以沪深300交易日为主日历
close = close.reindex(cal).ffill()
cs = pd.Series(cal)
month_last = cs.groupby(cs.str.slice(0, 6)).last().tolist() # 原帖 run_monthly(-1)：每月最后交易日
rb = [d for d in month_last if START <= d <= END]
print("有数据的候选:", df["instrument"].nunique(), "/", len(UNIVERSE))
print("回测 %s ~ %s, %d 次月末调仓" % (START, END, len(rb)))有数据的候选: 23 / 23
回测 20210101 ~ 20260613, 66 次月末调仓
动量打分、选择、定权
momentum=近 63 日日均收益 − 近 252 日日均收益（需满 252 日历史）。select：常驻类别全纳入，其余按动量 top-k。prev=cal[i-1] 无未来函数。
python
def momentum(prev, code, window=63, lookback=252):
s = close[code].loc[:prev].dropna()
if len(s) < lookback:
return None
s = s.tail(lookback)
return s.tail(window).pct_change().dropna().mean() - s.pct_change().dropna().mean()
def select(prev, use_momentum):
by_cat = {}
for code in cat_map:
if code not in close.columns:
continue
m = momentum(prev, code)
if m is not None:
by_cat.setdefault(cat_map[code], []).append((code, m))
chosen = []
for cat in ANCHORS: # 常驻类别全纳入
chosen += [c for c, _ in by_cat.get(cat, [])]
for cat, k in SELECT_K.items(): # 其余按动量 top-k
ranked = sorted(by_cat.get(cat, []), key=lambda x: -x[1])
chosen += [c for c, _ in (ranked[:k] if use_momentum else ranked)]
return chosen
def make_weight_fn(use_momentum, cov_days=126):
pos = {d: i for i, d in enumerate(cal)}
def wfn(today):
prev = cal[pos[today] - 1] if pos[today] > 0 else today
sel = select(prev, use_momentum)
rets = close[sel].loc[:prev].pct_change().dropna().tail(cov_days)
return risk_parity_weights(rets, w_bounds=(0.02, 0.15)) # 单权重 [2%,15%]
return wfn
def perf(returns, w):
return dict(ann=annualized_return(returns), vol=annualized_volatility(returns),
sr=sharpe_ratio(returns), mdd=max_drawdown(returns), to=turnover(w))消融回测：动量选择 ON vs OFF
python
results = {}
for label, use_mom in [("动量选择+capped风险平价", True), ("无动量(全体capped风险平价)", False)]:
results[label] = run_rebalance_backtest(
close, rb, make_weight_fn(use_mom),
account=1e6, invest_ratio=0.98, commission=0.0003, weight_band=0.0)
bench_px = close.loc[results["动量选择+capped风险平价"]["nav"].index, "510300.SH"]
bench_ret = bench_px.pct_change()
bench_nav = bench_px / bench_px.iloc[0]
print("回测完成,", len(results["动量选择+capped风险平价"]["nav"]), "个交易日")回测完成, 1298 个交易日
绩效对比
python
print(f"{'策略':28s}{'年化':>9s}{'波动':>9s}{'夏普':>7s}{'最大回撤':>10s}{'月均换手':>9s}{'净值':>8s}")
for label, bt in results.items():
p = perf(bt["returns"], bt["weights"])
print(f"{label:28s}{p['ann']*100:+8.2f}%{p['vol']*100:+8.2f}%{p['sr']:7.2f}"
f"{p['mdd']*100:+9.2f}%{p['to']*100:8.2f}%{bt['nav'].iloc[-1]:8.3f}")
bp = perf(bench_ret, pd.DataFrame())
print(f"{'沪深300买入持有':28s}{bp['ann']*100:+8.2f}%{bp['vol']*100:+8.2f}%{bp['sr']:7.2f}"
f"{bp['mdd']*100:+9.2f}%{'—':>9s}{bench_nav.iloc[-1]:8.3f}")
main_bt = results["动量选择+capped风险平价"]
lastw = main_bt["weights"].iloc[-1]
lastw = lastw[lastw > 1e-4].sort_values(ascending=False)
print("\n最近一次调仓 (%s) 选中 %d 只：" % (main_bt["weights"].index[-1], len(lastw)))
for c in lastw.index:
print(" %-10s %-10s %5.1f%%" % (name_map.get(c, c), cat_map.get(c, ""), lastw[c] * 100))策略 年化 波动 夏普 最大回撤 月均换手 净值
动量选择+capped风险平价 +6.61% +8.10% 0.83 -13.43% 32.98% 1.390
无动量(全体capped风险平价) +7.07% +8.71% 0.83 -11.52% 7.34% 1.421
沪深300买入持有 -0.08% +17.96% 0.09 -42.13% — 0.996
最近一次调仓 (20260612) 选中 12 只：
10年地方债 bond 14.7%
30年国债 bond 14.2%
银行 industry 14.2%
5年地方债 bond 14.0%
红利低波 style 13.7%
质量 style 6.0%
标普500 us 5.5%
创业板 size 3.5%
科技 industry 2.9%
科创50 size 2.8%
黄金 commodity 2.6%
半导体 industry 2.6%
净值（消融对比）与回撤
python
x = pd.to_datetime(main_bt["nav"].index).to_numpy()
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(11, 6.4), sharex=True,
gridspec_kw={"height_ratios": [3, 1]})
for label, bt in results.items():
ax1.plot(x, bt["nav"].values, lw=1.6 if "动量选择" in label else 1.2, label=label)
ax1.plot(x, bench_nav.values, lw=1.0, alpha=0.7, label="沪深300买入持有")
ax1.set_title("动量+风险平价复合策略 净值（消融对比）"); ax1.set_ylabel("NAV")
ax1.legend(fontsize=8); ax1.grid(alpha=0.3)
nav = main_bt["nav"]; dd = nav / nav.cummax() - 1
ax2.fill_between(x, dd.values, 0, color="crimson", alpha=0.4)
ax2.set_ylabel("回撤"); ax2.grid(alpha=0.3)
plt.show()