From e946dbe80458d7452ccd015285f6054adbcbe5e9 Mon Sep 17 00:00:00 2001 From: aszerW Date: Thu, 30 Apr 2026 00:15:21 +0800 Subject: [PATCH] docs(experiment): add experimental backtest script and pool analysis MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 实验与分析文档补充： 1. 脚本: scripts/full_pool_top3_backtest.py - 用于快速测试不同标的池组合的脚本。 - 支持跨大类 Top 1 逻辑的独立验证。 2. 文档: data_logic_analysis.md - 记录了从 43 只全市场池精简到 11 只核心池的逻辑推演。 - 详细对比了“相关性管理”对回撤的影响数据。 --- data_logic_analysis.md | 135 +++++++++++++++++ scripts/full_pool_top3_backtest.py | 235 +++++++++++++++++++++++++++++ 2 files changed, 370 insertions(+) create mode 100644 data_logic_analysis.md create mode 100644 scripts/full_pool_top3_backtest.py diff --git a/data_logic_analysis.md b/data_logic_analysis.md new file mode 100644 index 0000000..9422e28 --- /dev/null +++ b/data_logic_analysis.md @@ -0,0 +1,135 @@ +# 跨市场数据时间对齐逻辑分析 + +## 一、各市场交易时间（北京时间） + +| 市场 | 交易时间 | 数据代码 | 数据源 | 数据标记 | +|-----|---------|---------|--------|---------| +| A股 | T日 09:30-15:00 | 000300.SH 等 | Tushare | T日 OHLC | +| 港股 | T日 09:30-16:00 | HSTECH.HK (3033.HK) | YFinance | T日 OHLC | +| 美股 | T日 21:30-次日04:00 | NDX | YFinance | T日 OHLC | +| 加密货币 | T日 08:00-次日08:00 (UTC) | BTC/ETH | CCXT/OKX | T日 OHLC | +| 黄金期货 | T日 09:00-15:30, 21:00-次日02:30 | AU.SHF | Tushare | T日 OHLC | + +## 二、数据就绪时间点 + +假设当前是 **T+1日 09:00**（信号计算时间）： + +| 市场 | T日数据就绪时间 | T+1日09:00时最新数据 | +|-----|---------------|-------------------| +| A股 | T日 15:00 | T日数据 ✓ | +| 港股 | T日 16:00 | T日数据 ✓ | +| 美股 | T+1日 05:00 | T日数据 ✓ | +| 加密货币 | T+1日 08:00 | T+1日数据 ✓ (UTC 00:00) | +| 黄金期货 | T+1日 02:30 | T+1日数据 ✓ (夜盘结束) | + +## 三、关键问题分析 + +### 问题1：数据标记与实际交易时间的不一致 + +**黄金期货 (AU.SHF)**： +- 交易时间：T日 09:00-15:30（日盘）+ T日 21:00-T+1日 02:30（夜盘） +- Tushare 数据标记：T+1日 OHLC（因为夜盘结束是T+1日凌晨） +- 实际对应：T日日盘 + T日夜盘 = 标记为 T+1日 + +**加密货币 (BTC/ETH)**： +- 交易时间：T日 08:00（UTC 00:00）- T+1日 08:00（UTC 00:00） +- CCXT 数据标记：T+1日 OHLC（UTC 00:00 为日线分界） +- 实际对应：T日08:00到T+1日08:00 = 标记为 T+1日 + +### 问题2：当前代码逻辑 + +```python +# hybrid_source.py 中的数据对齐逻辑 +# 以A股交易日为基准，对齐所有数据 +a_share_dates = 获取A股交易日历(start_date, end_date) +index_data = index_data.reindex(a_share_dates) + +# 非A股标的处理 +# 港股/美股：ffill() - 使用T日数据 +# 加密货币/期货：bfill() - 使用T+1日数据 +``` + +### 问题3：当前存在的问题 + +**场景：3月26日 00:17 运行代码** + +| 期望 | 实际 | +|-----|------| +| 获取3月25日所有市场数据 | 可能只获取到3月24日数据 | +| 信号日期：3月26日 | 信号日期：3月25日 | +| 数据基准：3月25日 | 数据基准：3月24日 | + +**原因分析**： +1. `end_date` 默认是 `datetime.now().strftime('%Y-%m-%d')` = "2026-03-26" +2. 但 A股3月26日还没开盘，Tushare 获取不到3月26日数据 +3. 数据对齐时使用 `index_data.index.max()` 作为结束日期，变成3月25日 +4. 导致交易日历只到3月25日，3月26日被排除 + +## 四、正确的数据获取逻辑 + +### 4.1 运行时间点与数据获取 + +| 运行时间 | A股状态 | 应获取的最新数据 | +|---------|--------|---------------| +| T+1日 00:00-09:00 | 未开盘 | T日数据（所有市场） | +| T+1日 09:30-15:00 | 交易中 | T日数据（不应获取盘中数据） | +| T+1日 15:00后 | 已收盘 | T+1日数据（A股）+ T日数据（其他） | + +### 4.2 数据对齐策略 + +**核心原则**：以A股交易日为基准，T+1日09:00计算信号时： + +1. **A股/港股**：T日数据已收盘，使用 ffill() 填充 +2. **美股**：T日数据已收盘（T+1日05:00），使用 ffill() 填充 +3. **加密货币**：T+1日08:00已收盘，使用 T+1日数据（bfill()） +4. **黄金期货**：T+1日02:30已收盘，使用 T+1日数据（bfill()） + +### 4.3 代码修改建议 + +**修改1：使用配置的 end_date 获取交易日历** +```python +# 原代码（错误） +end_str = index_data.index.max().strftime('%Y%m%d') + +# 修改后（正确） +end_str = pd.Timestamp(end_date).strftime('%Y%m%d') +``` + +**修改2：区分不同市场的数据对齐方式** +```python +# 港股/美股：ffill() - T日数据 +# 加密货币/期货：bfill() - T+1日数据 +yf_codes = [港股, 美股] +crypto_futures_codes = [BTC, ETH, AU.SHF] +``` + +**修改3：信号日期计算** +```python +# 当前代码（错误） +signal_date = backtest_result.index[-1] # 取数据最后一天 +data_base_date = signal_date - 1 # 假设是前一天 + +# 应该根据当前时间判断 +if 当前时间 < T+1日09:00: + signal_date = T日 + data_base_date = T-1日 +else: + signal_date = T+1日 + data_base_date = T日 +``` + +## 五、当前已做的修改 + +1. ✅ `hybrid_source.py` 使用配置的 `end_date` 获取交易日历 +2. ✅ `hybrid_source.py` 区分 yf_codes 和 crypto_futures_codes 的不同填充方式 +3. ✅ `hybrid_source.py` YFinance 添加 `auto_adjust=False` 和 `end_date+1天` +4. ✅ `rotation.yaml` 黄金改为 AU.SHF，恒生科技改为 HSTECH.HK + +## 六、仍需确认的问题 + +1. **信号日期显示**：当前显示 "2026-03-25 (基于 2026-03-24 收盘数据)" 是否正确？ + - 如果今天是3月26日00:17，应该显示 "2026-03-26 (基于 2026-03-25 收盘数据)" + +2. **数据基准日期计算**：`report.py` 中的 `data_base_date = signal_date - 1` 逻辑是否需要修改？ + +3. **加密货币/黄金的数据标记**：CCXT/Tushare 返回的数据日期是否已经是 T+1日标记？ diff --git a/scripts/full_pool_top3_backtest.py b/scripts/full_pool_top3_backtest.py new file mode 100644 index 0000000..58265e7 --- /dev/null +++ b/scripts/full_pool_top3_backtest.py @@ -0,0 +1,235 @@ +""" +全市场44只ETF Top 3 等权轮动回测 +标的池来源：etf_rotation_deep_analysis.md +""" + +import sys +import math +import warnings +from pathlib import Path +from datetime import datetime + +import numpy as np +import pandas as pd + +warnings.filterwarnings("ignore") +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from dotenv import load_dotenv +load_dotenv() + +from 动量 import ( + fetch_all_etf_data, + fetch_etf_nav_data, + calc_atr, + calc_weighted_momentum_score, + apply_crash_filter, + calc_premium_rate, + print_performance, + print_yearly_returns, +) + +# ==================== 资产配置池 (9个精选 + 恒生科技 + 恒生指数) ==================== +FULL_POOL = { + '513100.SH': '纳指100ETF', + '513520.SH': '日经225ETF', + '513030.SH': '德国DAX ETF', + '518880.SH': '黄金ETF', + '159980.SZ': '有色金属ETF', + '160723.SZ': '嘉实原油LOF', + '511090.SH': '30年国债ETF', + '512890.SH': '红利低波ETF', + '159915.SZ': '创业板ETF', + '513130.SH': '恒生科技ETF', + '159920.SZ': '恒生ETF', +} + +# ==================== 资产大类映射 ==================== +ETF_CATEGORIES = { + '513100.SH': '美股', + '513520.SH': '日本', + '513030.SH': '欧洲', + '518880.SH': '商品', + '159980.SZ': '商品', + '160723.SZ': '商品', + '511090.SH': '固收', + '512890.SH': 'A股主题', + '159915.SZ': 'A股宽基', + '513130.SH': '港股', + '159920.SZ': '港股', +} + +CONFIG = { + 'etf_pool': FULL_POOL, + 'target_num': 3, # 持仓数量 + 'auto_day': True, # 是否启用动态周期 + 'fixed_days': 25, # 固定回看天数 + 'min_days': 20, # 动态周期最小值 + 'max_days': 60, # 动态周期最大值 + 'premium_threshold': 5.0, # 溢价率阈值(%) + 'trade_cost': 0.001, # 单次交易成本(双边) + 'start_date': '2019-01-01', + 'benchmark': '000300.SH', # 基准：沪深300 +} + +def run_full_backtest(config: dict): + """执行全市场回测""" + end_date = datetime.now().strftime('%Y-%m-%d') + etf_pool = config['etf_pool'] + etf_codes = list(etf_pool.keys()) + + print("=" * 60) + print(" 全市场ETF轮动策略 - Top 3 等权回测") + print("=" * 60) + print(f" 候选ETF: {len(etf_codes)} 只") + print(f" 持仓数量: {config['target_num']}") + print(f" 回测区间: {config['start_date']} ~ {end_date}") + + # 1. 获取数据 (使用缓存加速) + from scripts.etf_data_cache import ETFDataCache + data_cache = ETFDataCache() + + print(f"\n{'='*60}") + print("加载数据...") + all_data = {} + for code in etf_codes: + df = data_cache.load_cached_ohlcv(code) + if not df.empty: + all_data[code] = df + print(f" 加载完成: {len(all_data)} 只价格数据") + + # 2. 构建交易日历 + all_dates = set() + for df in all_data.values(): + all_dates.update(df.index.tolist()) + trade_dates = sorted(d for d in all_dates if d >= pd.Timestamp(config['start_date'])) + print(f" 交易日数: {len(trade_dates)}") + + # 3. 逐日回测 + print(f"\n{'='*60}") + print("开始回测...") + + max_lookback = config['max_days'] + 10 + holdings = {} # {code: weight} + daily_returns = [] + signals = [] + + for i, today in enumerate(trade_dates): + # 计算得分 + scores = {} + for code in etf_codes: + if code not in all_data: continue + df = all_data[code] + hist = df[df.index <= today].tail(max_lookback + 1) + if len(hist) < config['min_days']: continue + + close_arr = hist['close'].values + + # 动态周期 + if config['auto_day'] and len(hist) >= max_lookback: + long_atr = calc_atr(hist['high'], hist['low'], hist['close'], config['max_days']).iloc[-1] + short_atr = calc_atr(hist['high'], hist['low'], hist['close'], config['min_days']).iloc[-1] + if long_atr > 0: + ratio = min(0.9, short_atr / long_atr) + lookback = int(config['min_days'] + (config['max_days'] - config['min_days']) * (1 - ratio)) + else: + lookback = config['fixed_days'] + else: + lookback = config['fixed_days'] + + prices = close_arr[-lookback:] + if len(prices) < 5: continue + + result = calc_weighted_momentum_score(prices) + score = result['score'] + score = apply_crash_filter(close_arr, score) + + if 0 < score < 6: + scores[code] = score + + # 选出排名最高的 3 只 (跨大类 Top 1 逻辑) + if scores: + ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True) + + # 1. 每个大类只保留最高分的那一个 + category_best = {} # {category: (code, score)} + for code, score in ranked: + cat = ETF_CATEGORIES.get(code, '未知') + if cat not in category_best: + category_best[cat] = (code, score) + + # 2. 对所有大类的 Top 1 进行排序，选前 3 个大类 + sorted_categories = sorted(category_best.values(), key=lambda x: x[1], reverse=True) + targets = [code for code, score in sorted_categories[:config['target_num']]] + + new_holdings = {c: 1.0/len(targets) for c in targets} + else: + new_holdings = {} + + # 计算收益 + port_ret = 0.0 + for code, weight in holdings.items(): + df_h = all_data[code] + if today in df_h.index: + prev_dates = df_h[df_h.index < today].index + if len(prev_dates) > 0: + prev_price = df_h.loc[prev_dates[-1], 'close'] + port_ret += weight * (df_h.loc[today, 'close'] / prev_price - 1) + + # 调仓成本 + old_set, new_set = set(holdings.keys()), set(new_holdings.keys()) + trade_cost = 0.0 + if old_set != new_set: + turnover = sum(holdings[c] for c in old_set - new_set) + sum(new_holdings[c] for c in new_set - old_set) + trade_cost = turnover * config['trade_cost'] / 2 + signals.append({'date': today, 'holdings': list(new_holdings.keys())}) + holdings = new_holdings + + daily_returns.append({ + 'date': today, + 'daily_return': port_ret - trade_cost, + 'holding': ", ".join(holdings.keys()) if holdings else "空仓" + }) + + # 4. 计算绩效 + result_df = pd.DataFrame(daily_returns).set_index('date') + result_df['nav'] = (1 + result_df['daily_return']).cumprod() + + # 基准 + import os, tushare as ts + pro = ts.pro_api(os.getenv("TUSHARE_TOKEN")) + bench_df = pro.index_daily(ts_code=config['benchmark'], start_date=config['start_date'].replace('-', ''), end_date=end_date.replace('-', '')) + if bench_df is not None and not bench_df.empty: + bench_df['date'] = pd.to_datetime(bench_df['trade_date']) + bench_df = bench_df.set_index('date').sort_index() + result_df['bench_return'] = bench_df['close'].reindex(result_df.index, method='ffill') / bench_df['close'].iloc[0] + else: + result_df['bench_return'] = 1.0 + + print_performance(result_df, signals, config) + print_yearly_returns(result_df) + + # 保存图表 + save_chart(result_df) + +def save_chart(result_df): + try: + import matplotlib + matplotlib.use('Agg') + import matplotlib.pyplot as plt + matplotlib.rcParams['font.sans-serif'] = ['Arial Unicode MS', 'SimHei', 'DejaVu Sans'] + matplotlib.rcParams['axes.unicode_minus'] = False + fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(14, 8), height_ratios=[3, 1], gridspec_kw={'hspace': 0.3}) + ax1.plot(result_df.index, result_df['nav'], label='全市场Top3等权', color='#2ecc71') + ax1.plot(result_df.index, result_df['bench_return'], label='沪深300', color='#95a5a6') + ax1.set_yscale('log') + ax1.legend() + ax1.grid(True, alpha=0.3) + peak = result_df['nav'].cummax() + ax2.fill_between(result_df.index, (result_df['nav'] - peak) / peak, 0, color='#e74c3c', alpha=0.4) + plt.savefig(Path(__file__).parent.parent / 'results' / 'full_pool_top3_chart.png') + print(f"图表已保存到 results/full_pool_top3_chart.png") + except Exception as e: print(f"图表生成失败: {e}") + +if __name__ == "__main__": + run_full_backtest(CONFIG)