全天候风险平价（ETF 资产配置）· notebook 版

复刻 QMT《全天候风险平价版策略V1》（公众号【量化君也】），用本地 tushare 前复权数据离线回测。

• 资产池：沪深300 / 中证500 / 标普500 / 纳指 ETF、5 年 / 10 年国债 ETF、黄金 ETF、商品 ETF
• 定权：每月第一个交易日，用过去 120 个交易日收益率协方差做风险平价（各资产风险贡献相等，SLSQP 求解，权重∈[0,1] 且和为 1）
• 替代逻辑：上市不足 1.5×120 天的品种用替代品种顶上（10 年国债→5 年国债，其余→上证50ETF）
• 成交：前复权价、按 100 份整手、ETF 单边佣金 0.03%、预留 2% 现金缓冲；基准为沪深300ETF 买入持有

仅供研究学习，不构成投资建议。脚本版见 scripts/run_allweather_riskparity.py 与复现说明 docs/replications/2026-06-15-allweather-risk-parity.md。

%matplotlib inline
import os
while not os.path.exists("pyproject.toml") and os.getcwd() != "/":
    os.chdir("..")
from datetime import timedelta
import numpy as np, 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.data.sources.tushare_fund import TushareFundSource          # 只取 listing_dates 助手
from quant.strategy.risk_parity import risk_parity_weights, month_first_trading_days
from quant.backtest.portfolio import run_rebalance_backtest
from quant.reports.metrics import (annualized_return, annualized_volatility,
                                   sharpe_ratio, max_drawdown, turnover)

ASSET_DICT = {"stock": ["510300.SH", "510500.SH", "513500.SH", "513100.SH"],
              "middle_bond": ["511010.SH"], "long_bond": ["511260.SH"],
              "gold": ["518880.SH"], "commodity": ["510170.SH"]}
REPLACE_DICT = {"511260.SH": "511010.SH", "default": "510050.SH"}
NAME_CN = {"510300.SH": "沪深300ETF", "510500.SH": "中证500ETF", "513500.SH": "标普500ETF",
           "513100.SH": "纳指ETF", "511010.SH": "5年国债ETF", "511260.SH": "10年国债ETF",
           "518880.SH": "黄金ETF", "510170.SH": "商品ETF", "510050.SH": "上证50ETF(替代)"}
COV_DAYS, START, END = 120, "20140201", "20260613"

long_df = pd.read_parquet("data/raw/allweather_etf_qfq.parquet")        # 读缓存,离线无需 token
listing = TushareFundSource.listing_dates(long_df)
close = long_df.pivot(index="datetime", columns="instrument", values="close").sort_index()
calendar = close.index[close["510300.SH"].notna()].tolist()             # 以沪深300交易日为主日历
close = close.reindex(calendar).ffill()
rb_dates = [d for d in month_first_trading_days(calendar) if START <= d <= END]
print("各品种数据起始:", {k: listing[k] for k in sorted(listing)})
print("回测 %s ~ %s, 月度调仓 %d 次" % (START, END, len(rb_dates)))

各品种数据起始: {'510050.SH': '20120104', '510170.SH': '20120104', '510300.SH': '20120528', '510500.SH': '20130315', '511010.SH': '20130325', '511260.SH': '20170824', '513100.SH': '20130515', '513500.SH': '20140115', '518880.SH': '20130729'}
回测 20140201 ~ 20260613, 月度调仓 149 次

调仓函数与回测

每个调仓日：先把未上市够久的品种替换成替代品种，取上一交易日往前 120 日收益率，做风险平价定权。prev = calendar[i-1] 保证用的是已知数据，无未来函数。

pos = {d: i for i, d in enumerate(calendar)}
cols = set(close.columns)

def resolve(s, today):
    """上市不足 1.5×120 天就用替代品种顶上（与原码一致）。"""
    ld = listing.get(s)
    if ld is not None:
        need_until = (pd.to_datetime(ld) + timedelta(days=round(1.5 * COV_DAYS))).strftime("%Y%m%d")
        if today >= need_until:
            return s
    code = REPLACE_DICT.get(s, REPLACE_DICT["default"])
    return code if code in cols else REPLACE_DICT["default"]

def weight_fn(today):
    i = pos[today]
    prev = calendar[i - 1] if i > 0 else today            # 用上一交易日及之前的数据定权
    trade_codes = sorted({resolve(s, today) for codes in ASSET_DICT.values() for s in codes})
    window = close.loc[:prev, trade_codes].ffill()
    rets = window.pct_change().dropna().tail(COV_DAYS)
    return risk_parity_weights(rets)

bt = run_rebalance_backtest(close, rb_dates, weight_fn,
                            account=1e6, invest_ratio=0.98, commission=0.0003)
nav, rets, weights = bt["nav"], bt["returns"], bt["weights"]
print("回测完成,", len(nav), "个交易日")

回测完成, 3003 个交易日

绩效

bench_px = close.loc[nav.index, "510300.SH"]
bench_ret = bench_px.pct_change()
bench_nav = bench_px / bench_px.iloc[0]

def perf_row(name, r):
    return {"策略": name, "年化收益": annualized_return(r), "年化波动": annualized_volatility(r),
            "夏普": sharpe_ratio(r), "最大回撤": max_drawdown(r)}

perf = pd.DataFrame([perf_row("全天候风险平价", rets),
                     perf_row("沪深300ETF买入持有", bench_ret)]).set_index("策略")
fmt = perf.copy()
for c in ["年化收益", "年化波动", "最大回撤"]:
    fmt[c] = fmt[c].map(lambda v: "%+.2f%%" % (v * 100))
fmt["夏普"] = fmt["夏普"].map(lambda v: "%.2f" % v)
print(fmt.to_string())
print("\n累计净值: 策略 %.3f | 沪深300 %.3f" % (nav.iloc[-1], bench_nav.iloc[-1]))
print("月均换手(Σ|Δw|): %.2f%%" % (turnover(weights) * 100))
print("\n最近一次调仓权重 (%s):" % weights.index[-1])
last_w = weights.iloc[-1]
for c in last_w[last_w > 1e-4].sort_values(ascending=False).index:
    print("  %-16s %5.1f%%" % (NAME_CN.get(c, c), last_w[c] * 100))

                年化收益     年化波动    夏普     最大回撤
策略                                          
全天候风险平价       +6.55%   +3.67%  1.75   -6.74%
沪深300ETF买入持有  +8.60%  +22.39%  0.48  -45.49%

累计净值: 策略 2.130 | 沪深300 2.672
月均换手(Σ|Δw|): 6.91%

最近一次调仓权重 (20260601):
  5年国债ETF           60.1%
  10年国债ETF          29.2%
  标普500ETF           1.9%
  沪深300ETF           1.9%
  纳指ETF              1.6%
  中证500ETF           1.2%
  商品ETF              1.0%
  黄金ETF              0.8%

净值与回撤

风险平价以“风险贡献相等”分散，债券权重通常显著高于股票，因而波动和回撤远低于宽基指数——用更平滑的净值换更高的风险调整后收益（夏普）。

nav_idx = pd.to_datetime(nav.index).to_numpy()
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(11, 6.4), sharex=True,
                               gridspec_kw={"height_ratios": [3, 1]})
ax1.plot(nav_idx, nav.values, label="全天候风险平价", lw=1.6)
ax1.plot(nav_idx, bench_nav.values, label="沪深300ETF买入持有", lw=1.2, alpha=0.8)
ax1.set_title("全天候风险平价 净值 vs 沪深300"); ax1.set_ylabel("NAV")
ax1.legend(); ax1.grid(alpha=0.3)
dd = nav / nav.cummax() - 1.0
ax2.fill_between(nav_idx, dd.values, 0, color="crimson", alpha=0.4)
ax2.set_ylabel("回撤"); ax2.grid(alpha=0.3)
plt.show()

各资产权重（风险平价，月度）

w_plot = weights.rename(columns=NAME_CN)
w_idx = pd.to_datetime(w_plot.index).to_numpy()
fig, ax = plt.subplots(figsize=(11, 4.6))
ax.stackplot(w_idx, *[w_plot[c].values for c in w_plot.columns], labels=list(w_plot.columns))
ax.set_title("各资产权重（风险平价，月度再平衡）"); ax.set_ylabel("权重"); ax.set_ylim(0, 1)
ax.legend(loc="upper left", fontsize=8, ncol=2); plt.show()