fix: 串行化 Numba parallel 内核，避免策略页并发崩溃

并发 run_all 会触发 workqueue Concurrent access，导致后端进程中断。
为 matrix parallel 内核加进程锁，前端对 run_all 做 pending 去重。

backend/app/backtest/matrix.py

@@ -27,6 +27,7 @@ import pyarrow.compute as pc
 import pyarrow.dataset as pads
 
 from app.backtest.minute_trigger import build_minute_exit_reference
+from app.backtest.numba_runtime import run_numba_parallel
 from app.price_limits import (
     MAIN_BOARD_ST_LIMIT_CHANGE_DATE,
     numpy_limit_pct_vectors,
@@ -2925,12 +2926,14 @@ def valid_shift(
         "valid_shift",
         (source, valid, index.offsets, index.rows),
         {"periods": int(periods)},
-        lambda: _valid_shift_kernel(
-            source,
-            valid,
-            index.offsets,
-            index.rows,
-            int(periods),
+        lambda: run_numba_parallel(
+            lambda: _valid_shift_kernel(
+                source,
+                valid,
+                index.offsets,
+                index.rows,
+                int(periods),
+            )
         ),
     )
 
@@ -3072,14 +3075,16 @@ def _valid_rolling_reduce(
     if ddof < 0 or ddof >= window:
         raise ValueError("valid rolling ddof must be in [0, window)")
     index = _resolve_valid_bar_index(source, valid, bar_index)
-    return _valid_rolling_kernel(
-        source,
-        valid,
-        index.offsets,
-        index.rows,
-        int(window),
-        int(operation),
-        int(ddof),
+    return run_numba_parallel(
+        lambda: _valid_rolling_kernel(
+            source,
+            valid,
+            index.offsets,
+            index.rows,
+            int(window),
+            int(operation),
+            int(ddof),
+        )
     )
 
 
@@ -3307,12 +3312,14 @@ def valid_ewm_adjust_false(
         "valid_ewm_adjust_false",
         (source, valid, index.offsets, index.rows),
         {"alpha": alpha_value},
-        lambda: _valid_ewm_kernel(
-            source,
-            valid,
-            index.offsets,
-            index.rows,
-            alpha_value,
+        lambda: run_numba_parallel(
+            lambda: _valid_ewm_kernel(
+                source,
+                valid,
+                index.offsets,
+                index.rows,
+                alpha_value,
+            )
         ),
     )
 

backend/app/backtest/numba_runtime.py

@@ -0,0 +1,23 @@
+"""Serialize Numba ``@njit(parallel=True)`` kernels across request threads.
+
+Numba's default ``workqueue`` threading layer is not thread-safe. Concurrent
+calls from FastAPI worker threads terminate the process
+(``Concurrent access has been detected`` → socket hang up).
+
+A process-wide lock is enough: overlapping kernels queue instead of crashing.
+"""
+from __future__ import annotations
+
+import threading
+from collections.abc import Callable
+from typing import TypeVar
+
+_NUMBA_PARALLEL_LOCK = threading.RLock()
+
+T = TypeVar("T")
+
+
+def run_numba_parallel(fn: Callable[[], T]) -> T:
+    """Run a Numba parallel kernel under the process-wide lock."""
+    with _NUMBA_PARALLEL_LOCK:
+        return fn()

backend/tests/backtest/test_numba_runtime.py

@@ -0,0 +1,59 @@
+"""Numba parallel kernels must stay safe under concurrent callers."""
+from __future__ import annotations
+
+import threading
+from concurrent.futures import ThreadPoolExecutor
+
+import numpy as np
+import pytest
+
+from app.backtest.numba_runtime import run_numba_parallel
+
+
+def test_run_numba_parallel_serializes_concurrent_calls():
+    """Two threads must not overlap inside the parallel critical section."""
+    active = 0
+    max_active = 0
+    lock = threading.Lock()
+
+    def work() -> int:
+        nonlocal active, max_active
+        with lock:
+            active += 1
+            max_active = max(max_active, active)
+        try:
+            total = 0
+            for i in range(20_000):
+                total += i
+            return total
+        finally:
+            with lock:
+                active -= 1
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        list(pool.map(lambda _: run_numba_parallel(work), range(16)))
+
+    assert max_active == 1
+
+
+@pytest.mark.skipif(
+    __import__("importlib").util.find_spec("numba") is None,
+    reason="numba not installed on this platform",
+)
+def test_valid_shift_kernel_survives_concurrent_threads():
+    """Regression for workqueue 'Concurrent access has been detected' crashes."""
+    from app.backtest.matrix import valid_shift
+
+    rng = np.random.default_rng(0)
+    values = rng.normal(size=(64, 32)).astype(np.float32)
+    values[::7, ::3] = np.nan
+
+    def once() -> np.ndarray:
+        return valid_shift(values, 3)
+
+    with ThreadPoolExecutor(max_workers=4) as pool:
+        results = list(pool.map(lambda _: once(), range(8)))
+
+    baseline = results[0]
+    for other in results[1:]:
+        np.testing.assert_allclose(baseline, other, equal_nan=True)

frontend/src/pages/Screener.tsx

@@ -232,6 +232,24 @@ export function Screener() {
   )
   const cacheCoversPool = visiblePool.length > 0 && missingStrategyIds.length === 0
 
+  // 防止 reload / auto-run / StrictMode 叠出并发 run_all（后端 Numba 会崩溃）
+  // 用 ref 同步门闩，避免同一渲染周期内 isPending 尚未更新导致重复触发
+  const runAllPendingRef = useRef(false)
+  const requestRunAll = useCallback((
+    vars: { date?: string; strategyIds?: string[] } = {},
+    options?: Parameters<typeof runAll.mutate>[1],
+  ) => {
+    if (runAllPendingRef.current || runAll.isPending) return
+    runAllPendingRef.current = true
+    runAll.mutate(vars, {
+      ...options,
+      onSettled: (...args) => {
+        runAllPendingRef.current = false
+        options?.onSettled?.(...args)
+      },
+    })
+  }, [runAll])
+
   // 摘要只同步当前日期的卡片数量，避免旧日期缓存短暂显示成当前结果。
   useEffect(() => {
     if (!summaryQuery.data || !asOf) return
@@ -433,8 +451,8 @@ export function Screener() {
     // 未覆盖: 受系统开关控制
     if (!screenerAutoRun) return
     runAllDateRef.current = runKey
-    runAll.mutate({ date: asOf, strategyIds: missingStrategyIds })
-  }, [asOf, strategyPresets.length, summaryQuery.isSuccess, visiblePool, cacheCoversPool, missingStrategyIds, screenerAutoRun, assetType, runAll.isPending])
+    requestRunAll({ date: asOf, strategyIds: missingStrategyIds })
+  }, [asOf, strategyPresets.length, summaryQuery.isSuccess, visiblePool, cacheCoversPool, missingStrategyIds, screenerAutoRun, assetType, runAll.isPending, requestRunAll])
 
   const run = useMutation({
     mutationFn: ({ id, date }: { id: string; date: string }) =>
@@ -501,7 +519,7 @@ export function Screener() {
     mutationFn: api.strategyReload,
     onSuccess: () => {
       qc.invalidateQueries({ queryKey: ['screener-strategies'] })
-      if (asOf) runAll.mutate({ date: asOf })
+      if (asOf) requestRunAll({ date: asOf })
     },
   })