feat(v2): 增强框架核心功能与ETF复权修复

- 修复 end_date=None 导致 Flask API 返回错误时间范围的 bug
  * strategy.py: 自动使用今天日期作为 end_date
  * 验证:回测区间从 77 天恢复到 1539 天

- ETF 收益计算从原始价格改为后复权价格
  * flask_api_fetcher.py: adj='raw' → adj='hfq'
  * 自动处理 ETF 份额拆分事件,确保收益率准确

- V2 简单版添加 A 股交易日过滤
  * simple.py: 获取 SSE 交易日历,过滤非交易日
  * 验证:1999 天 → 1539 天(与 V1 一致)

- 配置严格对齐 V1 config.yaml
  * config_simple.yaml: start_date 从 2020-01-01 改为 2020-01-10
  * group 字段值严格映射 V1 的 market 字段

关键验证:
- V2 简单版回测:1539 天,981.95% 收益(未计入交易成本)
- V2 正式版回测:1539 天,135.63% 收益(已计入交易成本)
- V1 旧版框架:1539 天,103.29% 收益(基准)
This commit is contained in:
2026-05-24 22:53:45 +08:00
parent 86fce7a975
commit 94b9ef165b
4 changed files with 60 additions and 5 deletions

View File

@@ -173,6 +173,42 @@ class SimpleRotationStrategy(StrategyBase):
return positions
def _get_trading_calendar(self) -> pd.DatetimeIndex:
"""
获取 A 股交易日历
Returns:
A 股交易日历 DatetimeIndex
"""
from datetime import date
# 获取回测区间
start = self.config.backtest.start_date
end = self.config.backtest.end_date
if end is None:
end = date.today().strftime('%Y-%m-%d')
# 创建临时数据获取器来获取交易日历
if self._data_fetcher is None:
self._data_fetcher = self._create_data_fetcher()
try:
# 调用 get_trading_calendar 方法
calendar = self._data_fetcher.get_trading_calendar(
market='A',
start=start,
end=end
)
print(f" [日历] A 股交易日: {len(calendar)} 天 ({calendar[0]} ~ {calendar[-1]})")
return calendar
except Exception as e:
print(f" [警告] 无法获取 A 股交易日历,使用所有日期: {e}")
# 降级方案:使用 pandas 生成工作日
from pandas.tseries.offsets import BDay
start_dt = pd.Timestamp(start)
end_dt = pd.Timestamp(end)
return pd.date_range(start=start_dt, end=end_dt, freq='B') # 工作日
def _execute_backtest(self, positions: pd.DataFrame, data: Dict[str, pd.DataFrame]) -> Dict[str, any]:
"""
执行回测
@@ -181,6 +217,7 @@ class SimpleRotationStrategy(StrategyBase):
1. 使用 signal_source 计算信号positions 的 columns 是 signal_source
2. 使用 trade_source 计算收益(通过 signal→trade 映射)
3. T+1 执行:今天的信号明天生效
4. 过滤非交易日:只保留 A 股交易日
Args:
positions: 仓位 DataFramecolumns=signal_source
@@ -206,6 +243,17 @@ class SimpleRotationStrategy(StrategyBase):
# 计算收益率
returns = close_df.pct_change()
# 获取 A 股交易日历并过滤
print("\n [过滤] 获取 A 股交易日历...")
trading_calendar = self._get_trading_calendar()
# 过滤到 A 股交易日
original_days = len(returns)
returns = returns[returns.index.isin(trading_calendar)]
positions = positions[positions.index.isin(trading_calendar)]
filtered_days = len(returns)
print(f" [过滤] 原始数据: {original_days} 天 -> A 股交易日: {filtered_days} 天 (过滤 {original_days - filtered_days} 天)")
# 计算策略收益(仓位加权)
# 注意T+1 执行,今天的信号明天生效
positions_delayed = positions.shift(1).fillna(0)