fix(walkforward): 优化器吃用户当前策略配置 + EmptyState 构建修复

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 对齐前提) 且相同入参稳定一致。
2026-09-12 15:34:16 +08:00 · 2026-07-12 00:52:11 +08:00
parent 32d27a0a6e
commit 72697707a4
4 changed files with 90 additions and 18 deletions
@@ -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,
                )

@@ -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

@@ -59,6 +59,8 @@ export interface StartWalkForwardParams {
  train_days: number
  test_days: number
  step_days: number
+  params?: Record<string, any> | null       // 未扫描参数固定为用户当前值
+  overrides?: Record<string, any> | 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<void> {
+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 {
@@ -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 && (
          <EmptyState
            title="Walk-forward 优化"
-            description="每折在训练区间网格优化选最优参数，再在紧邻的测试区间做样本外(OOS)验证。样本内漂亮、样本外崩溃即过拟合。"
+            hint="每折在训练区间网格优化选最优参数，再在紧邻的测试区间做样本外(OOS)验证。样本内漂亮、样本外崩溃即过拟合。"
          />
        )}

        {result && result.n_folds === 0 && (
          <EmptyState title="未产生有效折"
-            description={`计划 ${result.n_planned_folds} 折, 但 ${result.n_skipped} 折因训练区间未优化出参数或 OOS 回测失败被跳过。请检查数据范围或放宽参数网格。`} />
+            hint={`计划 ${result.n_planned_folds} 折, 但 ${result.n_skipped} 折因训练区间未优化出参数或 OOS 回测失败被跳过。请检查数据范围或放宽参数网格。`} />
        )}

        {result && summary && result.n_folds > 0 && (