chore(archive): pr-479 审查带出的 strategy-prewarm 工作区内容存档

10 月中旬 checkout PR#479 分支时带出的预热线工作区内容, 当时按指示
撤回暂存 (stash 备注: 勿合入)。现按指示存档到独立分支保留, 不合入
main — 分支即终点, 仅供日后参考。内容: screener 预热线 (services/
strategy_prewarm + 接线 + 测试), 与 amber 插件无关。

backend/app/api/screener.py

@@ -17,8 +17,9 @@ 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, strategy_run_queue
+from app.services import strategy_cache, strategy_prewarm, strategy_run_queue
 from app.services.screener import ScreenerService
+from app.services.strategy_prewarm import coverage_warnings as _coverage_warnings
 from app.strategy import config as strategy_config
 
 logger = logging.getLogger(__name__)
@@ -55,27 +56,6 @@ def _safe(result_dict: dict) -> dict:
     return result_dict
 
 
-def _coverage_warnings(
-    svc, as_of, *, engine=None, strategy_ids=None, params_map=None, overrides_map=None,
-) -> list[str]:
-    """数据充足性提示 (#303): enriched 覆盖低于暖机需求时给出人话警告。
-
-    advisory 元数据: 任何计算失败都静默返回 [], 绝不影响策略主流程。
-    引擎/服务无该方法时 (测试 Fake) 同样跳过, 与 build_shared_matrix 的
-    可选能力探测同风格。
-    """
-    try:
-        required: int | None = None
-        rhb = getattr(engine, "required_history_bars", None)
-        if engine is not None and strategy_ids and callable(rhb):
-            required = rhb(strategy_ids, params_map=params_map, overrides_map=overrides_map)
-        cw = getattr(svc, "coverage_warnings", None)
-        if callable(cw):
-            return cw(as_of, required_bars=required)
-    except Exception:
-        logger.debug("coverage warnings unavailable", exc_info=True)
-    return []
-
 
 def _one_word_limit_expr(status_main: str, columns: list[str]) -> Any:
     required = {"open", "high", "low", "close", "status"}
@@ -607,126 +587,6 @@ 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,
-        )
-        # 逐策略 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:
-            t0 = time.perf_counter()
-            # 逐策略隔离: 单个策略崩溃 (如自定义代码的数据类型错误) 只记
-            # 错误跳过, 不让整批剩余策略陪葬 — 其余策略照常算完落缓存。
-            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),
-                "rows": _safe(asdict(result)).get("rows", []),
-                "computed_at": int(time.time() * 1000),
-            }
-            if timeframe == "1d":
-                w = _coverage_warnings(
-                    svc, as_of, engine=engine, strategy_ids=[sid],
-                    params_map=params_map, overrides_map=overrides_map,
-                )
-                if w:
-                    payload["warnings"] = w
-            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"],
-        "errors": snap["errors"],
-        "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):
@@ -809,7 +669,7 @@ def run_all(request: Request, body: Optional[dict] = None):
     # 仅日线 + 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:
-        progressive = _run_all_progressive(
+        progressive = strategy_prewarm.run_all_progressive(
             repo=repo,
             engine=engine,
             svc=svc,

backend/app/config.py

@@ -151,6 +151,10 @@ class Settings(BaseSettings):
     # 前端轮询 cached-summary 点亮卡片。0 = 关闭 (整段阻塞, 旧行为)。
     strategy_run_all_first_return_s: float = 15.0
 
+    # 盘后策略缓存预热: 数据落地 (定时管道/手动同步/启动预热完成) 后自动算满
+    # 当日策略缓存, 策略页打开即读缓存秒出。关闭后回到「页面打开补算缺失」旧行为。
+    strategy_prewarm_enabled: bool = True
+
     # Auth — 首次启动时预置访问密码(明文, 仅用于初始化, 详见 services/auth.bootstrap_from_env)
     # 公网服务器部署时免去 SSH 端口转发设密码的麻烦。写入 auth.json(哈希)后即不再读取。
     auth_password: str = ""

backend/app/main.py

@@ -368,9 +368,20 @@ async def _application_lifespan(app: FastAPI):
         if not matrix_prewarm_owner.schedule(_prewarm):
             logger.info("matrix cache prewarm already running or shutting down, skip")
 
-    repo._on_refresh_done = _schedule_matrix_cache_prewarm  # noqa: SLF001
+    # 策略缓存预热链在同一回调上: 手动/定时管道完成、启动后台预热完成都会触发,
+    # 数据落地后即算满当日策略缓存, 策略页打开秒出 (needs_prewarm 内部判重)。
+    _matrix_prewarm_callback = _schedule_matrix_cache_prewarm
+
+    def _on_refresh_done_with_strategy_prewarm() -> None:
+        if _matrix_prewarm_callback is not None:
+            _matrix_prewarm_callback()
+        from app.services.strategy_prewarm import schedule_prewarm
+
+        schedule_prewarm(app)
+
+    repo._on_refresh_done = _on_refresh_done_with_strategy_prewarm  # noqa: SLF001
     if repo.enriched_ready:
-        _schedule_matrix_cache_prewarm()
+        _on_refresh_done_with_strategy_prewarm()
 
     # 通用监控规则引擎: 启动时 reload 规则到内存态 (修复重启后告警失效)
     from app.services import preferences

backend/app/services/screener.py

@@ -46,7 +46,9 @@ def enriched_history_days(data_dir, asset_type: str = "stock", as_of: date | Non
 
 # ── 进程级历史数据缓存 (避免 run_all 每次重新扫描 parquet + 计算指标) ──
 _history_cache: dict[tuple[str, date, int], tuple[float, pl.DataFrame]] = {}
-_HISTORY_CACHE_TTL = 120.0  # 秒
+# TTL 与「盘后预热 + 页面只补缺失」节奏对齐: 覆盖数据落地到用户查看的典型窗口,
+# 同窗口内补跑缺失策略不再重付指标冷启动 (~5s)。
+_HISTORY_CACHE_TTL = 600.0  # 秒
 
 # load_prior_consecutive 最多回看多少个已存在的日分区 (缺列时继续往前找的上限)
 _PRIOR_PARTITION_SCAN = 10

backend/app/services/strategy_prewarm.py

@@ -0,0 +1,291 @@
+"""盘后策略缓存预热 — 数据落地后主动算满当日策略缓存, 策略页打开即读缓存。
+
+设计要点:
+- 与策略页 run_all 走同一条渐进式执行编排 (本模块的 run_all_progressive,
+  从 api/screener 提取): 同 key 单飞, 预热运行中用户打开页面会搭车同一执行,
+  不重复算; 逐策略算完即写 strategy_cache, 前端轮询可逐个点亮。
+- 触发点链在 repo._on_refresh_done 上 (定时盘后管道 / 数据页手动同步 /
+  启动预热三个完成点都会触发); needs_prewarm 按「缓存 as_of 与最新 enriched
+  日期一致 且 results 覆盖全部合格策略」判重, 已暖则跳过。
+"""
+from __future__ import annotations
+
+import contextlib
+import logging
+import math
+import time
+from collections.abc import Callable
+from dataclasses import asdict, replace
+from datetime import date
+from typing import Any
+
+from app.config import settings
+from app.services import strategy_cache, strategy_run_queue
+
+logger = logging.getLogger(__name__)
+
+
+# ── 轻量助手 (与 api/screener._safe 同语义; services 层不反向依赖 api 模块) ──
+
+def _safe_rows(result: Any) -> list[dict]:
+    rows = asdict(result).get("rows", [])
+    for r in rows:
+        for k, v in list(r.items()):
+            if isinstance(v, float) and not math.isfinite(v):
+                r[k] = None
+    return rows
+
+
+def coverage_warnings(
+    svc, as_of, *, engine=None, strategy_ids=None, params_map=None, overrides_map=None,
+) -> list[str]:
+    """数据充足性提示 (#303): enriched 覆盖低于暖机需求时给出人话警告。
+
+    advisory 元数据: 任何计算失败都静默返回 [], 绝不影响策略主流程。
+    引擎/服务无该方法时 (测试 Fake) 同样跳过, 与 build_shared_matrix 的
+    可选能力探测同风格。
+    """
+    try:
+        required: int | None = None
+        rhb = getattr(engine, "required_history_bars", None)
+        if engine is not None and strategy_ids and callable(rhb):
+            required = rhb(strategy_ids, params_map=params_map, overrides_map=overrides_map)
+        cw = getattr(svc, "coverage_warnings", None)
+        if callable(cw):
+            return cw(as_of, required_bars=required)
+    except Exception:
+        logger.debug("coverage warnings unavailable", exc_info=True)
+    return []
+
+
+def eligible_daily_ids(engine, asset_type: str = "stock", timeframe: str = "1d") -> list[str]:
+    """与 run_all 默认路径同口径的合格策略清单 (非 research_only / 匹配资产与周期)。"""
+    return [
+        meta["id"]
+        for meta in engine.list_strategies()
+        if not meta.get("research_only")
+        and asset_type in meta.get("asset_types", ["stock"])
+        and timeframe in meta.get("timeframes", ["1d"])
+    ]
+
+
+# ── 渐进式执行编排 (api/screener._run_all_progressive 提取, 端点与预热共用) ──
+
+def _build_job(
+    *, repo, engine, svc, as_of: date, timeframe: str,
+    all_ids: list[str], params_map: dict, overrides_map: dict,
+) -> Callable[[strategy_run_queue.StrategyRunHandle], None]:
+    data_dir = repo.store.data_dir
+
+    def job(handle: strategy_run_queue.StrategyRunHandle) -> None:
+        context = svc.build_strategy_context(
+            engine,
+            as_of,
+            all_ids,
+            timeframe=timeframe,
+            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 all_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 all_ids:
+            t0 = time.perf_counter()
+            # 逐策略隔离: 单个策略崩溃 (如自定义代码的数据类型错误) 只记
+            # 错误跳过, 不让整批剩余策略陪葬 — 其余策略照常算完落缓存。
+            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:  # noqa: BLE001
+                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),
+                "rows": _safe_rows(result),
+                "computed_at": int(time.time() * 1000),
+            }
+            if timeframe == "1d":
+                w = coverage_warnings(
+                    svc, as_of, engine=engine, strategy_ids=[sid],
+                    params_map=params_map, overrides_map=overrides_map,
+                )
+                if w:
+                    payload["warnings"] = w
+            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)
+
+    return job
+
+
+def submit_progressive(
+    *, repo, engine, svc, as_of: date, asset_type: str, timeframe: str,
+    all_ids: list[str], params_map: dict, overrides_map: dict,
+) -> strategy_run_queue.StrategyRunHandle:
+    """组装 key 与耗时排序, 提交到单飞执行器; 不等待 (fire-and-forget)。"""
+    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)
+    )
+    job = _build_job(
+        repo=repo, engine=engine, svc=svc, as_of=as_of, timeframe=timeframe,
+        all_ids=ordered_ids, params_map=params_map, overrides_map=overrides_map,
+    )
+    return strategy_run_queue.MANAGER.get_or_submit(key, ordered_ids, job)
+
+
+def run_all_progressive(
+    *,
+    repo,
+    engine,
+    svc,
+    as_of: date,
+    asset_type: str,
+    timeframe: str,
+    all_ids: list[str],
+    params_map: dict,
+    overrides_map: dict,
+    first_return_s: float,
+    t_total: float | None = None,
+) -> dict:
+    """run_all 渐进式执行: 快策略随响应先返回, 慢策略后台算完逐个落缓存。
+
+    执行全程在单飞执行器里 (见 services/strategy_run_queue.py): 相同请求
+    搭车现有执行, 不同请求排队; HTTP 侧只轮询状态快照到首返时限。
+    """
+    if t_total is None:
+        t_total = time.perf_counter()
+    handle = submit_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,
+    )
+    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:
+        from fastapi import HTTPException
+
+        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"],
+        "errors": snap["errors"],
+        "complete": snap["done"] and not snap["errors"],
+        "error": snap["error"],
+        "started_at": snap["started_at_ms"],
+    }
+
+
+# ── 盘后预热入口 ──────────────────────────────────────────────────────
+
+def needs_prewarm(
+    data_dir, engine, latest_date: date | None, *, asset_type: str = "stock",
+) -> bool:
+    """当日策略缓存是否需要预热。
+
+    与前端 cacheCoversPool 同语义: 缓存 as_of 等于最新 enriched 日期, 且
+    每个合格策略都有当日结果才算已暖; 无数据 / 无策略时 False (无事可做)。
+    """
+    if latest_date is None:
+        return False
+    ids = eligible_daily_ids(engine, asset_type)
+    if not ids:
+        return False
+    cached = strategy_cache.read_cache(data_dir)
+    as_of_s = str(latest_date)
+    if not cached or str(cached.get("as_of") or "") != as_of_s:
+        return True
+    results = cached.get("results") or {}
+    covered = {
+        sid for sid in ids
+        if isinstance(results.get(sid), dict) and results[sid].get("as_of") == as_of_s
+    }
+    return not set(ids).issubset(covered)
+
+
+def schedule_prewarm(app) -> None:
+    """数据落地后预热当日策略缓存; 由 repo._on_refresh_done 链式触发。
+
+    判重 + 提交都在调用线程完成 (毫秒级: 读一个 JSON + 列策略), 实际计算在
+    单飞执行器的 daemon 工作线程里跑, 与页面触发的 run_all 排同一条队列。
+    任何失败只记日志, 绝不影响数据管道本身。
+    """
+    if not settings.strategy_prewarm_enabled:
+        return
+    try:
+        from app.services import screener as screener_service
+        from app.strategy import config as strategy_config
+
+        repo = app.state.repo
+        engine = getattr(app.state, "strategy_engine", None)
+        if engine is None:
+            return
+        data_dir = repo.store.data_dir
+        svc = screener_service.ScreenerService(repo)
+        latest = svc.latest_date()
+        if not needs_prewarm(data_dir, engine, latest):
+            logger.debug("strategy prewarm skipped: cache already covers %s", latest)
+            return
+        all_ids = eligible_daily_ids(engine)
+        if not all_ids:
+            return
+        overrides = strategy_config.list_overrides(data_dir)
+        params_map = {
+            sid: dict((overrides.get(sid) or {}).get("params") or {})
+            for sid in all_ids
+        }
+        overrides_map = {sid: overrides.get(sid, {}) for sid in all_ids}
+        submit_progressive(
+            repo=repo, engine=engine, svc=svc, as_of=latest, asset_type="stock",
+            timeframe="1d", all_ids=all_ids, params_map=params_map,
+            overrides_map=overrides_map,
+        )
+        logger.info(
+            "策略缓存预热已提交: %d 个策略, as_of=%s", len(all_ids), latest,
+        )
+    except Exception:  # noqa: BLE001
+        logger.exception("策略缓存预热提交失败 (忽略)")

backend/tests/test_strategy_prewarm.py

@@ -0,0 +1,207 @@
+"""盘后策略缓存预热 — 判重语义、提交链路与开关行为。"""
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+from datetime import date
+from types import SimpleNamespace
+
+import pytest
+
+from app.config import settings
+from app.services import strategy_cache, strategy_prewarm, strategy_run_queue
+
+AS_OF = "2026-10-12"
+
+
+# ── Fakes (与 test_screener_run_all_progressive 同款风格) ──
+
+@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, ids: list[str]):
+        self._ids = ids
+        self.executed: list[str] = []
+
+    def list_strategies(self) -> list[dict]:
+        return [{"id": sid} for sid in self._ids]
+
+    def has(self, sid: str) -> bool:
+        return sid in self._ids
+
+    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)
+            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 _app(tmp_path, engine):
+    repo = SimpleNamespace(store=SimpleNamespace(data_dir=tmp_path))
+    state = SimpleNamespace(repo=repo, strategy_engine=engine, monitor_engine=None)
+    return SimpleNamespace(state=state)
+
+
+def _write_full_cache(tmp_path, ids, as_of=AS_OF):
+    strategy_cache.write_cache(
+        tmp_path, as_of,
+        {sid: {"total": 1, "as_of": as_of, "rows": []} for sid in ids},
+    )
+
+
+def _wait_executed(engine, want, timeout=8.0):
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        if set(want).issubset(set(engine.executed)):
+            return True
+        time.sleep(0.05)
+    return False
+
+
+# ── needs_prewarm 判重语义 (与前端 cacheCoversPool 同口径) ──
+
+def test_needs_prewarm_true_on_cold_cache(tmp_path):
+    engine = _FakeEngine(["a", "b"])
+    assert strategy_prewarm.needs_prewarm(tmp_path, engine, date.fromisoformat(AS_OF)) is True
+
+
+def test_needs_prewarm_false_when_fully_covered(tmp_path):
+    engine = _FakeEngine(["a", "b"])
+    _write_full_cache(tmp_path, ["a", "b"])
+    assert strategy_prewarm.needs_prewarm(tmp_path, engine, date.fromisoformat(AS_OF)) is False
+
+
+def test_needs_prewarm_true_on_partial_coverage(tmp_path):
+    engine = _FakeEngine(["a", "b", "c"])
+    _write_full_cache(tmp_path, ["a", "b"])
+    assert strategy_prewarm.needs_prewarm(tmp_path, engine, date.fromisoformat(AS_OF)) is True
+
+
+def test_needs_prewarm_true_when_cache_date_behind(tmp_path):
+    engine = _FakeEngine(["a"])
+    _write_full_cache(tmp_path, ["a"], as_of="2026-10-10")
+    assert strategy_prewarm.needs_prewarm(tmp_path, engine, date.fromisoformat(AS_OF)) is True
+
+
+def test_needs_prewarm_true_when_entry_as_of_mismatch(tmp_path):
+    # 顶层 as_of 一致但个别策略结果日期落后 (历史脏数据形态) → 该策略视为未覆盖
+    engine = _FakeEngine(["a", "b"])
+    strategy_cache.write_cache(
+        tmp_path, AS_OF,
+        {"a": {"total": 1, "as_of": AS_OF, "rows": []},
+         "b": {"total": 1, "as_of": "2026-10-10", "rows": []}},
+    )
+    assert strategy_prewarm.needs_prewarm(tmp_path, engine, date.fromisoformat(AS_OF)) is True
+
+
+def test_needs_prewarm_false_without_data_or_strategies(tmp_path):
+    engine = _FakeEngine(["a"])
+    assert strategy_prewarm.needs_prewarm(tmp_path, engine, None) is False
+    empty = _FakeEngine([])
+    assert strategy_prewarm.needs_prewarm(tmp_path, empty, date.fromisoformat(AS_OF)) is False
+
+
+def test_eligible_daily_ids_filters_research_and_mismatched_axes():
+    engine = SimpleNamespace(list_strategies=lambda: [
+        {"id": "ok"},
+        {"id": "research", "research_only": True},
+        {"id": "etf_only", "asset_types": ["etf"]},
+        {"id": "minute_only", "timeframes": ["1m"]},
+        {"id": "default_axes"},
+    ])
+    assert strategy_prewarm.eligible_daily_ids(engine) == ["ok", "default_axes"]
+
+
+# ── schedule_prewarm 提交链路 ──
+
+def test_schedule_prewarm_submits_and_fills_cache(monkeypatch, tmp_path):
+    from app.services import screener as screener_service
+    monkeypatch.setattr(screener_service, "ScreenerService", _FakeService)
+
+    engine = _FakeEngine(["a", "b"])
+    app = _app(tmp_path, engine)
+    strategy_prewarm.schedule_prewarm(app)
+
+    assert _wait_executed(engine, {"a", "b"})
+    results = (strategy_cache.read_cache(tmp_path) or {}).get("results") or {}
+    assert set(results) == {"a", "b"}
+    assert all(r.get("computed_at") for r in results.values())
+    # 耗时记录落盘 → 下次按耗时排序的既有机制同样适用于预热
+    assert set(strategy_run_queue.load_run_timings(tmp_path)) == {"a", "b"}
+
+
+def test_schedule_prewarm_skips_when_covered(monkeypatch, tmp_path):
+    from app.services import screener as screener_service
+    monkeypatch.setattr(screener_service, "ScreenerService", _FakeService)
+
+    engine = _FakeEngine(["a"])
+    _write_full_cache(tmp_path, ["a"])
+    strategy_prewarm.schedule_prewarm(_app(tmp_path, engine))
+
+    assert engine.executed == []
+    assert not (tmp_path / "user_data" / "strategy_run_timings.json").exists()
+
+
+def test_schedule_prewarm_disabled_by_config(monkeypatch, tmp_path):
+    from app.services import screener as screener_service
+    monkeypatch.setattr(screener_service, "ScreenerService", _FakeService)
+    monkeypatch.setattr(settings, "strategy_prewarm_enabled", False)
+
+    engine = _FakeEngine(["a"])
+    strategy_prewarm.schedule_prewarm(_app(tmp_path, engine))
+    assert engine.executed == []
+
+    # 脏状态隔离: 恢复开关后同一实例可正常提交
+    monkeypatch.setattr(settings, "strategy_prewarm_enabled", True)
+    strategy_prewarm.schedule_prewarm(_app(tmp_path, engine))
+    assert _wait_executed(engine, {"a"})
+
+
+def test_schedule_prewarm_swallows_errors(monkeypatch, tmp_path):
+    # 预热绝不反噬数据管道: 服务构造抛错也只静默返回
+    from app.services import screener as screener_service
+
+    def _boom(repo):
+        raise RuntimeError("repo unavailable")
+
+    monkeypatch.setattr(screener_service, "ScreenerService", _boom)
+    strategy_prewarm.schedule_prewarm(_app(tmp_path, _FakeEngine(["a"])))
+
+
+def test_prewarm_rides_same_execution_as_page_run(monkeypatch, tmp_path):
+    """预热与页面 run_all 同 key → 搭车同一执行, 不重复算。"""
+    from app.services import screener as screener_service
+    monkeypatch.setattr(screener_service, "ScreenerService", _FakeService)
+
+    engine = _FakeEngine(["a"])
+    app = _app(tmp_path, engine)
+    strategy_prewarm.schedule_prewarm(app)
+    # 页面在预热运行中发起同参数 run_all (等价 key) → 搭车
+    resp = strategy_prewarm.run_all_progressive(
+        repo=app.state.repo, engine=engine, svc=_FakeService(app.state.repo),
+        as_of=date.fromisoformat(AS_OF), asset_type="stock", timeframe="1d",
+        all_ids=["a"], params_map={}, overrides_map={}, first_return_s=3.0,
+    )
+    assert resp["complete"] is True or resp["results"].get("a", {}).get("total") == 1
+    # 只执行一次
+    assert engine.executed.count("a") == 1