From 586d9a469934bd72942994064b5ff8bbbe31d707 Mon Sep 17 00:00:00 2001 From: im47cn <67424112+im47cn@users.noreply.github.com> Date: Wed, 8 Jul 2026 23:13:00 +0800 Subject: [PATCH 1/9] =?UTF-8?q?feat(walkforward):=20=E7=9C=9F=C2=B7walk-fo?= =?UTF-8?q?rward=20=E4=BC=98=E5=8C=96=E5=90=8E=E7=AB=AF=20=E2=80=94=20?= =?UTF-8?q?=E6=BB=9A=E5=8A=A8=E7=AA=97=E5=8F=A3=20IS=20=E4=BC=98=E5=8C=96?= =?UTF-8?q?=20+=20OOS=20=E9=AA=8C=E8=AF=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR2b 后端 (叠在 PR2a 优化器上): app/backtest/walkforward.py: - generate_folds: 滚动窗口切分 (训练固定长度 + 紧邻测试窗口 + step 前移), 测试超出 end 即停; 放不下一折抛错; 非正窗口拒绝。 - aggregate_oos: 从各折 OOS 结果聚合 — 复利净值曲线 / IS-vs-OOS 退化 (样本内目标均值 - 样本外目标均值, 正值=过拟合信号) / 一致性 (OOS 目标为正的折占比)。 - WalkForwardService.run: 每折在训练区间调 PR2a 优化器选最优参数, 再在测试区间 用该参数做 OOS 回测。核心产出是纯 OOS 拼接 + 每折 IS/OOS 对比 — 单次样本内 回测的过拟合一眼看穿。支持进度回调 (fold i/N) 与 cancel。 API app/api/backtest.py: - GET /walkforward/stream (SSE, 复用 _BacktestJob + job_key 回吐) + POST /walkforward/cancel (按回吐 key 查表)。 测试 13 例: fold 切分 (滚动/边界/不足/非正) + OOS 聚合 (复利/退化/一致性/空) + 编排 (训练优化→测试OOS/退化上报/取消) + API (job_key 区分窗口/cancel按key)。 全量 169 测试通过。 --- backend/app/api/backtest.py | 155 +++++++++++++++ backend/app/backtest/walkforward.py | 216 +++++++++++++++++++++ backend/tests/backtest/test_walkforward.py | 209 ++++++++++++++++++++ 3 files changed, 580 insertions(+) create mode 100644 backend/app/backtest/walkforward.py create mode 100644 backend/tests/backtest/test_walkforward.py diff --git a/backend/app/api/backtest.py b/backend/app/api/backtest.py index 7d89704..7ed9a7c 100644 --- a/backend/app/api/backtest.py +++ b/backend/app/api/backtest.py @@ -719,3 +719,158 @@ async def optimize_cancel(request: Request): return {"ok": True} return {"ok": False, "message": "任务不存在或已完成"} + +# ══════════════════════════════════════════════════════════════ +# Walk-forward 优化 — 每折训练区间优化 + 测试区间 OOS 验证 (复用优化器 + job_key 回吐) +# ══════════════════════════════════════════════════════════════ + +def _make_wf_job_key(strategy_id, symbols, start, end, param_grid, objective, direction, windows, bt_sig) -> str: + raw = f"WF|{strategy_id}|{symbols}|{start}|{end}|{param_grid}|{objective}|{direction}|{windows}|{bt_sig}" + return hashlib.md5(raw.encode()).hexdigest()[:12] + + +@router.get("/walkforward/stream") +async def walkforward_stream( + request: Request, + strategy_id: str, + param_grid: str, + objective: str = "sortino", + direction: str | None = None, + train_days: int = 252, + test_days: int = 63, + step_days: int = 63, + max_workers: int = 4, + symbols: str | None = None, + start: str | None = None, + end: str | None = None, + matching: str = "open_t+1", + fees_pct: float = 0.0002, + commission_pct: float | None = None, + stamp_tax_pct: float | None = None, + slippage_bps: float = 5.0, + max_positions: int = 10, + max_exposure_pct: float = 1.0, + initial_capital: float = 1_000_000.0, + position_sizing: str = "equal", + mode: str = "position", + holding_days: int = 5, +): + """SSE 流式 walk-forward: 每折训练区间网格优化 -> 测试区间 OOS 回测。 + + 事件: job {key} / progress {type:walkforward_progress,done,total,fold} / done {result} / error {message} + """ + from app.backtest.optimizer import StrategyOptimizer + from app.backtest.strategy import StrategyBacktestService + from app.backtest.walkforward import WalkForwardConfig, WalkForwardService + + direction = direction or None + engine = _get_engine(request) + strategy_engine = request.app.state.strategy_engine + svc = StrategyBacktestService(engine, strategy_engine) + optimizer = StrategyOptimizer(svc, strategy_engine) + + end_date = date.fromisoformat(end) if end else date.today() + if start: + start_date = date.fromisoformat(start) + else: + earliest = request.app.state.repo.earliest_daily_date() + start_date = earliest or (end_date - timedelta(days=STRATEGY_DEFAULT_DAYS)) + + bt_kwargs = _opt_backtest_kwargs( + matching, fees_pct, commission_pct, stamp_tax_pct, slippage_bps, + max_positions, max_exposure_pct, initial_capital, position_sizing, mode, holding_days, + ) + bt_sig = "|".join(f"{k}={bt_kwargs[k]}" for k in _OPT_BT_FIELDS) + windows = f"{train_days}/{test_days}/{step_days}" + job_key = _make_wf_job_key(strategy_id, symbols, start, end, param_grid, objective, direction, windows, bt_sig) + + _cleanup_stale_jobs() + with _jobs_lock: + job = _running_jobs.get(job_key) + if job is None: + job = _BacktestJob(job_key) + _running_jobs[job_key] = job + is_new = True + else: + is_new = False + + async def event_generator(): + yield f"event: job\ndata: {json.dumps({'key': job_key}, ensure_ascii=False)}\n\n" + + if is_new and not job.done: + try: + grid = json.loads(param_grid) + except (json.JSONDecodeError, TypeError): + grid = None + if not isinstance(grid, dict) or not grid: + job.error = "param_grid 必须是非空的参数网格对象" + job.done = True + job.finish_ts = time.time() + grid = None + + if grid is not None: + wf_cfg = WalkForwardConfig( + strategy_id=strategy_id, + symbols=[s.strip() for s in symbols.split(",") if s.strip()] if symbols else None, + start=start_date, + end=end_date, + param_grid=grid, + objective=objective, + direction=direction, + train_days=int(train_days), + test_days=int(test_days), + step_days=int(step_days), + max_workers=int(max_workers), + backtest_kwargs=bt_kwargs, + ) + + def _run_wf(): + try: + wf = WalkForwardService(optimizer, svc, strategy_engine) + job.result = wf.run(wf_cfg, lambda d: job.progress.append(d), job.cancel_event) + job.done = True + job.finish_ts = time.time() + except Exception as e: + job.error = str(e) + job.done = True + job.finish_ts = time.time() + + threading.Thread(target=_run_wf, daemon=True).start() + + cursor = 0 + tick = 0 + try: + while True: + if job.done: + if job.error: + yield f"event: error\ndata: {json.dumps({'message': job.error}, ensure_ascii=False)}\n\n" + 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" + return + tick += 1 + if tick % 4 == 0 and await request.is_disconnected(): + break + while cursor < len(job.progress): + msg = job.progress[cursor] + cursor += 1 + yield f"event: progress\ndata: {json.dumps(msg, ensure_ascii=False, default=str)}\n\n" + await asyncio.sleep(0.5) + except asyncio.CancelledError: + raise + + return StreamingResponse(event_generator(), media_type="text/event-stream") + + +@router.post("/walkforward/cancel") +async def walkforward_cancel(request: Request): + """取消 walk-forward 任务 — 传 stream 首事件回吐的 job_key。""" + body = await request.json() + job_key = body.get("job_key", "") + job = _running_jobs.get(job_key) + if job and not job.done: + job.cancel_event.set() + return {"ok": True} + return {"ok": False, "message": "任务不存在或已完成"} + diff --git a/backend/app/backtest/walkforward.py b/backend/app/backtest/walkforward.py new file mode 100644 index 0000000..969521a --- /dev/null +++ b/backend/app/backtest/walkforward.py @@ -0,0 +1,216 @@ +"""Walk-forward 优化 — 滚动窗口的样本内优化 + 样本外验证。 + +每折在训练区间用参数网格优化选出最优参数, 再在紧邻的测试区间用该参数做样本外(OOS) +回测。滚动前移。核心产出是 OOS 拼接净值 + 每折 IS-vs-OOS 退化 —— 样本内漂亮、样本外 +崩溃即过拟合信号, 单次样本内回测看不到。 + +依赖 PR2a 的 StrategyOptimizer 做每折训练区间的网格优化。 +""" +from __future__ import annotations + +import logging +import time +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 + 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 aggregate_oos(fold_records: list[dict], objective: str) -> dict: + """从各折 OOS 结果聚合: 复利净值曲线 / IS-OOS 退化 / 一致性。 + + 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 的折占比 + """ + 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] = [] + for f in fold_records: + r = float(f["oos_stats"].get("total_return", 0.0) or 0.0) + equity *= (1 + r) + 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] + + 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 + + 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": consistency, + "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) + + +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 run( + self, + cfg: WalkForwardConfig, + progress_cb=None, + cancel_event=None, + ) -> dict: + from app.backtest.optimizer import OptimizeConfig + from app.backtest.strategy import StrategyBacktestConfig + + t0 = time.perf_counter() + folds = generate_folds(cfg.start, cfg.end, cfg.train_days, cfg.test_days, cfg.step_days) + n_total = len(folds) + + fold_records: list[dict] = [] + for f in folds: + if cancel_event is not None and cancel_event.is_set(): + break + + # 训练区间: 网格优化选最优参数 + 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=cfg.backtest_kwargs, + ) + 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") + + # 测试区间: 用最优参数做样本外回测 + merged = {**cfg.base_params, **(best_params or {})} + 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_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), + "best_params": best_params, + "is_score": is_score, + "oos_objective": oos_stats.get(cfg.objective), + "oos_stats": oos_stats, + }) + + if progress_cb is not None: + progress_cb({ + "type": "walkforward_progress", + "done": len(fold_records), + "total": n_total, + "fold": f.index, + }) + + summary = aggregate_oos(fold_records, cfg.objective) + return { + "objective": cfg.objective, + "n_folds": len(fold_records), + "n_planned_folds": n_total, + "folds": fold_records, + "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 new file mode 100644 index 0000000..9f2406f --- /dev/null +++ b/backend/tests/backtest/test_walkforward.py @@ -0,0 +1,209 @@ +"""Walk-forward 核心测试 — 滚动窗口 fold 生成 + OOS 聚合 + 编排。 + +被测: +- generate_folds: 滚动训练/测试窗口切分 +- aggregate_oos: 从各折 OOS 结果聚合 (复利净值/IS-OOS 退化/一致性) +- WalkForwardService.run: 每折 训练区间优化 -> 测试区间 OOS 验证 +""" +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date + +import pytest + +from app.backtest.walkforward import ( + WalkForwardConfig, + WalkForwardService, + aggregate_oos, + generate_folds, +) + +# --------------------------------------------------------------- +# fold 生成 +# --------------------------------------------------------------- + +def test_folds_rolling_windows(): + # 1 年数据, 训练 90d / 测试 30d / 步进 30d + folds = generate_folds(date(2024, 1, 1), date(2024, 12, 31), train_days=90, test_days=30, step_days=30) + assert len(folds) > 0 + 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 + # 滚动: 下一折训练起点 +step + assert folds[1].train_start == date(2024, 1, 31) # +30d + + +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: + assert f.test_end <= date(2024, 12, 31) + + +def test_folds_insufficient_span_raises(): + # 训练90+测试30=120d, 但只有 100d 数据 -> 0 折 + with pytest.raises(ValueError, match=r"数据区间不足|至少"): + generate_folds(date(2024, 1, 1), date(2024, 4, 10), train_days=90, test_days=30, step_days=30) + + +def test_folds_reject_nonpositive_windows(): + with pytest.raises(ValueError, match=r"必须为正"): + generate_folds(date(2024, 1, 1), date(2024, 12, 31), train_days=0, test_days=30, step_days=30) + + +# --------------------------------------------------------------- +# OOS 聚合 +# --------------------------------------------------------------- + +def _rec(index, is_score, total_return, obj): + return { + "index": index, + "test_end": date(2024, 1, 1), + "best_params": {"p": index}, + "is_score": is_score, + "oos_stats": {"total_return": total_return, "sortino": obj}, + } + + +def test_aggregate_compounds_oos_returns(): + recs = [_rec(0, 2.0, 0.10, 1.5), _rec(1, 2.0, -0.05, 0.8), _rec(2, 2.0, 0.08, 1.2)] + agg = aggregate_oos(recs, objective="sortino") + # 复利: 1.1 * 0.95 * 1.08 - 1 + assert abs(agg["compounded_oos_return"] - (1.10 * 0.95 * 1.08 - 1)) < 1e-9 + assert len(agg["oos_equity_curve"]) == 3 + + +def test_aggregate_is_oos_degradation(): + # IS 目标平均远高于 OOS -> 退化为正 (过拟合信号) + recs = [_rec(0, 3.0, 0.05, 0.5), _rec(1, 3.0, 0.02, 0.3)] + agg = aggregate_oos(recs, objective="sortino") + assert agg["avg_is_objective"] == 3.0 + assert abs(agg["avg_oos_objective"] - 0.4) < 1e-9 + assert agg["degradation"] > 0 # IS 3.0 - OOS 0.4 = 2.6 + + +def test_aggregate_consistency_fraction_positive(): + # 3 折 OOS sortino: 1.5>0, -0.2<=0, 0.8>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_empty_folds(): + agg = aggregate_oos([], objective="sortino") + assert agg["n_folds"] == 0 + assert agg["compounded_oos_return"] == 0.0 + + +# --------------------------------------------------------------- +# 编排 (假 optimizer / service) +# --------------------------------------------------------------- + +@dataclass +class _FakeResult: + stats: dict + error: str | None = None + + +class _FakeOptimizer: + """optimize 返回受控 best_params/best_score, 记录被优化的训练区间。""" + def __init__(self): + self.train_ranges = [] + + def optimize(self, cfg, progress_cb=None, cancel_event=None): + self.train_ranges.append((cfg.start, cfg.end)) + # best_params 随训练起点变化, best_score 固定 + return {"best_params": {"p": cfg.start.month}, "best_score": 2.0, "results": [], "n_completed": 1} + + +class _FakeService: + """run 返回受控 OOS stats, 记录测试区间 + 收到的 params。""" + def __init__(self): + self.calls = [] + + 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={"total_return": 0.05, "sortino": 1.0}) + + +def _wf_cfg(**kw): + base = dict( + strategy_id="s", symbols=None, start=date(2024, 1, 1), end=date(2024, 12, 31), + param_grid={"p": [1, 2]}, objective="sortino", + train_days=90, test_days=30, step_days=30, + ) + base.update(kw) + return WalkForwardConfig(**base) + + +def test_walkforward_optimizes_train_applies_oos(): + opt, svc = _FakeOptimizer(), _FakeService() + wf = WalkForwardService(opt, svc, strategy_engine=None) + out = wf.run(_wf_cfg()) + + assert out["n_folds"] > 0 + # 每折: optimizer 在训练区间跑, service 在测试区间用最优参数跑 + assert len(opt.train_ranges) == out["n_folds"] + assert len(svc.calls) == out["n_folds"] + # OOS 回测用的是该折优化出的 best_params (来自训练起点月份) + first_fold = out["folds"][0] + assert svc.calls[0]["params"] == first_fold["best_params"] + # 训练区间与测试区间不重叠 (测试在训练之后) + assert svc.calls[0]["start"] >= opt.train_ranges[0][1] + + +def test_walkforward_reports_degradation(): + opt, svc = _FakeOptimizer(), _FakeService() + wf = WalkForwardService(opt, svc, strategy_engine=None) + out = wf.run(_wf_cfg()) + # IS best_score=2.0, OOS sortino=1.0 -> 退化 1.0 + assert out["summary"]["avg_is_objective"] == 2.0 + assert out["summary"]["avg_oos_objective"] == 1.0 + assert abs(out["summary"]["degradation"] - 1.0) < 1e-9 + + +def test_walkforward_cancel_stops(): + import threading + ev = threading.Event() + ev.set() + opt, svc = _FakeOptimizer(), _FakeService() + wf = WalkForwardService(opt, svc, strategy_engine=None) + out = wf.run(_wf_cfg(), cancel_event=ev) + # 取消 -> 不跑任何折 + assert svc.calls == [] + assert out["n_folds"] == 0 + + +# --------------------------------------------------------------- +# API: job_key 回吐 + cancel 按 key 查表 +# --------------------------------------------------------------- + +def test_wf_job_key_distinguishes_windows(): + from app.api.backtest import _make_wf_job_key + base = _make_wf_job_key("s", None, None, None, '{"p":[1]}', "sortino", None, "252/63/63", "sig") + assert base != _make_wf_job_key("s", None, None, None, '{"p":[1]}', "sortino", None, "120/30/30", "sig") + + +def test_wf_cancel_by_echoed_key(): + import asyncio + + from app.api.backtest import _BacktestJob, _running_jobs, walkforward_cancel + + class _Req: + def __init__(self, body): + self._body = body + async def json(self): + return self._body + + key = "wfkey_test_1" + _running_jobs[key] = _BacktestJob(key) + try: + res = asyncio.run(walkforward_cancel(_Req({"job_key": key}))) + assert res["ok"] is True + assert _running_jobs[key].cancel_event.is_set() + res2 = asyncio.run(walkforward_cancel(_Req({"job_key": "nope"}))) + assert res2["ok"] is False + finally: + _running_jobs.pop(key, None) From 4c40c6b1bf67bbf1bb625a8fa5bc9bb8ff0e9814 Mon Sep 17 00:00:00 2001 From: im47cn <67424112+im47cn@users.noreply.github.com> Date: Wed, 8 Jul 2026 23:18:37 +0800 Subject: [PATCH 2/9] =?UTF-8?q?feat(walkforward):=20=E5=89=8D=E7=AB=AF=20w?= =?UTF-8?q?alk-forward=20=E9=9D=A2=E6=9D=BF=20+=20=E6=8A=BD=E5=87=BA?= =?UTF-8?q?=E5=85=B1=E4=BA=AB=E5=8F=82=E6=95=B0=E6=89=AB=E6=8F=8F=E7=BB=84?= =?UTF-8?q?=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR2b 前端 (完成 PR2b): 抽共享组件 (DRY, 优化器与 walk-forward 复用): - components/paramSweep.tsx: Sweep 类型/helper + useParamSweep hook (策略选择 + 各参数扫描配置 + 组合数 + buildGrid) + StrategySelect/SweepParamList/CombosHint 组件 + OBJECTIVES/INPUT_CLS 常量。 - StrategyOptimizer.tsx 重构复用该组件 (删除本地重复的 sweep 逻辑与 JSX)。 Walk-forward: - lib/walkforwardTask.ts: SSE 客户端 (镜像 optimizerTask + job_key 回吐 + 重连)。 - StrategyWalkForward.tsx: 配置 (策略/目标/日期/训练-测试-步进窗口/可扫参数) + 结果 (汇总卡: OOS复利收益/IS→OOS退化/一致性/折数; 每折表: 测试区间/最优参数/ IS目标/OOS目标(退化标红)/OOS收益; 过拟合告警条)。 - Backtest.tsx: 新增 'Walk-forward' 第四 tab。 前端 tsc 无新增类型错误。 --- frontend/src/lib/walkforwardTask.ts | 223 ++++++++++++++++ frontend/src/pages/Backtest.tsx | 14 +- .../src/pages/backtest/StrategyOptimizer.tsx | 189 ++------------ .../pages/backtest/StrategyWalkForward.tsx | 238 ++++++++++++++++++ .../pages/backtest/components/paramSweep.tsx | 185 ++++++++++++++ 5 files changed, 682 insertions(+), 167 deletions(-) create mode 100644 frontend/src/lib/walkforwardTask.ts create mode 100644 frontend/src/pages/backtest/StrategyWalkForward.tsx create mode 100644 frontend/src/pages/backtest/components/paramSweep.tsx diff --git a/frontend/src/lib/walkforwardTask.ts b/frontend/src/lib/walkforwardTask.ts new file mode 100644 index 0000000..7fdddde --- /dev/null +++ b/frontend/src/lib/walkforwardTask.ts @@ -0,0 +1,223 @@ +import { useSyncExternalStore } from 'react' + +/** Walk-forward 任务管理 (SSE + job_key 回吐 + 重连)。镜像 optimizerTask。 */ + +export interface WFProgress { + type: string + done: number + total: number + fold: number +} + +export interface WFFold { + index: number + train_start: string + train_end: string + test_start: string + test_end: string + best_params: Record | null + is_score: number | null + oos_objective: number | null + oos_stats: Record +} + +export interface WFSummary { + n_folds: number + compounded_oos_return: number + avg_is_objective: number | null + avg_oos_objective: number | null + degradation: number | null + consistency: number + oos_equity_curve: { fold: number; date: string; value: number }[] +} + +export interface WalkForwardResult { + objective: string + n_folds: number + n_planned_folds: number + folds: WFFold[] + summary: WFSummary + elapsed_ms: number +} + +export interface WalkForwardTask { + id: number + isPending: boolean + result: WalkForwardResult | null + progress: WFProgress | null + error: string | null +} + +export interface StartWalkForwardParams { + strategy_id: string + param_grid: Record + objective: string + train_days: number + test_days: number + step_days: number + symbols?: string[] | null + start?: string | null + end?: string | null + mode?: 'position' | 'full' +} + +let current: WalkForwardTask | null = null +const listeners = new Set<() => void>() +let taskSeq = 0 +let eventSource: EventSource | null = null +let currentJobKey: string | null = null + +const RECONNECT_KEY = 'walkforward_reconnect' +const JOB_KEY_KEY = 'walkforward_job_key' + +function emit() { + listeners.forEach(fn => fn()) +} + +function subscribe(fn: () => void) { + listeners.add(fn) + return () => listeners.delete(fn) +} + +function buildQuery(params: Record): string { + const sp = new URLSearchParams() + for (const [k, v] of Object.entries(params)) { + if (v != null && v !== '') sp.set(k, String(v)) + } + return sp.toString() +} + +function connectSSE(url: string): void { + const id = current?.id ?? ++taskSeq + + if (eventSource) { + eventSource.close() + eventSource = null + } + + const es = new EventSource(url) + eventSource = es + + es.addEventListener('job', (e: MessageEvent) => { + try { + const key = JSON.parse(e.data)?.key + if (key) { + currentJobKey = key + localStorage.setItem(JOB_KEY_KEY, key) + } + } catch { /* ignore */ } + }) + + es.addEventListener('progress', (e: MessageEvent) => { + if (current?.id !== id) return + try { + const prog = JSON.parse(e.data) as WFProgress + current = { ...current, progress: prog } + emit() + } catch { /* ignore */ } + }) + + es.addEventListener('done', (e: MessageEvent) => { + if (current?.id !== id) return + try { + const result = JSON.parse(e.data) as WalkForwardResult + current = { ...current, isPending: false, result, error: null } + emit() + } catch { + current = { ...current, isPending: false, error: '结果解析失败' } + emit() + } + es.close() + eventSource = null + currentJobKey = null + localStorage.removeItem(RECONNECT_KEY) + localStorage.removeItem(JOB_KEY_KEY) + }) + + es.addEventListener('error', (e: MessageEvent) => { + if (current?.id !== id) return + if (e.data) { + try { + const msg = JSON.parse(e.data)?.message ?? 'walk-forward 出错' + current = { ...current, isPending: false, error: msg } + emit() + } catch { + current = { ...current, isPending: false, error: 'walk-forward 出错' } + emit() + } + es.close() + eventSource = null + currentJobKey = null + localStorage.removeItem(RECONNECT_KEY) + localStorage.removeItem(JOB_KEY_KEY) + } + }) +} + +export function startWalkForward(params: StartWalkForwardParams): void { + if (eventSource) { + eventSource.close() + eventSource = null + } + + const id = ++taskSeq + current = { id, isPending: true, result: null, progress: null, error: null } + emit() + + const qs = buildQuery({ + strategy_id: params.strategy_id, + param_grid: JSON.stringify(params.param_grid), + objective: params.objective, + train_days: params.train_days, + test_days: params.test_days, + step_days: params.step_days, + symbols: params.symbols?.join(','), + start: params.start ?? undefined, + end: params.end ?? undefined, + mode: params.mode, + }) + + localStorage.setItem(RECONNECT_KEY, qs) + connectSSE(`/api/backtest/walkforward/stream?${qs}`) +} + +export async function stopWalkForward(): Promise { + const jobKey = currentJobKey ?? localStorage.getItem(JOB_KEY_KEY) + if (jobKey) { + await fetch('/api/backtest/walkforward/cancel', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ job_key: jobKey }), + }).catch(() => {}) + } + if (eventSource) { + eventSource.close() + eventSource = null + } + if (current?.isPending) { + current = { ...current, isPending: false, error: '已取消' } + emit() + } + currentJobKey = null + localStorage.removeItem(RECONNECT_KEY) + localStorage.removeItem(JOB_KEY_KEY) +} + +export function clearWalkForward(): void { + current = null + emit() +} + +export function tryReconnectWalkForward(): boolean { + const qs = localStorage.getItem(RECONNECT_KEY) + if (!qs) return false + const id = ++taskSeq + current = { id, isPending: true, result: null, progress: null, error: null } + emit() + connectSSE(`/api/backtest/walkforward/stream?${qs}`) + return true +} + +export function useWalkForwardTask(): WalkForwardTask | null { + return useSyncExternalStore(subscribe, () => current, () => null) +} diff --git a/frontend/src/pages/Backtest.tsx b/frontend/src/pages/Backtest.tsx index 4227f3e..fa44702 100644 --- a/frontend/src/pages/Backtest.tsx +++ b/frontend/src/pages/Backtest.tsx @@ -3,9 +3,10 @@ import { PageHeader } from '@/components/PageHeader' import { FactorBacktest } from './backtest/FactorBacktest' import { StrategyBacktest } from './backtest/StrategyBacktest' import { StrategyOptimizer } from './backtest/StrategyOptimizer' -import { BarChart3, FlaskConical, SlidersHorizontal } from 'lucide-react' +import { StrategyWalkForward } from './backtest/StrategyWalkForward' +import { BarChart3, FlaskConical, SlidersHorizontal, Waypoints } from 'lucide-react' -type Tab = 'factor' | 'strategy' | 'optimizer' +type Tab = 'factor' | 'strategy' | 'optimizer' | 'walkforward' const MODES: Record = { factor: { @@ -23,12 +24,18 @@ const MODES: Record = { subtitle: '网格搜索最优参数组合', hint: '并行回测所有参数组合，按夏普/索提诺等目标排序，找到最优参数。', }, + walkforward: { + title: 'Walk-forward', + subtitle: '滚动窗口样本外验证', + hint: '每折训练区间优化、测试区间验证，看样本外是否退化以识别过拟合。', + }, } const TAB_ICONS: Record = { factor: BarChart3, strategy: FlaskConical, optimizer: SlidersHorizontal, + walkforward: Waypoints, } export function Backtest() { @@ -36,7 +43,7 @@ export function Backtest() { const modeSwitch = (
- {(['factor', 'strategy', 'optimizer'] as const).map(tab => { + {(['factor', 'strategy', 'optimizer', 'walkforward'] as const).map(tab => { const Icon = TAB_ICONS[tab] const active = activeTab === tab return ( @@ -77,6 +84,7 @@ export function Backtest() { {activeTab === 'factor' && } {activeTab === 'strategy' && } {activeTab === 'optimizer' && } + {activeTab === 'walkforward' && }
) diff --git a/frontend/src/pages/backtest/StrategyOptimizer.tsx b/frontend/src/pages/backtest/StrategyOptimizer.tsx index cb8398d..5b2f151 100644 --- a/frontend/src/pages/backtest/StrategyOptimizer.tsx +++ b/frontend/src/pages/backtest/StrategyOptimizer.tsx @@ -1,7 +1,7 @@ -import { useEffect, useMemo, useState } from 'react' +import { useEffect, useState } from 'react' import { useQuery } from '@tanstack/react-query' import { Play, Square, Trophy } from 'lucide-react' -import { api, type StrategyDetail, type StrategyParamDef } from '@/lib/api' +import { api, type StrategyDetail } from '@/lib/api' import { fmtPct } from '@/lib/format' import { EmptyState } from '@/components/EmptyState' import { DatePicker } from '@/components/DatePicker' @@ -13,65 +13,15 @@ import { useOptimizerTask, } from '@/lib/optimizerTask' import { buildDefaultOverrides } from '@/lib/strategyOverrides' - -const INPUT_CLS = 'w-full px-2.5 py-1.5 rounded-input bg-surface border border-border text-xs focus:outline-none focus:border-accent' - -// 可选优化目标 (对齐后端 VALID_OBJECTIVES) + 中文标签 + 是否越小越好 -const OBJECTIVES: { id: string; label: string; min?: boolean }[] = [ - { id: 'sortino', label: '索提诺比率' }, - { id: 'sharpe', label: '夏普比率' }, - { id: 'calmar', label: 'Calmar 比率' }, - { id: 'total_return', label: '总收益' }, - { id: 'annual_return', label: '年化收益' }, - { id: 'win_rate', label: '胜率' }, - { id: 'profit_factor', label: '盈亏比' }, - { id: 'max_drawdown', label: '最大回撤(越小越好)' }, - { id: 'mc_maxdd_p95', label: '蒙卡回撤P95(越小越好)' }, - { id: 'avg_holding_days', label: '平均持仓天数', min: true }, -] - -// 单个可扫参数的网格配置 -interface Sweep { - enabled: boolean - min: string - max: string - step: string -} - -function defaultSweep(p: StrategyParamDef): Sweep { - return { - enabled: false, - min: String(p.min ?? p.default ?? 0), - max: String(p.max ?? p.default ?? 1), - step: String(p.step ?? (p.type === 'int' ? 1 : 0.01)), - } -} - -/** 从 sweep 配置估算某参数候选值个数 (与后端整数计数一致) */ -function candidateCount(p: StrategyParamDef, s: Sweep): number { - if (p.type === 'bool') return 2 - if (p.type === 'select') return p.options?.length ?? 1 - const lo = Number(s.min), hi = Number(s.max), step = Number(s.step) - if (!(step > 0) || hi < lo) return 0 - return Math.round((hi - lo) / step) + 1 -} - -/** 校验某数值参数的 sweep 是否会被后端拒绝 (与后端 _candidates_for 同口径)。 - * 后端按 lo+i*step 生成 (i=0..round((hi-lo)/step)), 任一值超出 [min,max] 即报错。 */ -function sweepError(p: StrategyParamDef, s: Sweep): string | null { - if (p.type === 'bool' || p.type === 'select') return null - const lo = Number(s.min), hi = Number(s.max), step = Number(s.step) - if (Number.isNaN(lo) || Number.isNaN(hi) || Number.isNaN(step)) return `${p.label}: 范围/步长非法` - if (!(step > 0)) return `${p.label}: 步长必须为正` - if (hi < lo) return `${p.label}: max < min` - if (p.min != null && lo < p.min - 1e-9) return `${p.label}: min 小于允许下限 ${p.min}` - if (p.max != null && hi > p.max + 1e-9) return `${p.label}: max 超出允许上限 ${p.max}` - // 后端生成的末值 lo + round((hi-lo)/step)*step 若 > max, 会被拒 - const nSteps = Math.round((hi - lo) / step) - const last = lo + nSteps * step - if (last > hi + 1e-9) return `${p.label}: 步长 ${step} 不整除区间, 末值 ${last.toFixed(4)} 超出 max ${hi}` - return null -} +import { + INPUT_CLS, + OBJECTIVES, + GRID_MAX_COMBINATIONS, + useParamSweep, + StrategySelect, + SweepParamList, + CombosHint, +} from './components/paramSweep' const TODAY = new Date().toISOString().slice(0, 10) const ONE_YEAR_AGO = new Date(Date.now() - 365 * 864e5).toISOString().slice(0, 10) @@ -81,15 +31,15 @@ export function StrategyOptimizer() { const { data: stratData } = useQuery({ queryKey: ['strategies'], queryFn: api.strategyList }) const strategies: StrategyDetail[] = stratData?.strategies ?? [] - const [strategyId, setStrategyId] = useState('') + // 切策略: 有任务在跑时先真正取消 (关 SSE + 后端 cancel + 清 localStorage), 不能静默丢 + const sweep = useParamSweep(strategies, () => { + if (task?.isPending) stopOptimize() + else clearOptimize() + }) const [objective, setObjective] = useState('sortino') const [start, setStart] = useState(ONE_YEAR_AGO) const [end, setEnd] = useState(TODAY) const [mode, setMode] = useState<'position' | 'full'>('position') - const [sweeps, setSweeps] = useState>({}) - - const selected = strategies.find(s => s.id === strategyId) - const params = selected?.params ?? [] // 刷新/切页后: 恢复未完成的优化任务 useEffect(() => { @@ -97,62 +47,20 @@ export function StrategyOptimizer() { // eslint-disable-next-line react-hooks/exhaustive-deps }, []) - // 切策略: 若有任务在跑, 先真正取消 (关 SSE + 后端 cancel + 清 localStorage), 不能静默丢。 - const onSelectStrategy = (id: string) => { - if (task?.isPending) stopOptimize() - else clearOptimize() - setStrategyId(id) - const s = strategies.find(x => x.id === id) - const init: Record = {} - for (const p of s?.params ?? []) init[p.id] = defaultSweep(p) - setSweeps(init) - } - - const updateSweep = (pid: string, patch: Partial) => - setSweeps(prev => ({ ...prev, [pid]: { ...prev[pid], ...patch } })) - - // 组合数预估 - const combos = useMemo(() => { - const enabled = params.filter(p => sweeps[p.id]?.enabled) - if (!enabled.length) return 0 - return enabled.reduce((acc, p) => acc * candidateCount(p, sweeps[p.id]), 1) - }, [params, sweeps]) - - // 网格合法性 (与后端展开同口径): 步长不整除/越界会被后端拒, 前端提前拦。 - const gridError = useMemo(() => { - for (const p of params) { - if (!sweeps[p.id]?.enabled) continue - const err = sweepError(p, sweeps[p.id]) - if (err) return err - } - return null - }, [params, sweeps]) - - const buildGrid = (): Record => { - const grid: Record = {} - for (const p of params) { - const s = sweeps[p.id] - if (!s?.enabled) continue - if (p.type === 'bool') grid[p.id] = [true, false] - else if (p.type === 'select') grid[p.id] = p.options ?? [] - else grid[p.id] = { min: Number(s.min), max: Number(s.max), step: Number(s.step) } - } - return grid - } - - const canRun = strategyId && combos > 0 && combos <= 2000 && !gridError && !task?.isPending + const canRun = sweep.strategyId && sweep.combos > 0 && sweep.combos <= GRID_MAX_COMBINATIONS + && !sweep.gridError && !task?.isPending const onRun = () => { if (!canRun) return clearOptimize() startOptimize({ - strategy_id: strategyId, - param_grid: buildGrid(), + strategy_id: sweep.strategyId, + param_grid: sweep.buildGrid(), objective, // 未扫描参数固定为策略当前默认值; overrides 让 basic_filter/信号/风控按当前策略参与, // 保证优化的就是用户实际回测的策略 (而非被剥离配置的裸策略)。 - params: selected?.params_defaults, - overrides: selected ? buildDefaultOverrides(selected) : undefined, + params: sweep.selected?.params_defaults, + overrides: sweep.selected ? buildDefaultOverrides(sweep.selected) : undefined, start, end, mode, @@ -168,10 +76,7 @@ export function StrategyOptimizer() {
- +
@@ -200,50 +105,8 @@ export function StrategyOptimizer() {
- {/* 可扫参数 */} - {params.length > 0 && ( -
-
扫描参数 (勾选后设范围)
-
- {params.map(p => { - const s = sweeps[p.id] ?? defaultSweep(p) - const numeric = p.type === 'float' || p.type === 'int' - return ( -
- - {s.enabled && numeric && ( -
- updateSweep(p.id, { min: e.target.value })} placeholder="min" className={INPUT_CLS} /> - updateSweep(p.id, { max: e.target.value })} placeholder="max" className={INPUT_CLS} /> - updateSweep(p.id, { step: e.target.value })} placeholder="step" className={INPUT_CLS} /> -
- )} - {s.enabled && !numeric && ( -
- {p.type === 'bool' ? '扫描 [是 / 否]' : `扫描全部选项 (${p.options?.length ?? 0})`} -
- )} -
- ) - })} -
-
- )} - - {/* 组合数 / 校验提示 */} - {strategyId && ( -
2000 || gridError) ? 'text-red-400' : 'text-secondary'}`}> - {gridError - ? gridError - : combos === 0 - ? '请至少勾选一个参数' - : `共 ${combos} 组参数组合${combos > 2000 ? ' — 超过上限 2000, 请增大 step 或缩小范围' : ''}`} -
- )} + + {task?.isPending ? (
+ ) +} diff --git a/frontend/src/pages/backtest/components/paramSweep.tsx b/frontend/src/pages/backtest/components/paramSweep.tsx new file mode 100644 index 0000000..4ab0215 --- /dev/null +++ b/frontend/src/pages/backtest/components/paramSweep.tsx @@ -0,0 +1,185 @@ +import { useMemo, useState } from 'react' +import type { StrategyDetail, StrategyParamDef } from '@/lib/api' + +/** 参数扫描配置的共享逻辑与 UI — 优化器与 walk-forward 复用。 */ + +export const INPUT_CLS = + 'w-full px-2.5 py-1.5 rounded-input bg-surface border border-border text-xs focus:outline-none focus:border-accent' + +// 可选优化目标 (对齐后端 VALID_OBJECTIVES) + 中文标签 +export const OBJECTIVES: { id: string; label: string }[] = [ + { id: 'sortino', label: '索提诺比率' }, + { id: 'sharpe', label: '夏普比率' }, + { id: 'calmar', label: 'Calmar 比率' }, + { id: 'total_return', label: '总收益' }, + { id: 'annual_return', label: '年化收益' }, + { id: 'win_rate', label: '胜率' }, + { id: 'profit_factor', label: '盈亏比' }, + { id: 'max_drawdown', label: '最大回撤(越小越好)' }, + { id: 'mc_maxdd_p95', label: '蒙卡回撤P95(越小越好)' }, + { id: 'avg_holding_days', label: '平均持仓天数' }, +] + +export const GRID_MAX_COMBINATIONS = 2000 + +export interface Sweep { + enabled: boolean + min: string + max: string + step: string +} + +function defaultSweep(p: StrategyParamDef): Sweep { + return { + enabled: false, + min: String(p.min ?? p.default ?? 0), + max: String(p.max ?? p.default ?? 1), + step: String(p.step ?? (p.type === 'int' ? 1 : 0.01)), + } +} + +/** 某参数候选值个数 (与后端整数计数一致)。 */ +function candidateCount(p: StrategyParamDef, s: Sweep): number { + if (p.type === 'bool') return 2 + if (p.type === 'select') return p.options?.length ?? 1 + const lo = Number(s.min), hi = Number(s.max), step = Number(s.step) + if (!(step > 0) || hi < lo) return 0 + return Math.round((hi - lo) / step) + 1 +} + +/** 校验某数值参数的 sweep 是否会被后端拒绝 (与后端 _candidates_for 同口径)。 + * 后端按 lo+i*step 生成 (i=0..round((hi-lo)/step)), 任一值超出 [min,max] 即报错。 */ +function sweepError(p: StrategyParamDef, s: Sweep): string | null { + if (p.type === 'bool' || p.type === 'select') return null + const lo = Number(s.min), hi = Number(s.max), step = Number(s.step) + if (Number.isNaN(lo) || Number.isNaN(hi) || Number.isNaN(step)) return `${p.label}: 范围/步长非法` + if (!(step > 0)) return `${p.label}: 步长必须为正` + if (hi < lo) return `${p.label}: max < min` + if (p.min != null && lo < p.min - 1e-9) return `${p.label}: min 小于允许下限 ${p.min}` + if (p.max != null && hi > p.max + 1e-9) return `${p.label}: max 超出允许上限 ${p.max}` + const nSteps = Math.round((hi - lo) / step) + const last = lo + nSteps * step + if (last > hi + 1e-9) return `${p.label}: 步长 ${step} 不整除区间, 末值 ${last.toFixed(4)} 超出 max ${hi}` + return null +} + +/** 管理策略选择 + 各参数扫描配置, 派生组合数 / 校验 / param_grid。 */ +export function useParamSweep(strategies: StrategyDetail[], onStrategyChange?: () => void) { + const [strategyId, setStrategyId] = useState('') + const [sweeps, setSweeps] = useState>({}) + + const selected = strategies.find(s => s.id === strategyId) + const params = selected?.params ?? [] + + const selectStrategy = (id: string) => { + setStrategyId(id) + onStrategyChange?.() + const s = strategies.find(x => x.id === id) + const init: Record = {} + for (const p of s?.params ?? []) init[p.id] = defaultSweep(p) + setSweeps(init) + } + + const updateSweep = (pid: string, patch: Partial) => + setSweeps(prev => ({ ...prev, [pid]: { ...prev[pid], ...patch } })) + + const combos = useMemo(() => { + const enabled = params.filter(p => sweeps[p.id]?.enabled) + if (!enabled.length) return 0 + return enabled.reduce((acc, p) => acc * candidateCount(p, sweeps[p.id]), 1) + }, [params, sweeps]) + + // 网格合法性 (与后端展开同口径): 步长不整除/越界会被后端拒, 前端提前拦。 + const gridError = useMemo(() => { + for (const p of params) { + if (!sweeps[p.id]?.enabled) continue + const err = sweepError(p, sweeps[p.id]) + if (err) return err + } + return null + }, [params, sweeps]) + + const buildGrid = (): Record => { + const grid: Record = {} + for (const p of params) { + const s = sweeps[p.id] + if (!s?.enabled) continue + if (p.type === 'bool') grid[p.id] = [true, false] + else if (p.type === 'select') grid[p.id] = p.options ?? [] + else grid[p.id] = { min: Number(s.min), max: Number(s.max), step: Number(s.step) } + } + return grid + } + + return { strategyId, selected, selectStrategy, params, sweeps, updateSweep, combos, gridError, buildGrid } +} + +/** 策略选择器。 */ +export function StrategySelect({ strategies, value, onChange }: { + strategies: StrategyDetail[] + value: string + onChange: (id: string) => void +}) { + return ( + + ) +} + +/** 可扫参数列表 (勾选 + min/max/step)。 */ +export function SweepParamList({ params, sweeps, updateSweep }: { + params: StrategyParamDef[] + sweeps: Record + updateSweep: (pid: string, patch: Partial) => void +}) { + if (!params.length) return null + return ( +
+
扫描参数 (勾选后设范围)
+
+ {params.map(p => { + const s = sweeps[p.id] ?? defaultSweep(p) + const numeric = p.type === 'float' || p.type === 'int' + return ( +
+ + {s.enabled && numeric && ( +
+ updateSweep(p.id, { min: e.target.value })} placeholder="min" className={INPUT_CLS} /> + updateSweep(p.id, { max: e.target.value })} placeholder="max" className={INPUT_CLS} /> + updateSweep(p.id, { step: e.target.value })} placeholder="step" className={INPUT_CLS} /> +
+ )} + {s.enabled && !numeric && ( +
+ {p.type === 'bool' ? '扫描 [是 / 否]' : `扫描全部选项 (${p.options?.length ?? 0})`} +
+ )} +
+ ) + })} +
+
+ ) +} + +/** 组合数 / 校验提示 (含上限与网格错误告警)。 */ +export function CombosHint({ show, combos, gridError }: { show: boolean; combos: number; gridError?: string | null }) { + if (!show) return null + const bad = combos > GRID_MAX_COMBINATIONS || !!gridError + return ( +
+ {gridError + ? gridError + : combos === 0 + ? '请至少勾选一个参数' + : `共 ${combos} 组参数组合${combos > GRID_MAX_COMBINATIONS ? ` — 超过上限 ${GRID_MAX_COMBINATIONS}, 请增大 step 或缩小范围` : ''}`} +
+ ) +} 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 3/9] =?UTF-8?q?fix(walkforward):=20=E5=AD=90=E4=BB=A3?= =?UTF-8?q?=E7=90=86=E5=AE=A1=E6=9F=A5=E4=BF=AE=E5=A4=8D=20=E2=80=94=20?= =?UTF-8?q?=E6=8A=98=E6=9C=89=E6=95=88=E6=80=A7/=E5=89=8D=E8=A7=86?= =?UTF-8?q?=E6=B3=84=E6=BC=8F/=E6=96=B9=E5=90=91=E9=80=80=E5=8C=96/NaN?= =?UTF-8?q?=E5=BA=8F=E5=88=97=E5=8C=96/=E5=87=80=E5=80=BC=E6=9B=B2?= =?UTF-8?q?=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 From 72697707a4499085a0699fc28e78e52ca11b6f56 Mon Sep 17 00:00:00 2001 From: im47cn <67424112+im47cn@users.noreply.github.com> Date: Sun, 12 Jul 2026 00:52:11 +0800 Subject: [PATCH 4/9] =?UTF-8?q?fix(walkforward):=20=E4=BC=98=E5=8C=96?= =?UTF-8?q?=E5=99=A8=E5=90=83=E7=94=A8=E6=88=B7=E5=BD=93=E5=89=8D=E7=AD=96?= =?UTF-8?q?=E7=95=A5=E9=85=8D=E7=BD=AE=20+=20EmptyState=20=E6=9E=84?= =?UTF-8?q?=E5=BB=BA=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WF 前端 onRun 只传了 grid/objective/窗口, 没传当前策略的 params/overrides: - 未扫描参数不固定为用户当前值 - basic_filter/entry_signals/exit_signals/止损止盈/评分等覆盖不参与优化 结果 walk-forward 优化的策略与用户实际回测的策略不一致 (与 PR #82 优化器 阻塞项同类问题)。修复: onRun 补 params=selected.params_defaults + overrides=buildDefaultOverrides(selected), 复用 optimizer 同一工具。 附带: - StrategyWalkForward.tsx 用了 EmptyState description= (该组件只接受 title/hint), 导致 tsc -b 构建失败, 2 处改为 hint。 - _make_wf_job_key 已含 params/overrides, 补测试断言二者不同产出不同 job_key (stream 与 cancel 对齐前提) 且相同入参稳定一致。 --- backend/app/api/backtest.py | 18 ++++- backend/tests/backtest/test_walkforward.py | 16 +++++ frontend/src/lib/walkforwardTask.ts | 65 +++++++++++++++---- .../pages/backtest/StrategyWalkForward.tsx | 9 ++- 4 files changed, 90 insertions(+), 18 deletions(-) diff --git a/backend/app/api/backtest.py b/backend/app/api/backtest.py index abe2e5a..96dfb4a 100644 --- a/backend/app/api/backtest.py +++ b/backend/app/api/backtest.py @@ -738,8 +738,8 @@ async def optimize_cancel(request: Request): # Walk-forward 优化 — 每折训练区间优化 + 测试区间 OOS 验证 (复用优化器 + job_key 回吐) # ══════════════════════════════════════════════════════════════ -def _make_wf_job_key(strategy_id, symbols, start, end, param_grid, objective, direction, windows, bt_sig) -> str: - raw = f"WF|{strategy_id}|{symbols}|{start}|{end}|{param_grid}|{objective}|{direction}|{windows}|{bt_sig}" +def _make_wf_job_key(strategy_id, symbols, start, end, param_grid, objective, direction, windows, bt_sig, params=None, overrides=None) -> str: + raw = f"WF|{strategy_id}|{symbols}|{start}|{end}|{param_grid}|{objective}|{direction}|{windows}|{bt_sig}|{params}|{overrides}" return hashlib.md5(raw.encode()).hexdigest()[:12] @@ -754,6 +754,8 @@ async def walkforward_stream( test_days: int = 63, step_days: int = 63, max_workers: int = 4, + params: str | None = None, # JSON: 未扫描参数固定为用户当前值 (base_params) + overrides: str | None = None, # JSON: 策略当前的 basic_filter/signals/风控等覆盖 symbols: str | None = None, start: str | None = None, end: str | None = None, @@ -796,7 +798,7 @@ async def walkforward_stream( ) bt_sig = "|".join(f"{k}={bt_kwargs[k]}" for k in _OPT_BT_FIELDS) windows = f"{train_days}/{test_days}/{step_days}" - job_key = _make_wf_job_key(strategy_id, symbols, start, end, param_grid, objective, direction, windows, bt_sig) + job_key = _make_wf_job_key(strategy_id, symbols, start, end, param_grid, objective, direction, windows, bt_sig, params, overrides) # guard 作用于单折窗口 (每折训练/测试各是一次回测), 而非总区间 —— WF 总区间可长达数年, # 按总区间拦会误杀; 真正的 OOM 风险在单折窗口过大。 @@ -835,6 +837,14 @@ async def walkforward_stream( grid = None if grid is not None: + try: + base_params = json.loads(params) if params else {} + except (json.JSONDecodeError, TypeError): + base_params = {} + try: + ov = json.loads(overrides) if overrides else None + except (json.JSONDecodeError, TypeError): + ov = None wf_cfg = WalkForwardConfig( strategy_id=strategy_id, symbols=[s.strip() for s in symbols.split(",") if s.strip()] if symbols else None, @@ -847,6 +857,8 @@ async def walkforward_stream( test_days=int(test_days), step_days=int(step_days), max_workers=int(max_workers), + base_params=base_params if isinstance(base_params, dict) else {}, + overrides=ov if isinstance(ov, dict) else None, backtest_kwargs=bt_kwargs, ) diff --git a/backend/tests/backtest/test_walkforward.py b/backend/tests/backtest/test_walkforward.py index 036c730..fb09d0c 100644 --- a/backend/tests/backtest/test_walkforward.py +++ b/backend/tests/backtest/test_walkforward.py @@ -240,6 +240,22 @@ def test_wf_job_key_distinguishes_windows(): assert base != _make_wf_job_key("s", None, None, None, '{"p":[1]}', "sortino", None, "120/30/30", "sig") +def test_wf_job_key_distinguishes_params_and_overrides(): + """params/overrides 不同必须产出不同 job_key —— 否则 stream 与 cancel 会错配到别的任务。""" + from app.api.backtest import _make_wf_job_key + base = _make_wf_job_key("s", None, None, None, '{"p":[1]}', "sortino", None, "252/63/63", "sig") + # params 不同 (未扫描参数固定值不同 -> 优化的策略不同) + assert base != _make_wf_job_key( + "s", None, None, None, '{"p":[1]}', "sortino", None, "252/63/63", "sig", params='{"x":1}') + # overrides 不同 (basic_filter/信号/风控 不同) + assert base != _make_wf_job_key( + "s", None, None, None, '{"p":[1]}', "sortino", None, "252/63/63", "sig", overrides='{"score_min":5}') + # 相同 params/overrides 必须稳定一致 (stream 端与 cancel 端对齐前提) + k = _make_wf_job_key("s", None, None, None, '{"p":[1]}', "sortino", None, "252/63/63", "sig", params='{"x":1}') + assert k == _make_wf_job_key( + "s", None, None, None, '{"p":[1]}', "sortino", None, "252/63/63", "sig", params='{"x":1}') + + def test_wf_cancel_by_echoed_key(): import asyncio diff --git a/frontend/src/lib/walkforwardTask.ts b/frontend/src/lib/walkforwardTask.ts index 50d48d9..4e26413 100644 --- a/frontend/src/lib/walkforwardTask.ts +++ b/frontend/src/lib/walkforwardTask.ts @@ -59,6 +59,8 @@ export interface StartWalkForwardParams { train_days: number test_days: number step_days: number + params?: Record | null // 未扫描参数固定为用户当前值 + overrides?: Record | null // 策略当前的 basic_filter/信号/风控覆盖 symbols?: string[] | null start?: string | null end?: string | null @@ -70,6 +72,9 @@ const listeners = new Set<() => void>() let taskSeq = 0 let eventSource: EventSource | null = null let currentJobKey: string | null = null +let cancelRequested = false +let reconnectAttempts = 0 +const MAX_RECONNECT = 5 const RECONNECT_KEY = 'walkforward_reconnect' const JOB_KEY_KEY = 'walkforward_job_key' @@ -103,17 +108,28 @@ function connectSSE(url: string): void { eventSource = es es.addEventListener('job', (e: MessageEvent) => { + reconnectAttempts = 0 try { const key = JSON.parse(e.data)?.key if (key) { currentJobKey = key localStorage.setItem(JOB_KEY_KEY, key) + // 竞态: stop 在拿到 key 前被点过 -> 补发 cancel 真正停后端任务, 再收尾关闭。 + if (cancelRequested) { + postCancel(key) + es.close() + eventSource = null + currentJobKey = null + localStorage.removeItem(RECONNECT_KEY) + localStorage.removeItem(JOB_KEY_KEY) + } } } catch { /* ignore */ } }) es.addEventListener('progress', (e: MessageEvent) => { if (current?.id !== id) return + reconnectAttempts = 0 try { const prog = JSON.parse(e.data) as WFProgress current = { ...current, progress: prog } @@ -154,16 +170,39 @@ function connectSSE(url: string): void { currentJobKey = null localStorage.removeItem(RECONNECT_KEY) localStorage.removeItem(JOB_KEY_KEY) + return + } + // 无 data: 连接异常断开。EventSource 自动重连, 设上限避免网络长断时无限 pending。 + if (current?.id === id) { + reconnectAttempts += 1 + if (reconnectAttempts > MAX_RECONNECT) { + es.close() + eventSource = null + current = { ...current, isPending: false, error: '连接中断, 重连多次失败' } + emit() + } } }) } +/** 调后端 cancel (按回吐的 job_key)。 */ +function postCancel(jobKey: string): void { + fetch('/api/backtest/walkforward/cancel', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ job_key: jobKey }), + }).catch(() => {}) +} + export function startWalkForward(params: StartWalkForwardParams): void { if (eventSource) { eventSource.close() eventSource = null } + cancelRequested = false + currentJobKey = null + reconnectAttempts = 0 const id = ++taskSeq current = { id, isPending: true, result: null, progress: null, error: null } emit() @@ -175,6 +214,8 @@ export function startWalkForward(params: StartWalkForwardParams): void { train_days: params.train_days, test_days: params.test_days, step_days: params.step_days, + params: params.params ? JSON.stringify(params.params) : undefined, + overrides: params.overrides ? JSON.stringify(params.overrides) : undefined, symbols: params.symbols?.join(','), start: params.start ?? undefined, end: params.end ?? undefined, @@ -185,26 +226,24 @@ export function startWalkForward(params: StartWalkForwardParams): void { connectSSE(`/api/backtest/walkforward/stream?${qs}`) } -export async function stopWalkForward(): Promise { +export function stopWalkForward(): void { + // 竞态: job_key 未到手时保持 SSE 打开, 等 job 事件补发 cancel (关 SSE 不停后端 daemon 线程)。 + cancelRequested = true const jobKey = currentJobKey ?? localStorage.getItem(JOB_KEY_KEY) if (jobKey) { - await fetch('/api/backtest/walkforward/cancel', { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ job_key: jobKey }), - }).catch(() => {}) - } - if (eventSource) { - eventSource.close() - eventSource = null + postCancel(jobKey) + if (eventSource) { eventSource.close(); eventSource = null } + currentJobKey = null + localStorage.removeItem(RECONNECT_KEY) + localStorage.removeItem(JOB_KEY_KEY) + } else if (eventSource) { + const es = eventSource + setTimeout(() => { if (es === eventSource) { es.close(); eventSource = null } }, 5000) } if (current?.isPending) { current = { ...current, isPending: false, error: '已取消' } emit() } - currentJobKey = null - localStorage.removeItem(RECONNECT_KEY) - localStorage.removeItem(JOB_KEY_KEY) } export function clearWalkForward(): void { diff --git a/frontend/src/pages/backtest/StrategyWalkForward.tsx b/frontend/src/pages/backtest/StrategyWalkForward.tsx index ee5787a..d61fda2 100644 --- a/frontend/src/pages/backtest/StrategyWalkForward.tsx +++ b/frontend/src/pages/backtest/StrategyWalkForward.tsx @@ -12,6 +12,7 @@ import { tryReconnectWalkForward, useWalkForwardTask, } from '@/lib/walkforwardTask' +import { buildDefaultOverrides } from '@/lib/strategyOverrides' import { INPUT_CLS, OBJECTIVES, @@ -93,6 +94,10 @@ export function StrategyWalkForward() { train_days: Number(trainDays), test_days: Number(testDays), step_days: Number(stepDays), + // 未扫描参数固定为策略当前默认值; overrides 让 basic_filter/信号/风控按当前策略参与, + // 否则 walk-forward 优化的策略与用户实际回测的不一致 (同 PR #82 优化器修复)。 + params: sweep.selected?.params_defaults, + overrides: sweep.selected ? buildDefaultOverrides(sweep.selected) : undefined, start, end, mode, @@ -189,13 +194,13 @@ export function StrategyWalkForward() { {!result && !task?.isPending && ( )} {result && result.n_folds === 0 && ( + hint={`计划 ${result.n_planned_folds} 折, 但 ${result.n_skipped} 折因训练区间未优化出参数或 OOS 回测失败被跳过。请检查数据范围或放宽参数网格。`} /> )} {result && summary && result.n_folds > 0 && ( From db535c8004edd860f2fd17d75e173a3b24887e33 Mon Sep 17 00:00:00 2001 From: im47cn <67424112+im47cn@users.noreply.github.com> Date: Sun, 12 Jul 2026 00:52:32 +0800 Subject: [PATCH 5/9] =?UTF-8?q?feat(panelcache):=20IO=20=E9=81=A5=E6=B5=8B?= =?UTF-8?q?=E9=87=8F=E5=8C=96=E8=B7=A8=E6=8A=98=E9=87=8D=E5=A4=8D=E6=89=AB?= =?UTF-8?q?=E7=9B=98=E5=8D=A0=E6=AF=94?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit walk-forward 各折 train 窗口日历重叠 ~75%, 但 PanelCache key 精确到区间, 跨折永不命中 -> 每折重复扫盘。为判断是否值得做"整区间加载一次+切片"优化, 先埋轻量遥测再决策 (避免盲优化)。 - PanelCache 累加真实扫盘耗时/次数 + 命中/single-flight 复用次数, 暴露 stats(); BacktestEngine.cache_stats() 透传 (不让上层伸手进私有 _cache)。 - WalkForwardService.run 首尾快照, 返回 cache_telemetry.load_panel_pct (扫盘耗时/WF总耗时) + scans/hits/reuses, 并 logger.info 一行。 - 测试: PanelCache 计数 (miss扫盘/命中/并发复用); WF 测试补最小 engine 桩 (生产路径不加 getattr 兜底, 避免静默吞掉真缺 engine 的 bug)。 --- backend/app/backtest/engine.py | 30 +++++++++++++++- backend/app/backtest/walkforward.py | 25 ++++++++++++- backend/tests/backtest/test_walkforward.py | 10 ++++++ backend/tests/test_backtest_etf.py | 41 ++++++++++++++++++++++ 4 files changed, 104 insertions(+), 2 deletions(-) diff --git a/backend/app/backtest/engine.py b/backend/app/backtest/engine.py index cd3b241..2394f34 100644 --- a/backend/app/backtest/engine.py +++ b/backend/app/backtest/engine.py @@ -144,8 +144,14 @@ class PanelCache: # 用实例锁守护所有 OrderedDict 变更; compute_fn (重扫盘) 放锁外避免串行化。 self._lock = threading.Lock() # single-flight: 同 key 只让一个线程 compute, 其余等其结果复用。 - # 否则优化器等场景下 max_workers 个线程冷启动同时 miss, 会并行加载 N 份同一面板。 + # 否则优化器/walk-forward 的 max_workers 个线程冷启动同时 miss, 会并行加载 N 份同一面板。 self._inflight: dict[str, _InFlight] = {} + # 轻量遥测: 累加真实扫盘耗时与命中/复用次数, 用于量化 IO 占比 (是否值得进一步优化)。 + # compute_seconds 只计 leader 的实际 compute_fn 耗时, follower 复用不计 —— 反映真实 IO。 + self._compute_seconds = 0.0 + self._compute_count = 0 # 实际扫盘次数 + self._hit_count = 0 # 缓存命中 (未扫盘) 次数 + self._reuse_count = 0 # single-flight 跟随者复用次数 def get_or_compute( self, @@ -164,6 +170,7 @@ class PanelCache: if entry is not None: if now - entry.ts < self._ttl: self._cache.move_to_end(key) + self._hit_count += 1 return entry.df del self._cache[key] # 过期, 丢弃后重算 # single-flight: 同 key 若已有线程在算, 登记为跟随者; 否则本线程当 leader。 @@ -176,21 +183,28 @@ class PanelCache: if not leader: # 跟随者: 等 leader 算完直接复用, 不重复 compute (消除缓存踩踏)。 flight.done.wait() + with self._lock: + self._reuse_count += 1 if flight.error is not None: raise flight.error return flight.df # leader: compute 放锁外 (不同 key 仍可并发, 保留原设计优点)。 + t_compute = time.perf_counter() try: df = compute_fn(symbols, start, end, columns, asset_type) except BaseException as e: # 失败不缓存: 摘除 inflight 让后续线程重试, 并把异常透传给已在等的跟随者。 with self._lock: + self._compute_seconds += time.perf_counter() - t_compute # 失败也花了 IO, 计入 + self._compute_count += 1 self._inflight.pop(key, None) flight.error = e flight.done.set() raise with self._lock: + self._compute_seconds += time.perf_counter() - t_compute + self._compute_count += 1 self._cache[key] = _CacheEntry(df=df, ts=now) if len(self._cache) > self._max_size: self._cache.popitem(last=False) @@ -199,6 +213,16 @@ class PanelCache: flight.done.set() return df + def stats(self) -> dict: + """遥测快照: 累计扫盘耗时/次数与命中/复用次数。首尾快照取差即区间内 IO 开销。""" + with self._lock: + return { + "compute_seconds": round(self._compute_seconds, 4), + "compute_count": self._compute_count, + "hit_count": self._hit_count, + "reuse_count": self._reuse_count, + } + def invalidate(self) -> None: with self._lock: self._cache.clear() @@ -237,6 +261,10 @@ class BacktestEngine: """加载 enriched 数据面板，带缓存。asset_type='etf' 时读 ETF enriched。""" return self._cache.get_or_compute(symbols, start, end, columns, self._load_panel_inner, asset_type=asset_type) + def cache_stats(self) -> dict: + """暴露 PanelCache 遥测快照 (扫盘耗时/次数/命中/复用), 供上层量化 IO 占比。""" + return self._cache.stats() + def _load_panel_inner( self, symbols: list[str] | None, diff --git a/backend/app/backtest/walkforward.py b/backend/app/backtest/walkforward.py index 5c27f31..bd083d1 100644 --- a/backend/app/backtest/walkforward.py +++ b/backend/app/backtest/walkforward.py @@ -160,6 +160,9 @@ class WalkForwardService: folds = generate_folds(cfg.start, cfg.end, cfg.train_days, cfg.test_days, cfg.step_days) n_total = len(folds) + # 遥测: 首尾快照 PanelCache, 量化跨折重叠区间重复扫盘的 IO 占比 (是否值得进一步优化)。 + cache_before = self.service.engine.cache_stats() + valid_records: list[dict] = [] # IS 与 OOS 都成功, 计入聚合 skipped: list[dict] = [] # 无优化结果 或 OOS 失败, 不计入聚合 (避免伪装成有效折) done = 0 @@ -240,6 +243,25 @@ class WalkForwardService: progress_cb({"type": "walkforward_progress", "done": done, "total": n_total, "fold": f.index}) summary = aggregate_oos(valid_records, cfg.objective, direction) + + # 遥测收尾: 本次 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, @@ -249,5 +271,6 @@ class WalkForwardService: "folds": valid_records, "skipped": skipped, "summary": summary, - "elapsed_ms": round((time.perf_counter() - t0) * 1000, 1), + "cache_telemetry": cache_telemetry, + "elapsed_ms": elapsed_ms, } diff --git a/backend/tests/backtest/test_walkforward.py b/backend/tests/backtest/test_walkforward.py index fb09d0c..8fbc95f 100644 --- a/backend/tests/backtest/test_walkforward.py +++ b/backend/tests/backtest/test_walkforward.py @@ -136,10 +136,20 @@ class _FakeOptimizer: return {"best_params": {"p": cfg.start.month}, "best_score": 2.0, "results": [], "n_completed": 1} +_ZERO_CACHE_STATS = {"compute_seconds": 0.0, "compute_count": 0, "hit_count": 0, "reuse_count": 0} + + +class _FakeEngine: + """最小引擎桩: 仅提供 WF 遥测所需的 cache_stats。""" + def cache_stats(self): + return dict(_ZERO_CACHE_STATS) + + class _FakeService: """run 返回受控 OOS stats, 记录测试区间 + 收到的 params。""" def __init__(self): self.calls = [] + self.engine = _FakeEngine() def run(self, config, progress_cb=None, cancel_event=None): self.calls.append({"start": config.start, "end": config.end, "params": dict(config.params or {})}) diff --git a/backend/tests/test_backtest_etf.py b/backend/tests/test_backtest_etf.py index ac8e866..f95e25e 100644 --- a/backend/tests/test_backtest_etf.py +++ b/backend/tests/test_backtest_etf.py @@ -150,6 +150,47 @@ def test_panel_cache_single_flight_error_propagates_and_retries(): assert got is df +def test_panel_cache_stats_counts_scans_hits_reuses(): + """遥测计数: 首次 miss 计扫盘, 二次同 key 计命中, 并发同 key 跟随者计复用。""" + import threading + + cache = PanelCache() + df = pl.DataFrame({"symbol": ["510300"]}) + args = (["510300"], date(2026, 1, 1), date(2026, 1, 2), None) + + # 1) 首次: 冷 miss -> 扫盘 1 次 + cache.get_or_compute(*args, lambda *a: df, "stock") + s = cache.stats() + assert s["compute_count"] == 1 and s["hit_count"] == 0 + + # 2) 二次同 key: 命中缓存, 不扫盘 + cache.get_or_compute(*args, lambda *a: df, "stock") + s = cache.stats() + assert s["compute_count"] == 1 and s["hit_count"] == 1 + + # 3) 新 key 并发踩踏: 1 次扫盘 + N-1 次 single-flight 复用 + barrier = threading.Barrier(5) + args2 = (["600000"], date(2026, 2, 1), date(2026, 2, 2), None) + + def slow(*a): + time.sleep(0.04) + return df + + def worker(): + barrier.wait() + cache.get_or_compute(*args2, slow, "stock") + + threads = [threading.Thread(target=worker) for _ in range(5)] + for t in threads: + t.start() + for t in threads: + t.join() + + s = cache.stats() + assert s["compute_count"] == 2, "并发同 key 应只扫盘 1 次" + assert s["reuse_count"] == 4, "其余 4 个跟随者应计为复用" + + def test_job_key_includes_asset_type_and_is_consistent(): """stream 与 cancel 必须用同一 job_key: asset_type 进 key 且相同入参产出相同 key。""" from app.api.backtest import _make_job_key From 482904b591da6a43003b95c4404e1a123c2fddb2 Mon Sep 17 00:00:00 2001 From: im47cn <67424112+im47cn@users.noreply.github.com> Date: Sun, 12 Jul 2026 13:52:49 +0800 Subject: [PATCH 6/9] =?UTF-8?q?test(panelcache):=20=E5=AD=90=E4=BB=A3?= =?UTF-8?q?=E7=90=86=E5=AE=A1=E6=9F=A5=E5=8A=A0=E5=9B=BA=20=E2=80=94=20?= =?UTF-8?q?=E5=8E=BB=20flaky=20+=20=E8=A1=A5=E9=81=A5=E6=B5=8B=E5=B7=AE?= =?UTF-8?q?=E5=80=BC=E8=A6=86=E7=9B=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 三处审查驱动的测试改进: - 并发计数测试放宽 reuse_count==4 为 reuse+hit==4 (保留"只扫盘1次"核心不变量), 消除慢调度下 leader 已写缓存致某线程走 hit 分支的 flaky。 - _FakeEngine 零值 dict 改从 PanelCache().stats() 取键 —— 字段重命名时桩自动跟随, 杜绝 test 绿而生产 cache_stats KeyError 的契约漂移。 - 新增 test_walkforward_cache_telemetry_computes_deltas: 递增桩验证 scans/hits/ reuses/秒数 = after-before, 覆盖此前 _FakeEngine 恒返 0 掩盖的差值/顺序逻辑。 --- backend/tests/backtest/test_walkforward.py | 34 ++++++++++++++++++++-- backend/tests/test_backtest_etf.py | 5 +++- 2 files changed, 36 insertions(+), 3 deletions(-) diff --git a/backend/tests/backtest/test_walkforward.py b/backend/tests/backtest/test_walkforward.py index 8fbc95f..df49071 100644 --- a/backend/tests/backtest/test_walkforward.py +++ b/backend/tests/backtest/test_walkforward.py @@ -136,11 +136,14 @@ class _FakeOptimizer: return {"best_params": {"p": cfg.start.month}, "best_score": 2.0, "results": [], "n_completed": 1} -_ZERO_CACHE_STATS = {"compute_seconds": 0.0, "compute_count": 0, "hit_count": 0, "reuse_count": 0} +# 从真实 PanelCache 取字段模板 —— 字段被重命名时本桩自动跟随, 避免 test 绿而生产 KeyError。 +from app.backtest.engine import PanelCache + +_ZERO_CACHE_STATS = {k: type(v)() for k, v in PanelCache().stats().items()} class _FakeEngine: - """最小引擎桩: 仅提供 WF 遥测所需的 cache_stats。""" + """最小引擎桩: 仅提供 WF 遥测所需的 cache_stats (字段同源自 PanelCache.stats)。""" def cache_stats(self): return dict(_ZERO_CACHE_STATS) @@ -182,6 +185,33 @@ def test_walkforward_optimizes_train_applies_oos(): assert svc.calls[0]["start"] >= opt.train_ranges[0][1] +class _CountingEngine: + """首尾两次 cache_stats 返回不同值, 用于验证 WF 遥测差值/顺序计算 (非全零掩盖)。""" + def __init__(self): + self._n = 0 + + def cache_stats(self): + self._n += 1 + if self._n == 1: # run 开头快照 (before) + return {"compute_seconds": 1.0, "compute_count": 2, "hit_count": 0, "reuse_count": 0} + return {"compute_seconds": 3.5, "compute_count": 7, "hit_count": 4, "reuse_count": 3} # after + + +def test_walkforward_cache_telemetry_computes_deltas(): + """cache_telemetry 用首尾快照差值: scans/hits/reuses/秒数 = after - before, 且方向正确。""" + opt, svc = _FakeOptimizer(), _FakeService() + svc.engine = _CountingEngine() + wf = WalkForwardService(opt, svc, strategy_engine=None) + out = wf.run(_wf_cfg()) + + tel = out["cache_telemetry"] + assert tel["scans"] == 5 # 7 - 2, 顺序写反会得 -5 + assert tel["hits"] == 4 # 4 - 0 + assert tel["single_flight_reuses"] == 3 # 3 - 0 + assert abs(tel["load_panel_seconds"] - 2.5) < 1e-9 # 3.5 - 1.0 + assert tel["load_panel_pct"] >= 0.0 # 扫盘耗时 / WF总耗时, 非负 + + def test_walkforward_reports_degradation(): opt, svc = _FakeOptimizer(), _FakeService() wf = WalkForwardService(opt, svc, strategy_engine=None) diff --git a/backend/tests/test_backtest_etf.py b/backend/tests/test_backtest_etf.py index f95e25e..3253e1a 100644 --- a/backend/tests/test_backtest_etf.py +++ b/backend/tests/test_backtest_etf.py @@ -187,8 +187,11 @@ def test_panel_cache_stats_counts_scans_hits_reuses(): t.join() s = cache.stats() + # 核心不变量: 5 线程并发同 key 只扫盘 1 次 (args1 首次 + args2 一次 = 2)。 assert s["compute_count"] == 2, "并发同 key 应只扫盘 1 次" - assert s["reuse_count"] == 4, "其余 4 个跟随者应计为复用" + # 其余 4 线程要么 single-flight 复用, 要么(慢调度下 leader 已写缓存)命中 —— + # 二者之和恒为 4。不锁定 reuse/hit 具体分配, 避免时序 flaky。 + assert s["reuse_count"] + (s["hit_count"] - 1) == 4, "4 个非 leader 线程应复用或命中" def test_job_key_includes_asset_type_and_is_consistent(): From 8d93b8fe98e57bb2d0d1fd7752d286d746d3784c Mon Sep 17 00:00:00 2001 From: im47cn <67424112+im47cn@users.noreply.github.com> Date: Sun, 12 Jul 2026 14:10:15 +0800 Subject: [PATCH 7/9] =?UTF-8?q?fix(walkforward):=20=E5=89=8D=E7=AB=AF?= =?UTF-8?q?=E5=AF=B9=E9=BD=90=E4=BC=98=E5=8C=96=E5=99=A8=E5=B7=B2=E6=9C=89?= =?UTF-8?q?=E9=98=B2=E6=8A=A4=20(=E5=88=87=E7=AD=96=E7=95=A5=E5=81=9C?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=20+=20=E5=9D=8F=E7=BD=91=E6=A0=BC=E6=8B=A6?= =?UTF-8?q?=E6=88=AA)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 对抗式审查发现 WF 页面偏离了 StrategyOptimizer 已修好的两处防护: - 切策略时直接传 clearWalkForward (只清前端状态), 不停后端任务 —— 对齐优化器 改为 if(pending) stopWalkForward() else clearWalkForward(), 真正 cancel+关SSE+清storage。 - canRun 缺 !sweep.gridError, 且 CombosHint 未传 gridError prop —— 坏网格(步长越界等) 既不显示错误也能提交到后端。补齐, 与优化器一致。 --- frontend/src/pages/backtest/StrategyWalkForward.tsx | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/frontend/src/pages/backtest/StrategyWalkForward.tsx b/frontend/src/pages/backtest/StrategyWalkForward.tsx index d61fda2..cb9a205 100644 --- a/frontend/src/pages/backtest/StrategyWalkForward.tsx +++ b/frontend/src/pages/backtest/StrategyWalkForward.tsx @@ -67,7 +67,11 @@ export function StrategyWalkForward() { const { data: stratData } = useQuery({ queryKey: ['strategies'], queryFn: api.strategyList }) const strategies: StrategyDetail[] = stratData?.strategies ?? [] - const sweep = useParamSweep(strategies, clearWalkForward) + // 切策略: 有任务在跑时先真正取消 (关 SSE + 后端 cancel + 清 localStorage), 不能静默丢 + const sweep = useParamSweep(strategies, () => { + if (task?.isPending) stopWalkForward() + else clearWalkForward() + }) const [objective, setObjective] = useState('sortino') const [start, setStart] = useState(THREE_YEARS_AGO) const [end, setEnd] = useState(TODAY) @@ -82,6 +86,7 @@ export function StrategyWalkForward() { }, []) const canRun = sweep.strategyId && sweep.combos > 0 && sweep.combos <= GRID_MAX_COMBINATIONS + && !sweep.gridError && Number(trainDays) > 0 && Number(testDays) > 0 && Number(stepDays) > 0 && !task?.isPending const onRun = () => { @@ -160,7 +165,7 @@ export function StrategyWalkForward() { - +
每折跑 {sweep.combos || 0} 组优化 × N 折，耗时较长
{task?.isPending ? ( From e4262d9951cf0161c62e75e02717fbddd1b9f370 Mon Sep 17 00:00:00 2001 From: im47cn <67424112+im47cn@users.noreply.github.com> Date: Sun, 12 Jul 2026 21:48:13 +0800 Subject: [PATCH 8/9] =?UTF-8?q?fix(walkforward):=20IS=20=E8=AE=AD=E7=BB=83?= =?UTF-8?q?=E6=8A=98=E5=BC=BA=E5=88=B6=20position=20=E6=A8=A1=E5=BC=8F,=20?= =?UTF-8?q?=E5=A0=B5=E5=89=8D=E8=A7=86=E6=B3=84=E6=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 对抗式审查发现: WF UI 暴露 full 模式, 选中时训练折(end=train_end)未平仓的持仓 会用 train_end 之后(即 OOS 区间)的真实 K 线平仓 -> IS 分数被未来数据污染, 优化选出 的最优参数乐观偏移, 使过拟合被掩盖 (对专门检测过拟合的 WF 工具危害尤重)。 修复: IS 训练区间优化强制 mode="position" (只看正式区间表现), OOS 回测保留用户 所选 mode。新增测试: 用户选 full 时断言每折 IS 得到 position、OOS 保留 full。 --- backend/app/backtest/walkforward.py | 8 +++++++- backend/tests/backtest/test_walkforward.py | 19 ++++++++++++++++++- 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/backend/app/backtest/walkforward.py b/backend/app/backtest/walkforward.py index bd083d1..cd7361d 100644 --- a/backend/app/backtest/walkforward.py +++ b/backend/app/backtest/walkforward.py @@ -166,6 +166,12 @@ class WalkForwardService: 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 @@ -182,7 +188,7 @@ class WalkForwardService: max_workers=cfg.max_workers, base_params=cfg.base_params, overrides=cfg.overrides, - backtest_kwargs=cfg.backtest_kwargs, + backtest_kwargs=is_backtest_kwargs, # IS 强制 position, 堵前视泄漏 ) opt_res = self.optimizer.optimize(opt_cfg, cancel_event=cancel_event) best_params = opt_res.get("best_params") diff --git a/backend/tests/backtest/test_walkforward.py b/backend/tests/backtest/test_walkforward.py index df49071..4c0d02c 100644 --- a/backend/tests/backtest/test_walkforward.py +++ b/backend/tests/backtest/test_walkforward.py @@ -129,9 +129,11 @@ class _FakeOptimizer: """optimize 返回受控 best_params/best_score, 记录被优化的训练区间。""" def __init__(self): self.train_ranges = [] + self.opt_kwargs = [] # 记录每折 IS 优化收到的 backtest_kwargs (验证 mode 强制) def optimize(self, cfg, progress_cb=None, cancel_event=None): self.train_ranges.append((cfg.start, cfg.end)) + self.opt_kwargs.append(dict(cfg.backtest_kwargs)) # best_params 随训练起点变化, best_score 固定 return {"best_params": {"p": cfg.start.month}, "best_score": 2.0, "results": [], "n_completed": 1} @@ -155,7 +157,8 @@ class _FakeService: self.engine = _FakeEngine() def run(self, config, progress_cb=None, cancel_event=None): - self.calls.append({"start": config.start, "end": config.end, "params": dict(config.params or {})}) + self.calls.append({"start": config.start, "end": config.end, + "params": dict(config.params or {}), "mode": config.mode}) return _FakeResult(stats={"total_return": 0.05, "sortino": 1.0}) @@ -212,6 +215,20 @@ def test_walkforward_cache_telemetry_computes_deltas(): assert tel["load_panel_pct"] >= 0.0 # 扫盘耗时 / WF总耗时, 非负 +def test_walkforward_forces_position_mode_for_is_optimization(): + """训练折(IS)强制 position 防前视泄漏(full 会用 OOS 区间 K 线平仓污染 IS); + OOS 回测保留用户所选 mode。""" + opt, svc = _FakeOptimizer(), _FakeService() + wf = WalkForwardService(opt, svc, strategy_engine=None) + wf.run(_wf_cfg(backtest_kwargs={"mode": "full"})) + + assert len(opt.opt_kwargs) > 0 and len(svc.calls) > 0 + # 用户选了 full, 但每折 IS 优化都被强制 position + assert all(kw["mode"] == "position" for kw in opt.opt_kwargs), "IS 优化未强制 position" + # OOS 回测保留用户的 full + assert all(c["mode"] == "full" for c in svc.calls), "OOS 未保留用户 mode" + + def test_walkforward_reports_degradation(): opt, svc = _FakeOptimizer(), _FakeService() wf = WalkForwardService(opt, svc, strategy_engine=None) From 5ea58f6a8d7711be15bb507f2d418ffe8a271006 Mon Sep 17 00:00:00 2001 From: im47cn <67424112+im47cn@users.noreply.github.com> Date: Sun, 12 Jul 2026 22:18:11 +0800 Subject: [PATCH 9/9] =?UTF-8?q?fix(walkforward):=20SSE=20=E6=94=BE?= =?UTF-8?q?=E5=BC=83=E6=97=B6=E6=B8=85=20localStorage=20+=20JSON=20?= =?UTF-8?q?=E8=A7=A3=E6=9E=90=E5=A4=B1=E8=B4=A5=E5=91=8A=E8=AD=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 跨组件后续修复的 WF 侧 (优化器侧同款问题另起 PR): - stopWalkForward 延迟分支 + MAX_RECONNECT 超限退出漏清 localStorage, 导致刷新页面 tryReconnect 会重连到已放弃/未取消的任务。两处补 removeItem(RECONNECT_KEY/JOB_KEY_KEY)。 - walkforward_stream 的 params/overrides JSON 解析失败静默降级为 {}/None, 会让用户配置 丢失变成无声 bug。升级为 logger.warning (仍降级不崩流, 但可诊断)。 --- backend/app/api/backtest.py | 6 ++++++ frontend/src/lib/walkforwardTask.ts | 13 ++++++++++++- 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/backend/app/api/backtest.py b/backend/app/api/backtest.py index 96dfb4a..ad8364d 100644 --- a/backend/app/api/backtest.py +++ b/backend/app/api/backtest.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio import json +import logging import queue import threading from dataclasses import asdict @@ -21,6 +22,8 @@ from app.services.backtest import ( is_available, ) +logger = logging.getLogger(__name__) + router = APIRouter(prefix="/api/backtest", tags=["backtest"]) FACTOR_DEFAULT_DAYS = 180 @@ -840,10 +843,13 @@ async def walkforward_stream( try: base_params = json.loads(params) if params else {} except (json.JSONDecodeError, TypeError): + # 静默降级会让"用户配置丢失"变成无声 bug: 至少 warn 供诊断 (前端应传合法 JSON)。 + logger.warning("walkforward: params JSON 解析失败, 降级为空 params: %r", params) base_params = {} try: ov = json.loads(overrides) if overrides else None except (json.JSONDecodeError, TypeError): + logger.warning("walkforward: overrides JSON 解析失败, 降级为 None: %r", overrides) ov = None wf_cfg = WalkForwardConfig( strategy_id=strategy_id, diff --git a/frontend/src/lib/walkforwardTask.ts b/frontend/src/lib/walkforwardTask.ts index 4e26413..3b62979 100644 --- a/frontend/src/lib/walkforwardTask.ts +++ b/frontend/src/lib/walkforwardTask.ts @@ -178,6 +178,9 @@ function connectSSE(url: string): void { if (reconnectAttempts > MAX_RECONNECT) { es.close() eventSource = null + // 清 localStorage: 否则刷新页面 tryReconnect 会重连到这个已放弃的任务。 + localStorage.removeItem(RECONNECT_KEY) + localStorage.removeItem(JOB_KEY_KEY) current = { ...current, isPending: false, error: '连接中断, 重连多次失败' } emit() } @@ -238,7 +241,15 @@ export function stopWalkForward(): void { localStorage.removeItem(JOB_KEY_KEY) } else if (eventSource) { const es = eventSource - setTimeout(() => { if (es === eventSource) { es.close(); eventSource = null } }, 5000) + // job_key 始终没到手(job 事件未达): 5 秒后放弃并清 localStorage, 避免刷新重连到未取消任务。 + // (若期间 job 到达, job handler 已 postCancel+清storage 并置 eventSource=null, 下面条件不成立跳过) + setTimeout(() => { + if (es === eventSource) { + es.close(); eventSource = null + localStorage.removeItem(RECONNECT_KEY) + localStorage.removeItem(JOB_KEY_KEY) + } + }, 5000) } if (current?.isPending) { current = { ...current, isPending: false, error: '已取消' }
{f.index + 1}