diff --git a/framework_v2/scripts/export_backtest_detail.py b/framework_v2/scripts/export_backtest_detail.py new file mode 100644 index 0000000..c8b1017 --- /dev/null +++ b/framework_v2/scripts/export_backtest_detail.py @@ -0,0 +1,327 @@ +#!/usr/bin/env python3 +""" +导出 V2 框架回测逐日明细到 JSON,供 HTML 回放器加载。 + +适用于 GlobalRotationStrategy(V2 正式版) +- 指数信号 + ETF 收益 +- 动态短债阈值 +- 强制分散化 +- 交易成本 +- CrossMarketAligner 数据对齐 + +用法: + python framework_v2/scripts/export_backtest_detail.py +""" + +import sys +import json +import math +from pathlib import Path +from datetime import datetime + +import numpy as np +import pandas as pd + +project_root = Path(__file__).parent.parent.parent +sys.path.insert(0, str(project_root)) + +from dotenv import load_dotenv +load_dotenv() + +from framework_v2.config import load_config +from framework_v2.strategies.rotation.rotation import GlobalRotationStrategy +from framework_v2.shared.data.alignment import CrossMarketAligner + + +# ==================== 辅助函数 ==================== + +def safe_val(v, decimals=4): + """安全转换数值,处理 NaN/Inf""" + if v is None or (isinstance(v, float) and (math.isnan(v) or math.isinf(v))): + return None + if isinstance(v, (np.floating, float)): + return round(float(v), decimals) + if isinstance(v, (np.integer, int)): + return int(v) + return v + + +def main(): + print("=" * 80) + print(" V2 回测逐日明细导出(GlobalRotationStrategy)") + print("=" * 80) + + # 1. 加载配置 + config_file = project_root / 'framework_v2' / 'strategies' / 'rotation' / 'config_simple.yaml' + print(f"\n[1] 加载配置: {config_file}") + config = load_config(str(config_file)) + + # 2. 初始化策略 + print("[2] 初始化策略...") + strategy = GlobalRotationStrategy(config) + + # 3. 获取数据 + print("[3] 获取数据...") + data = strategy.get_data() + print(f" 获取 {len(data)} 个标的") + + # 4. 计算因子 + print("[4] 计算因子...") + factors = strategy.compute_factors(data) + print(f" 计算 {len(factors)} 个因子") + + # 5. 生成信号 + print("[5] 生成信号...") + signals = strategy.generate_signals(factors) + print(f" 生成 {signals.shape[0]} 个信号") + + # 6. 仓位管理 + print("[6] 仓位管理...") + positions = strategy.manage_positions(signals) + + # 7. 准备收益率数据(使用 CrossMarketAligner) + print("[7] 准备收益率数据...") + signal_to_trade = config.asset_pools.get_signal_to_trade_mapping() + + # 获取 A 股交易日历 + trading_calendar = strategy._get_trading_calendar() + print(f" A 股交易日: {len(trading_calendar)} 天") + + # 创建对齐器 + aligner = CrossMarketAligner(target_calendar=trading_calendar) + + # 提取收盘价 + close_dict = {} + for signal_code, trade_code in signal_to_trade.items(): + if trade_code in data: + close_dict[signal_code] = data[trade_code]['close'] + + # 对齐收益率 + returns_df = aligner.align_multi_asset(close_dict) + print(f" 收益率数据: {len(returns_df)} 天, {len(returns_df.columns)} 个标的") + + # 8. 计算策略收益和净值 + print("[8] 计算策略收益...") + positions_aligned = positions.reindex(trading_calendar, method='ffill') + positions_delayed = positions_aligned.shift(1).fillna(0) + strategy_returns = (positions_delayed * returns_df).sum(axis=1) + + # 扣除交易成本 + strategy_returns_clean, rebalance_count = strategy._apply_trade_cost( + strategy_returns, positions_aligned + ) + print(f" 调仓次数: {rebalance_count}") + + # 计算净值 + equity_curve = (1 + strategy_returns_clean).cumprod() + print(f" 最终净值: {equity_curve.iloc[-1]:.4f}") + + # 9. 构建逐日明细 + print("[9] 构建逐日明细...") + + # 因子数据(DataFrame 格式) + factor_df = pd.DataFrame(factors) + + # 持仓状态跟踪 + holdings_state = {} # {code: {'entry_date': str, 'entry_price': float}} + prev_holdings = set() + + days_list = [] + common_dates = equity_curve.index + + # 获取配置信息 + bond_code = strategy.bond_code if strategy.use_dynamic_threshold else None + bond_ratio = strategy.bond_ratio + + for i, date in enumerate(common_dates): + # 当前持仓 + pos_row = positions_aligned.loc[date] + current_holdings = set(pos_row[pos_row > 0].index.tolist()) + + # 调仓检测 + added = list(current_holdings - prev_holdings) + removed = list(prev_holdings - current_holdings) + is_rebalance = len(added) > 0 or len(removed) > 0 + + # 更新持仓状态 + for code in removed: + holdings_state.pop(code, None) + for code in added: + # 获取入场价格 + entry_price = None + if code in close_dict: + ep = close_dict[code].get(date) + if pd.notna(ep): + entry_price = float(ep) + + holdings_state[code] = { + 'entry_date': date.strftime('%Y-%m-%d'), + 'entry_price': entry_price, + } + + # 动态阈值 + factor_scores = {} + if date in factor_df.index: + for code in factor_df.columns: + v = factor_df.loc[date, code] + if pd.notna(v): + factor_scores[code] = float(v) + + bond_score = factor_scores.get(bond_code) if bond_code else None + if bond_score is not None: + threshold = bond_score * bond_ratio + else: + threshold = 0.0 + + # 排名(按动量降序,排除 BOND) + groups = config.asset_pools.by_group + bond_assets = groups.get('BOND', {}) + bond_codes = set(bond_assets.keys()) + + non_bond_scores = {k: v for k, v in factor_scores.items() if k not in bond_codes} + sorted_codes = sorted(non_bond_scores.keys(), + key=lambda c: non_bond_scores[c], reverse=True) + rank_map = {c: r + 1 for r, c in enumerate(sorted_codes)} + + # BOND 不参与排名 + for code in bond_codes: + if code in factor_scores: + rank_map[code] = None + + # 每标的详情 + assets = {} + all_codes = factor_df.columns.tolist() + + for code in all_codes: + asset = {} + + # 动量得分 + mom = factor_scores.get(code) + asset['momentum'] = safe_val(mom, 4) + + # 排名 + asset['rank'] = rank_map.get(code) + + # 阈值 + asset['threshold'] = safe_val(threshold, 4) + asset['above_threshold'] = mom >= threshold if mom is not None else False + + # 持仓状态 + is_held = code in current_holdings + asset['is_held'] = is_held + asset['weight'] = safe_val(pos_row.get(code, 0), 4) if is_held else 0.0 + + if is_held and code in holdings_state: + hs = holdings_state[code] + asset['entry_date'] = hs['entry_date'] + asset['entry_price'] = safe_val(hs['entry_price'], 4) + + entry_dt = pd.Timestamp(hs['entry_date']) + trading_days_held = len(common_dates[(common_dates >= entry_dt) & (common_dates <= date)]) + asset['holding_days'] = trading_days_held + + # 累计收益 + if hs['entry_price'] and hs['entry_price'] > 0: + if code in close_dict: + cur = close_dict[code].get(date) + if cur and pd.notna(cur): + asset['cum_return'] = safe_val(float(cur) / hs['entry_price'] - 1, 4) + else: + asset['cum_return'] = None + else: + asset['cum_return'] = None + else: + asset['cum_return'] = None + + # 当日收益贡献 + if code in returns_df.columns: + asset['daily_return'] = safe_val(returns_df.loc[date, code], 6) + asset['return_contribution'] = safe_val( + pos_row.get(code, 0) * returns_df.loc[date, code], 6 + ) + else: + asset['daily_return'] = None + asset['return_contribution'] = None + else: + asset['entry_date'] = None + asset['entry_price'] = None + asset['holding_days'] = 0 + asset['cum_return'] = None + asset['daily_return'] = None + asset['return_contribution'] = None + + assets[code] = asset + + # 构建当天记录 + nav_val = equity_curve.loc[date] if date in equity_curve.index else None + ret_val = strategy_returns_clean.loc[date] if date in strategy_returns_clean.index else None + + day_record = { + 'date': date.strftime('%Y-%m-%d'), + 'nav': safe_val(nav_val, 4), + 'daily_return': safe_val(ret_val, 6), + 'is_rebalance': is_rebalance, + 'holdings': sorted(list(current_holdings)), + 'added': sorted(added), + 'removed': sorted(removed), + 'assets': assets + } + days_list.append(day_record) + prev_holdings = current_holdings + + # 10. 构建元数据 + codes_meta = {} + for signal_code, asset in config.asset_pools.assets.items(): + codes_meta[signal_code] = { + 'name': asset.name, + 'etf': asset.trade_source, + 'group': asset.group + } + + output = { + 'meta': { + 'version': 'V2', + 'strategy': 'GlobalRotationStrategy', + 'mode': '指数信号 + ETF收益', + 'start_date': common_dates[0].strftime('%Y-%m-%d'), + 'end_date': common_dates[-1].strftime('%Y-%m-%d'), + 'total_days': len(common_dates), + 'trade_cost': strategy.trade_cost, + 'dynamic_threshold': { + 'enabled': strategy.use_dynamic_threshold, + 'bond_code': bond_code, + 'ratio': bond_ratio + }, + 'diversified': strategy.diversified, + 'rebalance_count': int(rebalance_count), + 'final_nav': safe_val(equity_curve.iloc[-1], 4), + 'codes': codes_meta + }, + 'days': days_list + } + + # 11. 输出 + output_path = project_root / 'framework_v2' / 'results' / 'backtest_detail_v2.json' + print(f"\n[10] 写入 {output_path}...") + with open(output_path, 'w', encoding='utf-8') as f: + json.dump(output, f, ensure_ascii=False) + + file_size_mb = output_path.stat().st_size / 1024 / 1024 + print(f" 大小: {file_size_mb:.1f} MB") + print(f" 天数: {len(days_list)}") + print(f" 标的: {len(all_codes)}") + print(" 完成!") + + # 打印汇总统计 + print("\n" + "=" * 80) + print(" 回测汇总") + print("=" * 80) + print(f" 总收益: {(equity_curve.iloc[-1] - 1) * 100:.2f}%") + print(f" 年化收益: {((equity_curve.iloc[-1]) ** (252 / len(common_dates)) - 1) * 100:.2f}%") + print(f" 调仓次数: {rebalance_count}") + print(f" 交易天数: {len(common_dates)}") + print(f" 输出文件: {output_path}") + + +if __name__ == '__main__': + main()