From 6345eb93ab4b15c058d2d31a671a73dbfdeb9661 Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Wed, 9 Sep 2026 15:49:21 +0800 Subject: [PATCH] =?UTF-8?q?fix(strategy):=20=E6=B8=90=E8=BF=9B=E5=BC=8F=20?= =?UTF-8?q?run=5Fall=20=E9=80=90=E7=AD=96=E7=95=A5=E9=9A=94=E7=A6=BB?= =?UTF-8?q?=E5=A4=B1=E8=B4=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 单个策略执行崩溃 (如自定义策略 filter_history 的数据类型错误) 会杀掉 整批剩余策略: handle 整体 fail, 后台线程结束, 页面剩余卡片永远不亮。 线上实证: 70 策略池跑到第 35 个 (custom_1782999589 pl.DataFrame 构造 schema 混杂) 崩溃, 后 36 个全部没算。 改为单策略 try/except: 失败记入 handle.errors 并移出待算队列, 其余 策略照常算完落缓存; 响应新增 errors 字段 (前端可忽略)。job 级失败 (context 构建崩溃) 仍走整体 500 语义不变。 --- backend/app/api/screener.py | 24 ++++++--- backend/app/services/strategy_run_queue.py | 11 +++- .../test_screener_run_all_progressive.py | 51 +++++++++++++++++-- 3 files changed, 72 insertions(+), 14 deletions(-) diff --git a/backend/app/api/screener.py b/backend/app/api/screener.py index 2930a2a..63daa55 100644 --- a/backend/app/api/screener.py +++ b/backend/app/api/screener.py @@ -554,14 +554,21 @@ def _run_all_progressive( 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] + # 逐策略隔离: 单个策略崩溃 (如自定义代码的数据类型错误) 只记 + # 错误跳过, 不让整批剩余策略陪葬 — 其余策略照常算完落缓存。 + try: + single = engine.run_all( + context, + params_map=params_map, + overrides_map=overrides_map, + strategy_ids=[sid], + parallel=False, + ) + result = single[sid] + except Exception as e: + logger.warning("run_all: 策略 %s 执行失败, 跳过: %s", sid, e, exc_info=True) + handle.fail_one(sid, str(e)) + continue payload = { "total": result.total, "as_of": str(as_of), @@ -602,6 +609,7 @@ def _run_all_progressive( "as_of": str(as_of), "results": done_results, "pending": snap["pending"], + "errors": snap["errors"], "complete": snap["done"] and not snap["error"], "error": snap["error"], "started_at": snap["started_at_ms"], diff --git a/backend/app/services/strategy_run_queue.py b/backend/app/services/strategy_run_queue.py index 8129a98..b828dd9 100644 --- a/backend/app/services/strategy_run_queue.py +++ b/backend/app/services/strategy_run_queue.py @@ -83,6 +83,7 @@ class StrategyRunHandle: self._lock = threading.Lock() self._results: dict[str, dict] = {} self._remaining: list[str] = list(ordered_ids) + self._errors: dict[str, str] = {} self._error: str | None = None self._done = False @@ -92,6 +93,13 @@ class StrategyRunHandle: if sid in self._remaining: self._remaining.remove(sid) + def fail_one(self, sid: str, message: str) -> None: + """单个策略失败: 记错误并移出待算队列, 不影响其余策略继续。""" + with self._lock: + self._errors[sid] = message + if sid in self._remaining: + self._remaining.remove(sid) + def fail(self, message: str) -> None: with self._lock: self._error = message @@ -102,11 +110,12 @@ class StrategyRunHandle: self._done = True def snapshot(self) -> dict: - """线程安全快照: 结果拷贝 + 剩余/错误/完成状态。""" + """线程安全快照: 结果拷贝 + 剩余/逐策略错误/整体错误/完成状态。""" with self._lock: return { "results": dict(self._results), "pending": list(self._remaining), + "errors": dict(self._errors), "error": self._error, "done": self._done, "started_at_ms": self.started_at_ms, diff --git a/backend/tests/test_screener_run_all_progressive.py b/backend/tests/test_screener_run_all_progressive.py index fbebfcd..33c1d8d 100644 --- a/backend/tests/test_screener_run_all_progressive.py +++ b/backend/tests/test_screener_run_all_progressive.py @@ -177,17 +177,23 @@ def test_run_all_same_key_piggybacks_running_execution( assert engine.executed.count("fast_a") == 1 -def test_run_all_background_error_without_results_is_500( +def test_run_all_job_level_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): + """job 级失败 (如 context 构建崩溃) 且无任何结果 → 500, 语义不变。 + + 策略级失败 (engine.run_all 对单个 sid 抛错) 已改为逐策略隔离, 见 + test_run_all_isolates_single_strategy_failure。 + """ + + class _BoomCtxService(_FakeService): + def build_strategy_context(self, *args, **kwargs): raise ValueError("缺少列: volume") - monkeypatch.setattr(screener_api, "ScreenerService", _FakeService) + monkeypatch.setattr(screener_api, "ScreenerService", _BoomCtxService) with pytest.raises(HTTPException) as excinfo: screener_api.run_all( - _request(tmp_path, _BoomEngine({"bad_a": 0.01})), + _request(tmp_path, _FakeEngine({"bad_a": 0.01})), { "as_of": AS_OF, "strategy_ids": ["bad_a"], @@ -291,3 +297,38 @@ def test_run_all_progressive_builds_matrix_once_and_shares_it( assert engine.matrix_builds == 1 assert engine.seen_markets and all(m == {"fields": 3} for m in engine.seen_markets) assert len(engine.seen_markets) == 3 + + +def test_run_all_isolates_single_strategy_failure( + monkeypatch, tmp_path, fast_first_return +): + """单个策略执行崩溃只跳过它自己: 其余策略照常算完落缓存, 整批不失败。""" + + class _FlakyEngine(_FakeEngine): + def run_all(self, context, params_map=None, overrides_map=None, *, strategy_ids=None, parallel=True): + for sid in strategy_ids or []: + if sid == "broken": + raise ValueError("boom: schema mismatch") + return super().run_all( + context, params_map=params_map, overrides_map=overrides_map, + strategy_ids=strategy_ids, parallel=parallel, + ) + + engine = _FlakyEngine({"ok_a": 0.01, "broken": 0.01, "ok_b": 0.5}) + monkeypatch.setattr(screener_api, "ScreenerService", _FakeService) + + resp = screener_api.run_all( + _request(tmp_path, engine), + { + "as_of": AS_OF, + "strategy_ids": ["ok_a", "broken", "ok_b"], + "asset_type": "stock", + "timeframe": "1d", + "summary_only": True, + }, + ) + # 后台继续: 好策略都落缓存; broken 不在结果也不在 pending, 而是进 errors + results = _wait_cache_results(tmp_path, ["ok_a", "ok_b"]) + assert set(results) == {"ok_a", "ok_b"} + assert "broken" not in results + assert "boom: schema mismatch" in (resp["errors"] or {}).get("broken", "")