849 lines
40 KiB
Python
849 lines
40 KiB
Python
"""
|
||
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")
|