diff --git a/backend/app/api/backtest.py b/backend/app/api/backtest.py index 1350252..7dfcf49 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 @@ -536,6 +539,20 @@ async def strategy_cancel(request: Request): # 参数网格优化器 — 复用 _BacktestJob SSE 框架 (多组参数并行回测 + 排序) # ══════════════════════════════════════════════════════════════ +def _json_safe(obj): + """递归把 nan/inf 置 None —— json.dumps(default=str) 处理不了它们, 会输出非法 JSON + 字面量 NaN/Infinity 让前端 JSON.parse 崩。优化器/WF 结果嵌套深 (逐组/逐折的 + sortino 等零波动场景可能算出 nan), 序列化前统一清洗。""" + import math + if isinstance(obj, float): + return obj if math.isfinite(obj) else None + if isinstance(obj, dict): + return {k: _json_safe(v) for k, v in obj.items()} + if isinstance(obj, list): + return [_json_safe(v) for v in obj] + return obj + + # 透传给每组回测的 StrategyBacktestConfig 字段 (作为 backtest_kwargs)。 _OPT_BT_FIELDS = [ "matching", "fees_pct", "commission_pct", "stamp_tax_pct", "slippage_bps", @@ -707,7 +724,7 @@ async def optimize_stream( # 取消时优化器把每组记为 cancelled 并正常返回, 需在此分流为取消提示而非"完成"。 yield f"event: error\ndata: {json.dumps({'message': '优化已取消'}, 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(): @@ -739,3 +756,185 @@ 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, 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] + + +@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, + 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, + 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, params, overrides) + + # guard 作用于单折窗口 (每折训练/测试各是一次回测), 而非总区间 —— WF 总区间可长达数年, + # 按总区间拦会误杀; 真正的 OOM 风险在单折窗口过大。 + wf_guard_violated = ( + settings.backtest_range_guard + and max(int(train_days), int(test_days)) > BACKTEST_MAX_SERVER_DAYS + ) + + _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 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) + 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: + 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, + 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), + base_params=base_params if isinstance(base_params, dict) else {}, + overrides=ov if isinstance(ov, dict) else None, + 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(_json_safe(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/engine.py b/backend/app/backtest/engine.py index a9ad18a..b3b43e8 100644 --- a/backend/app/backtest/engine.py +++ b/backend/app/backtest/engine.py @@ -171,8 +171,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, @@ -191,6 +197,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。 @@ -203,21 +210,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) @@ -226,6 +240,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() @@ -264,6 +288,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 new file mode 100644 index 0000000..cd7361d --- /dev/null +++ b/backend/app/backtest/walkforward.py @@ -0,0 +1,282 @@ +"""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 则 + # 该日 K 线同时进训练优化与 OOS 首日, 构成前视泄漏。后移一天隔断。 + test_start = train_end + timedelta(days=1) + test_end = test_start + timedelta(days=test_days) + if test_end > end: + break + folds.append(Fold(i, train_start, train_end, test_start, test_end)) + i += 1 + train_start = train_start + timedelta(days=step_days) + + if not folds: + raise ValueError( + f"数据区间不足以切出至少一折 (需 train+test={train_days + test_days}天, " + f"实有 {(end - start).days}天)" + ) + return folds + + +def _norm(v: float, direction: str) -> float: + """把目标值归一到"越大越好"空间, 以便跨目标一致地算退化 (min 类目标取负)。""" + return -v if direction == "min" else v + + +def aggregate_oos(fold_records: list[dict], objective: str, direction: str = "max") -> dict: + """从**有效折** (IS 与 OOS 都成功) 聚合: 复利净值 / IS-OOS 退化 / 一致性。 + + 调用方只传有效折 (best_params 非空且 OOS 未 error), 故此处每折 is_score/oos_objective + 均有值, 无需 .get 默认兜底 —— 无效折被伪装成 0 收益混入曾是 H1/H2 的根因。 + + - compounded_oos_return: 各折 OOS 总收益复利 + - degradation: 归一空间下 IS 目标均值 - OOS 目标均值, 正值 = 样本外退化 (过拟合信号), + 对"越小越好"目标 (max_drawdown 等) 方向也正确 + - consistency: OOS 总收益 > 0 的折占比 (与目标方向无关, 直观) + """ + n = len(fold_records) + if n == 0: + return { + "n_folds": 0, + "compounded_oos_return": 0.0, + "avg_is_objective": None, + "avg_oos_objective": None, + "degradation": None, + "consistency": 0.0, + "oos_equity_curve": [], + } + + equity = 1.0 + curve: list[dict] = [] + n_positive = 0 + for f in fold_records: + r = float(f["oos_stats"].get("total_return", 0.0) or 0.0) + equity *= (1 + r) + if r > 0: + n_positive += 1 + curve.append({"fold": f["index"], "date": str(f["test_end"]), "value": round(equity, 4)}) + + is_vals = [f["is_score"] for f in fold_records if f["is_score"] is not None] + oos_vals = [f["oos_objective"] for f in fold_records if f["oos_objective"] is not None] + avg_is = round(float(sum(is_vals) / len(is_vals)), 4) if is_vals else None + avg_oos = round(float(sum(oos_vals) / len(oos_vals)), 4) if oos_vals else None + degradation = ( + round(_norm(avg_is, direction) - _norm(avg_oos, direction), 4) + if (avg_is is not None and avg_oos is not None) else None + ) + + return { + "n_folds": n, + "compounded_oos_return": round(equity - 1.0, 4), + "avg_is_objective": avg_is, + "avg_oos_objective": avg_oos, + "degradation": degradation, + "consistency": round(n_positive / n, 4), + "oos_equity_curve": curve, + } + + +@dataclass +class WalkForwardConfig: + strategy_id: str + symbols: list[str] | None + start: date + end: date + param_grid: dict + objective: str = "sortino" + train_days: int = 252 + test_days: int = 63 + step_days: int = 63 + direction: str | None = None + max_workers: int = 4 + base_params: dict = field(default_factory=dict) + overrides: dict | None = None + backtest_kwargs: dict = field(default_factory=dict) + + +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, 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) + + # 遥测: 首尾快照 PanelCache, 量化跨折重叠区间重复扫盘的 IO 占比 (是否值得进一步优化)。 + cache_before = self.service.engine.cache_stats() + + valid_records: list[dict] = [] # IS 与 OOS 都成功, 计入聚合 + skipped: list[dict] = [] # 无优化结果 或 OOS 失败, 不计入聚合 (避免伪装成有效折) + done = 0 + + # IS 训练区间强制 position 模式: full 模式会让训练折未平仓持仓用 train_end 之后 + # (即 OOS 区间) 的真实 K 线平仓, IS 分数被未来数据污染 -> 优化选参乐观偏移, 使过拟合 + # 被掩盖。OOS 回测保留用户所选 mode。参数扫描优化只看正式区间内的表现即可。 + is_backtest_kwargs = {**cfg.backtest_kwargs, "mode": "position"} + + for f in folds: + if cancel_event is not None and cancel_event.is_set(): + break + + # 训练区间: 网格优化选最优参数 + opt_cfg = OptimizeConfig( + strategy_id=cfg.strategy_id, + symbols=cfg.symbols, + start=f.train_start, + end=f.train_end, + param_grid=cfg.param_grid, + objective=cfg.objective, + direction=cfg.direction, + max_workers=cfg.max_workers, + base_params=cfg.base_params, + overrides=cfg.overrides, + backtest_kwargs=is_backtest_kwargs, # IS 强制 position, 堵前视泄漏 + ) + 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} + 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 失败 (含 cancelled) -> 跳过, 不把空/0 收益混入复利曲线 + if oos_res.error: + skipped.append({**base, "best_params": best_params, "reason": f"OOS 回测失败: {oos_res.error}"}) + if progress_cb is not None: + progress_cb({"type": "walkforward_progress", "done": done, "total": n_total, "fold": f.index}) + continue + + oos_objective = oos_res.stats.get(cfg.objective) + # 该折 OOS 是否较 IS 退化 (方向感知: min 类目标数值更大才是退化) + oos_degraded = ( + _norm(oos_objective, direction) < _norm(is_score, direction) + if (oos_objective is not None and is_score is not None) else None + ) + valid_records.append({ + **base, + "best_params": best_params, + "is_score": is_score, + "oos_objective": oos_objective, + "oos_degraded": oos_degraded, + "oos_stats": oos_res.stats, + }) + + if progress_cb is not None: + progress_cb({"type": "walkforward_progress", "done": done, "total": n_total, "fold": f.index}) + + summary = aggregate_oos(valid_records, cfg.objective, direction) + + # 遥测收尾: 本次 WF 累计扫盘耗时 / 命中 / 复用, 与总耗时对比得出 load_panel 占比。 + cache_after = self.service.engine.cache_stats() + elapsed_ms = round((time.perf_counter() - t0) * 1000, 1) + io_seconds = round(cache_after["compute_seconds"] - cache_before["compute_seconds"], 4) + io_pct = round(io_seconds * 1000 / elapsed_ms * 100, 1) if elapsed_ms > 0 else 0.0 + cache_telemetry = { + "load_panel_seconds": io_seconds, + "load_panel_pct": io_pct, # 扫盘耗时 / WF 总耗时 + "scans": cache_after["compute_count"] - cache_before["compute_count"], + "hits": cache_after["hit_count"] - cache_before["hit_count"], + "single_flight_reuses": cache_after["reuse_count"] - cache_before["reuse_count"], + } + logger.info( + "walk-forward IO 占比: load_panel %.3fs (%.1f%% of %.1fms) | 扫盘 %d 次 命中 %d 复用 %d", + io_seconds, io_pct, elapsed_ms, + cache_telemetry["scans"], cache_telemetry["hits"], cache_telemetry["single_flight_reuses"], + ) + + return { + "objective": cfg.objective, + "direction": direction, + "n_folds": len(valid_records), + "n_skipped": len(skipped), + "n_planned_folds": n_total, + "folds": valid_records, + "skipped": skipped, + "summary": summary, + "cache_telemetry": cache_telemetry, + "elapsed_ms": elapsed_ms, + } diff --git a/backend/tests/backtest/test_walkforward.py b/backend/tests/backtest/test_walkforward.py new file mode 100644 index 0000000..4c0d02c --- /dev/null +++ b/backend/tests/backtest/test_walkforward.py @@ -0,0 +1,336 @@ +"""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, 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: + 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_objective": obj, + "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(): + # 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 + 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 = [] + 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} + + +# 从真实 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 (字段同源自 PanelCache.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 {}), "mode": config.mode}) + 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] + + +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_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) + 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 + + +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() + 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_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 + + 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) diff --git a/backend/tests/test_backtest_etf.py b/backend/tests/test_backtest_etf.py index ac8e866..3253e1a 100644 --- a/backend/tests/test_backtest_etf.py +++ b/backend/tests/test_backtest_etf.py @@ -150,6 +150,50 @@ 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() + # 核心不变量: 5 线程并发同 key 只扫盘 1 次 (args1 首次 + args2 一次 = 2)。 + assert s["compute_count"] == 2, "并发同 key 应只扫盘 1 次" + # 其余 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(): """stream 与 cancel 必须用同一 job_key: asset_type 进 key 且相同入参产出相同 key。""" from app.api.backtest import _make_job_key diff --git a/frontend/src/lib/walkforwardTask.ts b/frontend/src/lib/walkforwardTask.ts new file mode 100644 index 0000000..3b62979 --- /dev/null +++ b/frontend/src/lib/walkforwardTask.ts @@ -0,0 +1,277 @@ +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_degraded: boolean | 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 + 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 +} + +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 + params?: Record | null // 未扫描参数固定为用户当前值 + overrides?: Record | null // 策略当前的 basic_filter/信号/风控覆盖 + 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 +let cancelRequested = false +let reconnectAttempts = 0 +const MAX_RECONNECT = 5 + +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) => { + 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 } + 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) + return + } + // 无 data: 连接异常断开。EventSource 自动重连, 设上限避免网络长断时无限 pending。 + if (current?.id === id) { + reconnectAttempts += 1 + 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() + } + } + }) +} + +/** 调后端 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() + + 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, + 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, + mode: params.mode, + }) + + localStorage.setItem(RECONNECT_KEY, qs) + connectSSE(`/api/backtest/walkforward/stream?${qs}`) +} + +export function stopWalkForward(): void { + // 竞态: job_key 未到手时保持 SSE 打开, 等 job 事件补发 cancel (关 SSE 不停后端 daemon 线程)。 + cancelRequested = true + const jobKey = currentJobKey ?? localStorage.getItem(JOB_KEY_KEY) + if (jobKey) { + 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 + // 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: '已取消' } + emit() + } +} + +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 88328fa..562ef9f 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 或缩小范围` : ''}`} +
+ ) +}