""" Task 6: 回撤诊断 分析维度： 6.1 最大回撤复盘 - 2022-05 前后持仓与动量变化 6.2 近期回撤趋势 - 2026 年回撤分析 6.3 极端尾部风险 - 极端日归因 """ import sys from pathlib import Path from collections import defaultdict from typing import Dict, List import numpy as np import pandas as pd sys.path.insert(0, str(Path(__file__).parent.parent.parent)) from rotation.experiments.common import ( load_nav, load_signals, load_detail_days, load_detail_meta, print_section, compute_drawdown, ) def analyze_max_drawdown(days: List[dict], nav: pd.DataFrame): """6.1 最大回撤复盘 (2022-05)""" print_section("6.1 最大回撤复盘 (2022-04 ~ 2022-06)") # 找到最大回撤的谷底 nav_df = nav.copy() dd = compute_drawdown(nav_df['nav']) trough_idx = dd.idxmin() trough_date = nav_df.iloc[trough_idx]['date'] trough_dd = dd.iloc[trough_idx] print(f" 最大回撤谷底: {trough_date.date()}") print(f" 最大回撤: {trough_dd:.2%}") # 找到回撤开始点（前高点） peak_idx = nav_df['nav'][:trough_idx+1].idxmax() peak_date = nav_df.iloc[peak_idx]['date'] peak_nav = nav_df.iloc[peak_idx]['nav'] print(f" 回撤起点(前高): {peak_date.date()}, NAV={peak_nav:.4f}") print(f" 回撤持续: {(trough_date - peak_date).days} 天") # 找到恢复点 recovery_mask = nav_df['nav'].iloc[trough_idx:] >= peak_nav if recovery_mask.any(): recovery_idx = recovery_mask.idxmax() recovery_date = nav_df.iloc[recovery_idx]['date'] print(f" 回撤恢复: {recovery_date.date()}, 恢复耗时: {(recovery_date - trough_date).days} 天") else: print(f" 回撤未恢复（截至数据结束）") # 2022-04 ~ 2022-06 详细持仓变化 print(f"\n 2022-04 ~ 2022-06 持仓与动量变化:") days_2022 = [d for d in days if d['date'] >= '2022-04-01' and d['date'] <= '2022-06-30'] prev_holdings = [] key_events = [] for day in days_2022: holdings = day.get('holdings', []) is_rebal = day.get('is_rebalance', False) if is_rebal: added = set(holdings) - set(prev_holdings) removed = set(prev_holdings) - set(holdings) # 收集动量得分 momentums = {} for code, asset in day.get('assets', {}).items(): m = asset.get('momentum') if m is not None: momentums[code] = m key_events.append({ 'date': day['date'], 'nav': day['nav'], 'daily_return': day['daily_return'], 'added': added, 'removed': removed, 'holdings': holdings, 'momentums': momentums, }) prev_holdings = holdings print(f" 期间调仓事件数: {len(key_events)}") for evt in key_events: mom_str = ', '.join(f"{c}={v:.4f}" for c, v in sorted(evt['momentums'].items(), key=lambda x: -x[1])[:5]) print(f" {evt['date']}: NAV={evt['nav']:.4f}, " f"日收益={evt['daily_return']:+.4%}") print(f" 调仓: +{evt['added']} -{evt['removed']}") print(f" 持仓: {evt['holdings']}") print(f" Top5动量: {mom_str}") # 分析为什么动态阈值没有触发防御 print(f"\n 动态阈值分析 (2022-04 ~ 2022-06):") for day in days_2022: assets = day.get('assets', {}) bond = assets.get('931862.CSI', {}) bond_m = bond.get('momentum') threshold = bond.get('threshold', 0) if bond_m is not None: holdings = day.get('holdings', []) has_bond = '931862.CSI' in holdings # 只在调仓日或每周输出一次 if day.get('is_rebalance') or day['date'].endswith('-01') or day['date'].endswith('-15'): non_bond_above = sum(1 for c, a in assets.items() if c != '931862.CSI' and a.get('above_threshold', False)) print(f" {day['date']}: 短债动量={bond_m:.6f}, " f"阈值={threshold:.6f}, 持仓={holdings}, " f"非债券>阈值: {non_bond_above}") return {'trough_date': str(trough_date.date()), 'max_dd': float(trough_dd)} def analyze_recent_drawdown(days: List[dict], nav: pd.DataFrame): """6.2 2026 年近期回撤分析""" print_section("6.2 2026 年近期回撤分析") nav_2026 = nav[nav['date'].dt.year == 2026].copy() if len(nav_2026) == 0: print(" 无 2026 年数据") return {} dd_2026 = compute_drawdown(nav_2026['nav']) max_dd_2026 = dd_2026.min() trough_idx = dd_2026.idxmin() trough_date = nav_2026.iloc[trough_idx - nav_2026.index[0]]['date'] if trough_idx >= nav_2026.index[0] else None print(f" 2026 年最大回撤: {max_dd_2026:.2%}") # 2026 年月度收益 nav_2026['month'] = nav_2026['date'].dt.month print(f"\n 2026 年月度收益:") for month, grp in nav_2026.groupby('month'): m_ret = grp['nav'].iloc[-1] / grp['nav'].iloc[0] - 1 m_dd = compute_drawdown(grp['nav']).min() print(f" {month}月: 收益={m_ret:+.2%}, 月内最大回撤={m_dd:.2%}") # 2026 年极端日 extreme_2026 = nav_2026[(nav_2026['daily_return'] > 0.03) | (nav_2026['daily_return'] < -0.03)] print(f"\n 2026 年极端日(|收益|>3%):") days_2026_detail = [d for d in days if d['date'].startswith('2026')] for _, row in extreme_2026.iterrows(): date_str = row['date'].strftime('%Y-%m-%d') day_detail = None for d in days_2026_detail: if d['date'] == date_str: day_detail = d break holdings_info = "" if day_detail: for code, asset in day_detail.get('assets', {}).items(): if asset.get('is_held'): etf_ret = asset.get('etf_return_ctc', 0) holdings_info += f" {code}({etf_ret:+.2%})" print(f" {date_str}: {row['daily_return']:+.4%} |{holdings_info}") # NAV 绝对回撤金额分析 print(f"\n 绝对回撤金额分析:") initial_nav = nav['nav'].iloc[0] current_nav = nav['nav'].iloc[-1] peak_nav = nav['nav'].max() print(f" 初始 NAV: {initial_nav:.4f}") print(f" 当前 NAV: {current_nav:.4f}") print(f" 峰值 NAV: {peak_nav:.4f}") print(f" 峰值回撤绝对值: {peak_nav * abs(max_dd_2026):.4f} NAV 单位") print(f" 对比: 2020 年全年收益 NAV = " f"{nav[nav['date'].dt.year == 2020]['nav'].iloc[-1] - nav[nav['date'].dt.year == 2020]['nav'].iloc[0]:.4f}") return {'max_dd_2026': float(max_dd_2026)} def analyze_extreme_tail(days: List[dict], nav: pd.DataFrame): """6.3 极端尾部风险""" print_section("6.3 极端尾部风险分析") returns = nav['daily_return'].values # 收益分布统计 print(" 日收益分布:") for pct in [1, 5, 10, 25, 50, 75, 90, 95, 99]: val = np.percentile(returns, pct) print(f" P{pct}: {val:+.4%}") # 尾部风险指标 tail_5pct = returns[returns <= np.percentile(returns, 5)] print(f"\n 尾部风险 (5% 分位以下):") print(f" CVaR(5%): {np.mean(tail_5pct):+.4%}") print(f" Worst: {np.min(returns):+.4%}") # 极端日归因 print(f"\n 极端亏损日归因 (日收益 < -5%):") extreme_loss = nav[nav['daily_return'] < -0.05] for _, row in extreme_loss.iterrows(): date_str = row['date'].strftime('%Y-%m-%d') day_detail = None for d in days: if d['date'] == date_str: day_detail = d break if day_detail: print(f"\n {date_str}: 日收益={row['daily_return']:+.4%}, NAV={row['nav']:.4f}") for code, asset in day_detail.get('assets', {}).items(): if asset.get('is_held'): etf_ret = asset.get('etf_return_ctc', 0) idx_ret = asset.get('index_return', 0) premium = asset.get('premium', 0) print(f" {code}: ETF收益={etf_ret:+.4%}, " f"指数收益={idx_ret:+.4%}, 溢价率={premium:+.4%}" if premium else f" {code}: ETF收益={etf_ret:+.4%}, 指数收益={idx_ret:+.4%}") # 连续亏损天数分析 streak = 0 max_streak = 0 streaks = [] for r in returns: if r < 0: streak += 1 else: if streak > 0: streaks.append(streak) streak = 0 if streak > 0: streaks.append(streak) max_streak = max(streaks) if streaks else 0 print(f"\n 连续亏损分析:") print(f" 最大连续亏损天数: {max_streak}") print(f" 连续亏损>=3天的次数: {sum(1 for s in streaks if s >= 3)}") print(f" 连续亏损>=5天的次数: {sum(1 for s in streaks if s >= 5)}") # 连续亏损段的累计收益 if streaks: loss_streak_rets = [] current_streak_ret = 0 in_streak = False for r in returns: if r < 0: current_streak_ret += r in_streak = True else: if in_streak and current_streak_ret < -0.05: loss_streak_rets.append(current_streak_ret) current_streak_ret = 0 in_streak = False if loss_streak_rets: print(f"\n 显著连续亏损段(累计<-5%):") for ret in sorted(loss_streak_rets): print(f" 累计亏损: {ret:+.4%}") return {'max_streak': max_streak, 'cvar_5pct': float(np.mean(tail_5pct))} def main(): print_section("Task 6: 回撤诊断") nav = load_nav() signals = load_signals() days = load_detail_days() meta = load_detail_meta() print(f" 数据期间: {meta['start_date']} ~ {meta['end_date']}") results = {} # 6.1 最大回撤复盘 results['max_dd'] = analyze_max_drawdown(days, nav) # 6.2 近期回撤 results['recent_dd'] = analyze_recent_drawdown(days, nav) # 6.3 尾部风险 results['tail'] = analyze_extreme_tail(days, nav) print_section("Task 6 总结") print(f" 1. 最大回撤 {results['max_dd']['max_dd']:.2%} 发生在 {results['max_dd']['trough_date']}") print(f" 2. 2026 年最大回撤: {results['recent_dd'].get('max_dd_2026', 0):.2%}") print(f" 3. CVaR(5%): {results['tail']['cvar_5pct']:+.4%}, " f"最大连续亏损: {results['tail']['max_streak']} 天") return results if __name__ == '__main__': main()