xorbitlab
/
trading
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: add A-Share Sector ETF rotation strategy demo with synthetic data 

- Implemented a comprehensive backtesting framework for ETF rotation strategies including relative momentum, dual momentum, and equal weight benchmark.
- Generated synthetic price data for various A-share sector ETFs and a bond ETF.
- Calculated momentum scores using different lookback periods for robustness.
- Developed a transaction cost model specific to A-share ETFs, accounting for commissions, stamp duty, and slippage.
- Conducted backtests for three strategies and calculated performance metrics including CAGR, volatility, Sharpe ratio, and maximum drawdown.
- Added visualizations for cumulative NAV, performance radar, drawdown curves, and monthly turnover.
- Summarized key takeaways and next steps for further development.
This commit is contained in:
tigerenwork 2026-06-02 00:36:50 +08:00
parent d4bf60767e
commit 7a77d2bf71
3 changed files with 1562 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

714
doc_07_etf_rotation.md Normal file
View File

@ -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轮动）***

BIN
etf_rotation_demo.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 664 KiB

848
quant_etf_rotation_demo.py Normal file
View File

@ -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")