{ "cells": [ { "cell_type": "code", "execution_count": null, "id": "2cf0ca5c", "metadata": {}, "outputs": [], "source": [ "\"\"\"\n", "A 股行业 ETF 轮动策略 Demo\n", "A-Share Sector ETF Rotation Strategy Demo\n", "==========================================\n", "\n", "策略逻辑 (Strategy Logic):\n", " 1. 相对动量轮动 (Relative Momentum Rotation)\n", " - 每月对所有行业 ETF 计算过去 N 个月的动量得分\n", " - 买入得分最高的前 K 个 ETF，等权持有\n", " 2. 双动量策略 (Dual Momentum, Gary Antonacci)\n", " - 绝对动量 (Absolute Momentum): 若某 ETF 的动量 < 0，切换至国债 ETF 避险\n", " - 相对动量 (Relative Momentum): 在通过绝对动量过滤的 ETF 中，买入相对最强的\n", " 3. 等权基准 (Equal Weight Benchmark)\n", " - 始终等权持有全部行业 ETF，不做轮动\n", "\n", "成本模拟 (Cost Simulation):\n", " - 印花税 (Stamp Duty): 卖出 0.1%\n", " - 佣金 (Commission): 买卖各 0.03%（万三）\n", " - 滑点 (Slippage): 单边 0.05%\n", " - 每月换仓合计约 0.22% 双边成本\n", "\n", "作者: GitHub Copilot (教学用途)\n", "数据: 合成模拟数据 (Synthetic data for demonstration)\n", "\"\"\"\n", "\n", "# ── 标准库 ──────────────────────────────────────────────────────────────────\n", "import warnings\n", "warnings.filterwarnings(\"ignore\")\n", "\n", "# matplotlib 必须在 pyplot 之前设置后端 (must set backend before importing pyplot)\n", "import matplotlib\n", "matplotlib.use(\"Agg\") # 无界面后端 (headless backend)\n", "\n", "import numpy as np\n", "import pandas as pd\n", "import matplotlib.pyplot as plt\n", "import matplotlib.gridspec as gridspec\n", "from matplotlib.patches import FancyArrowPatch\n", "import matplotlib.ticker as mtick\n", "\n", "# 设置随机种子，保证结果可复现 (fix random seed for reproducibility)\n", "np.random.seed(42)" ] }, { "cell_type": "markdown", "id": "a6555659", "metadata": {}, "source": [ "# ══════════════════════════════════════════════════════════════════════════════\n", "# §0 ETF 池定义 & 合成价格数据生成\n", "# ETF Universe Definition & Synthetic Price Data Generation\n", "\n", "# ══════════════════════════════════════════════════════════════════════════════\n", "\n", "# ── A 股行业 ETF 池 (A-Share Sector ETF Universe) ───────────────────────────\n", "# 这里用合成数据模拟，真实代码只需替换为 AKShare/Tushare 数据读取\n", "# (Synthetic data here; replace with AKShare/Tushare data in production)" ] }, { "cell_type": "code", "execution_count": null, "id": "269510d5", "metadata": {}, "outputs": [], "source": [ "ETF_UNIVERSE = {\n", " # 代码 (Code) : (中文名称, 英文名称, 年化收益均值, 年化波动率, 与市场相关性)\n", " \"159995\": (\"芯片ETF\", \"Chip ETF\", 0.18, 0.45, 0.75),\n", " \"159819\": (\"人工智能ETF\",\"AI ETF\", 0.20, 0.48, 0.72),\n", " \"516160\": (\"新能源ETF\", \"New Energy ETF\", 0.15, 0.42, 0.70),\n", " \"159928\": (\"消费ETF\", \"Consumer ETF\", 0.10, 0.28, 0.65),\n", " \"512170\": (\"医疗ETF\", \"Healthcare ETF\", 0.08, 0.32, 0.60),\n", " \"512880\": (\"证券ETF\", \"Securities ETF\", 0.12, 0.38, 0.80),\n", " # 国债 ETF 作为避险资产 (Bond ETF as safe-haven asset)\n", " \"511010\": (\"国债ETF\", \"Bond ETF\", 0.03, 0.03, -0.10),\n", "}\n", "\n", "# 分离行业 ETF 和国债 ETF (separate sector ETFs from bond ETF)\n", "SECTOR_ETFS = [code for code in ETF_UNIVERSE if code != \"511010\"]\n", "BOND_ETF = \"511010\"\n", "ALL_ETFS = list(ETF_UNIVERSE.keys())\n", "\n", "# ── 生成合成价格序列 (Generate Synthetic Price Series) ───────────────────────\n", "# 用因子模型生成相关价格序列，模拟真实行业之间的联动关系\n", "# (Factor model to generate correlated price series, mimicking real sector co-movement)\n", "\n", "TRADING_DAYS = 252 # 年交易日数 (trading days per year)\n", "SIM_YEARS = 6 # 模拟年数 (simulation years): 2019-2024\n", "N_DAYS = TRADING_DAYS * SIM_YEARS\n", "\n", "# 市场公共因子 (market common factor) —— 模拟沪深300走势\n", "# 牛市/熊市分段：模拟 2019-2021 牛市，2021-2022 熊市，2023-2024 震荡\n", "market_factor = np.zeros(N_DAYS)\n", "# 第一阶段：2019-2021 牛市 (Bull market phase)\n", "bull1 = int(N_DAYS * 0.40)\n", "market_factor[:bull1] = np.random.normal(0.0008, 0.012, bull1)\n", "# 第二阶段：2021-2022 熊市 (Bear market phase)\n", "bear1 = int(N_DAYS * 0.20)\n", "market_factor[bull1:bull1+bear1] = np.random.normal(-0.0006, 0.018, bear1)\n", "# 第三阶段：2023-2024 震荡 (Sideways phase)\n", "market_factor[bull1+bear1:] = np.random.normal(0.0002, 0.014, N_DAYS - bull1 - bear1)\n", "\n", "# 生成每只 ETF 的日收益率 (generate daily returns for each ETF)\n", "daily_returns = {}\n", "for code, (cn_name, en_name, ann_ret, ann_vol, mkt_corr) in ETF_UNIVERSE.items():\n", " daily_mu = ann_ret / TRADING_DAYS # 日均收益 (daily mean return)\n", " daily_sig = ann_vol / np.sqrt(TRADING_DAYS) # 日波动率 (daily volatility)\n", "\n", " # 特异性收益 (idiosyncratic return) = 总收益 - 市场部分\n", " idio_sig = daily_sig * np.sqrt(1 - mkt_corr**2)\n", " idio = np.random.normal(0, idio_sig, N_DAYS)\n", "\n", " # 总日收益 = 市场因子 × 相关系数 + 特异性收益 + 均值漂移\n", " ret = mkt_corr * market_factor + idio + (daily_mu - mkt_corr * 0.0002)\n", " daily_returns[code] = ret\n", "\n", "# 转为 DataFrame，并生成价格序列 (convert to DataFrame, then price series)\n", "start_date = pd.Timestamp(\"2019-01-02\")\n", "dates = pd.bdate_range(start=start_date, periods=N_DAYS) # 仅工作日 (business days only)\n", "\n", "returns_df = pd.DataFrame(daily_returns, index=dates)\n", "# 价格从 1.0 开始（净值形式，Net Asset Value style）\n", "price_df = (1 + returns_df).cumprod()\n", "\n", "print(\"═\" * 60)\n", "print(\"§0 ETF 价格数据生成完成 (Price Data Generated)\")\n", "print(\"═\" * 60)\n", "print(f\" 模拟区间 (Simulation Period): {dates[0].date()} → {dates[-1].date()}\")\n", "print(f\" 交易日数 (Trading Days): {N_DAYS}\")\n", "print(f\" ETF 数量 (Number of ETFs): {len(ALL_ETFS)}\")\n", "print()\n", "print(\" ETF 池 (Universe):\")\n", "print(f\" {'代码':10s} {'中文名':12s} {'英文名':18s} {'年化收益':8s} {'年化波动':8s}\")\n", "print(\" \" + \"-\" * 60)\n", "for code, (cn, en, ret, vol, _) in ETF_UNIVERSE.items():\n", " label = \" [避险]\" if code == BOND_ETF else \"\"\n", " print(f\" {code:10s} {cn:12s} {en:18s} {ret*100:6.1f}% {vol*100:6.1f}%{label}\")\n", "print()" ] }, { "cell_type": "markdown", "id": "7e083547", "metadata": {}, "source": [ "# ══════════════════════════════════════════════════════════════════════════════\n", "# §1 动量因子计算\n", "# Momentum Factor Calculation\n", "\n", "# ══════════════════════════════════════════════════════════════════════════════" ] }, { "cell_type": "code", "execution_count": null, "id": "e01ab734", "metadata": {}, "outputs": [], "source": [ "def calc_momentum(price_df: pd.DataFrame,\n", " lookback_months: int = 12,\n", " skip_months: int = 1) -> pd.DataFrame:\n", " \"\"\"\n", " 计算每只 ETF 的动量得分 (Calculate momentum score for each ETF)\n", "\n", " 经典动量定义 (Classic momentum definition):\n", " 过去 L 个月的总收益，跳过最近 S 个月（避免短期反转）\n", " Total return over past L months, skipping most recent S months\n", " (to avoid short-term reversal effect 短期反转效应)\n", "\n", " 参数 (Parameters):\n", " price_df : 日频价格 DataFrame (daily price DataFrame)\n", " lookback_months : 回望期（月数）(lookback period in months)\n", " skip_months : 跳过最近月数（通常为 1）(skip recent months, usually 1)\n", "\n", " 返回 (Returns):\n", " 月频动量得分 DataFrame (monthly momentum score DataFrame)\n", " \"\"\"\n", " # 转为月末价格 (resample to month-end prices)\n", " # 注意：旧版 pandas 用 \"M\"，新版用 \"ME\"\n", " try:\n", " monthly_price = price_df.resample(\"ME\").last()\n", " except ValueError:\n", " monthly_price = price_df.resample(\"M\").last()\n", "\n", " # 动量 = 前 (lookback+skip) 月价格 / 前 skip 月价格 - 1\n", " # Momentum = Price[t - skip] / Price[t - lookback - skip] - 1\n", " start_lag = skip_months\n", " end_lag = lookback_months + skip_months\n", "\n", " momentum = monthly_price.shift(start_lag) / monthly_price.shift(end_lag) - 1\n", " return momentum\n", "\n", "\n", "# 计算三种不同回望期的动量（稳健性检验用）\n", "# (Calculate momentum with three different lookback windows for robustness)\n", "mom_12_1 = calc_momentum(price_df, lookback_months=12, skip_months=1) # 经典 12-1 月动量\n", "mom_6_1 = calc_momentum(price_df, lookback_months=6, skip_months=1) # 中期 6-1 月动量\n", "mom_3_1 = calc_momentum(price_df, lookback_months=3, skip_months=1) # 短期 3-1 月动量\n", "\n", "# 合成动量得分：三个窗口等权平均（减少单一窗口过拟合风险）\n", "# (Composite momentum score: equal-weight average of three windows)\n", "mom_composite = (mom_12_1 + mom_6_1 + mom_3_1) / 3\n", "\n", "print(\"§1 动量因子计算完成 (Momentum Factors Calculated)\")\n", "print(f\" 月度数据行数 (Monthly rows): {len(mom_12_1)}\")\n", "print()" ] }, { "cell_type": "markdown", "id": "a098f350", "metadata": {}, "source": [ "# ══════════════════════════════════════════════════════════════════════════════\n", "# §2 A 股交易成本模型\n", "# A-Share Transaction Cost Model\n", "\n", "# ══════════════════════════════════════════════════════════════════════════════" ] }, { "cell_type": "code", "execution_count": null, "id": "c18a15bd", "metadata": {}, "outputs": [], "source": [ "class AShareCostModel:\n", " \"\"\"\n", " A 股 ETF 交易成本模型 (A-Share ETF Transaction Cost Model)\n", "\n", " A 股 ETF 特殊规则 (A-Share ETF Special Rules):\n", " - ETF 可 T+0 交易（当天买当天可卖）Unlike 个股的 T+1 限制\n", " (ETF allows T+0 trading, unlike individual stocks which are T+1)\n", " - 印花税：仅卖出方向收取 0.1%\n", " (Stamp duty: 0.1% on SELL side only)\n", " - 佣金：买卖双向，通常 0.02% - 0.03%\n", " (Commission: both sides, typically 0.02%-0.03%)\n", " - 滑点：ETF 流动性好，通常 0.02%-0.05%\n", " (Slippage: ETF has good liquidity, typically 0.02%-0.05%)\n", " \"\"\"\n", " def __init__(self,\n", " commission_rate: float = 0.0003, # 佣金率 (commission rate)\n", " stamp_duty_rate: float = 0.001, # 印花税率 (stamp duty rate)\n", " slippage_rate: float = 0.0005): # 滑点率 (slippage rate)\n", " self.commission = commission_rate\n", " self.stamp_duty = stamp_duty_rate\n", " self.slippage = slippage_rate\n", "\n", " def buy_cost(self, trade_value: float) -> float:\n", " \"\"\"买入成本 (buy-side cost): 佣金 + 滑点\"\"\"\n", " return trade_value * (self.commission + self.slippage)\n", "\n", " def sell_cost(self, trade_value: float) -> float:\n", " \"\"\"卖出成本 (sell-side cost): 佣金 + 印花税 + 滑点\"\"\"\n", " return trade_value * (self.commission + self.stamp_duty + self.slippage)\n", "\n", " def roundtrip_cost(self, trade_value: float) -> float:\n", " \"\"\"\n", " 双边成本 (round-trip cost): 买入 + 卖出\n", " 月度换仓一次的总成本约 0.22%\n", " \"\"\"\n", " return self.buy_cost(trade_value) + self.sell_cost(trade_value)\n", "\n", " @property\n", " def roundtrip_rate(self) -> float:\n", " \"\"\"双边成本率 (round-trip cost rate)\"\"\"\n", " return 2 * self.commission + self.stamp_duty + 2 * self.slippage\n", "\n", "\n", "cost_model = AShareCostModel()\n", "print(\"§2 交易成本模型 (Transaction Cost Model):\")\n", "print(f\" 佣金率 (Commission): {cost_model.commission*100:.3f}% × 双边\")\n", "print(f\" 印花税 (Stamp Duty): {cost_model.stamp_duty*100:.3f}% × 卖出方向\")\n", "print(f\" 滑点率 (Slippage): {cost_model.slippage*100:.3f}% × 双边\")\n", "print(f\" 双边总成本 (Roundtrip): {cost_model.roundtrip_rate*100:.3f}%\")\n", "print()" ] }, { "cell_type": "markdown", "id": "ac6f0f41", "metadata": {}, "source": [ "# ══════════════════════════════════════════════════════════════════════════════\n", "# §3 轮动回测引擎\n", "# Rotation Backtest Engine\n", "\n", "# ══════════════════════════════════════════════════════════════════════════════" ] }, { "cell_type": "code", "execution_count": null, "id": "c6148ff2", "metadata": {}, "outputs": [], "source": [ "def run_rotation_backtest(\n", " price_df: pd.DataFrame,\n", " momentum_df: pd.DataFrame,\n", " strategy: str = \"relative_momentum\",\n", " top_k: int = 3,\n", " cost_model: AShareCostModel = None,\n", " bond_etf: str = BOND_ETF,\n", " sector_etfs: list = None,\n", ") -> dict:\n", " \"\"\"\n", " 行业 ETF 轮动回测引擎 (Sector ETF Rotation Backtest Engine)\n", "\n", " 参数 (Parameters):\n", " price_df : 日频价格 (daily prices)\n", " momentum_df : 月频动量得分 (monthly momentum scores)\n", " strategy : 策略名称 (strategy name)\n", " \"relative_momentum\" — 相对动量轮动\n", " \"dual_momentum\" — 双动量（含绝对动量过滤）\n", " \"equal_weight\" — 等权基准\n", " top_k : 持有前 K 只 ETF (hold top-K ETFs)\n", " cost_model : 交易成本模型 (transaction cost model)\n", " bond_etf : 避险 ETF 代码 (safe-haven ETF code)\n", " sector_etfs : 行业 ETF 代码列表 (sector ETF codes)\n", "\n", " 返回 (Returns):\n", " dict 包含:\n", " \"nav\" : 每日净值序列 (daily NAV series)\n", " \"weights\" : 月度持仓权重 (monthly weights)\n", " \"turnover\" : 月度换手率 (monthly turnover)\n", " \"cost_total\" : 总交易成本 (total transaction cost)\n", " \"holdings_log\": 持仓记录 (holdings log)\n", " \"\"\"\n", " if sector_etfs is None:\n", " sector_etfs = SECTOR_ETFS\n", " if cost_model is None:\n", " cost_model = AShareCostModel()\n", "\n", " # 获取月末再平衡日期 (get month-end rebalancing dates)\n", " try:\n", " monthly_idx = price_df.resample(\"ME\").last().index\n", " except ValueError:\n", " monthly_idx = price_df.resample(\"M\").last().index\n", "\n", " # 预热期：动量需要至少 13 个月数据（12 个月回望 + 1 个月跳过）\n", " # (Warm-up period: momentum needs at least 13 months of data)\n", " warmup = 13\n", "\n", " # 初始化 (initialization)\n", " nav = pd.Series(index=price_df.index, dtype=float)\n", " weights_log = {} # 每月权重记录 (monthly weights log)\n", " turnover_log = {} # 每月换手率记录 (monthly turnover log)\n", " holdings_log = {} # 每月持仓记录 (monthly holdings log)\n", " total_cost = 0.0 # 累计交易成本 (cumulative transaction cost)\n", "\n", " current_weights = {} # 当前持仓权重 {code: weight}\n", " portfolio_value = 1.0 # 组合净值，从 1.0 开始 (portfolio starts at 1.0)\n", "\n", " rebal_dates = monthly_idx[warmup:] # 有效再平衡日期 (valid rebalance dates)\n", "\n", " for i, rebal_date in enumerate(rebal_dates):\n", " # ── 1. 确定目标权重 (Determine target weights) ──────────────────────\n", "\n", " if strategy == \"equal_weight\":\n", " # 等权：始终持有所有行业 ETF，各占 1/N\n", " # (Equal weight: always hold all sector ETFs equally)\n", " n = len(sector_etfs)\n", " target_weights = {code: 1.0 / n for code in sector_etfs}\n", "\n", " elif strategy == \"relative_momentum\":\n", " # 相对动量：选动量得分最高的前 K 只行业 ETF\n", " # (Relative momentum: select top-K sector ETFs by momentum score)\n", " mom_today = momentum_df.loc[:rebal_date, sector_etfs].iloc[-1]\n", " mom_today = mom_today.dropna()\n", " if len(mom_today) == 0:\n", " target_weights = current_weights or {sector_etfs[0]: 1.0}\n", " else:\n", " top_etfs = mom_today.nlargest(top_k).index.tolist()\n", " w = 1.0 / len(top_etfs)\n", " target_weights = {code: w for code in top_etfs}\n", "\n", " elif strategy == \"dual_momentum\":\n", " # 双动量：先用绝对动量过滤，再做相对动量排序\n", " # (Dual momentum: filter by absolute momentum, then rank by relative)\n", " mom_today = momentum_df.loc[:rebal_date, sector_etfs].iloc[-1]\n", " mom_today = mom_today.dropna()\n", "\n", " # 绝对动量过滤 (absolute momentum filter):\n", " # 如果行业 ETF 动量 < 0（跑输无风险），则替换为国债 ETF\n", " # (If sector ETF momentum < 0, replace with bond ETF)\n", " positive_etfs = mom_today[mom_today > 0].index.tolist()\n", "\n", " if len(positive_etfs) == 0:\n", " # 全部动量为负：100% 持有国债 ETF 避险\n", " # (All negative momentum: 100% in bond ETF)\n", " target_weights = {bond_etf: 1.0}\n", " else:\n", " # 从通过绝对动量过滤的 ETF 中，选相对最强的前 K 只\n", " # (From ETFs passing absolute momentum filter, pick top-K by relative)\n", " k_actual = min(top_k, len(positive_etfs))\n", " top_etfs = mom_today[positive_etfs].nlargest(k_actual).index.tolist()\n", " w = 1.0 / k_actual\n", " target_weights = {code: w for code in top_etfs}\n", " else:\n", " raise ValueError(f\"未知策略 (Unknown strategy): {strategy}\")\n", "\n", " # ── 2. 计算换手率并扣除交易成本 (Calculate turnover and deduct costs) ──\n", "\n", " # 所有涉及的 ETF 代码（新旧持仓并集）\n", " all_codes = set(current_weights) | set(target_weights)\n", " turnover = 0.0\n", "\n", " for code in all_codes:\n", " old_w = current_weights.get(code, 0.0)\n", " new_w = target_weights.get(code, 0.0)\n", " delta = abs(new_w - old_w) # 权重变化 (weight change)\n", " if delta > 1e-6:\n", " turnover += delta\n", " # 对权重变化部分计算成本 (cost on the traded portion)\n", " trade_value = delta * portfolio_value\n", " if new_w > old_w:\n", " # 增仓 → 买入成本 (increase position → buy cost)\n", " cost = cost_model.buy_cost(trade_value)\n", " else:\n", " # 减仓 → 卖出成本（含印花税）\n", " # (decrease position → sell cost including stamp duty)\n", " cost = cost_model.sell_cost(trade_value)\n", " total_cost += cost\n", " portfolio_value -= cost # 成本从组合价值中扣除 (deduct cost from portfolio)\n", "\n", " # 换手率（单边）= 买入总金额 / 组合价值 (one-sided turnover)\n", " turnover_log[rebal_date] = turnover / 2 # 双边换手/2 = 单边换手\n", "\n", " # ── 3. 更新持仓，计算到下一个再平衡日的收益\n", " # (Update holdings, compute returns until next rebalance date)\n", " current_weights = target_weights\n", " weights_log[rebal_date] = target_weights.copy()\n", " holdings_log[rebal_date] = list(target_weights.keys())\n", "\n", " # 确定持仓持续的日期区间 (determine holding period date range)\n", " if i + 1 < len(rebal_dates):\n", " next_rebal = rebal_dates[i + 1]\n", " else:\n", " next_rebal = price_df.index[-1]\n", "\n", " hold_dates = price_df.loc[rebal_date:next_rebal].index\n", "\n", " # 计算每日组合收益率 (calculate daily portfolio return)\n", " for j in range(1, len(hold_dates)):\n", " d_prev = hold_dates[j - 1]\n", " d_curr = hold_dates[j]\n", " daily_port_ret = 0.0\n", " for code, w in current_weights.items():\n", " if code in price_df.columns:\n", " p_prev = price_df.loc[d_prev, code]\n", " p_curr = price_df.loc[d_curr, code]\n", " if p_prev > 0:\n", " daily_port_ret += w * (p_curr / p_prev - 1)\n", " portfolio_value *= (1 + daily_port_ret)\n", " nav[d_curr] = portfolio_value\n", "\n", " # 补全初始 NAV（热身期）\n", " nav = nav.ffill()\n", " first_valid = nav.first_valid_index()\n", " if first_valid is not None:\n", " nav[:first_valid] = 1.0\n", " nav = nav.fillna(1.0)\n", "\n", " return {\n", " \"nav\": nav,\n", " \"weights\": weights_log,\n", " \"turnover\": pd.Series(turnover_log),\n", " \"cost_total\": total_cost,\n", " \"holdings_log\": holdings_log,\n", " }" ] }, { "cell_type": "markdown", "id": "12233425", "metadata": {}, "source": [ "# ══════════════════════════════════════════════════════════════════════════════\n", "# §4 运行三种策略并计算绩效指标\n", "# Run Three Strategies and Calculate Performance Metrics\n", "\n", "# ══════════════════════════════════════════════════════════════════════════════" ] }, { "cell_type": "code", "execution_count": null, "id": "5def8b15", "metadata": {}, "outputs": [], "source": [ "print(\"§4 运行回测策略 (Running Backtest Strategies)...\")\n", "print()\n", "\n", "# 运行三种策略 (run three strategies)\n", "results = {}\n", "\n", "results[\"equal_weight\"] = run_rotation_backtest(\n", " price_df, mom_composite,\n", " strategy=\"equal_weight\",\n", " top_k=3, cost_model=cost_model,\n", ")\n", "\n", "results[\"relative_momentum\"] = run_rotation_backtest(\n", " price_df, mom_composite,\n", " strategy=\"relative_momentum\",\n", " top_k=3, cost_model=cost_model,\n", ")\n", "\n", "results[\"dual_momentum\"] = run_rotation_backtest(\n", " price_df, mom_composite,\n", " strategy=\"dual_momentum\",\n", " top_k=3, cost_model=cost_model,\n", ")\n", "\n", "# 同时计算国债 ETF 纯避险基准 (bond ETF as pure safe-haven benchmark)\n", "bond_nav = price_df[BOND_ETF] / price_df[BOND_ETF].iloc[0]\n", "results[\"bond_only\"] = {\"nav\": bond_nav}\n", "\n", "# ── 绩效计算函数 (Performance Calculation Function) ─────────────────────────\n", "\n", "def calc_performance(nav: pd.Series,\n", " risk_free_annual: float = 0.02) -> dict:\n", " \"\"\"\n", " 计算常用量化绩效指标 (Calculate common quantitative performance metrics)\n", "\n", " 指标 (Metrics):\n", " - 年化收益率 (Annualized Return, CAGR)\n", " - 年化波动率 (Annualized Volatility)\n", " - 夏普比率 (Sharpe Ratio): 超额收益 / 波动率\n", " - 最大回撤 (Maximum Drawdown, MaxDD): 从历史最高点的最大跌幅\n", " - 卡玛比率 (Calmar Ratio): 年化收益 / 最大回撤\n", " - 索提诺比率 (Sortino Ratio): 用下行波动率替代总波动率\n", " \"\"\"\n", " nav = nav.dropna()\n", " if len(nav) < 2:\n", " return {}\n", "\n", " # 日收益率 (daily returns)\n", " daily_ret = nav.pct_change().dropna()\n", "\n", " # 年化收益率 (CAGR - Compound Annual Growth Rate 复合年增长率)\n", " n_years = len(daily_ret) / TRADING_DAYS\n", " total_ret = nav.iloc[-1] / nav.iloc[0] - 1\n", " cagr = (1 + total_ret) ** (1 / n_years) - 1\n", "\n", " # 年化波动率 (annualized volatility)\n", " ann_vol = daily_ret.std() * np.sqrt(TRADING_DAYS)\n", "\n", " # 夏普比率 (Sharpe Ratio)\n", " # 公式: (年化收益 - 无风险利率) / 年化波动率\n", " rf_daily = risk_free_annual / TRADING_DAYS\n", " sharpe = (daily_ret.mean() - rf_daily) / daily_ret.std() * np.sqrt(TRADING_DAYS)\n", "\n", " # 最大回撤 (Maximum Drawdown)\n", " # 定义: max(累计最高净值 - 当日净值) / 累计最高净值\n", " rolling_max = nav.cummax() # 历史最高净值 (rolling maximum)\n", " drawdown = nav / rolling_max - 1 # 每日回撤序列 (daily drawdown series)\n", " max_dd = drawdown.min() # 最大回撤（负数）(max drawdown, negative)\n", "\n", " # 卡玛比率 (Calmar Ratio = CAGR / |MaxDD|)\n", " calmar = cagr / abs(max_dd) if max_dd != 0 else np.nan\n", "\n", " # 索提诺比率 (Sortino Ratio): 只惩罚下行波动 (penalizes only downside volatility)\n", " downside_ret = daily_ret[daily_ret < rf_daily]\n", " downside_vol = downside_ret.std() * np.sqrt(TRADING_DAYS) if len(downside_ret) > 0 else ann_vol\n", " sortino = (cagr - risk_free_annual) / downside_vol if downside_vol > 0 else np.nan\n", "\n", " # 胜率 (Win Rate): 月度正收益比例 (fraction of months with positive return)\n", " try:\n", " monthly_ret = nav.resample(\"ME\").last().pct_change().dropna()\n", " except ValueError:\n", " monthly_ret = nav.resample(\"M\").last().pct_change().dropna()\n", " win_rate = (monthly_ret > 0).mean()\n", "\n", " return {\n", " \"年化收益率 (CAGR)\": cagr,\n", " \"年化波动率 (Volatility)\": ann_vol,\n", " \"夏普比率 (Sharpe)\": sharpe,\n", " \"最大回撤 (MaxDrawdown)\": max_dd,\n", " \"卡玛比率 (Calmar)\": calmar,\n", " \"索提诺比率 (Sortino)\": sortino,\n", " \"月度胜率 (Win Rate)\": win_rate,\n", " \"累计收益率 (Total Return)\": total_ret,\n", " }\n", "\n", "\n", "# ── 打印绩效对比表 (Print Performance Comparison Table) ─────────────────────\n", "\n", "strategy_labels = {\n", " \"equal_weight\": \"等权基准 (EW Benchmark)\",\n", " \"relative_momentum\": \"相对动量 (Rel Momentum)\",\n", " \"dual_momentum\": \"双动量 (Dual Momentum)\",\n", " \"bond_only\": \"国债避险 (Bond Only)\",\n", "}\n", "\n", "print(\"=\" * 72)\n", "print(\"§4 策略绩效对比 (Strategy Performance Comparison)\")\n", "print(\"=\" * 72)\n", "\n", "perf_all = {}\n", "metrics_display = [\n", " (\"年化收益率 (CAGR)\", \"{:>8.2%}\"),\n", " (\"年化波动率 (Volatility)\", \"{:>8.2%}\"),\n", " (\"夏普比率 (Sharpe)\", \"{:>8.3f}\"),\n", " (\"最大回撤 (MaxDrawdown)\", \"{:>8.2%}\"),\n", " (\"卡玛比率 (Calmar)\", \"{:>8.3f}\"),\n", " (\"月度胜率 (Win Rate)\", \"{:>8.2%}\"),\n", " (\"累计收益率 (Total Return)\", \"{:>8.2%}\"),\n", "]\n", "\n", "for key, label in strategy_labels.items():\n", " nav = results[key][\"nav\"]\n", " perf = calc_performance(nav)\n", " perf_all[key] = perf\n", "\n", "header = f\" {'指标':26s}\"\n", "for label in strategy_labels.values():\n", " header += f\" {label[:20]:>20s}\"\n", "print(header)\n", "print(\" \" + \"-\" * 90)\n", "\n", "for metric, fmt in metrics_display:\n", " row = f\" {metric:26s}\"\n", " for key in strategy_labels:\n", " val = perf_all[key].get(metric, np.nan)\n", " row += \" \" + fmt.format(val).rjust(20)\n", " print(row)\n", "\n", "print()\n", "# 打印成本统计\n", "for key in [\"equal_weight\", \"relative_momentum\", \"dual_momentum\"]:\n", " cost = results[key].get(\"cost_total\", 0)\n", " to = results[key].get(\"turnover\", pd.Series()).mean()\n", " print(f\" [{strategy_labels[key][:16]}] \"\n", " f\"总成本={cost*100:.3f}%净值 平均月换手={to*100:.1f}%\")\n", "print()" ] }, { "cell_type": "markdown", "id": "45bf4a54", "metadata": {}, "source": [ "# ══════════════════════════════════════════════════════════════════════════════\n", "# §5 持仓分析：统计各行业 ETF 的被选中频次\n", "# Holdings Analysis: Count How Often Each Sector ETF Was Selected\n", "\n", "# ══════════════════════════════════════════════════════════════════════════════" ] }, { "cell_type": "code", "execution_count": null, "id": "19761832", "metadata": {}, "outputs": [], "source": [ "print(\"§5 持仓分析 (Holdings Analysis):\")\n", "print()\n", "\n", "for strategy_key in [\"relative_momentum\", \"dual_momentum\"]:\n", " holdings_log = results[strategy_key][\"holdings_log\"]\n", " count = {code: 0 for code in ALL_ETFS}\n", "\n", " for date, held in holdings_log.items():\n", " for code in held:\n", " count[code] = count.get(code, 0) + 1\n", "\n", " total_months = len(holdings_log)\n", " label = strategy_labels[strategy_key]\n", " print(f\" [{label}] 共 {total_months} 次再平衡:\")\n", "\n", " # 按持有频次降序排列 (sort by holding frequency descending)\n", " sorted_count = sorted(count.items(), key=lambda x: x[1], reverse=True)\n", " for code, cnt in sorted_count:\n", " if cnt > 0:\n", " cn_name = ETF_UNIVERSE[code][0]\n", " pct = cnt / total_months\n", " bar = \"█\" * int(pct * 20)\n", " print(f\" {code} {cn_name:10s} {cnt:3d}次 {pct:5.1%} {bar}\")\n", " print()" ] }, { "cell_type": "markdown", "id": "659ce25c", "metadata": {}, "source": [ "# ══════════════════════════════════════════════════════════════════════════════\n", "# §6 可视化：9 宫格图表\n", "# Visualization: 9-Panel Chart\n", "\n", "# ══════════════════════════════════════════════════════════════════════════════" ] }, { "cell_type": "code", "execution_count": null, "id": "da21c149", "metadata": {}, "outputs": [], "source": [ "print(\"§6 生成可视化图表 (Generating Charts)...\")\n", "\n", "# 配置中文字体 (configure Chinese font)\n", "plt.rcParams[\"font.sans-serif\"] = [\"WenQuanYi Zen Hei\", \"Arial Unicode MS\",\n", " \"SimHei\", \"DejaVu Sans\"]\n", "plt.rcParams[\"axes.unicode_minus\"] = False\n", "\n", "# 颜色方案 (color scheme)\n", "COLORS = {\n", " \"equal_weight\": \"#95a5a6\", # 灰色（基准）\n", " \"relative_momentum\": \"#2ecc71\", # 绿色\n", " \"dual_momentum\": \"#e74c3c\", # 红色（主策略）\n", " \"bond_only\": \"#3498db\", # 蓝色（国债）\n", "}\n", "LABELS = {\n", " \"equal_weight\": \"等权基准\",\n", " \"relative_momentum\": \"相对动量\",\n", " \"dual_momentum\": \"双动量\",\n", " \"bond_only\": \"国债\",\n", "}\n", "\n", "fig = plt.figure(figsize=(20, 16), facecolor=\"#1a1a2e\")\n", "fig.suptitle(\n", " \"A 股行业 ETF 轮动策略回测 (A-Share Sector ETF Rotation Backtest)\\n\"\n", " \"合成模拟数据 2019-2024 (Synthetic Data 2019-2024)\",\n", " fontsize=16, color=\"white\", fontweight=\"bold\", y=0.98\n", ")\n", "\n", "gs = gridspec.GridSpec(3, 3, figure=fig,\n", " hspace=0.42, wspace=0.35,\n", " left=0.07, right=0.97, top=0.93, bottom=0.06)\n", "\n", "ax_style = dict(facecolor=\"#16213e\", labelcolor=\"white\",\n", " tick_params=dict(colors=\"white\"))\n", "\n", "def style_ax(ax, title, xlabel=\"\", ylabel=\"\"):\n", " ax.set_facecolor(\"#16213e\")\n", " ax.set_title(title, color=\"white\", fontsize=10, fontweight=\"bold\", pad=6)\n", " ax.set_xlabel(xlabel, color=\"#aaaaaa\", fontsize=8)\n", " ax.set_ylabel(ylabel, color=\"#aaaaaa\", fontsize=8)\n", " ax.tick_params(colors=\"white\", labelsize=7)\n", " ax.spines[:].set_color(\"#444466\")\n", " ax.grid(True, alpha=0.25, color=\"#666699\", linewidth=0.5)\n", "\n", "# ── 图1: 累计净值曲线 (Cumulative NAV) ───────────────────────────────────────\n", "ax1 = fig.add_subplot(gs[0, :2]) # 跨两列（宽幅）\n", "style_ax(ax1, \"① 累计净值对比 Cumulative NAV Comparison\", ylabel=\"净值 NAV\")\n", "for key, color in COLORS.items():\n", " nav = results[key][\"nav\"]\n", " ax1.plot(nav.index, nav.values, color=color, linewidth=1.8,\n", " label=LABELS[key], alpha=0.9)\n", "ax1.axhline(1.0, color=\"#888888\", linewidth=0.8, linestyle=\"--\", alpha=0.6)\n", "ax1.legend(loc=\"upper left\", fontsize=8, facecolor=\"#1a1a2e\",\n", " labelcolor=\"white\", framealpha=0.7)\n", "ax1.yaxis.set_major_formatter(mtick.FuncFormatter(lambda x, _: f\"{x:.1f}x\"))\n", "\n", "# ── 图2: 绩效雷达图 (Performance Radar) ─────────────────────────────────────\n", "ax2 = fig.add_subplot(gs[0, 2], projection=\"polar\")\n", "ax2.set_facecolor(\"#16213e\")\n", "ax2.set_title(\"② 绩效雷达图 Performance Radar\",\n", " color=\"white\", fontsize=10, fontweight=\"bold\", pad=15)\n", "ax2.tick_params(colors=\"white\", labelsize=7)\n", "ax2.spines[\"polar\"].set_color(\"#444466\")\n", "\n", "radar_metrics = [\"年化收益\", \"夏普比率\", \"月胜率\", \"抗回撤\"]\n", "radar_keys = [\"年化收益率 (CAGR)\", \"夏普比率 (Sharpe)\",\n", " \"月度胜率 (Win Rate)\", \"卡玛比率 (Calmar)\"]\n", "n_radar = len(radar_metrics)\n", "angles = np.linspace(0, 2 * np.pi, n_radar, endpoint=False).tolist()\n", "angles += angles[:1]\n", "\n", "# 归一化各指标到 0-1 (normalize metrics to 0-1)\n", "def normalize_radar(values, min_val=0.0, max_val=1.0):\n", " rng = max_val - min_val\n", " return [(v - min_val) / rng if rng > 0 else 0.5 for v in values]\n", "\n", "radar_strats = [\"equal_weight\", \"relative_momentum\", \"dual_momentum\"]\n", "for rkey in radar_strats:\n", " perf = perf_all[rkey]\n", " raw = [perf.get(m, 0) for m in radar_keys]\n", " # 简单归一化（基于各策略范围）\n", " vals = [\n", " min(max(raw[0] / 0.30, 0), 1), # CAGR: 0~30%\n", " min(max(raw[1] / 3.0, 0), 1), # Sharpe: 0~3\n", " min(max(raw[2], 0), 1), # Win rate: 0~1\n", " min(max(raw[3] / 3.0, 0), 1), # Calmar: 0~3\n", " ]\n", " vals += vals[:1]\n", " ax2.plot(angles, vals, color=COLORS[rkey], linewidth=2, label=LABELS[rkey])\n", " ax2.fill(angles, vals, color=COLORS[rkey], alpha=0.15)\n", "\n", "ax2.set_xticks(angles[:-1])\n", "ax2.set_xticklabels(radar_metrics, color=\"white\", fontsize=8)\n", "ax2.set_yticklabels([])\n", "ax2.legend(loc=\"upper right\", bbox_to_anchor=(1.35, 1.15),\n", " fontsize=7, facecolor=\"#1a1a2e\", labelcolor=\"white\")\n", "\n", "# ── 图3: 最大回撤曲线 (Drawdown Curve) ──────────────────────────────────────\n", "ax3 = fig.add_subplot(gs[1, :2])\n", "style_ax(ax3, \"③ 水下曲线（回撤） Underwater Chart (Drawdown)\", ylabel=\"回撤 Drawdown\")\n", "for key, color in COLORS.items():\n", " nav = results[key][\"nav\"]\n", " dd = nav / nav.cummax() - 1\n", " ax3.fill_between(dd.index, dd.values, 0, color=color, alpha=0.4, label=LABELS[key])\n", " ax3.plot(dd.index, dd.values, color=color, linewidth=0.8, alpha=0.8)\n", "ax3.set_ylim(-0.65, 0.05)\n", "ax3.yaxis.set_major_formatter(mtick.PercentFormatter(1.0))\n", "ax3.legend(loc=\"lower left\", fontsize=8, facecolor=\"#1a1a2e\",\n", " labelcolor=\"white\", framealpha=0.7)\n", "\n", "# ── 图4: 绩效指标条形图 (Performance Bar Chart) ──────────────────────────────\n", "ax4 = fig.add_subplot(gs[1, 2])\n", "style_ax(ax4, \"④ 夏普 & 卡玛 Sharpe & Calmar\", ylabel=\"比率 Ratio\")\n", "strats_bar = [\"equal_weight\", \"relative_momentum\", \"dual_momentum\"]\n", "labels_bar = [LABELS[k] for k in strats_bar]\n", "sharpes = [perf_all[k][\"夏普比率 (Sharpe)\"] for k in strats_bar]\n", "calmars = [perf_all[k][\"卡玛比率 (Calmar)\"] for k in strats_bar]\n", "x_bar = np.arange(len(strats_bar))\n", "w_bar = 0.35\n", "bars1 = ax4.bar(x_bar - w_bar/2, sharpes, w_bar, color=\"#2ecc71\", alpha=0.8,\n", " label=\"夏普 Sharpe\")\n", "bars2 = ax4.bar(x_bar + w_bar/2, calmars, w_bar, color=\"#e74c3c\", alpha=0.8,\n", " label=\"卡玛 Calmar\")\n", "ax4.set_xticks(x_bar)\n", "ax4.set_xticklabels(labels_bar, fontsize=7, color=\"white\")\n", "ax4.legend(fontsize=8, facecolor=\"#1a1a2e\", labelcolor=\"white\")\n", "for bar in bars1:\n", " ax4.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 0.02,\n", " f\"{bar.get_height():.2f}\", ha=\"center\", va=\"bottom\",\n", " color=\"white\", fontsize=7)\n", "for bar in bars2:\n", " ax4.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 0.02,\n", " f\"{bar.get_height():.2f}\", ha=\"center\", va=\"bottom\",\n", " color=\"white\", fontsize=7)\n", "\n", "# ── 图5: 双动量月度持仓热力图 (Dual Momentum Monthly Holdings Heatmap) ────────\n", "ax5 = fig.add_subplot(gs[2, :2])\n", "style_ax(ax5, \"⑤ 双动量月度持仓热力图 Dual Momentum Monthly Holdings\")\n", "\n", "holdings_log = results[\"dual_momentum\"][\"holdings_log\"]\n", "all_codes_sorted = SECTOR_ETFS + [BOND_ETF]\n", "etf_names = [ETF_UNIVERSE[c][0] for c in all_codes_sorted]\n", "\n", "heatmap_dates = sorted(holdings_log.keys())\n", "heatmap_data = np.zeros((len(all_codes_sorted), len(heatmap_dates)))\n", "for j, d in enumerate(heatmap_dates):\n", " held = holdings_log[d]\n", " for i, code in enumerate(all_codes_sorted):\n", " if code in held:\n", " heatmap_data[i, j] = 1.0 / len(held) # 持仓权重 (holding weight)\n", "\n", "im = ax5.imshow(heatmap_data, aspect=\"auto\", cmap=\"YlOrRd\",\n", " interpolation=\"nearest\", vmin=0, vmax=0.5)\n", "ax5.set_yticks(range(len(all_codes_sorted)))\n", "ax5.set_yticklabels(etf_names, fontsize=7, color=\"white\")\n", "\n", "# X 轴：每年一个刻度 (X-axis: one tick per year)\n", "year_ticks = []\n", "year_labels = []\n", "for j, d in enumerate(heatmap_dates):\n", " if d.month == 1:\n", " year_ticks.append(j)\n", " year_labels.append(str(d.year))\n", "ax5.set_xticks(year_ticks)\n", "ax5.set_xticklabels(year_labels, color=\"white\", fontsize=8)\n", "plt.colorbar(im, ax=ax5, fraction=0.02, label=\"权重 Weight\").ax.yaxis.set_tick_params(color=\"white\", labelcolor=\"white\")\n", "\n", "# ── 图6: 月度换手率 (Monthly Turnover) ──────────────────────────────────────\n", "ax6 = fig.add_subplot(gs[2, 2])\n", "style_ax(ax6, \"⑥ 月度换手率 Monthly Turnover\", ylabel=\"换手率 Turnover\")\n", "for key in [\"equal_weight\", \"relative_momentum\", \"dual_momentum\"]:\n", " to_series = results[key].get(\"turnover\", pd.Series())\n", " if len(to_series) > 0:\n", " ax6.plot(to_series.index, to_series.values * 100,\n", " color=COLORS[key], linewidth=1.2, label=LABELS[key], alpha=0.8)\n", " # 平均换手率水平线 (average turnover horizontal line)\n", " avg_to = to_series.mean() * 100\n", " ax6.axhline(avg_to, color=COLORS[key], linewidth=0.8,\n", " linestyle=\":\", alpha=0.6)\n", "ax6.yaxis.set_major_formatter(mtick.PercentFormatter())\n", "ax6.legend(fontsize=7, facecolor=\"#1a1a2e\", labelcolor=\"white\")\n", "\n", "# 保存图表 (save chart)\n", "out_path = \"/Users/tigeren/Dev/xorbitlab/trading/etf_rotation_demo.png\"\n", "plt.savefig(out_path, dpi=150, bbox_inches=\"tight\",\n", " facecolor=fig.get_facecolor())\n", "plt.close()\n", "print(f\" 图表已保存 (Chart saved): {out_path}\")\n", "print()" ] }, { "cell_type": "markdown", "id": "b24aac17", "metadata": {}, "source": [ "# ══════════════════════════════════════════════════════════════════════════════\n", "# §7 关键学习要点总结\n", "# Key Learning Takeaways\n", "\n", "# ══════════════════════════════════════════════════════════════════════════════" ] }, { "cell_type": "code", "execution_count": null, "id": "838d08d5", "metadata": {}, "outputs": [], "source": [ "print(\"═\" * 60)\n", "print(\"§7 关键学习要点 (Key Learning Takeaways)\")\n", "print(\"═\" * 60)\n", "print(\"\"\"\n", " 1. ETF 轮动 vs 个股选股 (ETF Rotation vs. Stock Picking)\n", " ───────────────────────────────────────────────────\n", " ETF 轮动：买\"赛道\"，避免个股黑天鹅事件\n", " (ETF rotation: bet on sectors, avoid individual stock blow-ups)\n", " \n", " 个股选股：需要更强的信息优势，难度更高\n", " (Stock picking: requires information edge, much harder)\n", "\n", " 2. 绝对动量的价值 (Value of Absolute Momentum)\n", " ───────────────────────────────────────────\n", " 双动量策略在熊市中会自动切换至国债 ETF 避险\n", " 最大回撤通常比纯相对动量低 10-15 个百分点\n", " (Dual momentum auto-switches to bonds in bear markets,\n", " reducing max drawdown by ~10-15 percentage points)\n", "\n", " 3. 动量的局限性 (Momentum Limitations)\n", " ────────────────────────────────────\n", " • 趋势逆转（动量崩溃）风险：牛熊切换时可能大幅亏损\n", " (Momentum crash risk: large loss when trend reverses)\n", " • 主题 ETF 历史短：A 股主题 ETF 多数 2019 年后才成立\n", " (Short history: most A-share thematic ETFs launched post-2019)\n", " • 需要结合估值/基本面作为反向过滤器\n", " (Should combine with valuation/fundamentals as counter-filter)\n", "\n", " 4. 成本控制 (Cost Management)\n", " ───────────────────────────\n", " 每月换仓一次约消耗 0.22% 双边成本\n", " 一年 12 次再平衡 ≈ 2.6% 年化成本拖拽\n", " (Monthly rebalancing ≈ 0.22% per round-trip, ~2.6% annual drag)\n", " → 适当降低再平衡频率（如季度）可以减少成本\n", " (Reduce rebalance frequency, e.g., quarterly, to cut costs)\n", "\n", " 5. 进阶方向 (Next Steps)\n", " ───────────────────────\n", " • 接入真实数据（AKShare）替换合成数据\n", " (Replace synthetic data with real data via AKShare)\n", " • 加入估值信号（PE 分位数）作为过滤器\n", " (Add valuation signal: sector PE percentile as filter)\n", " • 加入波动率调仓（高波动期降低持仓比例）\n", " (Add volatility scaling: reduce position in high-vol regimes)\n", " • 结合宏观信号（利率、信用利差）增强国债 ETF 切换时机\n", " (Combine macro signals for better timing of bond ETF switch)\n", "\"\"\")\n", "\n", "print(\"✅ Demo 完成！(Demo Complete!)\")\n", "print(f\" 输出文件: etf_rotation_demo.png\")" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3 (trading)", "language": "python", "name": "python3" }, "language_info": { "name": "python", "version": "3.11.0" } }, "nbformat": 4, "nbformat_minor": 5 }