From 01347b72bc0b999383bdd3bdc9207b46587958ce Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Wed, 9 Sep 2026 15:29:52 +0800 Subject: [PATCH] =?UTF-8?q?perf(strategy):=20=E6=B8=90=E8=BF=9B=E5=BC=8F?= =?UTF-8?q?=20run=5Fall=20=E5=8F=AA=E6=9E=84=E5=BB=BA=E4=B8=80=E6=AC=A1?= =?UTF-8?q?=E5=85=B1=E4=BA=AB=E7=9F=A9=E9=98=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _run_all_progressive 逐策略调用 engine.run_all(strategy_ids=[sid]), run_all 不回写 context.market → 每个 matrix_native 策略都重建一次 全市场矩阵 (39 万行, 小服务器单次数秒到十余秒), 6 策略一轮多耗约 1 分钟。 从 run_all 抽出 build_shared_matrix (字段并集语义不变, 非渐进路径 行为一致), 渐进式循环外一次建好放入 context.market 复用。 --- backend/app/api/screener.py | 16 +++- backend/app/strategy/engine.py | 77 +++++++++++-------- .../test_screener_run_all_progressive.py | 57 ++++++++++++++ 3 files changed, 118 insertions(+), 32 deletions(-) diff --git a/backend/app/api/screener.py b/backend/app/api/screener.py index 2302449..2930a2a 100644 --- a/backend/app/api/screener.py +++ b/backend/app/api/screener.py @@ -8,7 +8,7 @@ import math import os import re import time -from dataclasses import asdict +from dataclasses import asdict, replace from datetime import date, datetime from typing import Any, Optional @@ -536,6 +536,20 @@ def _run_all_progressive( params_map=params_map, overrides_map=overrides_map, ) + # 逐策略 run_all 不会把矩阵回写 context.market → 每个矩阵策略都会重建 + # 全市场矩阵 (小服务器上单次数秒到十余秒)。这里按字段并集一次建好复用; + # FakeEngine 等无该方法的实现跳过 (保持旧行为)。 + if getattr(context, "market", None) is None: + build_matrix = getattr(engine, "build_shared_matrix", None) + if callable(build_matrix): + matrix = build_matrix( + context, + [(sid, engine.get(sid)) for sid in ordered_ids], + params_map, + overrides_map, + ) + if matrix is not None: + context = replace(context, market=matrix) all_results: dict[str, dict] = {} elapsed_map: dict[str, float] = {} for sid in ordered_ids: diff --git a/backend/app/strategy/engine.py b/backend/app/strategy/engine.py index 3082133..a9f3883 100644 --- a/backend/app/strategy/engine.py +++ b/backend/app/strategy/engine.py @@ -1151,37 +1151,9 @@ class StrategyEngine: raise ValueError("selected strategies require history data") shared_matrix = context.market - matrix_strats = [ - (sid, strategy) - for sid, strategy in selected - if strategy.execution_backend == "matrix_native" - ] - if ( - shared_matrix is None - and matrix_strats - and shared_history is not None - and not shared_history.is_empty() - ): - from app.backtest.matrix import build_market_data_matrix - - field_columns: set[str] = set() - for sid, strategy in matrix_strats: - field_columns.update( - self._matrix_field_columns( - strategy, - overrides_map.get(sid), - params_map.get(sid), - ) - ) - matrix_t0 = time.perf_counter() - shared_matrix = build_market_data_matrix( - shared_history, - field_columns=field_columns, - ) - logger.info( - "run_all: shared matrix built in %.0fms (fields=%d)", - (time.perf_counter() - matrix_t0) * 1000, - len(field_columns), + if shared_matrix is None: + shared_matrix = self.build_shared_matrix( + context, selected, params_map, overrides_map ) results: dict[str, StrategyResult] = {} @@ -1227,6 +1199,49 @@ class StrategyEngine: return results + def build_shared_matrix( + self, + context: StrategyDataContext, + selected: list[tuple[str, StrategyDef]], + params_map: dict | None = None, + overrides_map: dict | None = None, + ): + """按所选策略的字段并集构建市场数据矩阵; 无矩阵策略或无历史时返回 None。 + + 渐进式 run_all (逐策略执行) 也用它一次建好并集矩阵后放入 context.market, + 避免每个 matrix_native 策略重复构建同一份大矩阵 (全市场历史, 秒级)。 + """ + params_map = params_map or {} + overrides_map = overrides_map or {} + matrix_strats = [ + (sid, strategy) + for sid, strategy in selected + if strategy.execution_backend == "matrix_native" + ] + history = context.history + if not matrix_strats or history is None or history.is_empty(): + return None + + from app.backtest.matrix import build_market_data_matrix + + field_columns: set[str] = set() + for sid, strategy in matrix_strats: + field_columns.update( + self._matrix_field_columns( + strategy, + overrides_map.get(sid), + params_map.get(sid), + ) + ) + matrix_t0 = time.perf_counter() + matrix = build_market_data_matrix(history, field_columns=field_columns) + logger.info( + "run_all: shared matrix built in %.0fms (fields=%d)", + (time.perf_counter() - matrix_t0) * 1000, + len(field_columns), + ) + return matrix + @staticmethod def _matrix_field_columns( strategy: StrategyDef, diff --git a/backend/tests/test_screener_run_all_progressive.py b/backend/tests/test_screener_run_all_progressive.py index f2081e7..fbebfcd 100644 --- a/backend/tests/test_screener_run_all_progressive.py +++ b/backend/tests/test_screener_run_all_progressive.py @@ -234,3 +234,60 @@ def test_run_all_full_detail_stays_blocking(monkeypatch, tmp_path, fast_first_re # 非 summary 请求: 保持整段阻塞并返回明细 assert resp["results"]["d1"]["rows"][0]["symbol"] == "000001.SZ" assert "pending" not in resp + + +def test_run_all_progressive_builds_matrix_once_and_shares_it( + monkeypatch, tmp_path, fast_first_return +): + """渐进式逐策略执行前, 并集矩阵只建一次并放进 context.market 复用。 + + 旧行为: 每个策略单独 run_all, context.market 不回写 → 每个矩阵策略都 + 重建全市场矩阵 (小服务器上单次数秒到十余秒)。 + """ + + @dataclass + class _Context: + market: object = None + + class _MatrixEngine(_FakeEngine): + def __init__(self, delays): + super().__init__(delays) + self.matrix_builds = 0 + self.seen_markets: list[object] = [] + + def build_shared_matrix(self, context, selected, params_map=None, overrides_map=None): + self.matrix_builds += 1 + return {"fields": len(selected)} + + def run_all(self, context, params_map=None, overrides_map=None, *, strategy_ids=None, parallel=True): + self.seen_markets.append(context.market) + return super().run_all( + context, params_map=params_map, overrides_map=overrides_map, + strategy_ids=strategy_ids, parallel=parallel, + ) + + class _CtxService(_FakeService): + def build_strategy_context(self, *args, **kwargs): + return _Context() + + engine = _MatrixEngine({"s1": 0.01, "s2": 0.01, "s3": 0.01}) + monkeypatch.setattr(screener_api, "ScreenerService", _CtxService) + + resp = screener_api.run_all( + _request(tmp_path, engine), + { + "as_of": AS_OF, + "strategy_ids": ["s1", "s2", "s3"], + "asset_type": "stock", + "timeframe": "1d", + "summary_only": True, + }, + ) + results = _wait_cache_results(tmp_path, ["s1", "s2", "s3"]) + assert set(results) == {"s1", "s2", "s3"} + assert resp["complete"] is True or resp["pending"] == [] + + # 矩阵只构建一次; 每个策略拿到的都是同一个 market 对象 + 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