From f03bc38a1612cfc79819ce42ce079faf730e73ca Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Mon, 7 Sep 2026 15:25:54 +0800 Subject: [PATCH] =?UTF-8?q?feat(screener):=20run=5Fall=20=E6=B8=90?= =?UTF-8?q?=E8=BF=9B=E5=BC=8F=E8=BF=94=E5=9B=9E,=20=E5=BF=AB=E7=AD=96?= =?UTF-8?q?=E7=95=A5=E5=85=88=E5=87=BA=E3=80=81=E6=85=A2=E7=AD=96=E7=95=A5?= =?UTF-8?q?=E5=90=8E=E5=8F=B0=E8=A1=A5=E7=AE=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 策略页全量 run_all 需 ~2 分钟, 期间卡片全空。现在按历史耗时升序执行, 首返时限 (strategy_run_all_first_return_s, 默认 15s) 内算完的策略随 响应返回, 慢策略转后台继续算并逐个写入策略缓存, 前端轮询 cached-summary 逐个点亮卡片数字 (未出的显示脉冲占位)。 后端: - services/strategy_run_queue: 单飞 daemon 工作线程 + handle 状态。 相同 key (资产/周期/日期/策略集) 且未完成的请求搭车现有执行不重算; 已完成的重跑即新执行。后端全局同时只跑一个 run_all, 补上前端 防重入之外的第二道 Numba 并发防线 - run_all 渐进分支 (仅日线 + summary_only): 逐策略增量写缓存 (同日 按 sid 合并), 收尾整体重写保持旧口径; 结果带 computed_at 时间戳; 分钟周期与明细请求保持整段阻塞不变 - 历史耗时落盘 user_data/strategy_run_timings.json, 次日起快策略自动 排前; cached-summary 透传 computed_at 前端: - 请求通用 30s 超时 + 慢接口豁免清单 (run_all/run/backtest/factor 等 300s), 避免一个挂起请求占满 HTTP/1.1 连接拖死全站 - Screener: 收到 pending 后 summaryQuery 每 2s 轮询, 以 computed_at >= started_at 判新 (防同日旧缓存冒充), 8 分钟兜底; StrategyCard 三种尺寸新增 computing 脉冲占位 验证: 新增 13 测试 (排序/落盘/搭车/串行化/端点行为/旧路径兼容), 全量套件 1762 passed; pnpm build 通过。 --- backend/app/api/screener.py | 119 ++++++++- backend/app/services/strategy_run_queue.py | 173 +++++++++++++ backend/tests/test_screener_builtin_params.py | 15 +- backend/tests/test_screener_cache_api.py | 4 +- .../test_screener_run_all_progressive.py | 236 ++++++++++++++++++ backend/tests/test_strategy_run_queue.py | 129 ++++++++++ .../src/components/screener/StrategyCard.tsx | 13 +- frontend/src/lib/api.ts | 57 ++++- frontend/src/pages/Screener.tsx | 35 ++- 9 files changed, 768 insertions(+), 13 deletions(-) create mode 100644 backend/app/services/strategy_run_queue.py create mode 100644 backend/tests/test_screener_run_all_progressive.py create mode 100644 backend/tests/test_strategy_run_queue.py diff --git a/backend/app/api/screener.py b/backend/app/api/screener.py index fe6af72..1650ec2 100644 --- a/backend/app/api/screener.py +++ b/backend/app/api/screener.py @@ -1,6 +1,7 @@ """Screener API。""" from __future__ import annotations +import contextlib import glob as _glob import logging import math @@ -14,8 +15,9 @@ from typing import Any, Optional from fastapi import APIRouter, HTTPException, Query, Request from pydantic import BaseModel +from app.config import settings from app.db_safe import is_valid_ext_ident, quote_ident -from app.services import strategy_cache +from app.services import strategy_cache, strategy_run_queue from app.services.screener import ScreenerService from app.strategy import config as strategy_config @@ -379,6 +381,9 @@ def get_cached_summary(request: Request): sid: { "total": int(result.get("total") or 0), "as_of": result.get("as_of"), + # 渐进式 run_all 写入的计算时间戳; 监控实时叠加/旧缓存无此字段 → None, + # 前端视为新鲜 (有值即为最新一轮实时结果) + "computed_at": result.get("computed_at"), } for sid, result in results.items() if isinstance(result, dict) @@ -497,6 +502,98 @@ def market_snapshot(request: Request): return {"as_of": str(as_of), "rows": rows} +def _run_all_progressive( + *, + repo, + engine, + svc: ScreenerService, + as_of, + asset_type: str, + timeframe: str, + all_ids: list[str], + params_map: dict, + overrides_map: dict, + first_return_s: float, + t_total: float, +) -> dict: + """run_all 渐进式执行: 快策略随响应先返回, 慢策略后台算完逐个落缓存。 + + 执行全程在单飞执行器里 (见 services/strategy_run_queue.py): 相同请求 + 搭车现有执行, 不同请求排队; HTTP 侧只轮询状态快照到首返时限。 + """ + data_dir = repo.store.data_dir + key = (asset_type, timeframe, str(as_of), tuple(sorted(all_ids))) + ordered_ids = strategy_run_queue.order_strategy_ids( + all_ids, strategy_run_queue.load_run_timings(data_dir) + ) + + def job(handle: strategy_run_queue.StrategyRunHandle) -> None: + context = svc.build_strategy_context( + engine, + as_of, + ordered_ids, + timeframe=timeframe, + params_map=params_map, + overrides_map=overrides_map, + ) + all_results: dict[str, dict] = {} + elapsed_map: dict[str, float] = {} + for sid in ordered_ids: + t0 = time.perf_counter() + single = engine.run_all( + context, + params_map=params_map, + overrides_map=overrides_map, + strategy_ids=[sid], + parallel=False, + ) + result = single[sid] + payload = { + "total": result.total, + "as_of": str(as_of), + "rows": _safe(asdict(result)).get("rows", []), + "computed_at": int(time.time() * 1000), + } + all_results[sid] = payload + elapsed_map[sid] = (time.perf_counter() - t0) * 1000 + # 逐策略增量落盘 (write_cache 同日按 sid 合并), 前端轮询即可逐个看到 + try: + strategy_cache.write_cache(data_dir, str(as_of), {sid: payload}) + except Exception: + logger.warning("run_all 渐进写入缓存失败: %s", sid, exc_info=True) + handle.complete(sid, {k: v for k, v in payload.items() if k != "rows"}) + # 收尾: 与旧版口径一致的整体重写 + 耗时落盘供下次排序 + if all_results: + with contextlib.suppress(Exception): + strategy_cache.write_cache(data_dir, str(as_of), all_results) + strategy_run_queue.record_run_timings(data_dir, elapsed_map) + + handle = strategy_run_queue.MANAGER.get_or_submit(key, ordered_ids, job) + deadline = time.perf_counter() + first_return_s + snap = handle.snapshot() + while not snap["done"] and time.perf_counter() < deadline: + time.sleep(0.2) + snap = handle.snapshot() + + done_results = snap["results"] + if snap["error"] and not done_results: + raise HTTPException(status_code=500, detail=snap["error"]) + logger.info( + "run_all: first return %.1fms (%d done, %d pending)", + (time.perf_counter() - t_total) * 1000, + len(done_results), + len(snap["pending"]), + ) + return { + "as_of": str(as_of), + "results": done_results, + "pending": snap["pending"], + "complete": snap["done"] and not snap["error"], + "error": snap["error"], + "started_at": snap["started_at_ms"], + } + + @router.post("/run_all") def run_all(request: Request, body: Optional[dict] = None): """批量运行指定策略；注册、路由和执行均由 StrategyEngine 负责。""" @@ -556,6 +653,26 @@ def run_all(request: Request, body: Optional[dict] = None): for sid in all_ids } overrides_map = {sid: all_overrides.get(sid, {}) for sid in all_ids} + + # 渐进式返回 (页面首屏路径): 按历史耗时升序执行, 首返时限内算完的随响应 + # 返回, 慢策略转后台继续算并逐个写入策略缓存, 前端轮询 cached-summary 点亮。 + # 仅日线 + summary_only (策略页卡片) 启用; 分钟/明细请求保持整段阻塞。 + first_return_s = settings.strategy_run_all_first_return_s + if body.get("summary_only") and timeframe == "1d" and first_return_s > 0: + return _run_all_progressive( + repo=repo, + engine=engine, + svc=svc, + as_of=as_of, + asset_type=asset_type, + timeframe=timeframe, + all_ids=all_ids, + params_map=params_map, + overrides_map=overrides_map, + first_return_s=first_return_s, + t_total=t_total, + ) + try: context = svc.build_strategy_context( engine, diff --git a/backend/app/services/strategy_run_queue.py b/backend/app/services/strategy_run_queue.py new file mode 100644 index 0000000..8129a98 --- /dev/null +++ b/backend/app/services/strategy_run_queue.py @@ -0,0 +1,173 @@ +"""策略 run_all 渐进式执行 — 单飞后台执行 + 快策略先返回。 + +页面进入策略页时 run_all 全量跑需要 ~2 分钟, 用户只能盯着空卡片等。此模块把 +执行拆成「同步等一小段 + 后台继续算」: + +- 全局同一时刻只执行一个 run_all (polars/Numba 并发跑两份有崩死风险), + 请求先到先得, 后来者排队; 相同 key (资产/周期/日期/策略集) 的重复请求 + 直接搭车现有执行, 不重复算。 +- 按历史耗时升序执行: 快策略 (秒级) 在首返时限内完成并随 HTTP 响应返回, + 慢策略 (分钟级) 留在后台慢慢算。 +- 每个策略算完立刻增量写入 strategy_cache, 前端轮询 cached-summary + 逐个点亮卡片数字。 +""" +from __future__ import annotations + +import json +import logging +import os +import queue +import threading +import time +from collections.abc import Callable +from pathlib import Path + +logger = logging.getLogger(__name__) + +_TIMINGS_FILENAME = "strategy_run_timings.json" +_timings_lock = threading.Lock() + + +def _timings_path(data_dir: Path) -> Path: + return data_dir / "user_data" / _TIMINGS_FILENAME + + +def load_run_timings(data_dir: Path) -> dict[str, float]: + """读取各策略上次执行耗时 (ms); 无文件/损坏时返回空。""" + with _timings_lock: + try: + data = json.loads(_timings_path(data_dir).read_text(encoding="utf-8")) + except (FileNotFoundError, ValueError, OSError): + return {} + if not isinstance(data, dict): + return {} + return {str(k): float(v) for k, v in data.items() if isinstance(v, (int, float))} + + +def record_run_timings(data_dir: Path, elapsed_ms: dict[str, float]) -> None: + """批量记录策略耗时 (ms), 与已有文件合并后原子重写。""" + if not elapsed_ms: + return + with _timings_lock: + path = _timings_path(data_dir) + path.parent.mkdir(parents=True, exist_ok=True) + merged: dict[str, float] = {} + try: + old = json.loads(path.read_text(encoding="utf-8")) + if isinstance(old, dict): + merged = {str(k): float(v) for k, v in old.items() if isinstance(v, (int, float))} + except (FileNotFoundError, ValueError, OSError): + pass + merged.update({sid: float(ms) for sid, ms in elapsed_ms.items()}) + tmp = path.with_name(path.name + ".tmp") + tmp.write_text(json.dumps(merged, ensure_ascii=False), encoding="utf-8") + os.replace(tmp, path) + + +def order_strategy_ids(all_ids: list[str], timings: dict[str, float]) -> list[str]: + """快策略先算: 有历史耗时的按耗时升序, 未知耗时的保持原顺序排在后面。""" + known = sorted( + (timings[sid], i, sid) for i, sid in enumerate(all_ids) if sid in timings + ) + known_ids = {sid for _, _, sid in known} + unknown = [sid for sid in all_ids if sid not in known_ids] + return [sid for _, _, sid in known] + unknown + + +class StrategyRunHandle: + """一次 run_all 的执行状态; 端点线程 (读) 与后台执行线程 (写) 共享。""" + + def __init__(self, key: tuple, ordered_ids: list[str]) -> None: + self.key = key + self.started_at_ms = int(time.time() * 1000) + self._lock = threading.Lock() + self._results: dict[str, dict] = {} + self._remaining: list[str] = list(ordered_ids) + self._error: str | None = None + self._done = False + + def complete(self, sid: str, payload: dict) -> None: + with self._lock: + self._results[sid] = payload + if sid in self._remaining: + self._remaining.remove(sid) + + def fail(self, message: str) -> None: + with self._lock: + self._error = message + self._done = True + + def finish(self) -> None: + with self._lock: + self._done = True + + def snapshot(self) -> dict: + """线程安全快照: 结果拷贝 + 剩余/错误/完成状态。""" + with self._lock: + return { + "results": dict(self._results), + "pending": list(self._remaining), + "error": self._error, + "done": self._done, + "started_at_ms": self.started_at_ms, + } + + +class StrategyRunManager: + """run_all 单飞管理器。 + + - 相同 key 且仍在执行 (含排队中) 的重复请求搭车现有执行, 不重复算 + (页面 reload / StrictMode / 反复切换); 已完成的不再搭车, 重跑即新执行。 + - 不同 key 在唯一 daemon 工作线程里排队; 端点在首返时限内等不到也只能 + 先返回 pending, 前端靠轮询缓存拿最终结果。 + - 工作线程为 daemon: 进程退出不等待剩余计算 (缓存写入均为原子替换, + 中断只留部分结果, 下次进入页面补算)。 + """ + + def __init__(self) -> None: + self._lock = threading.Lock() + self._handles: dict[tuple, StrategyRunHandle] = {} + self._queue: queue.Queue[tuple[StrategyRunHandle, Callable]] = queue.Queue() + self._worker: threading.Thread | None = None + + def get_or_submit( + self, + key: tuple, + ordered_ids: list[str], + job: Callable[[StrategyRunHandle], None], + ) -> StrategyRunHandle: + with self._lock: + # 顺手清理已完成的 handle, 防止字典随不同 key 无限增长 + for k in [k for k, h in self._handles.items() if h.snapshot()["done"]]: + del self._handles[k] + existing = self._handles.get(key) + if existing is not None: + return existing + handle = StrategyRunHandle(key, ordered_ids) + self._handles[key] = handle + self._ensure_worker() + self._queue.put((handle, job)) + return handle + + def _ensure_worker(self) -> None: + with self._lock: + if self._worker is None or not self._worker.is_alive(): + self._worker = threading.Thread( + target=self._run_loop, name="runall", daemon=True + ) + self._worker.start() + + def _run_loop(self) -> None: + while True: + handle, job = self._queue.get() + try: + job(handle) + except Exception as e: + logger.exception("run_all 后台执行失败: %s", e) + handle.fail(str(e)) + else: + handle.finish() + + +# 进程级单例: 与 strategy_cache 的模块级锁同风格, 生命周期跟随进程 +MANAGER = StrategyRunManager() diff --git a/backend/tests/test_screener_builtin_params.py b/backend/tests/test_screener_builtin_params.py index 660d9d2..aff4077 100644 --- a/backend/tests/test_screener_builtin_params.py +++ b/backend/tests/test_screener_builtin_params.py @@ -59,7 +59,7 @@ class _CapturingStrategyEngine: }) return ScreenerResult(as_of=context.as_of, strategy=strategy_id) - def run_all(self, context, *, params_map=None, overrides_map=None, strategy_ids=None): + def run_all(self, context, *, params_map=None, overrides_map=None, strategy_ids=None, parallel=True): self.calls.append({ "kind": "run_all", "params_map": params_map, @@ -153,8 +153,13 @@ def test_batch_summary_response_still_writes_full_cache(monkeypatch, tmp_path): }, ) - assert payload == { - "as_of": "2026-07-15", - "results": {"builtin_strategy": {"total": 0, "as_of": "2026-07-15"}}, - } + assert payload["as_of"] == "2026-07-15" + assert payload["results"]["builtin_strategy"]["total"] == 0 + assert payload["results"]["builtin_strategy"]["as_of"] == "2026-07-15" + # 渐进式路径: 全部算完 → complete 且无 pending + assert payload["pending"] == [] + assert payload["complete"] is True + assert payload["error"] is None + # 渐进式增量写 + 收尾全量写都带 rows, 缓存口径不变 assert written[0][2]["builtin_strategy"]["rows"] == [] + assert written[-1][2]["builtin_strategy"]["rows"] == [] diff --git a/backend/tests/test_screener_cache_api.py b/backend/tests/test_screener_cache_api.py index 8d748b4..d00a7f6 100644 --- a/backend/tests/test_screener_cache_api.py +++ b/backend/tests/test_screener_cache_api.py @@ -48,7 +48,9 @@ def test_cached_summary_omits_rows_and_counts_realtime_expirations(monkeypatch, payload = screener_api.get_cached_summary(_request(tmp_path, realtime)) - assert payload["results"] == {"strategy_a": {"total": 2, "as_of": "2026-07-20"}} + assert payload["results"] == { + "strategy_a": {"total": 2, "as_of": "2026-07-20", "computed_at": None} + } assert payload["today_ever_counts"] == {"strategy_a": 4} assert "rows" not in payload["results"]["strategy_a"] diff --git a/backend/tests/test_screener_run_all_progressive.py b/backend/tests/test_screener_run_all_progressive.py new file mode 100644 index 0000000..f2081e7 --- /dev/null +++ b/backend/tests/test_screener_run_all_progressive.py @@ -0,0 +1,236 @@ +"""run_all 渐进式返回的端点行为 — 快策略先返回、慢策略后台落缓存、搭车与旧路径兼容。""" +from __future__ import annotations + +import time +from dataclasses import dataclass, field +from datetime import date +from types import SimpleNamespace + +import pytest +from fastapi import HTTPException + +from app.api import screener as screener_api +from app.config import settings +from app.services import strategy_cache, strategy_run_queue + +AS_OF = "2026-09-04" + + +@dataclass +class _FakeResult: + total: int = 1 + rows: list = field(default_factory=lambda: [{"symbol": "000001.SZ", "close": 1.0}]) + as_of: str = AS_OF + + +class _FakeEngine: + def __init__(self, delays: dict[str, float]): + self._delays = delays + self.executed: list[str] = [] + + def has(self, sid: str) -> bool: + return sid in self._delays + + def get(self, sid: str): + return SimpleNamespace(meta={}) + + def run_all(self, context, params_map=None, overrides_map=None, *, strategy_ids=None, parallel=True): + out = {} + for sid in strategy_ids or []: + self.executed.append(sid) + time.sleep(self._delays[sid]) + out[sid] = _FakeResult() + return out + + +class _FakeService: + def __init__(self, repo, asset_type="stock"): + pass + + def latest_date(self): + return date.fromisoformat(AS_OF) + + def build_strategy_context(self, *args, **kwargs): + return SimpleNamespace() + + +def _request(tmp_path, engine): + repo = SimpleNamespace(store=SimpleNamespace(data_dir=tmp_path)) + state = SimpleNamespace(repo=repo, strategy_engine=engine, monitor_engine=None) + return SimpleNamespace(app=SimpleNamespace(state=state)) + + +def _wait_cache_results(tmp_path, want_ids, timeout=8.0) -> dict: + deadline = time.time() + timeout + while time.time() < deadline: + results = (strategy_cache.read_cache(tmp_path) or {}).get("results") or {} + if all(i in results for i in want_ids): + return results + time.sleep(0.05) + return (strategy_cache.read_cache(tmp_path) or {}).get("results") or {} + + +@pytest.fixture() +def fast_first_return(monkeypatch): + monkeypatch.setattr(settings, "strategy_run_all_first_return_s", 0.4) + + +def test_run_all_returns_fast_first_then_background_fills_cache( + monkeypatch, tmp_path, fast_first_return +): + engine = _FakeEngine({"fast_a": 0.02, "fast_b": 0.02, "slow_c": 0.8}) + monkeypatch.setattr(screener_api, "ScreenerService", _FakeService) + + resp = screener_api.run_all( + _request(tmp_path, engine), + { + "as_of": AS_OF, + "strategy_ids": ["fast_a", "fast_b", "slow_c"], + "asset_type": "stock", + "timeframe": "1d", + "summary_only": True, + }, + ) + # 首返: 快策略已完成 (带 total, 无 rows), 慢策略 pending + assert set(resp["results"]) == {"fast_a", "fast_b"} + assert resp["results"]["fast_a"]["total"] == 1 + assert "rows" not in resp["results"]["fast_a"] + assert resp["pending"] == ["slow_c"] + assert resp["complete"] is False + assert isinstance(resp["started_at"], int) + + # 后台继续: 慢策略最终也落进缓存, 且带 computed_at + results = _wait_cache_results(tmp_path, ["fast_a", "fast_b", "slow_c"]) + assert set(results) == {"fast_a", "fast_b", "slow_c"} + assert all(r.get("computed_at") for r in results.values()) + + # 耗时已记录 → 下次按耗时升序 (快策略先算) + timings = strategy_run_queue.load_run_timings(tmp_path) + assert set(timings) == {"fast_a", "fast_b", "slow_c"} + assert timings["slow_c"] > timings["fast_a"] + + +def test_run_all_second_run_orders_by_recorded_timings( + monkeypatch, tmp_path, fast_first_return +): + monkeypatch.setattr(screener_api, "ScreenerService", _FakeService) + engine = _FakeEngine({"slow_a": 0.5, "fast_b": 0.01}) + body = { + "as_of": AS_OF, + "strategy_ids": ["slow_a", "fast_b"], + "asset_type": "stock", + "timeframe": "1d", + "summary_only": True, + } + req = _request(tmp_path, engine) + + def _wait_executed(count, timeout=8.0): + deadline = time.time() + timeout + while time.time() < deadline: + if len(engine.executed) >= count: + return + time.sleep(0.05) + raise AssertionError(f"执行未到 {count} 个: {engine.executed}") + + # 第一轮: 无耗时记录 → 按传入顺序 slow_a 先算 + screener_api.run_all(req, body) + _wait_executed(2) + assert engine.executed == ["slow_a", "fast_b"] + + # 第二轮: 有耗时记录 → fast_b (快) 升序在前, 且首返带上 fast_b。 + # 若第一轮 handle 还在收尾 (终写缓存/记录耗时), 请求会搭车旧 handle; + # 重试直到真正触发新一轮执行。 + resp2 = None + for _ in range(40): + engine.executed.clear() + resp2 = screener_api.run_all(req, body) + if engine.executed: + break + time.sleep(0.05) + assert resp2 is not None and engine.executed, "第二轮未触发新执行" + assert engine.executed[0] == "fast_b" + assert set(resp2["results"]) == {"fast_b"} + _wait_executed(2) + assert engine.executed == ["fast_b", "slow_a"] + + +def test_run_all_same_key_piggybacks_running_execution( + monkeypatch, tmp_path, fast_first_return +): + engine = _FakeEngine({"fast_a": 0.02, "slow_c": 1.2}) + monkeypatch.setattr(screener_api, "ScreenerService", _FakeService) + body = { + "as_of": AS_OF, + "strategy_ids": ["fast_a", "slow_c"], + "asset_type": "stock", + "timeframe": "1d", + "summary_only": True, + } + req = _request(tmp_path, engine) + resp1 = screener_api.run_all(req, body) + # 第一笔仍在后台跑 slow_c 时, 相同请求搭车: 同一起点, 不重复执行 + resp2 = screener_api.run_all(_request(tmp_path, engine), body) + assert resp2["started_at"] == resp1["started_at"] + assert engine.executed.count("fast_a") == 1 + _wait_cache_results(tmp_path, ["fast_a", "slow_c"]) + assert engine.executed.count("slow_c") == 1 + assert engine.executed.count("fast_a") == 1 + + +def test_run_all_background_error_without_results_is_500( + monkeypatch, tmp_path, fast_first_return +): + class _BoomEngine(_FakeEngine): + def run_all(self, context, params_map=None, overrides_map=None, *, strategy_ids=None, parallel=True): + raise ValueError("缺少列: volume") + + monkeypatch.setattr(screener_api, "ScreenerService", _FakeService) + with pytest.raises(HTTPException) as excinfo: + screener_api.run_all( + _request(tmp_path, _BoomEngine({"bad_a": 0.01})), + { + "as_of": AS_OF, + "strategy_ids": ["bad_a"], + "asset_type": "stock", + "timeframe": "1d", + "summary_only": True, + }, + ) + assert excinfo.value.status_code == 500 + assert "volume" in excinfo.value.detail + + +def test_run_all_minute_timeframe_stays_blocking(monkeypatch, tmp_path, fast_first_return): + engine = _FakeEngine({"m1": 0.01}) + monkeypatch.setattr(screener_api, "ScreenerService", _FakeService) + resp = screener_api.run_all( + _request(tmp_path, engine), + { + "as_of": AS_OF, + "strategy_ids": ["m1"], + "asset_type": "stock", + "timeframe": "1m", + "summary_only": True, + }, + ) + # 分钟周期: 整段阻塞、旧响应形状 (无 pending), 且不写日线缓存 + assert set(resp["results"]) == {"m1"} + assert "pending" not in resp + assert not ((strategy_cache.read_cache(tmp_path) or {}).get("results") or {}) + + +def test_run_all_full_detail_stays_blocking(monkeypatch, tmp_path, fast_first_return): + engine = _FakeEngine({"d1": 0.01}) + monkeypatch.setattr(screener_api, "ScreenerService", _FakeService) + resp = screener_api.run_all( + _request(tmp_path, engine), + { + "as_of": AS_OF, + "strategy_ids": ["d1"], + "asset_type": "stock", + "timeframe": "1d", + }, + ) + # 非 summary 请求: 保持整段阻塞并返回明细 + assert resp["results"]["d1"]["rows"][0]["symbol"] == "000001.SZ" + assert "pending" not in resp diff --git a/backend/tests/test_strategy_run_queue.py b/backend/tests/test_strategy_run_queue.py new file mode 100644 index 0000000..d1f8437 --- /dev/null +++ b/backend/tests/test_strategy_run_queue.py @@ -0,0 +1,129 @@ +"""strategy_run_queue 单测 — 耗时排序、耗时落盘、handle 状态、单飞管理器。""" +from __future__ import annotations + +import threading +import time + +from app.services import strategy_run_queue as q + + +def test_order_strategy_ids_fast_first_unknown_last(): + ids = ["slow_a", "mid_b", "fast_c", "new_d", "new_e"] + timings = {"slow_a": 5000.0, "fast_c": 100.0, "mid_b": 1000.0} + assert q.order_strategy_ids(ids, timings) == ["fast_c", "mid_b", "slow_a", "new_d", "new_e"] + + +def test_run_timings_record_merge_and_load(tmp_path): + assert q.load_run_timings(tmp_path) == {} + q.record_run_timings(tmp_path, {"a": 100.0}) + q.record_run_timings(tmp_path, {"b": 200.0, "a": 50.0}) + assert q.load_run_timings(tmp_path) == {"a": 50.0, "b": 200.0} + + +def test_run_timings_survives_corrupt_file(tmp_path): + path = q._timings_path(tmp_path) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("not json", encoding="utf-8") + assert q.load_run_timings(tmp_path) == {} + q.record_run_timings(tmp_path, {"a": 1.0}) + assert q.load_run_timings(tmp_path) == {"a": 1.0} + + +def test_handle_snapshot_lifecycle(): + h = q.StrategyRunHandle(("k",), ["s1", "s2"]) + snap = h.snapshot() + assert snap["pending"] == ["s1", "s2"] + assert snap["done"] is False + + h.complete("s1", {"total": 3}) + snap = h.snapshot() + assert snap["results"] == {"s1": {"total": 3}} + assert snap["pending"] == ["s2"] + + h.fail("boom") + snap = h.snapshot() + assert snap["error"] == "boom" + assert snap["done"] is True + + +def _wait_done(handle, timeout=5.0) -> None: + deadline = time.time() + timeout + while time.time() < deadline: + if handle.snapshot()["done"]: + return + time.sleep(0.01) + raise AssertionError("handle 未在时限内完成") + + +def test_manager_piggybacks_same_running_key(): + mgr = q.StrategyRunManager() + started = threading.Event() + release = threading.Event() + calls: list[int] = [] + + def job(handle): + calls.append(1) + started.set() + assert release.wait(timeout=5) + + h1 = mgr.get_or_submit(("k",), ["s"], job) + assert started.wait(timeout=5) + # 执行中: 相同 key 搭车, 不重复提交 + h2 = mgr.get_or_submit(("k",), ["s"], job) + assert h2 is h1 + + release.set() + _wait_done(h1) + assert calls == [1] + + # 已完成后: 同 key 再来 → 新执行 (重跑语义) + h3 = mgr.get_or_submit(("k",), ["s"], job) + assert h3 is not h1 + _wait_done(h3) + assert calls == [1, 1] + + +def test_manager_piggybacks_same_queued_key(): + mgr = q.StrategyRunManager() + started = threading.Event() + release = threading.Event() + + def job(handle): + started.set() + assert release.wait(timeout=5) + + h1 = mgr.get_or_submit(("a",), ["s"], job) # 占住唯一 worker + assert started.wait(timeout=5) + # key b 排队 (未开始), 此时 b 的重复请求应搭车排队中的 handle + h2 = mgr.get_or_submit(("b",), ["s"], job) + h3 = mgr.get_or_submit(("b",), ["s"], job) + assert h2 is h3 + + release.set() + _wait_done(h1) + _wait_done(h2, timeout=10) + + +def test_manager_serializes_different_keys(): + mgr = q.StrategyRunManager() + lock = threading.Lock() + active: list[str] = [] + overlap: list[list[str]] = [] + + def make_job(name): + def job(handle): + with lock: + active.append(name) + if len(active) > 1: + overlap.append(list(active)) + time.sleep(0.1) + with lock: + active.remove(name) + + return job + + h1 = mgr.get_or_submit(("a",), ["s"], make_job("a")) + h2 = mgr.get_or_submit(("b",), ["s"], make_job("b")) + _wait_done(h1, timeout=10) + _wait_done(h2, timeout=10) + assert overlap == [] diff --git a/frontend/src/components/screener/StrategyCard.tsx b/frontend/src/components/screener/StrategyCard.tsx index 7a9a580..9d7d5a6 100644 --- a/frontend/src/components/screener/StrategyCard.tsx +++ b/frontend/src/components/screener/StrategyCard.tsx @@ -94,12 +94,14 @@ interface StrategyCardProps { onToggleMonitor?: () => void /** 周期徽章 (如 '分钟'); 日线策略不传 */ timeframeBadge?: string + /** 后台计算中 (渐进式 run_all): 数字未出时显示脉冲占位 */ + computing?: boolean } export function StrategyCard({ name, description, source, active, count, expiredCount, loading, cardSize, - onRun, disabled, onSettings, monitored, onToggleMonitor, timeframeBadge, + onRun, disabled, onSettings, monitored, onToggleMonitor, timeframeBadge, computing, }: StrategyCardProps) { const cs = CARD_STYLES[cardSize] const activeCls = active @@ -149,6 +151,9 @@ export function StrategyCard({ )} )} + {count == null && !loading && computing && ( + ··· + )} {loading &&
}
@@ -209,6 +217,9 @@ export function StrategyCard({ {count != null && !loading && ( {count} )} + {count == null && !loading && computing && ( + ··· + )} {hasExpired && ( {'-' + expiredCount} )} diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 0f20fa5..205518d 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -10,6 +10,9 @@ const BASE = '' type RequestOptions = RequestInit & { /** 为 true 时不弹错误 toast（由调用方自行汇总提示，如多图串行队列） */ quiet?: boolean + /** 请求超时毫秒数; null 关闭。默认 30s — 后端依赖 polars, 偶发挂起时无超时 + * 会占满浏览器同源连接池, 拖垮整页所有请求 (表现为全部排队"已停止")。 */ + timeoutMs?: number | null } export class ApiError extends Error { @@ -22,14 +25,39 @@ export class ApiError extends Error { } } +const DEFAULT_REQUEST_TIMEOUT_MS = 30_000 +/** 同步计算型接口 (回测/筛选等) 的放宽超时: 合法耗时可能远超轮询类接口。 */ +const COMPUTE_REQUEST_TIMEOUT_MS = 300_000 + async function request(path: string, init?: RequestOptions): Promise { - const { quiet, ...fetchInit } = init ?? {} + const { quiet, timeoutMs = DEFAULT_REQUEST_TIMEOUT_MS, ...fetchInit } = init ?? {} const isFormData = fetchInit.body instanceof FormData const headers: Record = {} if (!isFormData) headers['Content-Type'] = 'application/json' // 合并调用方传入的 headers (此前会被整体覆盖丢弃) Object.assign(headers, fetchInit.headers as Record | undefined) - const res = await fetch(`${BASE}${path}`, { ...fetchInit, headers }) + // 自带 signal 的调用方 (上传/串行队列) 由其自行控制中止; 其余走默认超时。 + const ctl = timeoutMs == null || fetchInit.signal ? undefined : new AbortController() + const timeoutSeconds = Math.round((timeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS) / 1000) + let timer: number | undefined + if (ctl && timeoutMs != null) timer = window.setTimeout(() => ctl.abort(), timeoutMs) + let res: Response + try { + res = await fetch(`${BASE}${path}`, { + ...fetchInit, + headers, + ...(ctl ? { signal: ctl.signal } : {}), + }) + } catch (err) { + if (ctl && err instanceof DOMException && err.name === 'AbortError') { + const msg = `请求超时（${timeoutSeconds}s）· ${path.split('?')[0]}` + if (!quiet) toast(msg, 'error') + throw new ApiError(msg, 0) + } + throw err + } finally { + if (timer !== undefined) window.clearTimeout(timer) + } if (!res.ok) { let detail = '' try { @@ -364,6 +392,8 @@ export interface ScreenerResult { export interface ScreenerResultSummary { total: number as_of: string + /** 渐进式 run_all 写入的计算时间戳 (Unix ms); 监控实时叠加等来源无此字段 */ + computed_at?: number | null } export interface ScreenerCachedSummary { @@ -373,6 +403,20 @@ export interface ScreenerCachedSummary { updated_at: number | null } +/** run_all 渐进式返回: 快策略已算完, 慢策略后台继续算 */ +export interface ScreenerRunAllSummary { + as_of: string | null + results: Record + /** 尚未算完的策略 (后台继续, 逐个写入缓存) */ + pending?: string[] + /** 全部算完时为 true */ + complete?: boolean + /** 后台执行出错时的错误信息 (部分结果仍会返回) */ + error?: string | null + /** 本次执行起点 (Unix ms, 后端时钟), 用于判断缓存结果是否属于本轮 */ + started_at?: number | null +} + export interface ScreenerCachedResult { result: ScreenerResult | null today_ever_rows: Record | null @@ -2464,16 +2508,18 @@ export const api = { screenerRunPreset: (strategy_id: string, pool?: string[], asOf?: string, extColumns?: string, assetType: 'stock' | 'etf' = 'stock', timeframe: '1d' | '1m' = '1d') => request('/api/screener/run_preset', { method: 'POST', + timeoutMs: COMPUTE_REQUEST_TIMEOUT_MS, body: JSON.stringify({ strategy_id, pool, as_of: asOf ?? null, ext_columns: extColumns || null, asset_type: assetType, timeframe }), }), screenerRunCustom: (conditions: string[], orderBy?: string, limit = 30, pool?: string[], extColumns?: string, assetType: 'stock' | 'etf' = 'stock') => request('/api/screener/run', { method: 'POST', + timeoutMs: COMPUTE_REQUEST_TIMEOUT_MS, body: JSON.stringify({ conditions, order_by: orderBy, limit, pool, ext_columns: extColumns || null, asset_type: assetType }), }), screenerRunAll: (asOf?: string, strategyIds?: string[], assetType: 'stock' | 'etf' = 'stock') => - request<{ as_of: string | null; results: Record }>( - '/api/screener/run_all', { method: 'POST', body: JSON.stringify({ as_of: asOf ?? null, strategy_ids: strategyIds ?? null, asset_type: assetType, timeframe: '1d', summary_only: true }) }, + request( + '/api/screener/run_all', { method: 'POST', timeoutMs: COMPUTE_REQUEST_TIMEOUT_MS, body: JSON.stringify({ as_of: asOf ?? null, strategy_ids: strategyIds ?? null, asset_type: assetType, timeframe: '1d', summary_only: true }) }, ), screenerCachedSummary: () => request('/api/screener/cached-summary'), @@ -2563,6 +2609,7 @@ export const api = { }) => request('/api/backtest/run', { method: 'POST', + timeoutMs: COMPUTE_REQUEST_TIMEOUT_MS, body: JSON.stringify(payload), }), @@ -2656,6 +2703,7 @@ export const api = { }) => request('/api/backtest/factor/run', { method: 'POST', + timeoutMs: COMPUTE_REQUEST_TIMEOUT_MS, body: JSON.stringify(payload), }), @@ -2673,6 +2721,7 @@ export const api = { }) => request('/api/backtest/factor/batch', { method: 'POST', + timeoutMs: COMPUTE_REQUEST_TIMEOUT_MS, body: JSON.stringify(payload), }), diff --git a/frontend/src/pages/Screener.tsx b/frontend/src/pages/Screener.tsx index d221ff9..8318077 100644 --- a/frontend/src/pages/Screener.tsx +++ b/frontend/src/pages/Screener.tsx @@ -124,6 +124,9 @@ export function Screener() { const [expiredCounts, setExpiredCounts] = useState>({}) // 各策略显示上限 (null = 全部) const [strategyLimits, setStrategyLimits] = useState>({}) + // run_all 渐进式返回后仍在后台计算的策略 (startedAt 为后端时钟, 用于判断缓存新旧) + const [pendingRun, setPendingRun] = useState<{ ids: string[]; startedAt: number } | null>(null) + const pendingRunIds = useMemo(() => new Set(pendingRun?.ids ?? []), [pendingRun]) // 筛选条件变化时同步到 map（供切换策略时读取最新值） useEffect(() => { @@ -161,10 +164,12 @@ export function Screener() { // 卡片首屏只读取轻量摘要；明细在点击策略或“全部”时按需加载。 // 摘要只覆盖日线缓存; 分钟策略命中数来自手动单跑。 + // run_all 渐进式返回后后台仍在算 → 轮询摘要, 算完的策略逐个点亮。 const summaryQuery = useQuery({ queryKey: QK.screenerCachedSummary, queryFn: api.screenerCachedSummary, enabled: assetType === 'stock', + refetchInterval: pendingRun ? 2000 : false, }) const fullCachedQuery = useQuery({ @@ -273,6 +278,13 @@ export function Screener() { counts[id] = item.total } setHitCounts(prev => ({ ...prev, ...counts })) + // 渐进式返回: 慢策略后台继续算, 开启摘要轮询逐个点亮 + setPendingRun( + data.pending?.length + ? { ids: [...data.pending], startedAt: data.started_at ?? 0 } + : null, + ) + if (data.error) toast(`策略计算失败：${data.error}`, 'error') qc.invalidateQueries({ queryKey: ['screener-cached'] }) }, }) @@ -315,7 +327,27 @@ export function Screener() { } setHitCounts(counts) setExpiredCounts(expired) - }, [summaryQuery.data, asOf]) + // 渐进式: computed_at 晚于本轮起点的策略已算完, 从 pending 中移除; + // 无 computed_at (监控实时叠加/旧缓存) 视为新鲜。容差吸收前后端时钟差。 + if (pendingRun) { + const arrived = (id: string) => { + const r = summaryQuery.data!.results[id] + if (!r || r.as_of !== asOf) return false + return r.computed_at == null || r.computed_at >= pendingRun.startedAt - 2000 + } + const rest = pendingRun.ids.filter(id => !arrived(id)) + if (rest.length !== pendingRun.ids.length) { + setPendingRun(rest.length ? { ...pendingRun, ids: rest } : null) + } + } + }, [summaryQuery.data, asOf, pendingRun]) + + // 渐进式兜底: 后台计算最长等 8 分钟, 防止异常时无限轮询 + useEffect(() => { + if (!pendingRun) return + const t = setTimeout(() => setPendingRun(null), 8 * 60 * 1000) + return () => clearTimeout(t) + }, [pendingRun]) // 当前单策略缓存更新后同步明细；参数保存的强制重算结果仍由 run 直接覆盖。 useEffect(() => { @@ -817,6 +849,7 @@ export function Screener() { count={hitCounts[id]} expiredCount={expiredCounts[id]} loading={runAll.isPending} + computing={pendingRunIds.has(id)} cardSize={cardSize} onRun={() => handleRun(s)} disabled={run.isPending && activeStrategy === s.id}