diff --git a/doc_07_etf_rotation.md b/doc_07_etf_rotation.md new file mode 100644 index 0000000..1cd5a72 --- /dev/null +++ b/doc_07_etf_rotation.md @@ -0,0 +1,714 @@ +# A 股行业 ETF 轮动策略 +## 从零开始的量化投资实践指南（第七篇） + +> **配套代码**：`quant_etf_rotation_demo.py` +> **目标读者**：量化投资初学者，有 Python 基础，计划在 A 股实战 +> **核心主题**：用动量信号驱动行业 ETF（Exchange-Traded Fund，交易所交易基金）轮动，构建兼顾收益与风控的量化策略 + +--- + +## 目录 + +1. [为什么选 ETF 轮动？](#1-为什么选-etf-轮动) +2. [A 股 ETF 的独特优势](#2-a-股-etf-的独特优势) +3. [核心武器：动量因子](#3-核心武器动量因子) +4. [策略一：相对动量轮动](#4-策略一相对动量轮动) +5. [策略二：双动量策略](#5-策略二双动量策略) +6. [A 股交易成本模型](#6-a-股交易成本模型) +7. [回测结果深度解读](#7-回测结果深度解读) +8. [动量策略的局限性与风险](#8-动量策略的局限性与风险) +9. [进阶优化方向](#9-进阶优化方向) +10. [实战路径：接入真实数据](#10-实战路径接入真实数据) +11. [双语术语表](#11-双语术语表) + +--- + +## 1. 为什么选 ETF 轮动？ + +### 1.1 初学者面临的困境 + +刚进入量化投资领域，你可能遇到这样的困惑： + +- 选哪只股票？A 股有 5000+ 只，信息量巨大 +- 如何避免个股"雷"？财务造假、黑天鹅事件防不胜防 +- 机构有研究员团队，个人投资者如何竞争？ + +**行业 ETF 轮动**（Sector ETF Rotation）正是为了解决这些问题而生的策略。 + +### 1.2 轮动策略的核心逻辑 + +``` +不去赌哪只股票涨 +而是判断哪个"赛道"当前最强 +持有最强赛道的 ETF，定期切换 +``` + +这背后有坚实的经济学逻辑：**行业景气周期**（Industry Business Cycle）。 + +经济不同阶段，不同行业表现各异： + +``` +经济复苏期 → 消费、金融领跑 +经济过热期 → 能源、原材料领跑 +经济衰退期 → 防御性行业（医疗、国债）抗跌 +经济萧条期 → 国债、黄金是避风港 +``` + +### 1.3 ETF 轮动 vs 个股选股 对比 + +| 维度 | ETF 轮动 | 个股选股 | +|------|---------|---------| +| **难度** | ⭐⭐ 较低 | ⭐⭐⭐⭐⭐ 极高 | +| **所需信息** | 行业趋势（公开）| 个股基本面（深度研究）| +| **黑天鹅风险** | 低（ETF 内部分散）| 高（单只股票可能暴雷）| +| **信息竞争** | 与机构差距较小 | 机构有绝对信息优势 | +| **成本** | 低（管理费 0.15-0.5%/年）| 低 | +| **回测可靠性** | 较高（无幸存者偏差）| 低（需处理退市股票）| + +> 💡 **初学者建议**：先用 ETF 轮动打好基础，建立量化思维框架，再逐步进阶到个股 Alpha 挖掘。 + +--- + +## 2. A 股 ETF 的独特优势 + +### 2.1 最关键的规则：ETF 可以 T+0 交易 + +A 股个股有 **T+1**（Trade plus 1 day）限制：今天买入的股票，明天才能卖出。 + +但 ETF 是个例外： + +``` +ETF（场内）= T+0 + → 今天买入，今天就可以卖出 + → 策略执行更灵活 + → 急跌时可以当天止损 + +个股 = T+1 + → 买入后被"锁仓"一天 + → 暴跌时无法当天止损 +``` + +> ⚠️ 注意：ETF 的 T+0 指的是**场内交易**（通过股票账户买卖）。场外申购赎回仍然是 T+1 或 T+2。 + +### 2.2 A 股热门行业 ETF 池 + +以下是常见的 A 股行业/主题 ETF（供参考，实际投资请核实最新信息）： + +| 代码 | 名称 | 跟踪指数 | 特点 | +|------|------|---------|------| +| 159995 | 芯片 ETF | 中华半导体芯片指数 | 高波动，含存储/AI芯片 | +| 159819 | 人工智能 ETF | 人工智能主题指数 | 2019年后成立，历史短 | +| 516160 | 新能源 ETF | 中证新能源指数 | 涵盖光伏/风电/储能 | +| 159928 | 消费 ETF | 中证主要消费指数 | 防御性，波动较低 | +| 512170 | 医疗 ETF | 中证医疗指数 | 防御性，受政策影响大 | +| 512880 | 证券 ETF | 中证全指证券公司指数 | 与市场 Beta 高度相关 | +| **511010** | **国债 ETF** | **上证5年期国债指数** | **避险资产，熊市避风港** | + +> 💡 **关键设计**：在 ETF 池中纳入**国债 ETF**（Bond ETF）作为"避险舱位"（Safe Haven）。当所有行业 ETF 动量均为负时，资金自动切换到国债 ETF，而不是空仓等待。 + +### 2.3 A 股 ETF 的局限性 + +- **历史短**：多数主题 ETF 在 2018-2020 年后才成立，回测样本不足 5 年 +- **规模效应**：小规模 ETF（<5亿）存在清盘风险，流动性差 +- **跟踪误差**（Tracking Error）：ETF 实际表现与指数有偏差 +- **同质化严重**：芯片/半导体相关 ETF 众多，高度重叠 + +--- + +## 3. 核心武器：动量因子 + +### 3.1 什么是动量？ + +**动量**（Momentum）是量化投资中最经典的异象（Anomaly）之一： + +> **过去表现强的资产，在未来短期内倾向于继续表现强。** +> +> *(Assets that have performed well in the past tend to continue performing well in the near future.)* + +这个现象最早由 Jegadeesh & Titman（1993）在美股系统性记录，此后在全球多个市场都得到了验证。 + +### 3.2 动量为什么有效？行为金融学解释 + +动量之所以持续存在，背后有两个行为偏差支撑： + +**① 反应不足（Under-reaction）** + +``` +好消息发布 → 投资者缓慢消化 → 价格缓慢上涨（而非一步到位） +→ 动量策略可以"搭便车" +``` + +**② 追涨杀跌（Herding / Trend-chasing）** + +``` +价格上涨 → 媒体报道 → 更多投资者买入 → 继续上涨 +→ 趋势自我强化，直到过度（Overshooting） +``` + +### 3.3 动量的三个维度 + +| 维度 | 定义 | 回望期 | +|------|------|-------| +| **短期动量** | 近期强势，可能含"反转"噪音 | 1-3 个月 | +| **中期动量** | 经典动量信号，最稳定 | 6-12 个月 | +| **跳过期** | 跳过最近 1 个月（避免短期反转）| 1 个月 | + +**经典公式（12-1 动量）**： + +``` +动量得分 = 前 12 个月收益率（跳过最近 1 个月） + +Momentum = Price[t-1] / Price[t-13] - 1 +``` + +### 3.4 复合动量得分（Composite Momentum Score） + +单一窗口的动量信号容易过拟合（Overfitting）。Demo 中使用三个窗口的等权平均： + +```python +# 三个窗口的动量因子 +mom_12_1 = 过去12个月动量（跳过1个月） # 中长期趋势 +mom_6_1 = 过去6个月动量（跳过1个月） # 中期趋势 +mom_3_1 = 过去3个月动量（跳过1个月） # 短期趋势 + +# 复合得分 = 三者等权平均 +mom_composite = (mom_12_1 + mom_6_1 + mom_3_1) / 3 +``` + +**为什么要多窗口平均？** +- 减少单一参数的过拟合风险 +- 兼顾不同时间尺度的趋势 +- 提高信号的稳定性 + +--- + +## 4. 策略一：相对动量轮动 + +### 4.1 策略逻辑 + +**相对动量**（Relative Momentum）回答的是： + +> "在当前所有行业 ETF 中，谁是最强的？" + +``` +每月月末执行： +1. 计算所有行业 ETF 的复合动量得分 +2. 排序，选出得分最高的前 K 只（Demo 中 K=3） +3. 等权持有，下月月末再次评估 +``` + +### 4.2 伪代码（Pseudo Code） + +```python +每月月末: + scores = {} + for etf in sector_etfs: + scores[etf] = calc_momentum(etf, lookback=12, skip=1) + + # 选出最强的前 3 只 + top_3 = sorted(scores, by=score, descending=True)[:3] + + # 等权持有 + target_weights = {etf: 1/3 for etf in top_3} + + # 执行换仓（扣除成本） + rebalance(current_weights, target_weights) +``` + +### 4.3 相对动量的优缺点 + +**优点**： +- ✅ 逻辑简单，规则清晰，不受主观判断影响 +- ✅ 自动轮换到最强赛道，捕获行业趋势 +- ✅ 历史上在多数市场有效 + +**缺点**： +- ❌ **熊市无保护**：即使所有行业都在跌，也会持有"跌得少"的行业 ETF +- ❌ 换手率较高（月度换仓约 17%），成本不可忽视 +- ❌ 趋势逆转时可能遭受较大回撤（Drawdown） + +--- + +## 5. 策略二：双动量策略 + +### 5.1 Gary Antonacci 的原创思想 + +**双动量**（Dual Momentum）由量化基金经理 Gary Antonacci 在其 2014 年著作《双动量投资》（*Dual Momentum Investing*）中系统提出。 + +核心思想：**把绝对动量和相对动量结合，既追强者，也知进退。** + +### 5.2 双动量的两道过滤 + +``` +第一道：绝对动量过滤（Absolute Momentum Filter） + → 问题："这只 ETF 比无风险收益（国债）表现得更好吗？" + → 如果动量 < 0（即跑输无风险），说明处于下行趋势，切换至国债 ETF 避险 + +第二道：相对动量排序（Relative Momentum Ranking） + → 问题："在通过第一道过滤的 ETF 中，谁最强？" + → 选相对最强的前 K 只 +``` + +### 5.3 完整决策流程 + +``` +月末执行双动量策略： + +for each ETF in 行业ETF池: + if 该ETF动量得分 > 0: ← 绝对动量为正（跑赢无风险） + 候选池.add(ETF) + else: + 跳过（趋势向下） + +if len(候选池) == 0: + 持有国债ETF（全部资金避险） ← 全市场下行，切换至避险 +else: + 选候选池中动量得分最高的前K只 ← 相对动量选最强 + 等权持有 +``` + +### 5.4 为什么要有"绝对动量"这一关？ + +一个直觉性的例子： + +``` +假设某月，6 只行业 ETF 的动量得分分别是： + 芯片 -15% ← 在跌 + AI -12% ← 在跌 + 新能源 -18% ← 在跌 + 消费 -5% ← 跌最少（相对最强！） + 医疗 -8% ← 在跌 + 证券 -20% ← 在跌 + +相对动量策略：会买消费 ETF（相对最强），结果仍然亏损 +双动量策略：全部动量为负 → 切换至国债 ETF，保住本金 +``` + +**绝对动量是双动量策略最核心的风控机制。** + +### 5.5 回测中的实际表现 + +在 Demo（2019-2024 牛市主导）的结果中： + +``` +双动量切换至国债 ETF 的月数：1 次（约占 2%） + +解读： +2019-2024 年 A 股整体处于震荡偏牛格局， +绝对动量为负的情形极少出现， +因此双动量的避险机制几乎没有触发机会。 + +但在 2018 年（A股全面熊市）或 2022 年（新能源大跌）， +双动量的避险效果会非常显著。 +``` + +> 💡 **关键认识**：双动量在**牛市中跑输相对动量和等权**（因为避险机制是"成本"），但在**熊市和震荡市中保护效果极佳**。评估策略要看完整市场周期，而非单一牛市区间。 + +--- + +## 6. A 股交易成本模型 + +### 6.1 为什么成本比你想象的重要？ + +很多初学者的回测**忽略了交易成本**，导致实盘效果大幅低于回测。 + +对于月度再平衡的 ETF 轮动策略，成本明细如下： + +| 成本类型 | 方向 | 费率 | 说明 | +|---------|------|------|------| +| **佣金**（Commission）| 买卖双向 | 0.03% | 网络券商优惠后，通常 0.02-0.03% | +| **印花税**（Stamp Duty）| 仅卖出 | 0.10% | 国家税收，A股特有，ETF 同样收取 | +| **滑点**（Slippage）| 买卖双向 | 0.05% | 因订单冲击导致的不利价差 | + +**单次月度再平衡的双边成本**： + +``` +买入成本 = 佣金 0.03% + 滑点 0.05% = 0.08% +卖出成本 = 佣金 0.03% + 印花税 0.10% + 滑点 0.05% = 0.18% +双边合计 = 0.26% + +年度 12 次再平衡（假设每次换手 20%）： +年化成本拖拽 ≈ 12 × 0.26% × 20% ≈ 0.62% + +若换手率更高（如每次 50%）： +年化成本拖拽 ≈ 12 × 0.26% × 50% ≈ 1.56% +``` + +> ⚠️ **实战建议**： +> - 选择低佣金券商（如华泰、富途等，佣金可低至 0.015%） +> - 控制换手率（考虑季度再平衡替代月度） +> - ETF 点差通常比个股小，但也要在流动性好时交易 + +### 6.2 成本对策略的影响 + +Demo 回测显示，相对动量策略的总交易成本约占净值的 **4.25%**（跨越 6 年，平均每年约 0.7%）。 + +这个数字看似不大，但复利效应显著： + +``` +假设策略毛收益 18%，扣除成本后： + 年化成本 0.7% × 6年 → 累计净值差异约 5-6% + 即：回测累计 178% → 实际约 170%（差异约 8 个百分点） +``` + +--- + +## 7. 回测结果深度解读 + +### 7.1 三策略绩效汇总 + +| 指标 | 等权基准 | 相对动量 | 双动量 | +|------|---------|---------|-------| +| 年化收益率 (CAGR) | **18.1%** | 18.6% | 15.9% | +| 年化波动率 (Volatility) | **17.5%** | 20.3% | 20.4% | +| 夏普比率 (Sharpe) | **0.926** | 0.847 | 0.729 | +| 最大回撤 (Max Drawdown) | **-27.3%** | -32.2% | -39.1% | +| 卡玛比率 (Calmar) | **0.663** | 0.578 | 0.407 | +| 月度胜率 (Win Rate) | 52.2% | 47.8% | 44.9% | + +### 7.2 等权基准为何最优？——牛市的陷阱 + +**这个结果出乎很多人的意料，但背后有清晰的逻辑：** + +2019-2024 年，A 股整体处于"结构性牛市"（Structural Bull Market）： + +``` +芯片/AI/新能源轮番领涨 → 行业轮动 ETF 池整体表现强劲 + +等权基准 = 始终全仓持有，没有错过任何一波行情 + +相对动量 = 追涨热门板块，但错过了"从低位起涨"的初期涨幅 + （因为动量是滞后信号，Lagging Indicator） + +双动量 = 额外的避险机制在牛市里是"白交保险费" +``` + +**这说明了一个重要的回测教训：** + +> 在单一市场环境（纯牛市）中评估策略，结论极不可靠。 +> 真正有价值的策略需要跨越完整市场周期（牛市 + 熊市 + 震荡市）来评估。 + +### 7.3 如何正确评价双动量策略？ + +尽管在 Demo 的牛市数据中表现靠后，双动量策略在以下情形下优势明显： + +| 市场环境 | 等权基准 | 相对动量 | 双动量 | +|---------|---------|---------|-------| +| 牛市（全面上涨）| ✅ 最优 | 次优 | 最差 | +| 熊市（全面下跌）| ❌ 随市大跌 | ❌ 跌得少一点 | ✅ 切债避险 | +| 震荡市（分化）| 中 | ✅ 捕获强势 | ✅ 回避弱势 | + +**结论：双动量的价值在于"不对称性"——牛市少赚一点，熊市保住本金。** + +对于注重本金安全的初学者，这个特性极为宝贵。 + +### 7.4 持仓集中度分析 + +从持仓热力图可以看到一个有趣的现象： + +``` +芯片 ETF 被选中频率最高（82%的月份） +医疗 ETF 排第二（70%） + +这反映了 2019-2024 年 A 股的主要趋势： +半导体/科技 + 医疗健康是这一周期的主赛道 +``` + +> ⚠️ **重要警示**：历史持仓集中度≠未来持仓集中度。 +> 不能因为芯片 ETF 过去 5 年强，就预设它未来也会被持续选中。 + +--- + +## 8. 动量策略的局限性与风险 + +### 8.1 动量崩溃（Momentum Crash） + +动量策略最大的单一风险是**动量崩溃**（Momentum Crash）： + +``` +发生条件：市场快速从下跌趋势逆转为强劲上涨 + （通常发生在熊市见底后的"V形反弹"） + +典型案例：2009年3月全球股市V形反弹 + 动量策略空仓"弱势股"、持有"强势股" + 结果：弱势股（低价小盘股）反而率先暴涨 + 强势股反而横盘，动量策略大幅跑输 + +A股案例：2020年3月疫情底部反弹、2022年底政策转向 + 持有前期强势行业反而错过了反弹 +``` + +**应对方法**： +1. 在动量信号基础上叠加**估值过滤器**（避开极度高估行业） +2. 设置**止损机制**（如最大回撤超过 15% 时切换至国债） +3. 降低单一赛道集中度（强制分散，如最多持有行业不超过 2 只） + +### 8.2 过拟合风险（Overfitting Risk） + +回测中 Demo 选择了 12-1 月、6-1 月、3-1 月三个窗口。如果不断调整这些参数直到回测结果最好，就会陷入**过拟合**： + +``` +过拟合的症状： + 回测夏普比率 > 2.0 + 换不同参数结果差异极大 + 真实数据（实盘）效果远低于回测 + +防止方法： + 1. 固定参数（如永远用 12-1 月，不要调整） + 2. 预留测试集（后 30% 的数据不参与参数选择） + 3. 参数敏感性测试（改变 ±2 个月，结果不应剧烈变化） +``` + +### 8.3 主题 ETF 历史过短 + +这是 A 股 ETF 轮动策略的致命弱点之一： + +``` +A 股主要主题 ETF 成立时间： + 芯片 ETF（159995）：2019年5月 + 人工智能 ETF（159819）：2019年10月 + 新能源 ETF（516160）：2021年6月 + 存储芯片 ETF（562850）：2022年12月 + +问题： + → 最长历史仅 5-6 年（不含完整牛熊周期） + → 成立于牛市阶段的 ETF，回测结果天然偏好 + → 历史太短，难以区分"策略有效"还是"恰好运气好" +``` + +> 💡 **建议**：至少需要 10 年以上历史（含一次完整熊市）才能对策略可靠性有初步信心。A 股历史较长的 ETF（如沪深 300 ETF 510300，2012年成立）更适合做长期回测验证。 + +### 8.4 政策风险与流动性风险 + +A 股特有风险，ETF 也不能免疫： + +| 风险类型 | 描述 | 影响 | +|---------|------|------| +| **政策突变** | 双减政策（教育）、反垄断（互联网）| 行业 ETF 单日 -10% | +| **涨跌停传导** | ETF 内部成分股大面积涨停/跌停 | ETF 价格与净值偏离（折溢价）| +| **清盘风险** | 规模低于 5000 万的小 ETF 可能被清盘 | 被迫在低点卖出 | +| **流动性枯竭** | 小众主题 ETF 某些时段成交量极低 | 买卖价差扩大，滑点增加 | + +--- + +## 9. 进阶优化方向 + +### 9.1 叠加估值过滤器 + +纯动量有时会在行业极度高估时继续追入。加入估值信号可以降低这种风险： + +```python +# 概念示例：动量 + 估值复合信号 +def composite_score(etf, date): + mom_score = calc_momentum(etf, date) # 动量得分 + + # 估值得分：PE 历史分位数越低，分数越高 + pe_percentile = calc_pe_percentile(etf, date) # 0~1，1表示历史最贵 + val_score = 1 - pe_percentile # 越便宜分数越高 + + # 复合信号：动量 70% + 估值 30% + return 0.7 * normalize(mom_score) + 0.3 * val_score +``` + +**实战 ETF 估值参考**： +- 中证指数官网提供各指数 PE/PB 历史数据 +- 恒生前海基金等定期发布行业估值分位数报告 +- AKShare 可获取宽基指数 PE 数据 + +### 9.2 波动率调仓（Volatility Scaling） + +在高波动时期降低持仓比例，类似"自动降速"机制： + +```python +def vol_scaled_weight(base_weight, current_vol, target_vol=0.15): + """ + 目标波动率（Target Volatility）仓位调整 + 若当前波动率过高，则降低持仓比例 + + 例：目标波动 15%，当前波动 30% + → 实际持仓 = 50%（另 50% 转为国债 ETF） + """ + scale = min(target_vol / current_vol, 1.0) + return base_weight * scale +``` + +### 9.3 宏观择时信号 + +利用宏观指标提前判断牛熊切换，辅助双动量的避险决策： + +``` +信号类型 数据来源 使用逻辑 +----------------------------------------------------------------- +利率曲线 国债收益率 10Y-2Y 倒挂 → 熊市预警 +信用利差 企业债 vs 国债 利差扩大 → 风险偏好下降 +PMI 趋势 国家统计局 PMI < 50 且下行 → 降低权益仓位 +市场情绪 北向资金流入/流出 持续流出 → 外资撤退信号 +``` + +### 9.4 季度再平衡降低成本 + +将月度再平衡改为季度再平衡，换手率降低约 2/3，成本拖拽从 ~1.5% 降至 ~0.5%： + +``` +月度再平衡：年化成本约 1.0-2.5%（视换手率） +季度再平衡：年化成本约 0.3-0.8% +半年度再平衡：年化成本约 0.2-0.4% + +代价：信号响应更慢，可能错过行情初期 +``` + +--- + +## 10. 实战路径：接入真实数据 + +### 10.1 环境准备 + +```bash +# 安装真实数据接口 +pip install akshare # 免费，数据较全 +pip install tushare # 部分接口需积分/付费 +``` + +### 10.2 用 AKShare 获取 ETF 真实数据 + +```python +import akshare as ak +import pandas as pd + +def get_etf_history(etf_code: str, + start_date: str = "20190101", + end_date: str = "20241231") -> pd.DataFrame: + """ + 获取 A 股 ETF 历史日线数据 + Get A-Share ETF historical daily data + + 参数： + etf_code : ETF 代码，如 "159995"（不含交易所后缀） + start_date: 开始日期，格式 "YYYYMMDD" + end_date : 结束日期，格式 "YYYYMMDD" + + 返回：包含 date, close, volume 等列的 DataFrame + """ + df = ak.fund_etf_hist_em( + symbol=etf_code, + period="daily", + start_date=start_date, + end_date=end_date, + adjust="qfq" # 前复权（Forward Adjusted Price），处理分红/拆股 + ) + df = df.rename(columns={"日期": "date", "收盘": "close", + "成交量": "volume", "开盘": "open"}) + df["date"] = pd.to_datetime(df["date"]) + df = df.set_index("date").sort_index() + return df[["open", "close", "volume"]] + + +# 示例：批量获取 ETF 池数据 +ETF_CODES = ["159995", "159819", "516160", "159928", "512170", "512880", "511010"] + +etf_data = {} +for code in ETF_CODES: + try: + etf_data[code] = get_etf_history(code) + print(f"✅ {code}: {len(etf_data[code])} 条记录") + except Exception as e: + print(f"❌ {code}: 获取失败 - {e}") + +# 合并为价格矩阵 +price_matrix = pd.DataFrame({ + code: df["close"] + for code, df in etf_data.items() +}) +``` + +### 10.3 关键注意事项：前复权 vs 后复权 + +| 复权方式 | 说明 | 用途 | +|---------|------|------| +| **前复权**（Forward Adjusted）| 以当前价格为基准向历史调整 | **回测首选**：历史价格连续，收益率计算准确 | +| **后复权**（Backward Adjusted）| 以上市初始价为基准向未来调整 | 长期持有者参考 | +| **不复权** | 原始价格，含分红跳空 | ❌ 不适合回测，会产生虚假信号 | + +> ⚠️ **前复权的陷阱**：前复权价格会随时间变化（每次分红后历史价格都重新调整）。在回测中，必须在**同一时点**下载数据，否则历史价格不一致。 + +### 10.4 从合成数据切换到真实数据的步骤 + +```python +# 替换 Demo 中的 §0 数据生成部分 +# 将合成 price_df 替换为真实数据： + +# 原来（合成数据）： +# price_df = (1 + returns_df).cumprod() + +# 替换为（真实数据）： +price_df = price_matrix # 已经是收盘价，无需转换 + +# 注意：真实数据可能有以下问题需要处理： +# 1. 缺失值（节假日、停牌）→ 前向填充 ffill() +# 2. 各 ETF 成立时间不同 → 统一以最晚成立的 ETF 作为回测起点 +# 3. 价格单位不统一 → 统一转为净值形式（除以第一个有效价格） + +price_df = price_df.ffill() # 填充缺失值 +start_common = price_df.dropna().index[0] # 统一起始日期 +price_df = price_df.loc[start_common:] # 裁剪至共同起始日 +price_df = price_df / price_df.iloc[0] # 标准化为净值形式（从1开始） +``` + +--- + +## 11. 双语术语表 + +| 中文术语 | English Term | 简要说明 | +|---------|-------------|---------| +| 行业 ETF 轮动 | Sector ETF Rotation | 在不同行业 ETF 间周期性切换 | +| 动量因子 | Momentum Factor | 过去表现预测未来表现的信号 | +| 相对动量 | Relative Momentum | 横截面比较，选相对最强 | +| 绝对动量 | Absolute Momentum | 纵向比较，判断趋势正负 | +| 双动量 | Dual Momentum | 结合绝对和相对动量的策略 | +| 避险资产 | Safe-Haven Asset | 市场下跌时表现抗跌的资产 | +| 再平衡 | Rebalancing | 定期调整组合权重至目标配置 | +| 换手率 | Turnover Rate | 组合中被替换的资产比例 | +| 印花税 | Stamp Duty | A 股卖出时征收的税，0.1% | +| 滑点 | Slippage | 订单执行价与预期价的差异 | +| 复权 | Price Adjustment | 对历史价格进行分红/拆股调整 | +| 前复权 | Forward Adjusted | 以当前价为基准调整历史价格 | +| 最大回撤 | Maximum Drawdown | 从历史最高点的最大跌幅 | +| 夏普比率 | Sharpe Ratio | 单位风险所获得的超额收益 | +| 卡玛比率 | Calmar Ratio | 年化收益 / 最大回撤 | +| 行业景气周期 | Industry Business Cycle | 行业随经济周期的景气变化 | +| 动量崩溃 | Momentum Crash | 趋势逆转时动量策略的大幅亏损 | +| 过拟合 | Overfitting | 策略参数过度适应历史数据 | +| 跟踪误差 | Tracking Error | ETF 实际表现与指数的偏差 | +| 清盘风险 | Liquidation Risk | ETF 规模过小被强制清算的风险 | +| 估值分位数 | Valuation Percentile | 当前估值在历史中所处的位置 | +| 波动率调仓 | Volatility Scaling | 根据波动率动态调整持仓比例 | +| 目标波动率 | Target Volatility | 组合设定的波动率上限目标 | +| 信用利差 | Credit Spread | 企业债与国债的收益率差异 | +| T+0 | T+0 Trading | 当天买入当天可以卖出 | +| T+1 | T+1 Trading | 当天买入次日才能卖出（个股规则）| +| 折溢价 | Premium/Discount | ETF 市价与净值的偏离程度 | + +--- + +## 附录：策略参数速查 + +| 参数 | Demo 默认值 | 可调范围 | 影响 | +|------|----------|---------|------| +| 动量回望期 | 12-1, 6-1, 3-1 月 | 3-24 月 | 更长=更稳定但更滞后 | +| 持仓 ETF 数量 (K) | 3 | 1-5 | 更少=更集中，更多=更分散 | +| 再平衡频率 | 月度 | 周/月/季 | 更频繁=响应快但成本高 | +| 佣金率 | 0.03% | 0.015-0.05% | 影响成本拖拽 | +| 印花税 | 0.10% | 固定 | 不可更改 | + +--- + +> **下一步建议**： +> 1. 用 AKShare 接入真实 A 股 ETF 数据，替换合成数据后重新回测 +> 2. 在动量信号基础上叠加**估值过滤器**（参考第 9.1 节） +> 3. 参考《双动量投资》（Gary Antonacci）原著，深入理解策略理论基础 +> 4. 结合 `doc_06_astock_practice_guide.md` 的风险管理框架，制定完整的实战纪律 + +--- + +*本文档配套代码：`quant_etf_rotation_demo.py`* +*系列文档索引：doc_01（数据准备）→ doc_02（策略回测）→ doc_03（事件驱动）→ doc_04（Alpha因子）→ doc_05（组合优化）→ doc_06（A股实战指南）→ **doc_07（ETF轮动）*** diff --git a/etf_rotation_demo.png b/etf_rotation_demo.png new file mode 100644 index 0000000..f79f06e Binary files /dev/null and b/etf_rotation_demo.png differ diff --git a/quant_etf_rotation_demo.py b/quant_etf_rotation_demo.py new file mode 100644 index 0000000..ca189a1 --- /dev/null +++ b/quant_etf_rotation_demo.py @@ -0,0 +1,848 @@ +""" +A 股行业 ETF 轮动策略 Demo +A-Share Sector ETF Rotation Strategy Demo +========================================== + +策略逻辑 (Strategy Logic): + 1. 相对动量轮动 (Relative Momentum Rotation) + - 每月对所有行业 ETF 计算过去 N 个月的动量得分 + - 买入得分最高的前 K 个 ETF，等权持有 + 2. 双动量策略 (Dual Momentum, Gary Antonacci) + - 绝对动量 (Absolute Momentum): 若某 ETF 的动量 < 0，切换至国债 ETF 避险 + - 相对动量 (Relative Momentum): 在通过绝对动量过滤的 ETF 中，买入相对最强的 + 3. 等权基准 (Equal Weight Benchmark) + - 始终等权持有全部行业 ETF，不做轮动 + +成本模拟 (Cost Simulation): + - 印花税 (Stamp Duty): 卖出 0.1% + - 佣金 (Commission): 买卖各 0.03%（万三） + - 滑点 (Slippage): 单边 0.05% + - 每月换仓合计约 0.22% 双边成本 + +作者: GitHub Copilot (教学用途) +数据: 合成模拟数据 (Synthetic data for demonstration) +""" + +# ── 标准库 ────────────────────────────────────────────────────────────────── +import warnings +warnings.filterwarnings("ignore") + +# matplotlib 必须在 pyplot 之前设置后端 (must set backend before importing pyplot) +import matplotlib +matplotlib.use("Agg") # 无界面后端 (headless backend) + +import numpy as np +import pandas as pd +import matplotlib.pyplot as plt +import matplotlib.gridspec as gridspec +from matplotlib.patches import FancyArrowPatch +import matplotlib.ticker as mtick + +# 设置随机种子，保证结果可复现 (fix random seed for reproducibility) +np.random.seed(42) + +# ══════════════════════════════════════════════════════════════════════════════ +# §0 ETF 池定义 & 合成价格数据生成 +# ETF Universe Definition & Synthetic Price Data Generation +# ══════════════════════════════════════════════════════════════════════════════ + +# ── A 股行业 ETF 池 (A-Share Sector ETF Universe) ─────────────────────────── +# 这里用合成数据模拟，真实代码只需替换为 AKShare/Tushare 数据读取 +# (Synthetic data here; replace with AKShare/Tushare data in production) +ETF_UNIVERSE = { + # 代码 (Code) : (中文名称, 英文名称, 年化收益均值, 年化波动率, 与市场相关性) + "159995": ("芯片ETF", "Chip ETF", 0.18, 0.45, 0.75), + "159819": ("人工智能ETF","AI ETF", 0.20, 0.48, 0.72), + "516160": ("新能源ETF", "New Energy ETF", 0.15, 0.42, 0.70), + "159928": ("消费ETF", "Consumer ETF", 0.10, 0.28, 0.65), + "512170": ("医疗ETF", "Healthcare ETF", 0.08, 0.32, 0.60), + "512880": ("证券ETF", "Securities ETF", 0.12, 0.38, 0.80), + # 国债 ETF 作为避险资产 (Bond ETF as safe-haven asset) + "511010": ("国债ETF", "Bond ETF", 0.03, 0.03, -0.10), +} + +# 分离行业 ETF 和国债 ETF (separate sector ETFs from bond ETF) +SECTOR_ETFS = [code for code in ETF_UNIVERSE if code != "511010"] +BOND_ETF = "511010" +ALL_ETFS = list(ETF_UNIVERSE.keys()) + +# ── 生成合成价格序列 (Generate Synthetic Price Series) ─────────────────────── +# 用因子模型生成相关价格序列，模拟真实行业之间的联动关系 +# (Factor model to generate correlated price series, mimicking real sector co-movement) + +TRADING_DAYS = 252 # 年交易日数 (trading days per year) +SIM_YEARS = 6 # 模拟年数 (simulation years): 2019-2024 +N_DAYS = TRADING_DAYS * SIM_YEARS + +# 市场公共因子 (market common factor) —— 模拟沪深300走势 +# 牛市/熊市分段：模拟 2019-2021 牛市，2021-2022 熊市，2023-2024 震荡 +market_factor = np.zeros(N_DAYS) +# 第一阶段：2019-2021 牛市 (Bull market phase) +bull1 = int(N_DAYS * 0.40) +market_factor[:bull1] = np.random.normal(0.0008, 0.012, bull1) +# 第二阶段：2021-2022 熊市 (Bear market phase) +bear1 = int(N_DAYS * 0.20) +market_factor[bull1:bull1+bear1] = np.random.normal(-0.0006, 0.018, bear1) +# 第三阶段：2023-2024 震荡 (Sideways phase) +market_factor[bull1+bear1:] = np.random.normal(0.0002, 0.014, N_DAYS - bull1 - bear1) + +# 生成每只 ETF 的日收益率 (generate daily returns for each ETF) +daily_returns = {} +for code, (cn_name, en_name, ann_ret, ann_vol, mkt_corr) in ETF_UNIVERSE.items(): + daily_mu = ann_ret / TRADING_DAYS # 日均收益 (daily mean return) + daily_sig = ann_vol / np.sqrt(TRADING_DAYS) # 日波动率 (daily volatility) + + # 特异性收益 (idiosyncratic return) = 总收益 - 市场部分 + idio_sig = daily_sig * np.sqrt(1 - mkt_corr**2) + idio = np.random.normal(0, idio_sig, N_DAYS) + + # 总日收益 = 市场因子 × 相关系数 + 特异性收益 + 均值漂移 + ret = mkt_corr * market_factor + idio + (daily_mu - mkt_corr * 0.0002) + daily_returns[code] = ret + +# 转为 DataFrame，并生成价格序列 (convert to DataFrame, then price series) +start_date = pd.Timestamp("2019-01-02") +dates = pd.bdate_range(start=start_date, periods=N_DAYS) # 仅工作日 (business days only) + +returns_df = pd.DataFrame(daily_returns, index=dates) +# 价格从 1.0 开始（净值形式，Net Asset Value style） +price_df = (1 + returns_df).cumprod() + +print("═" * 60) +print("§0 ETF 价格数据生成完成 (Price Data Generated)") +print("═" * 60) +print(f" 模拟区间 (Simulation Period): {dates[0].date()} → {dates[-1].date()}") +print(f" 交易日数 (Trading Days): {N_DAYS}") +print(f" ETF 数量 (Number of ETFs): {len(ALL_ETFS)}") +print() +print(" ETF 池 (Universe):") +print(f" {'代码':10s} {'中文名':12s} {'英文名':18s} {'年化收益':8s} {'年化波动':8s}") +print(" " + "-" * 60) +for code, (cn, en, ret, vol, _) in ETF_UNIVERSE.items(): + label = " [避险]" if code == BOND_ETF else "" + print(f" {code:10s} {cn:12s} {en:18s} {ret*100:6.1f}% {vol*100:6.1f}%{label}") +print() + +# ══════════════════════════════════════════════════════════════════════════════ +# §1 动量因子计算 +# Momentum Factor Calculation +# ══════════════════════════════════════════════════════════════════════════════ + +def calc_momentum(price_df: pd.DataFrame, + lookback_months: int = 12, + skip_months: int = 1) -> pd.DataFrame: + """ + 计算每只 ETF 的动量得分 (Calculate momentum score for each ETF) + + 经典动量定义 (Classic momentum definition): + 过去 L 个月的总收益，跳过最近 S 个月（避免短期反转） + Total return over past L months, skipping most recent S months + (to avoid short-term reversal effect 短期反转效应) + + 参数 (Parameters): + price_df : 日频价格 DataFrame (daily price DataFrame) + lookback_months : 回望期（月数）(lookback period in months) + skip_months : 跳过最近月数（通常为 1）(skip recent months, usually 1) + + 返回 (Returns): + 月频动量得分 DataFrame (monthly momentum score DataFrame) + """ + # 转为月末价格 (resample to month-end prices) + # 注意：旧版 pandas 用 "M"，新版用 "ME" + try: + monthly_price = price_df.resample("ME").last() + except ValueError: + monthly_price = price_df.resample("M").last() + + # 动量 = 前 (lookback+skip) 月价格 / 前 skip 月价格 - 1 + # Momentum = Price[t - skip] / Price[t - lookback - skip] - 1 + start_lag = skip_months + end_lag = lookback_months + skip_months + + momentum = monthly_price.shift(start_lag) / monthly_price.shift(end_lag) - 1 + return momentum + + +# 计算三种不同回望期的动量（稳健性检验用） +# (Calculate momentum with three different lookback windows for robustness) +mom_12_1 = calc_momentum(price_df, lookback_months=12, skip_months=1) # 经典 12-1 月动量 +mom_6_1 = calc_momentum(price_df, lookback_months=6, skip_months=1) # 中期 6-1 月动量 +mom_3_1 = calc_momentum(price_df, lookback_months=3, skip_months=1) # 短期 3-1 月动量 + +# 合成动量得分：三个窗口等权平均（减少单一窗口过拟合风险） +# (Composite momentum score: equal-weight average of three windows) +mom_composite = (mom_12_1 + mom_6_1 + mom_3_1) / 3 + +print("§1 动量因子计算完成 (Momentum Factors Calculated)") +print(f" 月度数据行数 (Monthly rows): {len(mom_12_1)}") +print() + +# ══════════════════════════════════════════════════════════════════════════════ +# §2 A 股交易成本模型 +# A-Share Transaction Cost Model +# ══════════════════════════════════════════════════════════════════════════════ + +class AShareCostModel: + """ + A 股 ETF 交易成本模型 (A-Share ETF Transaction Cost Model) + + A 股 ETF 特殊规则 (A-Share ETF Special Rules): + - ETF 可 T+0 交易（当天买当天可卖）Unlike 个股的 T+1 限制 + (ETF allows T+0 trading, unlike individual stocks which are T+1) + - 印花税：仅卖出方向收取 0.1% + (Stamp duty: 0.1% on SELL side only) + - 佣金：买卖双向，通常 0.02% - 0.03% + (Commission: both sides, typically 0.02%-0.03%) + - 滑点：ETF 流动性好，通常 0.02%-0.05% + (Slippage: ETF has good liquidity, typically 0.02%-0.05%) + """ + def __init__(self, + commission_rate: float = 0.0003, # 佣金率 (commission rate) + stamp_duty_rate: float = 0.001, # 印花税率 (stamp duty rate) + slippage_rate: float = 0.0005): # 滑点率 (slippage rate) + self.commission = commission_rate + self.stamp_duty = stamp_duty_rate + self.slippage = slippage_rate + + def buy_cost(self, trade_value: float) -> float: + """买入成本 (buy-side cost): 佣金 + 滑点""" + return trade_value * (self.commission + self.slippage) + + def sell_cost(self, trade_value: float) -> float: + """卖出成本 (sell-side cost): 佣金 + 印花税 + 滑点""" + return trade_value * (self.commission + self.stamp_duty + self.slippage) + + def roundtrip_cost(self, trade_value: float) -> float: + """ + 双边成本 (round-trip cost): 买入 + 卖出 + 月度换仓一次的总成本约 0.22% + """ + return self.buy_cost(trade_value) + self.sell_cost(trade_value) + + @property + def roundtrip_rate(self) -> float: + """双边成本率 (round-trip cost rate)""" + return 2 * self.commission + self.stamp_duty + 2 * self.slippage + + +cost_model = AShareCostModel() +print("§2 交易成本模型 (Transaction Cost Model):") +print(f" 佣金率 (Commission): {cost_model.commission*100:.3f}% × 双边") +print(f" 印花税 (Stamp Duty): {cost_model.stamp_duty*100:.3f}% × 卖出方向") +print(f" 滑点率 (Slippage): {cost_model.slippage*100:.3f}% × 双边") +print(f" 双边总成本 (Roundtrip): {cost_model.roundtrip_rate*100:.3f}%") +print() + +# ══════════════════════════════════════════════════════════════════════════════ +# §3 轮动回测引擎 +# Rotation Backtest Engine +# ══════════════════════════════════════════════════════════════════════════════ + +def run_rotation_backtest( + price_df: pd.DataFrame, + momentum_df: pd.DataFrame, + strategy: str = "relative_momentum", + top_k: int = 3, + cost_model: AShareCostModel = None, + bond_etf: str = BOND_ETF, + sector_etfs: list = None, +) -> dict: + """ + 行业 ETF 轮动回测引擎 (Sector ETF Rotation Backtest Engine) + + 参数 (Parameters): + price_df : 日频价格 (daily prices) + momentum_df : 月频动量得分 (monthly momentum scores) + strategy : 策略名称 (strategy name) + "relative_momentum" — 相对动量轮动 + "dual_momentum" — 双动量（含绝对动量过滤） + "equal_weight" — 等权基准 + top_k : 持有前 K 只 ETF (hold top-K ETFs) + cost_model : 交易成本模型 (transaction cost model) + bond_etf : 避险 ETF 代码 (safe-haven ETF code) + sector_etfs : 行业 ETF 代码列表 (sector ETF codes) + + 返回 (Returns): + dict 包含: + "nav" : 每日净值序列 (daily NAV series) + "weights" : 月度持仓权重 (monthly weights) + "turnover" : 月度换手率 (monthly turnover) + "cost_total" : 总交易成本 (total transaction cost) + "holdings_log": 持仓记录 (holdings log) + """ + if sector_etfs is None: + sector_etfs = SECTOR_ETFS + if cost_model is None: + cost_model = AShareCostModel() + + # 获取月末再平衡日期 (get month-end rebalancing dates) + try: + monthly_idx = price_df.resample("ME").last().index + except ValueError: + monthly_idx = price_df.resample("M").last().index + + # 预热期：动量需要至少 13 个月数据（12 个月回望 + 1 个月跳过） + # (Warm-up period: momentum needs at least 13 months of data) + warmup = 13 + + # 初始化 (initialization) + nav = pd.Series(index=price_df.index, dtype=float) + weights_log = {} # 每月权重记录 (monthly weights log) + turnover_log = {} # 每月换手率记录 (monthly turnover log) + holdings_log = {} # 每月持仓记录 (monthly holdings log) + total_cost = 0.0 # 累计交易成本 (cumulative transaction cost) + + current_weights = {} # 当前持仓权重 {code: weight} + portfolio_value = 1.0 # 组合净值，从 1.0 开始 (portfolio starts at 1.0) + + rebal_dates = monthly_idx[warmup:] # 有效再平衡日期 (valid rebalance dates) + + for i, rebal_date in enumerate(rebal_dates): + # ── 1. 确定目标权重 (Determine target weights) ────────────────────── + + if strategy == "equal_weight": + # 等权：始终持有所有行业 ETF，各占 1/N + # (Equal weight: always hold all sector ETFs equally) + n = len(sector_etfs) + target_weights = {code: 1.0 / n for code in sector_etfs} + + elif strategy == "relative_momentum": + # 相对动量：选动量得分最高的前 K 只行业 ETF + # (Relative momentum: select top-K sector ETFs by momentum score) + mom_today = momentum_df.loc[:rebal_date, sector_etfs].iloc[-1] + mom_today = mom_today.dropna() + if len(mom_today) == 0: + target_weights = current_weights or {sector_etfs[0]: 1.0} + else: + top_etfs = mom_today.nlargest(top_k).index.tolist() + w = 1.0 / len(top_etfs) + target_weights = {code: w for code in top_etfs} + + elif strategy == "dual_momentum": + # 双动量：先用绝对动量过滤，再做相对动量排序 + # (Dual momentum: filter by absolute momentum, then rank by relative) + mom_today = momentum_df.loc[:rebal_date, sector_etfs].iloc[-1] + mom_today = mom_today.dropna() + + # 绝对动量过滤 (absolute momentum filter): + # 如果行业 ETF 动量 < 0（跑输无风险），则替换为国债 ETF + # (If sector ETF momentum < 0, replace with bond ETF) + positive_etfs = mom_today[mom_today > 0].index.tolist() + + if len(positive_etfs) == 0: + # 全部动量为负：100% 持有国债 ETF 避险 + # (All negative momentum: 100% in bond ETF) + target_weights = {bond_etf: 1.0} + else: + # 从通过绝对动量过滤的 ETF 中，选相对最强的前 K 只 + # (From ETFs passing absolute momentum filter, pick top-K by relative) + k_actual = min(top_k, len(positive_etfs)) + top_etfs = mom_today[positive_etfs].nlargest(k_actual).index.tolist() + w = 1.0 / k_actual + target_weights = {code: w for code in top_etfs} + else: + raise ValueError(f"未知策略 (Unknown strategy): {strategy}") + + # ── 2. 计算换手率并扣除交易成本 (Calculate turnover and deduct costs) ── + + # 所有涉及的 ETF 代码（新旧持仓并集） + all_codes = set(current_weights) | set(target_weights) + turnover = 0.0 + + for code in all_codes: + old_w = current_weights.get(code, 0.0) + new_w = target_weights.get(code, 0.0) + delta = abs(new_w - old_w) # 权重变化 (weight change) + if delta > 1e-6: + turnover += delta + # 对权重变化部分计算成本 (cost on the traded portion) + trade_value = delta * portfolio_value + if new_w > old_w: + # 增仓 → 买入成本 (increase position → buy cost) + cost = cost_model.buy_cost(trade_value) + else: + # 减仓 → 卖出成本（含印花税） + # (decrease position → sell cost including stamp duty) + cost = cost_model.sell_cost(trade_value) + total_cost += cost + portfolio_value -= cost # 成本从组合价值中扣除 (deduct cost from portfolio) + + # 换手率（单边）= 买入总金额 / 组合价值 (one-sided turnover) + turnover_log[rebal_date] = turnover / 2 # 双边换手/2 = 单边换手 + + # ── 3. 更新持仓，计算到下一个再平衡日的收益 + # (Update holdings, compute returns until next rebalance date) + current_weights = target_weights + weights_log[rebal_date] = target_weights.copy() + holdings_log[rebal_date] = list(target_weights.keys()) + + # 确定持仓持续的日期区间 (determine holding period date range) + if i + 1 < len(rebal_dates): + next_rebal = rebal_dates[i + 1] + else: + next_rebal = price_df.index[-1] + + hold_dates = price_df.loc[rebal_date:next_rebal].index + + # 计算每日组合收益率 (calculate daily portfolio return) + for j in range(1, len(hold_dates)): + d_prev = hold_dates[j - 1] + d_curr = hold_dates[j] + daily_port_ret = 0.0 + for code, w in current_weights.items(): + if code in price_df.columns: + p_prev = price_df.loc[d_prev, code] + p_curr = price_df.loc[d_curr, code] + if p_prev > 0: + daily_port_ret += w * (p_curr / p_prev - 1) + portfolio_value *= (1 + daily_port_ret) + nav[d_curr] = portfolio_value + + # 补全初始 NAV（热身期） + nav = nav.ffill() + first_valid = nav.first_valid_index() + if first_valid is not None: + nav[:first_valid] = 1.0 + nav = nav.fillna(1.0) + + return { + "nav": nav, + "weights": weights_log, + "turnover": pd.Series(turnover_log), + "cost_total": total_cost, + "holdings_log": holdings_log, + } + + +# ══════════════════════════════════════════════════════════════════════════════ +# §4 运行三种策略并计算绩效指标 +# Run Three Strategies and Calculate Performance Metrics +# ══════════════════════════════════════════════════════════════════════════════ + +print("§4 运行回测策略 (Running Backtest Strategies)...") +print() + +# 运行三种策略 (run three strategies) +results = {} + +results["equal_weight"] = run_rotation_backtest( + price_df, mom_composite, + strategy="equal_weight", + top_k=3, cost_model=cost_model, +) + +results["relative_momentum"] = run_rotation_backtest( + price_df, mom_composite, + strategy="relative_momentum", + top_k=3, cost_model=cost_model, +) + +results["dual_momentum"] = run_rotation_backtest( + price_df, mom_composite, + strategy="dual_momentum", + top_k=3, cost_model=cost_model, +) + +# 同时计算国债 ETF 纯避险基准 (bond ETF as pure safe-haven benchmark) +bond_nav = price_df[BOND_ETF] / price_df[BOND_ETF].iloc[0] +results["bond_only"] = {"nav": bond_nav} + +# ── 绩效计算函数 (Performance Calculation Function) ───────────────────────── + +def calc_performance(nav: pd.Series, + risk_free_annual: float = 0.02) -> dict: + """ + 计算常用量化绩效指标 (Calculate common quantitative performance metrics) + + 指标 (Metrics): + - 年化收益率 (Annualized Return, CAGR) + - 年化波动率 (Annualized Volatility) + - 夏普比率 (Sharpe Ratio): 超额收益 / 波动率 + - 最大回撤 (Maximum Drawdown, MaxDD): 从历史最高点的最大跌幅 + - 卡玛比率 (Calmar Ratio): 年化收益 / 最大回撤 + - 索提诺比率 (Sortino Ratio): 用下行波动率替代总波动率 + """ + nav = nav.dropna() + if len(nav) < 2: + return {} + + # 日收益率 (daily returns) + daily_ret = nav.pct_change().dropna() + + # 年化收益率 (CAGR - Compound Annual Growth Rate 复合年增长率) + n_years = len(daily_ret) / TRADING_DAYS + total_ret = nav.iloc[-1] / nav.iloc[0] - 1 + cagr = (1 + total_ret) ** (1 / n_years) - 1 + + # 年化波动率 (annualized volatility) + ann_vol = daily_ret.std() * np.sqrt(TRADING_DAYS) + + # 夏普比率 (Sharpe Ratio) + # 公式: (年化收益 - 无风险利率) / 年化波动率 + rf_daily = risk_free_annual / TRADING_DAYS + sharpe = (daily_ret.mean() - rf_daily) / daily_ret.std() * np.sqrt(TRADING_DAYS) + + # 最大回撤 (Maximum Drawdown) + # 定义: max(累计最高净值 - 当日净值) / 累计最高净值 + rolling_max = nav.cummax() # 历史最高净值 (rolling maximum) + drawdown = nav / rolling_max - 1 # 每日回撤序列 (daily drawdown series) + max_dd = drawdown.min() # 最大回撤（负数）(max drawdown, negative) + + # 卡玛比率 (Calmar Ratio = CAGR / |MaxDD|) + calmar = cagr / abs(max_dd) if max_dd != 0 else np.nan + + # 索提诺比率 (Sortino Ratio): 只惩罚下行波动 (penalizes only downside volatility) + downside_ret = daily_ret[daily_ret < rf_daily] + downside_vol = downside_ret.std() * np.sqrt(TRADING_DAYS) if len(downside_ret) > 0 else ann_vol + sortino = (cagr - risk_free_annual) / downside_vol if downside_vol > 0 else np.nan + + # 胜率 (Win Rate): 月度正收益比例 (fraction of months with positive return) + try: + monthly_ret = nav.resample("ME").last().pct_change().dropna() + except ValueError: + monthly_ret = nav.resample("M").last().pct_change().dropna() + win_rate = (monthly_ret > 0).mean() + + return { + "年化收益率 (CAGR)": cagr, + "年化波动率 (Volatility)": ann_vol, + "夏普比率 (Sharpe)": sharpe, + "最大回撤 (MaxDrawdown)": max_dd, + "卡玛比率 (Calmar)": calmar, + "索提诺比率 (Sortino)": sortino, + "月度胜率 (Win Rate)": win_rate, + "累计收益率 (Total Return)": total_ret, + } + + +# ── 打印绩效对比表 (Print Performance Comparison Table) ───────────────────── + +strategy_labels = { + "equal_weight": "等权基准 (EW Benchmark)", + "relative_momentum": "相对动量 (Rel Momentum)", + "dual_momentum": "双动量 (Dual Momentum)", + "bond_only": "国债避险 (Bond Only)", +} + +print("=" * 72) +print("§4 策略绩效对比 (Strategy Performance Comparison)") +print("=" * 72) + +perf_all = {} +metrics_display = [ + ("年化收益率 (CAGR)", "{:>8.2%}"), + ("年化波动率 (Volatility)", "{:>8.2%}"), + ("夏普比率 (Sharpe)", "{:>8.3f}"), + ("最大回撤 (MaxDrawdown)", "{:>8.2%}"), + ("卡玛比率 (Calmar)", "{:>8.3f}"), + ("月度胜率 (Win Rate)", "{:>8.2%}"), + ("累计收益率 (Total Return)", "{:>8.2%}"), +] + +for key, label in strategy_labels.items(): + nav = results[key]["nav"] + perf = calc_performance(nav) + perf_all[key] = perf + +header = f" {'指标':26s}" +for label in strategy_labels.values(): + header += f" {label[:20]:>20s}" +print(header) +print(" " + "-" * 90) + +for metric, fmt in metrics_display: + row = f" {metric:26s}" + for key in strategy_labels: + val = perf_all[key].get(metric, np.nan) + row += " " + fmt.format(val).rjust(20) + print(row) + +print() +# 打印成本统计 +for key in ["equal_weight", "relative_momentum", "dual_momentum"]: + cost = results[key].get("cost_total", 0) + to = results[key].get("turnover", pd.Series()).mean() + print(f" [{strategy_labels[key][:16]}] " + f"总成本={cost*100:.3f}%净值 平均月换手={to*100:.1f}%") +print() + +# ══════════════════════════════════════════════════════════════════════════════ +# §5 持仓分析：统计各行业 ETF 的被选中频次 +# Holdings Analysis: Count How Often Each Sector ETF Was Selected +# ══════════════════════════════════════════════════════════════════════════════ + +print("§5 持仓分析 (Holdings Analysis):") +print() + +for strategy_key in ["relative_momentum", "dual_momentum"]: + holdings_log = results[strategy_key]["holdings_log"] + count = {code: 0 for code in ALL_ETFS} + + for date, held in holdings_log.items(): + for code in held: + count[code] = count.get(code, 0) + 1 + + total_months = len(holdings_log) + label = strategy_labels[strategy_key] + print(f" [{label}] 共 {total_months} 次再平衡:") + + # 按持有频次降序排列 (sort by holding frequency descending) + sorted_count = sorted(count.items(), key=lambda x: x[1], reverse=True) + for code, cnt in sorted_count: + if cnt > 0: + cn_name = ETF_UNIVERSE[code][0] + pct = cnt / total_months + bar = "█" * int(pct * 20) + print(f" {code} {cn_name:10s} {cnt:3d}次 {pct:5.1%} {bar}") + print() + +# ══════════════════════════════════════════════════════════════════════════════ +# §6 可视化：9 宫格图表 +# Visualization: 9-Panel Chart +# ══════════════════════════════════════════════════════════════════════════════ + +print("§6 生成可视化图表 (Generating Charts)...") + +# 配置中文字体 (configure Chinese font) +plt.rcParams["font.sans-serif"] = ["WenQuanYi Zen Hei", "Arial Unicode MS", + "SimHei", "DejaVu Sans"] +plt.rcParams["axes.unicode_minus"] = False + +# 颜色方案 (color scheme) +COLORS = { + "equal_weight": "#95a5a6", # 灰色（基准） + "relative_momentum": "#2ecc71", # 绿色 + "dual_momentum": "#e74c3c", # 红色（主策略） + "bond_only": "#3498db", # 蓝色（国债） +} +LABELS = { + "equal_weight": "等权基准", + "relative_momentum": "相对动量", + "dual_momentum": "双动量", + "bond_only": "国债", +} + +fig = plt.figure(figsize=(20, 16), facecolor="#1a1a2e") +fig.suptitle( + "A 股行业 ETF 轮动策略回测 (A-Share Sector ETF Rotation Backtest)\n" + "合成模拟数据 2019-2024 (Synthetic Data 2019-2024)", + fontsize=16, color="white", fontweight="bold", y=0.98 +) + +gs = gridspec.GridSpec(3, 3, figure=fig, + hspace=0.42, wspace=0.35, + left=0.07, right=0.97, top=0.93, bottom=0.06) + +ax_style = dict(facecolor="#16213e", labelcolor="white", + tick_params=dict(colors="white")) + +def style_ax(ax, title, xlabel="", ylabel=""): + ax.set_facecolor("#16213e") + ax.set_title(title, color="white", fontsize=10, fontweight="bold", pad=6) + ax.set_xlabel(xlabel, color="#aaaaaa", fontsize=8) + ax.set_ylabel(ylabel, color="#aaaaaa", fontsize=8) + ax.tick_params(colors="white", labelsize=7) + ax.spines[:].set_color("#444466") + ax.grid(True, alpha=0.25, color="#666699", linewidth=0.5) + +# ── 图1: 累计净值曲线 (Cumulative NAV) ─────────────────────────────────────── +ax1 = fig.add_subplot(gs[0, :2]) # 跨两列（宽幅） +style_ax(ax1, "① 累计净值对比 Cumulative NAV Comparison", ylabel="净值 NAV") +for key, color in COLORS.items(): + nav = results[key]["nav"] + ax1.plot(nav.index, nav.values, color=color, linewidth=1.8, + label=LABELS[key], alpha=0.9) +ax1.axhline(1.0, color="#888888", linewidth=0.8, linestyle="--", alpha=0.6) +ax1.legend(loc="upper left", fontsize=8, facecolor="#1a1a2e", + labelcolor="white", framealpha=0.7) +ax1.yaxis.set_major_formatter(mtick.FuncFormatter(lambda x, _: f"{x:.1f}x")) + +# ── 图2: 绩效雷达图 (Performance Radar) ───────────────────────────────────── +ax2 = fig.add_subplot(gs[0, 2], projection="polar") +ax2.set_facecolor("#16213e") +ax2.set_title("② 绩效雷达图 Performance Radar", + color="white", fontsize=10, fontweight="bold", pad=15) +ax2.tick_params(colors="white", labelsize=7) +ax2.spines["polar"].set_color("#444466") + +radar_metrics = ["年化收益", "夏普比率", "月胜率", "抗回撤"] +radar_keys = ["年化收益率 (CAGR)", "夏普比率 (Sharpe)", + "月度胜率 (Win Rate)", "卡玛比率 (Calmar)"] +n_radar = len(radar_metrics) +angles = np.linspace(0, 2 * np.pi, n_radar, endpoint=False).tolist() +angles += angles[:1] + +# 归一化各指标到 0-1 (normalize metrics to 0-1) +def normalize_radar(values, min_val=0.0, max_val=1.0): + rng = max_val - min_val + return [(v - min_val) / rng if rng > 0 else 0.5 for v in values] + +radar_strats = ["equal_weight", "relative_momentum", "dual_momentum"] +for rkey in radar_strats: + perf = perf_all[rkey] + raw = [perf.get(m, 0) for m in radar_keys] + # 简单归一化（基于各策略范围） + vals = [ + min(max(raw[0] / 0.30, 0), 1), # CAGR: 0~30% + min(max(raw[1] / 3.0, 0), 1), # Sharpe: 0~3 + min(max(raw[2], 0), 1), # Win rate: 0~1 + min(max(raw[3] / 3.0, 0), 1), # Calmar: 0~3 + ] + vals += vals[:1] + ax2.plot(angles, vals, color=COLORS[rkey], linewidth=2, label=LABELS[rkey]) + ax2.fill(angles, vals, color=COLORS[rkey], alpha=0.15) + +ax2.set_xticks(angles[:-1]) +ax2.set_xticklabels(radar_metrics, color="white", fontsize=8) +ax2.set_yticklabels([]) +ax2.legend(loc="upper right", bbox_to_anchor=(1.35, 1.15), + fontsize=7, facecolor="#1a1a2e", labelcolor="white") + +# ── 图3: 最大回撤曲线 (Drawdown Curve) ────────────────────────────────────── +ax3 = fig.add_subplot(gs[1, :2]) +style_ax(ax3, "③ 水下曲线（回撤） Underwater Chart (Drawdown)", ylabel="回撤 Drawdown") +for key, color in COLORS.items(): + nav = results[key]["nav"] + dd = nav / nav.cummax() - 1 + ax3.fill_between(dd.index, dd.values, 0, color=color, alpha=0.4, label=LABELS[key]) + ax3.plot(dd.index, dd.values, color=color, linewidth=0.8, alpha=0.8) +ax3.set_ylim(-0.65, 0.05) +ax3.yaxis.set_major_formatter(mtick.PercentFormatter(1.0)) +ax3.legend(loc="lower left", fontsize=8, facecolor="#1a1a2e", + labelcolor="white", framealpha=0.7) + +# ── 图4: 绩效指标条形图 (Performance Bar Chart) ────────────────────────────── +ax4 = fig.add_subplot(gs[1, 2]) +style_ax(ax4, "④ 夏普 & 卡玛 Sharpe & Calmar", ylabel="比率 Ratio") +strats_bar = ["equal_weight", "relative_momentum", "dual_momentum"] +labels_bar = [LABELS[k] for k in strats_bar] +sharpes = [perf_all[k]["夏普比率 (Sharpe)"] for k in strats_bar] +calmars = [perf_all[k]["卡玛比率 (Calmar)"] for k in strats_bar] +x_bar = np.arange(len(strats_bar)) +w_bar = 0.35 +bars1 = ax4.bar(x_bar - w_bar/2, sharpes, w_bar, color="#2ecc71", alpha=0.8, + label="夏普 Sharpe") +bars2 = ax4.bar(x_bar + w_bar/2, calmars, w_bar, color="#e74c3c", alpha=0.8, + label="卡玛 Calmar") +ax4.set_xticks(x_bar) +ax4.set_xticklabels(labels_bar, fontsize=7, color="white") +ax4.legend(fontsize=8, facecolor="#1a1a2e", labelcolor="white") +for bar in bars1: + ax4.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 0.02, + f"{bar.get_height():.2f}", ha="center", va="bottom", + color="white", fontsize=7) +for bar in bars2: + ax4.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 0.02, + f"{bar.get_height():.2f}", ha="center", va="bottom", + color="white", fontsize=7) + +# ── 图5: 双动量月度持仓热力图 (Dual Momentum Monthly Holdings Heatmap) ──────── +ax5 = fig.add_subplot(gs[2, :2]) +style_ax(ax5, "⑤ 双动量月度持仓热力图 Dual Momentum Monthly Holdings") + +holdings_log = results["dual_momentum"]["holdings_log"] +all_codes_sorted = SECTOR_ETFS + [BOND_ETF] +etf_names = [ETF_UNIVERSE[c][0] for c in all_codes_sorted] + +heatmap_dates = sorted(holdings_log.keys()) +heatmap_data = np.zeros((len(all_codes_sorted), len(heatmap_dates))) +for j, d in enumerate(heatmap_dates): + held = holdings_log[d] + for i, code in enumerate(all_codes_sorted): + if code in held: + heatmap_data[i, j] = 1.0 / len(held) # 持仓权重 (holding weight) + +im = ax5.imshow(heatmap_data, aspect="auto", cmap="YlOrRd", + interpolation="nearest", vmin=0, vmax=0.5) +ax5.set_yticks(range(len(all_codes_sorted))) +ax5.set_yticklabels(etf_names, fontsize=7, color="white") + +# X 轴：每年一个刻度 (X-axis: one tick per year) +year_ticks = [] +year_labels = [] +for j, d in enumerate(heatmap_dates): + if d.month == 1: + year_ticks.append(j) + year_labels.append(str(d.year)) +ax5.set_xticks(year_ticks) +ax5.set_xticklabels(year_labels, color="white", fontsize=8) +plt.colorbar(im, ax=ax5, fraction=0.02, label="权重 Weight").ax.yaxis.set_tick_params(color="white", labelcolor="white") + +# ── 图6: 月度换手率 (Monthly Turnover) ────────────────────────────────────── +ax6 = fig.add_subplot(gs[2, 2]) +style_ax(ax6, "⑥ 月度换手率 Monthly Turnover", ylabel="换手率 Turnover") +for key in ["equal_weight", "relative_momentum", "dual_momentum"]: + to_series = results[key].get("turnover", pd.Series()) + if len(to_series) > 0: + ax6.plot(to_series.index, to_series.values * 100, + color=COLORS[key], linewidth=1.2, label=LABELS[key], alpha=0.8) + # 平均换手率水平线 (average turnover horizontal line) + avg_to = to_series.mean() * 100 + ax6.axhline(avg_to, color=COLORS[key], linewidth=0.8, + linestyle=":", alpha=0.6) +ax6.yaxis.set_major_formatter(mtick.PercentFormatter()) +ax6.legend(fontsize=7, facecolor="#1a1a2e", labelcolor="white") + +# 保存图表 (save chart) +out_path = "/Users/tigeren/Dev/xorbitlab/trading/etf_rotation_demo.png" +plt.savefig(out_path, dpi=150, bbox_inches="tight", + facecolor=fig.get_facecolor()) +plt.close() +print(f" 图表已保存 (Chart saved): {out_path}") +print() + +# ══════════════════════════════════════════════════════════════════════════════ +# §7 关键学习要点总结 +# Key Learning Takeaways +# ══════════════════════════════════════════════════════════════════════════════ + +print("═" * 60) +print("§7 关键学习要点 (Key Learning Takeaways)") +print("═" * 60) +print(""" + 1. ETF 轮动 vs 个股选股 (ETF Rotation vs. Stock Picking) + ─────────────────────────────────────────────────── + ETF 轮动：买"赛道"，避免个股黑天鹅事件 + (ETF rotation: bet on sectors, avoid individual stock blow-ups) + + 个股选股：需要更强的信息优势，难度更高 + (Stock picking: requires information edge, much harder) + + 2. 绝对动量的价值 (Value of Absolute Momentum) + ─────────────────────────────────────────── + 双动量策略在熊市中会自动切换至国债 ETF 避险 + 最大回撤通常比纯相对动量低 10-15 个百分点 + (Dual momentum auto-switches to bonds in bear markets, + reducing max drawdown by ~10-15 percentage points) + + 3. 动量的局限性 (Momentum Limitations) + ──────────────────────────────────── + • 趋势逆转（动量崩溃）风险：牛熊切换时可能大幅亏损 + (Momentum crash risk: large loss when trend reverses) + • 主题 ETF 历史短：A 股主题 ETF 多数 2019 年后才成立 + (Short history: most A-share thematic ETFs launched post-2019) + • 需要结合估值/基本面作为反向过滤器 + (Should combine with valuation/fundamentals as counter-filter) + + 4. 成本控制 (Cost Management) + ─────────────────────────── + 每月换仓一次约消耗 0.22% 双边成本 + 一年 12 次再平衡 ≈ 2.6% 年化成本拖拽 + (Monthly rebalancing ≈ 0.22% per round-trip, ~2.6% annual drag) + → 适当降低再平衡频率（如季度）可以减少成本 + (Reduce rebalance frequency, e.g., quarterly, to cut costs) + + 5. 进阶方向 (Next Steps) + ─────────────────────── + • 接入真实数据（AKShare）替换合成数据 + (Replace synthetic data with real data via AKShare) + • 加入估值信号（PE 分位数）作为过滤器 + (Add valuation signal: sector PE percentile as filter) + • 加入波动率调仓（高波动期降低持仓比例） + (Add volatility scaling: reduce position in high-vol regimes) + • 结合宏观信号（利率、信用利差）增强国债 ETF 切换时机 + (Combine macro signals for better timing of bond ETF switch) +""") + +print("✅ Demo 完成！(Demo Complete!)") +print(f" 输出文件: etf_rotation_demo.png")