mirror of
https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git
synced 2026-09-12 17:54:15 +08:00
shy3130
c289ac5767
问题: WalkForwardService._prepare_shared_matrix 对非 matrix_native 策略直接抛 "步进优化暂仅支持矩阵策略", filter_history 类自定义策略 (如 everbloom 系列) 无法做步进优化。 根因: 闸门把非矩阵后端一刀切, 但 run() 内部本就存在 shared_market_data=None 的通用路径 (每折独立优化 + 独立 OOS 回测), 只是永远不会被走到。 方案: 闸门改分流 —— matrix_native 仍走共享矩阵加速路径; python_history_legacy 与 polars_expr 返回 None 走通用每折回测路径 (正确但无矩阵加速); composite / minute_filter 维持 fail-closed 拒绝, 错误文案同步更新。 兼容: matrix_native 行为不变; 结果字段 shared_market_data=false 标识通用路径。 性能: 非矩阵策略每组合×每折一次完整回测, 大网格耗时线性放大 (UI 已有耗时提示)。 验证: uv run --frozen pytest tests/backtest/test_walkforward.py -q (23 passed, 含新增 3 例: legacy 走通用路径/polars_expr 放行/minute_filter 拒绝); tests/backtest 全套 282 passed; ruff 无新增告警; 真实 API 端到端: everbloom_tupengpan 2x2 网格 120/30 步进 8/8 折全部完成。
392 lines
16 KiB
Python
392 lines
16 KiB
Python
"""Walk-forward 优化 — 滚动窗口的样本内优化 + 样本外验证。
|
|
|
|
每折在训练区间用参数网格优化选出最优参数, 再在紧邻的测试区间用该参数做样本外(OOS)
|
|
回测。滚动前移。核心产出是 OOS 拼接净值 + 每折 IS-vs-OOS 退化 —— 样本内漂亮、样本外
|
|
崩溃即过拟合信号, 单次样本内回测看不到。
|
|
|
|
依赖 PR2a 的 StrategyOptimizer 做每折训练区间的网格优化。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from bisect import bisect_left
|
|
from dataclasses import dataclass, field
|
|
from datetime import date, timedelta
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class Fold:
|
|
index: int
|
|
train_start: date
|
|
train_end: date
|
|
test_start: date
|
|
test_end: date
|
|
|
|
|
|
def generate_folds(
|
|
start: date,
|
|
end: date,
|
|
train_days: int,
|
|
test_days: int,
|
|
step_days: int,
|
|
) -> list[Fold]:
|
|
"""滚动窗口 fold 切分: 训练窗口固定长度, 测试窗口紧接其后, 按 step 前移。
|
|
|
|
测试区间超出 end 即停止。数据区间放不下一折则抛错。
|
|
"""
|
|
if train_days <= 0 or test_days <= 0 or step_days <= 0:
|
|
raise ValueError("train_days / test_days / step_days 必须为正")
|
|
|
|
folds: list[Fold] = []
|
|
i = 0
|
|
train_start = start
|
|
while True:
|
|
train_end = train_start + timedelta(days=train_days)
|
|
# 测试区间从训练末日的次日开始: 回测区间是闭区间, 若 test_start==train_end 则
|
|
# 该日 K 线同时进训练优化与 OOS 首日, 构成前视泄漏。后移一天隔断。
|
|
test_start = train_end + timedelta(days=1)
|
|
test_end = test_start + timedelta(days=test_days)
|
|
if test_end > end:
|
|
break
|
|
folds.append(Fold(i, train_start, train_end, test_start, test_end))
|
|
i += 1
|
|
train_start = train_start + timedelta(days=step_days)
|
|
|
|
if not folds:
|
|
raise ValueError(
|
|
f"数据区间不足以切出至少一折 (需 train+test={train_days + test_days}天, "
|
|
f"实有 {(end - start).days}天)"
|
|
)
|
|
return folds
|
|
|
|
|
|
def _norm(v: float, direction: str) -> float:
|
|
"""把目标值归一到"越大越好"空间, 以便跨目标一致地算退化 (min 类目标取负)。"""
|
|
return -v if direction == "min" else v
|
|
|
|
|
|
def aggregate_oos(fold_records: list[dict], objective: str, direction: str = "max") -> dict:
|
|
"""从**有效折** (IS 与 OOS 都成功) 聚合: 复利净值 / IS-OOS 退化 / 一致性。
|
|
|
|
调用方只传有效折 (best_params 非空且 OOS 未 error), 故此处每折 is_score/oos_objective
|
|
均有值, 无需 .get 默认兜底 —— 无效折被伪装成 0 收益混入曾是 H1/H2 的根因。
|
|
|
|
- compounded_oos_return: 各折 OOS 总收益复利
|
|
- degradation: 归一空间下 IS 目标均值 - OOS 目标均值, 正值 = 样本外退化 (过拟合信号),
|
|
对"越小越好"目标 (max_drawdown 等) 方向也正确
|
|
- consistency: OOS 总收益 > 0 的折占比 (与目标方向无关, 直观)
|
|
"""
|
|
n = len(fold_records)
|
|
if n == 0:
|
|
return {
|
|
"n_folds": 0,
|
|
"compounded_oos_return": 0.0,
|
|
"avg_is_objective": None,
|
|
"avg_oos_objective": None,
|
|
"degradation": None,
|
|
"consistency": 0.0,
|
|
"oos_equity_curve": [],
|
|
}
|
|
|
|
equity = 1.0
|
|
curve: list[dict] = []
|
|
n_positive = 0
|
|
for f in fold_records:
|
|
r = float(f["oos_stats"].get("total_return", 0.0) or 0.0)
|
|
equity *= (1 + r)
|
|
if r > 0:
|
|
n_positive += 1
|
|
curve.append({"fold": f["index"], "date": str(f["test_end"]), "value": round(equity, 4)})
|
|
|
|
is_vals = [f["is_score"] for f in fold_records if f["is_score"] is not None]
|
|
oos_vals = [f["oos_objective"] for f in fold_records if f["oos_objective"] is not None]
|
|
avg_is = round(float(sum(is_vals) / len(is_vals)), 4) if is_vals else None
|
|
avg_oos = round(float(sum(oos_vals) / len(oos_vals)), 4) if oos_vals else None
|
|
degradation = (
|
|
round(_norm(avg_is, direction) - _norm(avg_oos, direction), 4)
|
|
if (avg_is is not None and avg_oos is not None) else None
|
|
)
|
|
|
|
return {
|
|
"n_folds": n,
|
|
"compounded_oos_return": round(equity - 1.0, 4),
|
|
"avg_is_objective": avg_is,
|
|
"avg_oos_objective": avg_oos,
|
|
"degradation": degradation,
|
|
"consistency": round(n_positive / n, 4),
|
|
"oos_equity_curve": curve,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class WalkForwardConfig:
|
|
strategy_id: str
|
|
symbols: list[str] | None
|
|
start: date
|
|
end: date
|
|
param_grid: dict
|
|
objective: str = "sortino"
|
|
train_days: int = 252
|
|
test_days: int = 63
|
|
step_days: int = 63
|
|
direction: str | None = None
|
|
max_workers: int = 4
|
|
base_params: dict = field(default_factory=dict)
|
|
overrides: dict | None = None
|
|
backtest_kwargs: dict = field(default_factory=dict)
|
|
matrix_cache_max_mb: int = 512
|
|
|
|
|
|
class WalkForwardService:
|
|
"""滚动窗口 walk-forward: 每折训练区间优化 -> 测试区间 OOS 验证 -> 聚合。"""
|
|
|
|
def __init__(self, optimizer, service, strategy_engine) -> None:
|
|
self.optimizer = optimizer
|
|
self.service = service
|
|
self.strategy_engine = strategy_engine
|
|
|
|
def _prepare_shared_matrix(self, cfg: WalkForwardConfig, folds: list[Fold]):
|
|
"""Build one immutable superset matrix for every matrix-native fold.
|
|
|
|
返回 None 时 run() 走通用路径: 每折独立优化 + OOS 回测, 正确但无共享矩阵加速。
|
|
python_history_legacy (filter_history) 与 polars_expr (内置) 策略无法装入
|
|
共享矩阵, 走通用路径; composite / minute_filter 仍不支持, 保持 fail-closed。
|
|
"""
|
|
if self.strategy_engine is None or not folds:
|
|
return None
|
|
strategy = self.strategy_engine.get(cfg.strategy_id)
|
|
if strategy.execution_backend != "matrix_native":
|
|
if strategy.execution_backend in ("python_history_legacy", "polars_expr"):
|
|
return None
|
|
raise ValueError(
|
|
f"步进优化暂仅支持矩阵(matrix_native)/日线历史(python_history_legacy/"
|
|
f"polars_expr)策略; {cfg.strategy_id} 是 {strategy.execution_backend}"
|
|
)
|
|
|
|
from app.backtest.optimizer import expand_param_grid
|
|
from app.backtest.strategy import StrategyBacktestConfig
|
|
|
|
combos = expand_param_grid(strategy.meta.get("params", []), cfg.param_grid)
|
|
shared_start = min(fold.train_start for fold in folds)
|
|
shared_end = max(fold.test_end for fold in folds)
|
|
configs = [
|
|
StrategyBacktestConfig(
|
|
strategy_id=cfg.strategy_id,
|
|
symbols=cfg.symbols,
|
|
start=shared_start,
|
|
end=shared_end,
|
|
params={**cfg.base_params, **combo},
|
|
overrides=cfg.overrides,
|
|
**cfg.backtest_kwargs,
|
|
)
|
|
for combo in combos
|
|
]
|
|
return self.service.prepare_matrix_optimization(
|
|
configs,
|
|
matrix_cache_max_bytes=int(cfg.matrix_cache_max_mb) * 1024 * 1024,
|
|
)
|
|
|
|
def run(
|
|
self,
|
|
cfg: WalkForwardConfig,
|
|
progress_cb=None,
|
|
cancel_event=None,
|
|
) -> dict:
|
|
from app.backtest.optimizer import OptimizeConfig, default_direction
|
|
from app.backtest.strategy import StrategyBacktestConfig
|
|
|
|
t0 = time.perf_counter()
|
|
direction = cfg.direction or default_direction(cfg.objective)
|
|
folds = generate_folds(cfg.start, cfg.end, cfg.train_days, cfg.test_days, cfg.step_days)
|
|
n_total = len(folds)
|
|
|
|
shared_prepared = self._prepare_shared_matrix(cfg, folds)
|
|
shared_market_data = (
|
|
shared_prepared.market_data if shared_prepared is not None else None
|
|
)
|
|
shared_date_labels = (
|
|
tuple(label[:10] for label in shared_market_data.timestamp_labels)
|
|
if shared_market_data is not None
|
|
else ()
|
|
)
|
|
|
|
def _shared_window_has_data(window_start: date, window_end: date) -> bool:
|
|
index = bisect_left(shared_date_labels, window_start.isoformat())
|
|
return index < len(shared_date_labels) and shared_date_labels[index] <= window_end.isoformat()
|
|
|
|
# 遥测: 首尾快照 PanelCache, 量化跨折重叠区间重复扫盘的 IO 占比 (是否值得进一步优化)。
|
|
cache_before = self.service.engine.cache_stats()
|
|
|
|
valid_records: list[dict] = [] # IS 与 OOS 都成功, 计入聚合
|
|
skipped: list[dict] = [] # 无优化结果 或 OOS 失败, 不计入聚合 (避免伪装成有效折)
|
|
done = 0
|
|
|
|
# IS 训练区间强制 position 模式: full 模式会让训练折未平仓持仓用 train_end 之后
|
|
# (即 OOS 区间) 的真实 K 线平仓, IS 分数被未来数据污染 -> 优化选参乐观偏移, 使过拟合
|
|
# 被掩盖。OOS 回测保留用户所选 mode。参数扫描优化只看正式区间内的表现即可。
|
|
is_backtest_kwargs = {**cfg.backtest_kwargs, "mode": "position"}
|
|
|
|
for f in folds:
|
|
if cancel_event is not None and cancel_event.is_set():
|
|
break
|
|
|
|
base = {
|
|
"index": f.index,
|
|
"train_start": str(f.train_start),
|
|
"train_end": str(f.train_end),
|
|
"test_start": str(f.test_start),
|
|
"test_end": str(f.test_end),
|
|
}
|
|
missing_window = None
|
|
if shared_market_data is not None:
|
|
if not _shared_window_has_data(f.train_start, f.train_end):
|
|
missing_window = "训练区间无可用行情数据"
|
|
elif not _shared_window_has_data(f.test_start, f.test_end):
|
|
missing_window = "测试区间无可用行情数据"
|
|
if missing_window is not None:
|
|
skipped.append({**base, "reason": missing_window})
|
|
done += 1
|
|
if progress_cb is not None:
|
|
progress_cb({"type": "walkforward_progress", "done": done, "total": n_total, "fold": f.index})
|
|
continue
|
|
|
|
# 训练区间: 网格优化选最优参数
|
|
opt_cfg = OptimizeConfig(
|
|
strategy_id=cfg.strategy_id,
|
|
symbols=cfg.symbols,
|
|
start=f.train_start,
|
|
end=f.train_end,
|
|
param_grid=cfg.param_grid,
|
|
objective=cfg.objective,
|
|
direction=cfg.direction,
|
|
max_workers=cfg.max_workers,
|
|
base_params=cfg.base_params,
|
|
overrides=cfg.overrides,
|
|
backtest_kwargs=is_backtest_kwargs, # IS 强制 position, 堵前视泄漏
|
|
)
|
|
if shared_market_data is None:
|
|
opt_res = self.optimizer.optimize(opt_cfg, cancel_event=cancel_event)
|
|
else:
|
|
opt_res = self.optimizer.optimize(
|
|
opt_cfg,
|
|
cancel_event=cancel_event,
|
|
prepared_market_data=shared_market_data,
|
|
)
|
|
best_params = opt_res.get("best_params")
|
|
is_score = opt_res.get("best_score")
|
|
done += 1
|
|
|
|
# 训练区间没优化出参数 (全组失败/取消) -> 跳过, 不用默认参数硬跑 OOS 伪装成有效折
|
|
if best_params is None:
|
|
skipped.append({**base, "reason": "训练区间未优化出参数"})
|
|
if progress_cb is not None:
|
|
progress_cb({"type": "walkforward_progress", "done": done, "total": n_total, "fold": f.index})
|
|
continue
|
|
|
|
# 测试区间: 用最优参数做样本外回测
|
|
merged = {**cfg.base_params, **best_params}
|
|
oos_cfg = StrategyBacktestConfig(
|
|
strategy_id=cfg.strategy_id,
|
|
symbols=cfg.symbols,
|
|
start=f.test_start,
|
|
end=f.test_end,
|
|
params=merged,
|
|
overrides=cfg.overrides,
|
|
**cfg.backtest_kwargs,
|
|
)
|
|
oos_prepared = None
|
|
try:
|
|
if shared_market_data is not None:
|
|
oos_prepared = self.service.prepare_matrix_optimization(
|
|
[oos_cfg],
|
|
matrix_cache_max_bytes=int(cfg.matrix_cache_max_mb) * 1024 * 1024,
|
|
market_data_override=shared_market_data,
|
|
)
|
|
if oos_prepared is None:
|
|
oos_res = self.service.run(
|
|
oos_cfg,
|
|
cancel_event=cancel_event,
|
|
)
|
|
else:
|
|
oos_res = self.service.run(
|
|
oos_cfg,
|
|
cancel_event=cancel_event,
|
|
prepared=oos_prepared,
|
|
)
|
|
finally:
|
|
if oos_prepared is not None:
|
|
oos_prepared.compute_cache.close()
|
|
|
|
# OOS 失败 (含 cancelled) -> 跳过, 不把空/0 收益混入复利曲线
|
|
if oos_res.error:
|
|
skipped.append({**base, "best_params": best_params, "reason": f"OOS 回测失败: {oos_res.error}"})
|
|
if progress_cb is not None:
|
|
progress_cb({"type": "walkforward_progress", "done": done, "total": n_total, "fold": f.index})
|
|
continue
|
|
|
|
oos_objective = oos_res.stats.get(cfg.objective)
|
|
# 该折 OOS 是否较 IS 退化 (方向感知: min 类目标数值更大才是退化)
|
|
oos_degraded = (
|
|
_norm(oos_objective, direction) < _norm(is_score, direction)
|
|
if (oos_objective is not None and is_score is not None) else None
|
|
)
|
|
valid_records.append({
|
|
**base,
|
|
"best_params": best_params,
|
|
"is_score": is_score,
|
|
"oos_objective": oos_objective,
|
|
"oos_degraded": oos_degraded,
|
|
"oos_stats": oos_res.stats,
|
|
})
|
|
|
|
if progress_cb is not None:
|
|
progress_cb({"type": "walkforward_progress", "done": done, "total": n_total, "fold": f.index})
|
|
|
|
summary = aggregate_oos(valid_records, cfg.objective, direction)
|
|
|
|
shared_matrix_bytes = (
|
|
shared_prepared.market_data.nbytes if shared_prepared is not None else 0
|
|
)
|
|
shared_matrix_status = (
|
|
shared_prepared.market_data.cache_status if shared_prepared is not None else "none"
|
|
)
|
|
if shared_prepared is not None:
|
|
shared_prepared.compute_cache.close()
|
|
|
|
# 遥测收尾: 本次 WF 累计扫盘耗时 / 命中 / 复用, 与总耗时对比得出 load_panel 占比。
|
|
cache_after = self.service.engine.cache_stats()
|
|
elapsed_ms = round((time.perf_counter() - t0) * 1000, 1)
|
|
io_seconds = round(cache_after["compute_seconds"] - cache_before["compute_seconds"], 4)
|
|
io_pct = round(io_seconds * 1000 / elapsed_ms * 100, 1) if elapsed_ms > 0 else 0.0
|
|
cache_telemetry = {
|
|
"load_panel_seconds": io_seconds,
|
|
"load_panel_pct": io_pct, # 扫盘耗时 / WF 总耗时
|
|
"scans": cache_after["compute_count"] - cache_before["compute_count"],
|
|
"hits": cache_after["hit_count"] - cache_before["hit_count"],
|
|
"single_flight_reuses": cache_after["reuse_count"] - cache_before["reuse_count"],
|
|
}
|
|
logger.info(
|
|
"walk-forward IO 占比: load_panel %.3fs (%.1f%% of %.1fms) | 扫盘 %d 次 命中 %d 复用 %d",
|
|
io_seconds, io_pct, elapsed_ms,
|
|
cache_telemetry["scans"], cache_telemetry["hits"], cache_telemetry["single_flight_reuses"],
|
|
)
|
|
|
|
return {
|
|
"objective": cfg.objective,
|
|
"direction": direction,
|
|
"n_folds": len(valid_records),
|
|
"n_skipped": len(skipped),
|
|
"n_planned_folds": n_total,
|
|
"folds": valid_records,
|
|
"skipped": skipped,
|
|
"summary": summary,
|
|
"cache_telemetry": cache_telemetry,
|
|
"shared_market_data": shared_prepared is not None,
|
|
"shared_market_data_bytes": shared_matrix_bytes,
|
|
"shared_market_data_status": shared_matrix_status,
|
|
"elapsed_ms": elapsed_ms,
|
|
}
|