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

Compare commits

base: xorbitlab:7a77d2bf71b8dd16ed93d70e3a25fa2d1ca645f6
Branches Tags
xorbitlab:main
..
compare: xorbitlab:5519ab9735672d26e1328d8f7fe97c536ac83cb4
Branches Tags
xorbitlab:main

No commits in common. "7a77d2bf71b8dd16ed93d70e3a25fa2d1ca645f6" and "5519ab9735672d26e1328d8f7fe97c536ac83cb4" have entirely different histories.
7a77d2bf71 ... 5519ab9735

4 changed files with 0 additions and 2412 deletions
Show Stats Expand all files Collapse all files

850
doc_06_astock_practice_guide.md
View File

@ -1,850 +0,0 @@
# A 股量化交易：从学习到实盘的完整路径
## 学习路径规划与注意事项指南
> **定位**：本文档是整个演示系列的"导航图"，帮助已完成五篇 Demo 学习的入门者，规划从学习到 A 股实盘的完整路径，并系统指出当前 Demo 的局限性与实盘中必须注意的陷阱。
> **目标读者**：量化入门者，主要关注中国 A 股市场
> **关键提示**：本文中所有带 ⚠️ 标记的内容，是初学者最容易踩坑的地方
---
## 目录
1. [当前五篇 Demo 的知识地图](#1-当前五篇-demo-的知识地图)
2. [A 股特有规则：Demo 与现实的差距](#2-a-股特有规则demo-与现实的差距)
3. [回测七大陷阱](#3-回测七大陷阱)
4. [风险管理：能否活到盈利那天](#4-风险管理)
5. [真实数据接入](#5-真实数据接入)
6. [基本面因子：价格因子的补充](#6-基本面因子)
7. [执行层：从信号到成交](#7-执行层从信号到成交)
8. [推荐工具与平台](#8-推荐工具与平台)
9. [完整学习与实践路径](#9-完整学习与实践路径)
10. [心理与纪律：最后也是最重要的](#10-心理与纪律)
---
## 1. 当前五篇 Demo 的知识地图
### 1.1 已完成的内容
```
┌─────────────────────────────────────────────────────────────┐
│ 量化交易知识体系 │
│ │
│ ✅ 第1篇 数据管道 ← 原始数据清洗、复权、收益率计算 │
│ ✅ 第2篇 策略与向量化回测 ← 技术指标、策略逻辑、绩效评估 │
│ ✅ 第3篇 事件驱动回测 ← 模拟真实撮合、成本模型 │
│ ✅ 第4篇 Alpha 因子研究 ← IC/ICIR、分层回测、因子合成 │
│ ✅ 第5篇 组合优化 ← MVO、风险平价、Black-Litterman │
└─────────────────────────────────────────────────────────────┘
```
### 1.2 五篇 Demo 解决了什么问题？
| Demo | 解决的核心问题 |
|------|---------------|
| 数据管道 | 如何把原始价格数据处理成可用的研究数据 |
| 策略回测 | 如何验证一个交易规则在历史上是否有效 |
| 事件驱动回测 | 如何更真实地模拟实际交易过程 |
| Alpha 因子 | 如何发现能预测未来收益的"信号" |
| 组合优化 | 如何科学地分配多只股票的权重 |
### 1.3 五篇 Demo 没有解决的问题
```
❌ A 股特有的交易规则（T+1、涨跌停、印花税）
❌ 回测结果是否可信（幸存者偏差、过拟合等）
❌ 如果亏损了怎么控制损失（风险管理）
❌ 如何接入真实市场数据
❌ 实际下单时价格和预期不一样怎么办（执行摩擦）
❌ 用财报数据选股（基本面因子）
```
**这些缺失的内容，恰恰是从"Demo 世界"走向"真实 A 股"最关键的桥梁。**
---
## 2. A 股特有规则：Demo 与现实的差距
> ⚠️ **这是全文最重要的章节**。
> 当前五篇 Demo 均使用合成数据，在"理想世界"中运行，与 A 股真实交易规则存在根本性差异。
### 2.1 T+1 交割制度
**规则**：今天（T 日）买入的股票，**最早在 T+1 日才能卖出**。
```
对比:
美股 / 港股: T+0（当天买当天可以卖）
中国 A 股: T+1（今天买，明天才能卖）
例外：科创板、创业板可以融券做空（但普通投资者很难操作）
例外：ETF（交易型基金）支持 T+0 操作
```
**对量化策略的影响**：
| 策略类型 | 影响 |
|---------|------|
| 日内交易（Day Trading）| **完全不可行** — 日内无法平仓 |
| 日频换仓策略 | 需重新考虑成本，换仓频率受限 |
| 反转策略 | 信号当天无法止损，下跌风险暴露更大 |
| 动量策略 | 相对友好，持有周期通常 > 1 天 |
> 💡 **建议**：初学者应聚焦在**持仓周期 ≥ 5 天**的策略，让 T+1 的影响尽量小。
### 2.2 涨跌停板制度
**规则**：A 股每日价格变动有上限：
- 主板（沪深主板）：上下 **±10%**
- 科创板（STAR Market）/ 创业板（ChiNext）：上下 **±20%**
- ST 股（特别处理股票）：上下 **±5%**
**对回测的严重影响**：
```
场景1: 你的策略在 T 日发出"卖出"信号
但 T+1 日开盘直接跌停（-10%）
跌停板无买家，当天根本无法成交！
→ 回测用收盘价计算"已成交"是完全错误的
场景2: 你买入一只股票，连续多个涨停板
每天想追买却买不到
→ 回测"买入成功"但现实中不可能成功
```
**解决方案**：在回测中增加"涨跌停过滤"逻辑：
```python
# 示例：过滤掉当天涨停的股票（无法以涨停价买入）
def is_limit_up(price_today, price_yesterday, limit=0.10):
return price_today >= price_yesterday * (1 + limit - 0.001)
def is_limit_down(price_today, price_yesterday, limit=0.10):
return price_today <= price_yesterday * (1 - limit + 0.001)
# 如果要买入但已涨停 → 放弃或等待次日
# 如果要卖出但已跌停 → 无法成交，记录为"滞留持仓"
```
### 2.3 交易成本：远比想象的高
A 股的完整交易成本结构：
| 成本类型 | 费率 | 收取方 | 说明 |
|---------|------|--------|------|
| **印花税**（Stamp Duty）| **0.1%**（单边） | 国家 | **仅卖出时收取**，买入不收 |
| 证券交易佣金（Commission）| 0.02%~0.03% | 券商 | 买卖都收，部分券商有最低佣金（5 元/笔）|
| 过户费（Transfer Fee）| 0.002%（沪市）| 交易所 | 仅上海证券交易所收取 |
| 融资利率（Margin Interest）| 5%~8%（年化）| 券商 | 使用融资买入时额外收取 |
**单次完整交易的总成本（买入+卖出）**：
```
总成本 ≈ 0.1%（印花税卖） + 0.03%×2（佣金买卖） + 0.002%（过户费沪市）
≈ 0.164%（单次来回）
如果月换手1次（年化12次）：
年化交易成本 ≈ 0.164% × 12 ≈ 2.0%
如果月换手4次（每周换仓）：
年化交易成本 ≈ 0.164% × 48 ≈ 7.9% ← 严重侵蚀收益！
```
> ⚠️ **关键认知**：一个"回测年化收益 15%"的策略，如果实际年化换手率为 600%（每月换仓100%），交易成本就接近 10%，实盘收益只剩约 5%。
### 2.4 A 股没有做空机制（对普通投资者）
| 市场 | 做空机制 |
|------|---------|
| 美股 | 融券做空广泛可行，成本低 |
| 港股 | 大部分股票可以融券做空 |
| A 股主板 | **普通投资者基本无法做空个股** |
| A 股（股指期货）| 可以通过沪深300、中证500期货做空指数 |
**对 Demo 的直接影响**：第4篇（Alpha 因子）§9 展示的"多空组合年化 33%"在 A 股完全无法直接复制。
**A 股量化投资者的变通方案**：
1. **多头因子策略**：只做多高分股票，不做空低分股票（损失多空价差）
2. **股指期货对冲**：买入个股组合 + 做空沪深300期货（对冲系统性风险）
3. **中性化**：权重设计时尽量减少市场 Beta 暴露
### 2.5 其他 A 股特有规则
| 规则 | 内容 | 量化影响 |
|------|------|---------|
| **停牌制度**（Suspension）| 重大事件期间股票可长期停牌（数天至数月）| 持仓被锁，无法止损 |
| **退市制度**（Delisting）| 连续亏损等情形会退市 | 回测数据需包含已退市股票 |
| **集合竞价**（Call Auction）| 09:25前委托，09:30后才成交 | 开盘价 ≠ 信号时点价格 |
| **大宗交易**（Block Trade）| 大额交易需走特殊通道 | 大资金策略容量受限 |
| **北向资金**（Northbound）| 陆股通资金流向影响 A 股短期走势 | 可作为情绪指标 |
---
## 3. 回测七大陷阱
> ⚠️ 一个漂亮的回测曲线，可能只是一个"谎言"。在信任任何回测结果之前，必须对以下七个陷阱逐一排查。
### 陷阱1：幸存者偏差（Survivorship Bias）
**问题**：如果你的股票池只包含"今天仍在交易的股票"，就自动排除了所有在回测期间**退市、被 ST、长期停牌**的股票——而这些正是最差的股票。
```
错误做法（Demo 中的做法）:
2025年选取50只股票池 → 回测2020-2025年
这50只股票全部存活到2025年，已经是幸存者！
那些2022年退市的烂股，从股票池中消失了
正确做法:
回测2020年时，股票池应该是"2020年可投资的全部股票"
包括后来退市的股票
```
**A 股幸存者偏差的严重程度**：每年有数十只股票退市或被 ST，每年也有数百只新股上市（IPO）。忽略退市/ST 股票，策略的真实表现会被高估约 **1-3%/年**。
### 陷阱2：前视偏差（Look-Ahead Bias）
**问题**：在回测中使用了"当时还不可能知道"的数据。
```
常见案例:
① 财报前视偏差: 用"第三季度财报数据"买入，
但第三季度财报通常10月底才发布
→ 9月份就用了10月才有的数据 ✗
② 复权前视偏差: 用"前复权"价格回测时，
历史价格会因未来的分红而被修改
→ 用了未来分红信息 ✗（需用后复权或不复权）
③ 指数成分前视偏差: 用"当前沪深300成分股"回测历史，
但该成分股是现在才进入的
→ 历史上你不可能知道它会进入 ✗
```
**A 股财报披露时间表**（设计因子时必须遵守）：
| 财报类型 | 披露截止日 |
|---------|-----------|
| 年报（Annual Report）| 次年 **4月30日** 前 |
| 一季报（Q1）| **4月30日** 前 |
| 半年报（Interim）| **8月31日** 前 |
| 三季报（Q3）| **10月31日** 前 |
> 💡 **规则**：使用财报数据时，必须在**披露截止日 + 1 天**之后才能"知道"该数据。
### 陷阱3：过度拟合（Overfitting）
**问题**：在历史数据上反复调整参数，找到"完美契合历史"的组合，但这个组合实质上只是在记忆历史噪音，而非真正发现了规律。
```
典型案例:
参数空间: MA 短窗口 5~30，长窗口 20~120
遍历 26×101 = 2626 个参数组合
每次都在同一段历史数据上计算夏普比率
最终选出夏普最高的那对参数
→ 这对参数只是恰好适合这段历史，样本外可能是最差的
```
**检验方法**：
```
正确的参数验证流程:
│─────── 训练集 70% ─────────│── 验证集 15% ──│── 测试集 15% ──│
↑ ↑ ↑
调参优化 选择最终参数 只用一次！
（评估真实表现）
测试集只能使用一次：使用后就"消耗"了，不能再用它来调参
```
### 陷阱4：数据窥探偏差（Data Snooping Bias）
**问题**：学术界或行业内已发表了数百种"有效"因子，但这些因子大多是在相同的历史数据上发现并验证的。这种"挖矿"行为本身就引入了偏差——即使随机策略，也总能找到少数"偶然有效"的参数。
**量化估算**：如果测试 100 个随机策略，在 5% 显著性水平下，期望有 5 个"偶然"通过检验。
**防范建议**：
- 策略应有明确的经济学逻辑，而不仅是统计拟合
- 在不同市场/时间段进行样本外验证
- 使用更严格的统计显著性阈值（如 1%）
- 报告所有尝试过的策略，而非只报告成功的
### 陷阱5：市场冲击忽略（Ignoring Market Impact）
**问题**：回测假设可以在任何时间以收盘价无限量成交，但现实中大额买卖会推动价格。
```
例子:
某策略买入信号出现在收盘后
次日买入该股票 10,000 股（市值 50 万元）
该股票日均成交额 200 万元
→ 你的买单占了日成交量的 25%！
→ 你的买入行为会把价格推高 1-3%（市场冲击成本）
```
**A 股容量限制**：
- 小盘股（日均成交额 500 万以下）：策略容量 < 50 万
- 中盘股（日均成交额 5000 万）：策略容量 < 500 万
- 大盘股（日均成交额 1 亿+）：策略容量 > 1000 万
### 陷阱6：未来函数（Future Function）
**问题**：代码层面的 Bug，使得某个时刻用到了它"不应该知道"的数据。
```python
# 错误示例（常见 Bug）:
signal = df['close'].rolling(20).mean() # 计算20日均线
# 问题: pandas 默认使用当日收盘价在内的20天！
# 如果在开盘时交易，当日收盘价还不知道
# 正确写法: 信号基于前一日的完整数据
signal = df['close'].shift(1).rolling(20).mean() # shift(1) 使用昨日收盘
```
### 陷阱7：交易成本低估（Underestimating Costs）
如上文 §2.3 所述。很多初学者的回测根本不扣除成本，或只扣佣金而忽略印花税。
**建议的保守成本设定**（适用于 A 股散户量化）：
| 换仓频率 | 建议预留成本 | 对应净年化阈值 |
|---------|------------|--------------|
| 月频（每月一次）| 0.3%/次 | 策略年化 > 4% 才有意义 |
| 周频（每周一次）| 0.3%/次 | 策略年化 > 16% 才有意义 |
| 日频（每天换仓）| 0.3%/次 | 策略年化 > 78% 才有意义（基本无法盈利）|
---
## 4. 风险管理
> ⚠️ **风险管理是量化投资中最被低估、最重要的内容。**
> 一个 "普通好" 的策略 + 优秀的风险管理 > 一个 "非常好" 的策略 + 没有风险管理。
### 4.1 为什么风险管理是第一位的？
```
数学现实:
亏损 50% → 需要盈利 100% 才能回本
亏损 30% → 需要盈利 43% 才能回本
亏损 20% → 需要盈利 25% 才能回本
结论: 不亏钱比赚钱更重要。大亏损后的"恢复期"可能长达数年。
```
### 4.2 仓位管理（Position Sizing）
**固定比例法（Fixed Fractional）**：每笔交易不超过总资产的固定比例。
```python
# 每笔交易最多动用总资产的 N%
MAX_POSITION_PCT = 0.05 # 每只股票最多 5%
MAX_PORTFOLIO_PCT = 0.80 # 总仓位最高 80%（留 20% 现金）
shares_to_buy = int(capital * MAX_POSITION_PCT / price)
```
**Kelly 公式**（理论最优仓位）：
$$f^* = \frac{p \cdot b - q}{b}$$
其中 $p$ = 胜率，$q = 1-p$ = 败率，$b$ = 盈亏比（平均盈利/平均亏损）。
> 💡 **实践建议**：Kelly 仓位往往偏激进，实际使用"半 Kelly"（$0.5 \times f^*$）更稳健。
### 4.3 止损机制（Stop-Loss）
| 止损类型 | 触发条件 | 适用场景 |
|---------|---------|---------|
| 硬止损（Hard Stop）| 亏损超过固定百分比（如 -8%）立即止损 | 所有策略都应有 |
| 追踪止损（Trailing Stop）| 从最高点回落超过固定幅度（如 -15%）| 趋势策略 |
| 时间止损（Time Stop）| 持仓超过预期天数但未达到目标，离场 | 短期策略 |
| 组合止损（Portfolio Stop）| 整体组合单日/单月亏损超阈值，全部减仓 | 所有策略 |
**A 股止损的特殊考量**：
- 涨跌停板可能导致**止损单无法成交**（挂单但无法匹配）
- 解决方案：止损条件触发后，设置"次日开盘卖出"而非"即刻市价单"
### 4.4 最大回撤控制（Max Drawdown Control）
```python
# 监控组合回撤，超过阈值自动降仓
def compute_drawdown(nav_series):
rolling_max = nav_series.cummax()
drawdown = nav_series / rolling_max - 1.0
return drawdown
MAX_DD_THRESHOLD = -0.15 # 回撤超过 15% → 降至半仓
STOP_DD_THRESHOLD = -0.25 # 回撤超过 25% → 清仓，停止交易
current_dd = compute_drawdown(portfolio_nav)
if current_dd < STOP_DD_THRESHOLD:
# 发出清仓信号，停止策略运行
pass
elif current_dd < MAX_DD_THRESHOLD:
# 将所有仓位减半
pass
```
### 4.5 VaR 与 CVaR（风险度量）
**在险价值 VaR (Value at Risk)**：在给定置信水平下，**未来某段时间内可能损失的最大金额**。
$$\text{VaR}_{95\%,1D} = -\text{Percentile}_{5\%}(r_{daily}) \times \text{Portfolio Value}$$
含义：有 95% 的把握，明天的损失不超过这个金额。
**条件在险价值 CVaR / ES (Conditional VaR / Expected Shortfall)**：
$$\text{CVaR}_{95\%} = -E[r \mid r < \text{VaR}_{5\%}]$$
含义：当损失超过 VaR 时，**平均损失有多大**。CVaR 比 VaR 更保守，更适合度量尾部风险。
```python
import numpy as np
def compute_var_cvar(daily_returns, confidence=0.95):
var = -np.percentile(daily_returns, (1 - confidence) * 100)
cvar = -daily_returns[daily_returns < -var].mean()
return var, cvar
var_95, cvar_95 = compute_var_cvar(portfolio_returns)
print(f"日 VaR(95%): 损失超过 {var_95:.2%} 的概率 < 5%")
print(f"日 CVaR(95%): 当超过 VaR 时，平均损失 {cvar_95:.2%}")
```
---
## 5. 真实数据接入
### 5.1 主要数据源对比
| 数据源 | 费用 | 数据质量 | A 股覆盖 | 适合用途 |
|--------|------|---------|---------|---------|
| **AKShare** | 免费 | 中等 | 全覆盖 | 学习、价格/财务数据 |
| **Tushare Pro** | 免费（积分制） | 较高 | 全覆盖 | 学习到轻量实战 |
| **聚宽 JoinQuant** | 免费额度 | 高 | 全覆盖 | 回测研究 |
| **掘金量化** | 付费/免费模拟 | 高 | 全覆盖 | 模拟盘到实盘 |
| **Wind** | 付费（贵）| 最高 | 最全 | 专业机构 |
| **同花顺 iFinD** | 付费 | 高 | 全覆盖 | 专业机构 |
**入门推荐路线**：
```
第一阶段（纯学习）: AKShare（完全免费）
↓
第二阶段（研究）: Tushare Pro（免费注册，积分获取数据）
↓
第三阶段（模拟盘）: 聚宽 / 掘金量化（含模拟交易环境）
↓
第四阶段（实盘）: 券商 API（XTP / CTP）或 vnpy 框架
```
### 5.2 AKShare 快速上手示例
```python
import akshare as ak
# 获取 A 股日线行情（前复权）
df = ak.stock_zh_a_hist(
symbol="000001", # 平安银行
period="daily",
start_date="20230101",
end_date="20231231",
adjust="qfq" # qfq = 前复权 / hfq = 后复权
)
# 获取沪深300成分股列表
hs300 = ak.index_stock_cons(symbol="000300")
# 获取 A 股财务数据（资产负债表）
balance_sheet = ak.stock_financial_abstract_ths(symbol="000001", indicator="资产负债表")
```
### 5.3 Tushare Pro 快速上手示例
```python
import tushare as ts
# 初始化（需要注册获取 token）
pro = ts.pro_api('your_token_here')
# 获取日线行情
df = pro.daily(
ts_code='000001.SZ', # 深圳股票加 .SZ，上海加 .SH
start_date='20230101',
end_date='20231231'
)
# 获取所有 A 股列表（含已退市，解决幸存者偏差）
stocks = pro.stock_basic(
exchange='', # 不指定则获取全部
list_status='L', # L=上市中 D=退市 P=暂停
fields='ts_code,symbol,name,area,industry,list_date,delist_date'
)
# 获取财务数据（解决前视偏差，有 ann_date 披露日字段）
income = pro.income(
ts_code='000001.SZ',
fields='ts_code,ann_date,end_date,total_revenue,net_income'
)
# ann_date: 实际公告日期（用这个！）
# end_date: 报告期截止日（用于标识是哪个季度）
```
### 5.4 数据处理中必须处理的 A 股特有问题
**问题1：退市股票的幸存者偏差**
```python
# 获取所有股票（包括退市的），按照每个历史时点重建股票池
all_stocks = pro.stock_basic(list_status='L') # 上市中
delisted = pro.stock_basic(list_status='D') # 已退市
# 合并后，在回测中对每个历史日期，只使用"当时已上市未退市"的股票
```
**问题2：财报披露延迟**
```python
# 使用 ann_date（公告日）而非 end_date（报告期末）
# 这确保你只在数据公开披露后才能"看到"它
df = df.sort_values('ann_date')
# 在回测中，只使用 ann_date <= 当前回测日期 的财报数据
```
**问题3：ST 股过滤**
```python
# ST、*ST 股票风险极高，通常需要过滤掉
is_st = df['name'].str.contains('ST')
df_clean = df[~is_st]
```
---
## 6. 基本面因子
### 6.1 为什么需要基本面因子？
当前 Demo 中的五个因子（MOM、REV、LVOL、BAB、ILLIQ）全部基于**价格和成交量**，称为"技术因子"或"量价因子"。
基本面因子（Fundamental Factors）基于公司财务数据，反映公司的内在价值：
| 因子类型 | 数据来源 | 信号持续性 | 换仓频率 |
|---------|---------|----------|---------|
| 量价因子（技术因子）| 日线行情 | 1天~3个月 | 日频~月频 |
| 基本面因子 | 财务报表 | 3个月~数年 | 季频~年频 |
### 6.2 主要基本面因子一览
**价值类因子（Value Factors）**：
| 因子 | 公式 | 含义 |
|------|------|------|
| B/P（账净比）| 净资产 / 市值 | 越高越"便宜"（价值股）|
| E/P（盈市比）| 净利润 / 市值 | 越高越低估 |
| S/P（营收市值比）| 营业收入 / 市值 | 避免亏损股影响 |
| EBITDA/EV | EBITDA / 企业价值 | 更完整的估值指标 |
**质量类因子（Quality Factors）**：
| 因子 | 公式 | 含义 |
|------|------|------|
| ROE | 净利润 / 净资产 | 股东权益回报率 |
| ROA | 净利润 / 总资产 | 资产回报率 |
| 毛利率 | 毛利润 / 营业收入 | 定价能力 |
| 应计比率（Accrual）| (净利润 - 经营现金流) / 总资产 | 低值 = 利润质量高 |
**成长类因子（Growth Factors）**：
| 因子 | 含义 |
|------|------|
| 营收同比增速 | 当季营收 vs 去年同季 |
| 净利润 YoY | 净利润年同比增速 |
| 超预期盈利（SUE）| (实际盈利 - 预期盈利) / 波动率 |
### 6.3 A 股基本面因子的特殊性
1. **季节性强**：A 股年报集中在 3-4 月披露，导致每年 4 月底有明显的"财报效应"
2. **财务造假风险**：A 股部分公司财报质量较差，需要交叉验证（对比现金流）
3. **ROE 的 DuPont 分解**更有用：`ROE = 净利率 × 资产周转率 × 杠杆比率`
---
## 7. 执行层：从信号到成交
### 7.1 信号 → 成交的完整链路
```
信号生成 订单生成 执行 成交
策略逻辑 → "买入 S01" → 下市价单 → 发送到交易所 → 实际成交
↓
时间延迟 / 滑点 / 市场冲击
```
### 7.2 滑点（Slippage）
**定义**：实际成交价格与信号触发时预期价格之间的差异。
**主要来源**：
- 时间延迟：从信号生成到订单到达交易所需要时间
- 买卖价差（Bid-Ask Spread）：买价总高于卖价
- 订单流向：你的订单与其他订单竞争，推动价格
**A 股典型滑点估算**（中小盘股）：0.1%~0.3% 单边
**在回测中模拟滑点**：
```python
SLIPPAGE_RATE = 0.002 # 0.2% 单边滑点
# 买入时支付更高价格
actual_buy_price = signal_price * (1 + SLIPPAGE_RATE)
# 卖出时收到更低价格
actual_sell_price = signal_price * (1 - SLIPPAGE_RATE)
```
### 7.3 A 股的订单类型
| 订单类型 | 说明 | 优点 | 缺点 |
|---------|------|------|------|
| 限价单（Limit Order）| 指定价格，不成交则等待 | 价格确定 | 可能无法成交 |
| 市价单（Market Order）| 以当前最优价立即成交 | 必定成交 | 价格不确定，滑点大 |
| 最优五档即时成交 | A 股特有，匹配5档后剩余撤销 | 平衡 | 大单可能只成交部分 |
**量化策略推荐**：使用**限价单**（设置在当前卖一价或稍微高一点），避免市价单的大滑点。
### 7.4 算法交易（Algorithmic Execution）
对于资金量较大的策略，需要将大单拆分执行，避免冲击市场：
| 算法 | 全称 | 适用场景 |
|------|------|---------|
| VWAP | Volume-Weighted Average Price（成交量加权均价）| 跟踪市场均价，减少冲击 |
| TWAP | Time-Weighted Average Price（时间加权均价）| 均匀分散，简单易实现 |
| POV | Percentage of Volume（跟量执行）| 按市场成交量的固定比例跟随 |
---
## 8. 推荐工具与平台
### 8.1 编程框架
| 框架 | 语言 | 特点 | 适用阶段 |
|------|------|------|---------|
| **vnpy** | Python | 国内最流行的量化框架，支持多个券商 API | 进阶→实盘 |
| **backtrader** | Python | 成熟的回测框架，社区活跃 | 学习→回测 |
| **zipline** | Python | Quantopian 开源，学术界常用 | 学习→回测 |
| **qlib** | Python | 微软开源，机器学习量化框架 | 进阶研究 |
### 8.2 研究与回测平台（含数据）
| 平台 | 费用 | 特点 | 推荐程度 |
|------|------|------|---------|
| **聚宽 JoinQuant** | 免费额度 | 数据+回测一体，适合 A 股 | ⭐⭐⭐⭐⭐（首推）|
| **掘金量化** | 免费+付费 | 支持模拟盘/实盘，数据全 | ⭐⭐⭐⭐⭐（首推）|
| **优矿 UqerIO** | 部分免费 | 与万得数据合作，数据质量高 | ⭐⭐⭐⭐ |
| **米筐 RiceQuant** | 部分免费 | Ricequant 系，稳定 | ⭐⭐⭐ |
### 8.3 实盘接入（进阶）
| 接口 | 说明 |
|------|------|
| **XTP（极速交易平台）**| 中泰证券开放的高性能交易接口，vnpy 支持 |
| **CTP（综合交易平台）**| 主要用于期货/期权，上期所/中金所标准接口 |
| **QMT（量化迷你终端）**| 部分券商提供，适合散户量化（如华泰、国金）|
| **miniQMT** | QMT 的轻量版，支持 Python 调用，适合个人 |
### 8.4 必读书单
| 书名 | 作者 | 重点内容 | 适合阶段 |
|------|------|---------|---------|
| 《量化投资：以 Python 为工具》| 蔡立耑 | 入门，代码实践 | 入门 |
| 《主动投资组合管理》| Grinold & Kahn | 因子模型、信息比率 | 进阶 |
| 《Advances in Financial Machine Learning》| Marcos López | 机器学习 + 量化陷阱 | 进阶 |
| 《打败市场的小书》| Joel Greenblatt | 基本面量化思维 | 入门 |
| 《风险、不确定性与利润》| Frank Knight | 理解真正的风险 | 思想基础 |
---
## 9. 完整学习与实践路径
### 9.1 路径总览图
```
阶段0: 理论基础（已完成 ✅）
├── 数据管道
├── 策略与回测
├── 事件驱动回测
├── Alpha 因子研究
└── 组合优化
↓ 当前位置
阶段1: 补充关键知识（建议 1-2 个月）
├── 学习 A 股规则（本文档）
├── 回测陷阱专项学习
└── 风险管理基础
↓
阶段2: 真实数据研究（建议 2-3 个月）
├── 接入 Tushare/AKShare
├── 用真实 A 股数据复现 Demo
├── 验证因子在真实数据上的有效性
└── 发现并修正前视偏差/幸存者偏差
↓
阶段3: 模拟盘验证（建议 3-6 个月）
├── 在聚宽/掘金搭建完整策略
├── 上线模拟盘（实时信号，不用真钱）
├── 记录实际信号 vs 回测信号的差异
└── 评估交易成本的真实影响
↓
阶段4: 小资金实盘（建议先用 1-5 万元）
├── 目的：学习，不是赚钱
├── 体验 T+1、涨跌停的实际影响
├── 建立完整的交易记录与复盘习惯
└── 以月为单位评估策略，不要日内焦虑
↓
阶段5: 策略完善与扩大
├── 根据实盘经验修正回测假设
├── 逐步增加资金（只用"闲钱"）
└── 持续研究新因子，维持策略有效性
```
### 9.2 阶段1：补充关键知识（推荐 Demo 顺序）
| 优先级 | 建议主题 | 核心内容 |
|--------|---------|---------|
| 🔴 必做 | **A 股回测陷阱与真实成本** | T+1 模拟、涨跌停过滤、完整成本扣除 |
| 🔴 必做 | **风险管理专题** | 仓位管理、止损、VaR/CVaR、回撤控制 |
| 🟡 重要 | **基本面因子研究** | PE/PB/ROE 因子，财报时间处理 |
| 🟡 重要 | **真实数据管道** | Tushare/AKShare 接入，解决幸存者偏差 |
| 🟢 选做 | **机器学习选股** | 随机森林、GBDT 预测收益 |
| 🟢 选做 | **市场情绪与另类数据** | 新闻情感分析、北向资金、融资融券 |
### 9.3 阶段2：用真实数据验证
**验证流程**：
```
① 选择标准化 A 股股票池:
• 沪深300成分股（流动性好，容易成交）
• 中证500成分股（中盘，Alpha 机会更多）
• 全 A 股（完整，但需处理幸存者偏差）
② 数据准备 Checklist:
□ 使用后复权价格（避免复权带来的前视偏差）
□ 包含已退市股票（解决幸存者偏差）
□ 财报数据使用 ann_date（解决财报前视偏差）
□ 过滤 ST、停牌股票
③ 因子验证:
□ 计算 IC / ICIR（目标：|IC均值| > 0.03）
□ 分层回测（检验单调性）
□ 扣除完整交易成本后，多空收益是否为正
④ 如果 IC 接近 0 或收益为负:
→ 该因子在真实 A 股中无效，需要换因子或调整
→ 不要试图调参"拟合"出好看的结果（过拟合陷阱！）
```
### 9.4 阶段3：模拟盘的正确姿势
```
❌ 错误做法:
开模拟盘 → 发现和回测差距很大 → 调整回测参数 → 重新上线
（这是在用模拟盘的结果"优化"回测，过拟合链条更长了）
✅ 正确做法:
每天记录: 策略信号 + 实际执行价格 + 回测预期价格
每月复盘: 实盘 vs 回测的差距在哪里？是成本？还是执行？
不要修改策略参数（除非发现明确的 Bug）
坚持至少 3 个月，收集足够的统计样本
```
---
## 10. 心理与纪律
> 这是整个学习路径中唯一不涉及编程的部分，但却是**决定最终成败**的关键。
### 10.1 量化交易的心理误区
| 误区 | 真实情况 |
|------|---------|
| "回测好 = 实盘好" | 回测是过去，实盘是未来；市场会变化，策略会失效 |
| "亏损了是模型有问题" | 短期亏损可能只是正常的随机波动，不要频繁换策略 |
| "我找到了一个完美策略" | 不存在永远有效的策略，所有策略都会有失效期 |
| "只要技术够好就能赚钱" | 纪律执行、风险控制、情绪管理同样关键 |
### 10.2 量化投资者的核心纪律
```
① 严格执行信号，不用直觉覆盖（除非明确发现 Bug）
"我觉得今天大盘要跌，不买了" → 这是在破坏策略的统计基础
② 以月或季度为周期评估策略，不以日为单位焦虑
连续亏 3 天不代表策略失效；连续亏 3 个月可能需要检查
③ 只用"闲钱"做量化
用生活必需金做量化 → 心理压力导致执行变形 → 放大亏损
④ 先求"不亏大钱"，再求"稳定盈利"
设定严格的止损和最大回撤限制，在学习期保住本金
⑤ 保持持续学习，不断迭代
A 股市场在变化，因子会失效，策略需要更新
量化不是"一劳永逸"，而是持续的研究工作
```
### 10.3 资金管理的"三个账户"思维
```
生活资金（绝对不能动）
↓
投资储蓄（传统股票/基金，长期持有）
↓
量化研究资金（可以全部亏损的部分）
→ 初期建议：不超过总可投资资产的 20%
→ 入门期建议：1-5 万元
```
### 10.4 成功的量化投资者的特征
- **工程师思维**：系统化、规则化，不依赖感觉
- **科学家心态**：愿意承认假设错误，不断迭代
- **风险第一意识**：先问"最坏情况是什么"，再问"最好情况是什么"
- **长期主义**：量化策略的优势在于长期统计，不要被短期噪音干扰
- **持续复盘**：每笔交易、每月业绩都有完整记录
---
## 总结：最重要的 10 点
1. ⚠️ **T+1 + 涨跌停 + 印花税** — 这三个 A 股规则让所有 Demo 结果都需要打折扣
2. ⚠️ **幸存者偏差** — 必须使用包含已退市股票的完整股票池
3. ⚠️ **财报前视偏差** — 使用财报数据时必须遵守 ann_date
4. ⚠️ **交易成本** — 月频换仓约消耗年化 3-4%，日频策略几乎不可能盈利
5. ⚠️ **A 股无法做空** — 多空策略需要用股指期货替代，且有门槛
6. 🔑 **风险管理第一** — 止损、仓位控制、回撤限制，保住本金才有未来
7. 🔑 **模拟盘 3 个月以上** — 收集足够数据再判断策略是否有效
8. 🔑 **小资金验证** — 实盘最初只用 1-5 万，目标是学习而非赚钱
9. 🔑 **严格执行信号** — 不要用直觉覆盖量化信号
10. 🔑 **持续迭代** — 量化不是一劳永逸，市场会变化，策略需要更新
---
## 附录：本系列文档导航
| 篇 | 文件 | 文档 | 核心内容 |
|----|------|------|----------|
| 第 1 篇 | `quant_data_pipeline_demo.py` | `doc_01_data_pipeline.md` | 复权、收益率、缺失值、异常值 |
| 第 2 篇 | `quant_strategy_backtest_demo.py` | `doc_02_strategy_backtest.md` | 技术指标、策略逻辑、向量化回测 |
| 第 3 篇 | `quant_event_driven_backtest_demo.py` | `doc_03_event_driven_backtest.md` | 事件驱动架构、成本模型 |
| 第 4 篇 | `quant_alpha_factor_demo.py` | `doc_04_alpha_factor.md` | 因子构建、IC/ICIR、分层回测 |
| 第 5 篇 | `quant_portfolio_optimization_demo.py` | `doc_05_portfolio_optimization.md` | MVO、风险平价、Black-Litterman |
| **导航** | — | **`doc_06_astock_practice_guide.md`** | **A股规则、路径规划、实盘注意事项** |

714
doc_07_etf_rotation.md
View File

@ -1,714 +0,0 @@
# 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
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 664 KiB

848
quant_etf_rotation_demo.py
View File

@ -1,848 +0,0 @@
"""
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")