From 32d27a0a6e1dca501137eca29973dcf3cc515f77 Mon Sep 17 00:00:00 2001 From: im47cn <67424112+im47cn@users.noreply.github.com> Date: Wed, 8 Jul 2026 23:30:10 +0800 Subject: [PATCH] =?UTF-8?q?fix(walkforward):=20=E5=AD=90=E4=BB=A3=E7=90=86?= =?UTF-8?q?=E5=AE=A1=E6=9F=A5=E4=BF=AE=E5=A4=8D=20=E2=80=94=20=E6=8A=98?= =?UTF-8?q?=E6=9C=89=E6=95=88=E6=80=A7/=E5=89=8D=E8=A7=86=E6=B3=84?= =?UTF-8?q?=E6=BC=8F/=E6=96=B9=E5=90=91=E9=80=80=E5=8C=96/NaN=E5=BA=8F?= =?UTF-8?q?=E5=88=97=E5=8C=96/=E5=87=80=E5=80=BC=E6=9B=B2=E7=BA=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 三份子代理审查 (核心/API/前端) 发现的真实 bug: 核心正确性 (walkforward.py): - [高] 折有效性抽象: best_params=None (训练全组失败) 或 OOS error 的折原会用默认 参数硬跑/按0收益混入复利, 伪装成有效折污染 OOS 净值与退化指标。改为分流 —— 有效折进 folds, 无效折进 skipped(带原因), 聚合只看有效折。一次修掉 H1/H2/M1。 - [高] 前视泄漏: train_end==test_start 使该日 K 线同时进训练与 OOS 首日。test_start 后移一天隔断。 - [中] degradation 方向感知: 原 avg_is-avg_oos 对 min 类目标 (avg_holding_days) 符号 反了。归一到越大越好空间再相减; 每折加 oos_degraded 方向感知标志。 - consistency 改为 OOS 盈利折占比 (与目标方向无关, 更直观)。 API (backtest.py): - [高] 单折 guard: WF 缺 guard, start=None 默认拉 3 年, 每折训练窗口可能 OOM。按 单折窗口 (train/test) 而非总区间 guard —— 总区间长本是 WF 正常形态, 按总区间拦会误杀。 - [中] NaN/inf 序列化: json.dumps(default=str) 处理不了 nan/inf, 输出非法 JSON 崩前端。 加 _json_safe 递归清洗, 优化器与 WF 两处 done 分支都套上。 前端 (StrategyWalkForward.tsx): - n_folds=0 门控: 全跳过时不再渲染误导性全0卡, 改显示'未产生有效折'+跳过原因。 - 渲染 OOS 拼接净值曲线 (walk-forward 核心产出, 原后端算了前端没画)。 - 每折退化标红改用后端方向感知的 oos_degraded (min 类目标 oos BACKTEST_MAX_SERVER_DAYS + ) + _cleanup_stale_jobs() with _jobs_lock: job = _running_jobs.get(job_key) @@ -797,6 +818,11 @@ async def walkforward_stream( async def event_generator(): yield f"event: job\ndata: {json.dumps({'key': job_key}, ensure_ascii=False)}\n\n" + if wf_guard_violated: + msg = f"单折窗口最多 {BACKTEST_MAX_SERVER_DAYS} 天 (当前 train/test 更大), 请减小训练/测试窗口或在更大内存环境运行。" + yield f"event: error\ndata: {json.dumps({'message': msg}, ensure_ascii=False)}\n\n" + return + if is_new and not job.done: try: grid = json.loads(param_grid) @@ -847,7 +873,7 @@ async def walkforward_stream( elif job.cancel_event.is_set(): yield f"event: error\ndata: {json.dumps({'message': 'walk-forward 已取消'}, ensure_ascii=False)}\n\n" elif job.result is not None: - yield f"event: done\ndata: {json.dumps(job.result, ensure_ascii=False, default=str)}\n\n" + yield f"event: done\ndata: {json.dumps(_json_safe(job.result), ensure_ascii=False, default=str)}\n\n" return tick += 1 if tick % 4 == 0 and await request.is_disconnected(): diff --git a/backend/app/backtest/walkforward.py b/backend/app/backtest/walkforward.py index 969521a..5c27f31 100644 --- a/backend/app/backtest/walkforward.py +++ b/backend/app/backtest/walkforward.py @@ -44,7 +44,9 @@ def generate_folds( train_start = start while True: train_end = train_start + timedelta(days=train_days) - test_start = train_end + # 测试区间从训练末日的次日开始: 回测区间是闭区间, 若 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 @@ -60,14 +62,21 @@ def generate_folds( return folds -def aggregate_oos(fold_records: list[dict], objective: str) -> dict: - """从各折 OOS 结果聚合: 复利净值曲线 / IS-OOS 退化 / 一致性。 +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 的根因。 - fold_records: [{index, test_end, best_params, is_score, oos_stats}] - compounded_oos_return: 各折 OOS 总收益复利 - - avg_is_objective / avg_oos_objective / degradation: IS 目标均值 - OOS 目标均值, - 正值 = 样本外退化 = 过拟合信号 - - consistency: OOS 目标 > 0 的折占比 + - degradation: 归一空间下 IS 目标均值 - OOS 目标均值, 正值 = 样本外退化 (过拟合信号), + 对"越小越好"目标 (max_drawdown 等) 方向也正确 + - consistency: OOS 总收益 > 0 的折占比 (与目标方向无关, 直观) """ n = len(fold_records) if n == 0: @@ -83,20 +92,22 @@ def aggregate_oos(fold_records: list[dict], objective: str) -> dict: 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_stats"].get(objective) for f in fold_records] - oos_vals = [v for v in oos_vals if v 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(avg_is - avg_oos, 4) if (avg_is is not None and avg_oos is not None) else None - n_positive = sum(1 for v in oos_vals if v > 0) - consistency = round(n_positive / len(oos_vals), 4) if oos_vals else 0.0 + 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, @@ -104,7 +115,7 @@ def aggregate_oos(fold_records: list[dict], objective: str) -> dict: "avg_is_objective": avg_is, "avg_oos_objective": avg_oos, "degradation": degradation, - "consistency": consistency, + "consistency": round(n_positive / n, 4), "oos_equity_curve": curve, } @@ -141,14 +152,17 @@ class WalkForwardService: progress_cb=None, cancel_event=None, ) -> dict: - from app.backtest.optimizer import OptimizeConfig + 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) - fold_records: list[dict] = [] + valid_records: list[dict] = [] # IS 与 OOS 都成功, 计入聚合 + skipped: list[dict] = [] # 无优化结果 或 OOS 失败, 不计入聚合 (避免伪装成有效折) + done = 0 for f in folds: if cancel_event is not None and cancel_event.is_set(): break @@ -170,9 +184,25 @@ class WalkForwardService: opt_res = self.optimizer.optimize(opt_cfg, cancel_event=cancel_event) best_params = opt_res.get("best_params") is_score = opt_res.get("best_score") + done += 1 + + 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), + } + + # 训练区间没优化出参数 (全组失败/取消) -> 跳过, 不用默认参数硬跑 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 or {})} + merged = {**cfg.base_params, **best_params} oos_cfg = StrategyBacktestConfig( strategy_id=cfg.strategy_id, symbols=cfg.symbols, @@ -183,34 +213,41 @@ class WalkForwardService: **cfg.backtest_kwargs, ) oos_res = self.service.run(oos_cfg, cancel_event=cancel_event) - oos_stats = {} if oos_res.error else oos_res.stats - fold_records.append({ - "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), + # 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_stats.get(cfg.objective), - "oos_stats": oos_stats, + "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": len(fold_records), - "total": n_total, - "fold": f.index, - }) + progress_cb({"type": "walkforward_progress", "done": done, "total": n_total, "fold": f.index}) - summary = aggregate_oos(fold_records, cfg.objective) + summary = aggregate_oos(valid_records, cfg.objective, direction) return { "objective": cfg.objective, - "n_folds": len(fold_records), + "direction": direction, + "n_folds": len(valid_records), + "n_skipped": len(skipped), "n_planned_folds": n_total, - "folds": fold_records, + "folds": valid_records, + "skipped": skipped, "summary": summary, "elapsed_ms": round((time.perf_counter() - t0) * 1000, 1), } diff --git a/backend/tests/backtest/test_walkforward.py b/backend/tests/backtest/test_walkforward.py index 9f2406f..036c730 100644 --- a/backend/tests/backtest/test_walkforward.py +++ b/backend/tests/backtest/test_walkforward.py @@ -30,12 +30,19 @@ def test_folds_rolling_windows(): f0 = folds[0] assert f0.train_start == date(2024, 1, 1) assert f0.train_end == date(2024, 3, 31) # +90d (2024 闰年) - assert f0.test_start == date(2024, 3, 31) # 紧接训练 - assert f0.test_end == date(2024, 4, 30) # +30d + assert f0.test_start == date(2024, 4, 1) # train_end + 1天 (隔断前视泄漏) + assert f0.test_end == date(2024, 5, 1) # +30d # 滚动: 下一折训练起点 +step assert folds[1].train_start == date(2024, 1, 31) # +30d +def test_folds_test_starts_day_after_train_end(): + """无前视泄漏: 每折 test_start 严格晚于 train_end (不共享同一天)。""" + folds = generate_folds(date(2024, 1, 1), date(2024, 12, 31), train_days=90, test_days=30, step_days=30) + for f in folds: + assert f.test_start > f.train_end + + def test_folds_no_test_beyond_end(): folds = generate_folds(date(2024, 1, 1), date(2024, 12, 31), train_days=90, test_days=30, step_days=30) for f in folds: @@ -63,6 +70,7 @@ def _rec(index, is_score, total_return, obj): "test_end": date(2024, 1, 1), "best_params": {"p": index}, "is_score": is_score, + "oos_objective": obj, "oos_stats": {"total_return": total_return, "sortino": obj}, } @@ -85,12 +93,22 @@ def test_aggregate_is_oos_degradation(): def test_aggregate_consistency_fraction_positive(): - # 3 折 OOS sortino: 1.5>0, -0.2<=0, 0.8>0 -> 2/3 正 + # consistency 按 OOS 总收益 > 0 的折占比: total_return 0.1>0, -0.1<=0, 0.1>0 -> 2/3 recs = [_rec(0, 1, 0.1, 1.5), _rec(1, 1, -0.1, -0.2), _rec(2, 1, 0.1, 0.8)] agg = aggregate_oos(recs, objective="sortino") assert agg["consistency"] == round(2 / 3, 4) # 0.6667 +def test_aggregate_degradation_direction_aware_for_min_objective(): + """min 类目标 (avg_holding_days, 越小越好): OOS 持仓天数更大 = 退化, degradation>0。""" + # IS 持仓 3 天, OOS 持仓 5 天 (更长=更差) -> 退化 + recs = [{"index": 0, "test_end": date(2024, 1, 1), "is_score": 3.0, + "oos_objective": 5.0, "oos_stats": {"total_return": 0.05}}] + agg = aggregate_oos(recs, objective="avg_holding_days", direction="min") + # 归一空间: norm(3)=-3, norm(5)=-5 -> degradation = -3 - (-5) = 2 > 0 = 退化 + assert agg["degradation"] == round(2.0, 4) + + def test_aggregate_empty_folds(): agg = aggregate_oos([], objective="sortino") assert agg["n_folds"] == 0 @@ -164,6 +182,42 @@ def test_walkforward_reports_degradation(): assert abs(out["summary"]["degradation"] - 1.0) < 1e-9 +class _NoParamsOptimizer(_FakeOptimizer): + """模拟训练区间全组失败: best_params=None。""" + def optimize(self, cfg, progress_cb=None, cancel_event=None): + self.train_ranges.append((cfg.start, cfg.end)) + return {"best_params": None, "best_score": None, "results": [], "n_completed": 0} + + +class _ErrorService(_FakeService): + """模拟 OOS 回测失败。""" + def run(self, config, progress_cb=None, cancel_event=None): + self.calls.append({"start": config.start, "end": config.end, "params": dict(config.params or {})}) + return _FakeResult(stats={}, error="no data") + + +def test_walkforward_skips_folds_without_optimized_params(): + """训练区间没优化出参数 (best_params=None) -> 跳过, 不用默认参数硬跑 OOS 伪装成有效折。""" + opt, svc = _NoParamsOptimizer(), _FakeService() + wf = WalkForwardService(opt, svc, strategy_engine=None) + out = wf.run(_wf_cfg()) + assert out["n_folds"] == 0 # 无有效折 + assert out["n_skipped"] > 0 # 全部跳过 + assert svc.calls == [] # 不跑 OOS + assert out["summary"]["compounded_oos_return"] == 0.0 # 无效折不污染净值 + + +def test_walkforward_skips_oos_error_folds(): + """OOS 回测失败的折 -> 跳过, 不把空/0 收益混入复利曲线。""" + opt, svc = _FakeOptimizer(), _ErrorService() + wf = WalkForwardService(opt, svc, strategy_engine=None) + out = wf.run(_wf_cfg()) + assert out["n_folds"] == 0 + assert out["n_skipped"] > 0 + assert len(svc.calls) > 0 # OOS 跑了但失败 + assert out["summary"]["compounded_oos_return"] == 0.0 # 失败折不计入 + + def test_walkforward_cancel_stops(): import threading ev = threading.Event() diff --git a/frontend/src/lib/walkforwardTask.ts b/frontend/src/lib/walkforwardTask.ts index 7fdddde..50d48d9 100644 --- a/frontend/src/lib/walkforwardTask.ts +++ b/frontend/src/lib/walkforwardTask.ts @@ -18,6 +18,7 @@ export interface WFFold { best_params: Record | null is_score: number | null oos_objective: number | null + oos_degraded: boolean | null oos_stats: Record } @@ -33,9 +34,12 @@ export interface WFSummary { export interface WalkForwardResult { objective: string + direction: string n_folds: number + n_skipped: number n_planned_folds: number folds: WFFold[] + skipped: { index: number; test_start: string; test_end: string; reason: string }[] summary: WFSummary elapsed_ms: number } diff --git a/frontend/src/pages/backtest/StrategyWalkForward.tsx b/frontend/src/pages/backtest/StrategyWalkForward.tsx index 4ca0b76..ee5787a 100644 --- a/frontend/src/pages/backtest/StrategyWalkForward.tsx +++ b/frontend/src/pages/backtest/StrategyWalkForward.tsx @@ -35,6 +35,32 @@ function Stat({ label, value, hint, color }: { label: string; value: string; hin ) } +/** OOS 拼接净值曲线 (逐折复利) — walk-forward 核心产出的极简 SVG 折线。 */ +function OosEquityChart({ curve }: { curve: { fold: number; date: string; value: number }[] }) { + if (!curve.length) return null + const W = 600, H = 120, pad = 8 + const vals = curve.map(p => p.value) + const lo = Math.min(1, ...vals), hi = Math.max(1, ...vals) + const span = hi - lo || 1 + // 起点补一个 value=1 基准, 让曲线从 1.0 起步 + const pts = [1, ...vals] + const x = (i: number) => pad + (i / (pts.length - 1 || 1)) * (W - 2 * pad) + const y = (v: number) => pad + (1 - (v - lo) / span) * (H - 2 * pad) + const d = pts.map((v, i) => `${i === 0 ? 'M' : 'L'}${x(i).toFixed(1)},${y(v).toFixed(1)}`).join(' ') + const last = vals[vals.length - 1] + const up = last >= 1 + return ( +
+
OOS 拼接净值 (逐折复利)
+ + + + +
终值 {last.toFixed(4)} · {curve.length} 折
+
+ ) +} + export function StrategyWalkForward() { const task = useWalkForwardTask() const { data: stratData } = useQuery({ queryKey: ['strategies'], queryFn: api.strategyList }) @@ -167,7 +193,12 @@ export function StrategyWalkForward() { /> )} - {result && summary && ( + {result && result.n_folds === 0 && ( + + )} + + {result && summary && result.n_folds > 0 && (
{/* 汇总卡 */}
@@ -177,14 +208,17 @@ export function StrategyWalkForward() { value={summary.degradation != null ? summary.degradation.toFixed(3) : '—'} hint={summary.degradation != null && summary.degradation > 0 ? '样本外退化=过拟合' : '样本外未退化'} color={summary.degradation != null && summary.degradation > 0 ? '#f87171' : '#34d399'} /> - - + + 0 ? `${result.n_folds} (跳过${result.n_skipped})` : String(result.n_folds)} />
IS 目标均值 {summary.avg_is_objective ?? '—'} · OOS 目标均值 {summary.avg_oos_objective ?? '—'} · 耗时 {(result.elapsed_ms / 1000).toFixed(1)}s
+ {/* OOS 拼接净值曲线 (walk-forward 核心产出) */} + + {/* 每折表 */}
@@ -202,7 +236,8 @@ export function StrategyWalkForward() { {result.folds.map(f => { const is = f.is_score const oos = f.oos_objective - const degraded = is != null && oos != null && oos < is + // 用后端方向感知的退化标志 (min 类目标 oos
{f.index + 1}