回测「全部」范围与环境过滤冲突时自动顺延正式起点

正式首日 = 面板首日 (数据边界, 如「全部」范围从本地数据第一天开始) 时,
T-1 环境校验所需的上一交易日不在面板内, fail-closed 直接报错且用户
无法补更早数据。新增 clamp_formal_start_for_regime: 首日让渡为预热,
正式起点顺延到第二个交易日 (首日环境即成为次日 T-1), 仅损失 1 个
正式交易日。接入 run() 三条数据分支与优化器共享矩阵路径, 结果 config
回显实际生效起点。
This commit is contained in:
shy3130
2026-08-21 17:41:42 +08:00
parent 21260f1053
commit f856b4a8f8
3 changed files with 94 additions and 1 deletions
+30
View File
@@ -35,6 +35,36 @@ def _normalize_regime_point(value: Any) -> RegimePoint:
raise ValueError("市场环境数据格式无效")
def clamp_formal_start_for_regime(
labels: Sequence[str],
required_start: date | None,
regime_filter: Mapping[str, Any] | None,
) -> date | None:
"""正式首日无前驱交易日时, 把首日让渡为预热, 返回顺延后的正式起点。
触发条件: 环境过滤实际启用, 且面板首日 >= 正式起点 (数据边界即正式起点,
典型如「全部」范围从本地数据第一天开始)。此时首日的 T-1 环境所需的
上一交易日不在面板内, fail-closed 校验会直接拒绝; 把首日降级为预热、
正式起点顺延到第二个交易日, 首日环境即成为次日的 T-1。
其余情况 (有预热日 / 过滤未启用 / 标签不足两天) 原样返回 required_start,
无法顺延时由后续校验给出明确报错。
"""
if not regime_filter or required_start is None:
return required_start
has_states = bool(regime_filter.get("states") or [])
has_score = regime_filter.get("min_score") is not None
if not (has_states or has_score):
return required_start
if len(labels) < 2:
return required_start
if _date_text(labels[0]) < str(required_start):
return required_start
try:
return date.fromisoformat(_date_text(labels[1]))
except ValueError:
return required_start
def align_regime_t_minus_one(
labels: Sequence[str],
regime_by_date: Mapping[object, Any],
+36 -1
View File
@@ -11,7 +11,7 @@ import threading
import time
import uuid
from collections.abc import Callable, Mapping
from dataclasses import dataclass, field
from dataclasses import dataclass, field, replace
from datetime import date, timedelta
from pathlib import Path
from typing import Literal
@@ -873,6 +873,8 @@ class StrategyBacktestService:
timing_ms["load_panel"] = direct_load_ms
timing_ms["market_data_matrix_build"] = 0.0
timing_ms["market_data_direct_load"] = direct_load_ms
# 环境过滤下正式起点=矩阵首日时顺延 (首日让渡为预热), 见 _clamp_regime_formal_start
first = self._clamp_regime_formal_start(first, market_data.timestamp_labels)
formal_range = self._matrix_date_range_mask(
market_data.timestamp_labels,
first.start,
@@ -1068,6 +1070,11 @@ class StrategyBacktestService:
matrix_data_cache_status = prepared.market_data.cache_status
matrix_data_cache_hit = matrix_data_cache_status in {"exact", "covering"}
matrix_data_cache_timing_ms = prepared.market_data.cache_timing_ms
# 环境过滤下正式起点=矩阵首日时顺延 (首日让渡为预热)
if config.regime_filter:
config = self._clamp_regime_formal_start(
config, prepared.market_data.timestamp_labels
)
elif s.execution_backend in ("matrix_native", "composite"):
t_load = time.perf_counter()
max_hold_for_profile = self._override_value(
@@ -1113,6 +1120,11 @@ class StrategyBacktestService:
matrix_data_cache_status = market_data.cache_status
matrix_data_cache_hit = matrix_data_cache_status in {"exact", "covering"}
matrix_data_cache_timing_ms = market_data.cache_timing_ms
# 环境过滤下正式起点=矩阵首日时顺延 (首日让渡为预热)
if config.regime_filter:
config = self._clamp_regime_formal_start(
config, market_data.timestamp_labels
)
formal_time_mask = self._matrix_date_range_mask(
market_data.timestamp_labels,
config.start,
@@ -1136,6 +1148,13 @@ class StrategyBacktestService:
timing_ms["load_panel"] = round((time.perf_counter() - t_load) * 1000, 1)
if panel.is_empty():
return _err("无数据,请检查日期范围或先运行盘后管道")
# 环境过滤下正式起点=面板首日时顺延 (首日让渡为预热)
if config.regime_filter:
date_labels = tuple(
str(value)[:10]
for value in panel.get_column("date").unique().sort().to_list()
)
config = self._clamp_regime_formal_start(config, date_labels)
formal_range = self._date_range_mask(panel, config.start, config.end)
if not formal_range.any():
return _err("正式回测区间内无数据")
@@ -1715,6 +1734,22 @@ class StrategyBacktestService:
count=len(timestamp_labels),
)
@staticmethod
def _clamp_regime_formal_start(
config: StrategyBacktestConfig, labels: tuple[str, ...] | list[str]
) -> StrategyBacktestConfig:
"""环境过滤下正式起点=面板首日 (无前驱交易日) 时, 顺延到第二个交易日。
数据边界即正式起点 (如「全部」范围) 时, T-1 环境校验会 fail-closed 拒绝;
首日降级为预热后, 其环境即成为次日的 T-1, 仅损失 1 个正式交易日。
"""
from app.backtest.regime_alignment import clamp_formal_start_for_regime
shifted = clamp_formal_start_for_regime(labels, config.start, config.regime_filter)
if shifted is not None and shifted != config.start:
return replace(config, start=shifted)
return config
@staticmethod
def _build_regime_mask(
timestamp_labels: tuple[str, ...],
@@ -389,3 +389,31 @@ def test_align_regime_t_minus_one_is_pure_and_fail_closed_in_required_range():
required_start=date(2026, 1, 6),
required_end=date(2026, 1, 7),
)
def test_clamp_formal_start_for_regime():
from app.backtest.regime_alignment import clamp_formal_start_for_regime
# 复现「全部」范围: 正式起点 = 面板首日 (数据边界), 环境过滤启用 → 顺延到次日
labels = ("2025-08-18", "2025-08-19", "2025-08-20")
filt = {"states": ["strong", "lean_strong"]}
assert clamp_formal_start_for_regime(labels, date(2025, 8, 18), filt) == date(2025, 8, 19)
# 正式起点早于面板首日 (用户选的日期早于数据) 同样顺延
assert clamp_formal_start_for_regime(labels, date(2025, 1, 1), filt) == date(2025, 8, 19)
# 面板首日早于正式起点 (已有预热日) → 不动
assert clamp_formal_start_for_regime(labels, date(2025, 8, 19), filt) == date(2025, 8, 19)
# 过滤未启用 / 空 filter (无 states 无 min_score) / 起点 None → 不动
assert clamp_formal_start_for_regime(labels, date(2025, 8, 18), None) == date(2025, 8, 18)
assert clamp_formal_start_for_regime(labels, date(2025, 8, 18), {}) == date(2025, 8, 18)
assert clamp_formal_start_for_regime(labels, None, filt) is None
# 面板只有一天, 无法顺延 → 原样返回 (由 fail-closed 校验兜底)
assert clamp_formal_start_for_regime(("2025-08-18",), date(2025, 8, 18), filt) == date(2025, 8, 18)
# 顺延后 T-1 对齐不再报「正式首日」错误: 首日环境成为次日 T-1
regimes = {"2025-08-18": ("weak", 20), "2025-08-19": ("strong", 80)}
clamped = clamp_formal_start_for_regime(labels, date(2025, 8, 18), filt)
aligned = align_regime_t_minus_one(labels, regimes, required_start=clamped, required_end=None)
assert aligned[0] is None and aligned[1] == ("weak", 20.0)