fix(backtest): #517 盘中快照读失败 — 有界重试 + 400 指引

- factor run/run_batch 服务级重试 _SNAPSHOT_MAX_ATTEMPTS=3: 撞 EnrichedGenerationUnavailableError (盘中实时刷新换世代) sleep 1s 后重读新世代整体重跑, 耗尽原样上抛
- /factor/run、/factor/batch API 层捕获映射 400「行情数据正在更新(盘中实时刷新), 请稍后重试」(对齐 mining 既有模式), 不再裸 500

backend/app/api/backtest.py

@@ -14,6 +14,7 @@ from fastapi.responses import StreamingResponse
 from pydantic import BaseModel, Field
 
 from app.config import settings
+from app.enriched_generation import EnrichedGenerationUnavailableError
 from app.market_time import cn_today
 from app.services.backtest import (
     BacktestConfig,
@@ -181,7 +182,14 @@ def factor_run(req: FactorBacktestRequest, request: Request):
         slippage_bps=req.slippage_bps,
         asset_type=req.asset_type,
     )
-    result = svc.run(cfg)
+    try:
+        result = svc.run(cfg)
+    except EnrichedGenerationUnavailableError as exc:
+        # 盘中实时刷新换新 generation, 重试后仍撞上发布 (api/mining.py 同款映射)
+        raise HTTPException(
+            status_code=400,
+            detail="行情数据正在更新(盘中实时刷新), 请稍后重试",
+        ) from exc
     return asdict(result)
 
 
@@ -224,18 +232,24 @@ def factor_batch(req: FactorBatchRequest, request: Request):
         )
 
     svc = FactorBacktestService(_get_engine(request))
-    result = svc.run_batch(FactorBatchConfig(
-        factor_names=factor_names,
-        symbols=symbols,
-        start=start,
-        end=end,
-        n_groups=req.n_groups,
-        rebalance=req.rebalance,
-        weight=req.weight,
-        fees_pct=req.fees_pct,
-        slippage_bps=req.slippage_bps,
-        asset_type=req.asset_type,
-    ))
+    try:
+        result = svc.run_batch(FactorBatchConfig(
+            factor_names=factor_names,
+            symbols=symbols,
+            start=start,
+            end=end,
+            n_groups=req.n_groups,
+            rebalance=req.rebalance,
+            weight=req.weight,
+            fees_pct=req.fees_pct,
+            slippage_bps=req.slippage_bps,
+            asset_type=req.asset_type,
+        ))
+    except EnrichedGenerationUnavailableError as exc:
+        raise HTTPException(
+            status_code=400,
+            detail="行情数据正在更新(盘中实时刷新), 请稍后重试",
+        ) from exc
     return asdict(result)
 
 
@@ -340,7 +354,10 @@ class StrategyBacktestRequest(BaseModel):
     max_positions: int = 10
     max_exposure_pct: float = 1.0
     initial_capital: float = 1_000_000.0
-    position_sizing: Literal["equal", "score_weight"] = "equal"
+    position_sizing: Literal["equal", "score_weight", "policy"] = "equal"
+    # 自定义仓位策略: 注册表 id + 请求级参数覆盖 (不存在时 400, 不静默回退)
+    position_policy: str | None = None
+    position_policy_params: dict[str, float | int | bool | str] | None = None
     mode: Literal["position", "full"] = "position"
     holding_days: int = 5
     asset_type: str = "stock"
@@ -386,6 +403,16 @@ def strategy_run(req: StrategyBacktestRequest, request: Request):
     _guard_server_backtest_range(start, end)
     _guard_minute_strategy_backtest(request, req.strategy_id, start, req.asset_type)
 
+    if req.position_sizing == "policy":
+        from app.backtest import position_policies
+
+        if not req.position_policy or position_policies.get_policy(
+            settings.data_dir, req.position_policy
+        ) is None:
+            raise HTTPException(
+                status_code=400,
+                detail=f"仓位策略不存在或不可用: {req.position_policy!r}",
+            )
     cfg = StrategyBacktestConfig(
         strategy_id=req.strategy_id,
         symbols=req.symbols if req.symbols else None,
@@ -404,6 +431,8 @@ def strategy_run(req: StrategyBacktestRequest, request: Request):
         max_exposure_pct=req.max_exposure_pct,
         initial_capital=req.initial_capital,
         position_sizing=req.position_sizing,
+        position_policy=req.position_policy,
+        position_policy_params=req.position_policy_params,
         mode=req.mode,
         holding_days=req.holding_days,
         asset_type=req.asset_type,
@@ -483,8 +512,10 @@ def _make_job_key(
     asset_type: str = "stock",
     minute_fill: bool = False,
     regime_filter: str | None = None,
+    position_policy: str | None = None,
+    position_policy_params: str | None = None,
 ) -> str:
-    raw = f"{strategy_id}|{symbols}|{start}|{end}|{matching}|{entry_fill}|{exit_fill}|{fees_pct}|{slippage_bps}|{max_positions}|{max_exposure_pct}|{initial_capital}|{position_sizing}|{params}|{overrides}|{mode}|{holding_days}|{commission_pct}|{stamp_tax_pct}|{asset_type}|{minute_fill}|{regime_filter}"
+    raw = f"{strategy_id}|{symbols}|{start}|{end}|{matching}|{entry_fill}|{exit_fill}|{fees_pct}|{slippage_bps}|{max_positions}|{max_exposure_pct}|{initial_capital}|{position_sizing}|{params}|{overrides}|{mode}|{holding_days}|{commission_pct}|{stamp_tax_pct}|{asset_type}|{minute_fill}|{regime_filter}|{position_policy}|{position_policy_params}"
     return hashlib.md5(raw.encode()).hexdigest()[:12]
 
 
@@ -506,6 +537,8 @@ async def strategy_stream(
     max_exposure_pct: float = 1.0,
     initial_capital: float = 1_000_000.0,
     position_sizing: str = "equal",
+    position_policy: str | None = None,
+    position_policy_params: str | None = None,
     params: str | None = None,
     overrides: str | None = None,
     mode: str = "position",
@@ -556,6 +589,8 @@ async def strategy_stream(
         asset_type=asset_type,
         minute_fill=minute_fill,
         regime_filter=regime_filter,
+        position_policy=position_policy,
+        position_policy_params=position_policy_params,
     )
 
     _cleanup_stale_jobs()
@@ -612,6 +647,10 @@ async def strategy_stream(
                 max_exposure_pct=float(max_exposure_pct),
                 initial_capital=float(initial_capital),
                 position_sizing=position_sizing,
+                position_policy=position_policy,
+                position_policy_params=(
+                    json.loads(position_policy_params) if position_policy_params else None
+                ),
                 mode=mode,
                 holding_days=int(holding_days),
                 asset_type=asset_type,
@@ -722,6 +761,8 @@ async def strategy_cancel(request: Request):
         commission_pct=_get_opt_float("commission_pct"),
         stamp_tax_pct=_get_opt_float("stamp_tax_pct"),
         asset_type=_get("asset_type", "stock"),
+        position_policy=_get("position_policy") or None,
+        position_policy_params=_get("position_policy_params") or None,
     )
     # 持锁读任务表: 与 _cleanup_stale_jobs 的 pop、stream 的写入互斥
     with _jobs_lock:

backend/app/backtest/factor.py

@@ -161,6 +161,23 @@ class FactorBacktestService:
     def __init__(self, engine: BacktestEngine) -> None:
         self.engine = engine
 
+    # 盘中实时刷新 (默认 ~6s 一轮) 每轮换新 generation, 快照式读取撞上发布属常态
+    # (issue #517): 有界重读新世代整体重跑, 而不是 attempts=1 直接失败。
+    _SNAPSHOT_MAX_ATTEMPTS = 3
+    _SNAPSHOT_RETRY_SLEEP_S = 1.0
+
+    def _with_snapshot_retry(self, body):
+        from app.enriched_generation import EnrichedGenerationUnavailableError
+
+        for attempt in range(self._SNAPSHOT_MAX_ATTEMPTS):
+            if attempt:
+                time.sleep(self._SNAPSHOT_RETRY_SLEEP_S)
+            try:
+                return body()
+            except EnrichedGenerationUnavailableError:
+                if attempt + 1 >= self._SNAPSHOT_MAX_ATTEMPTS:
+                    raise
+
     def run(
         self,
         config: FactorConfig,
@@ -170,7 +187,9 @@ class FactorBacktestService:
         from app.services.heavy_job_limiter import shared_heavy_job_limiter
 
         with shared_heavy_job_limiter.slot("exclusive"):
-            return self._run(config, regime_by_date=regime_by_date)
+            return self._with_snapshot_retry(
+                lambda: self._run(config, regime_by_date=regime_by_date)
+            )
 
     def _run(
         self,
@@ -225,7 +244,9 @@ class FactorBacktestService:
         from app.services.heavy_job_limiter import shared_heavy_job_limiter
 
         with shared_heavy_job_limiter.slot("exclusive"):
-            return self._run_batch(config, regime_by_date=regime_by_date)
+            return self._with_snapshot_retry(
+                lambda: self._run_batch(config, regime_by_date=regime_by_date)
+            )
 
     def _run_batch(
         self,

backend/tests/test_factor_snapshot_retry.py

@@ -0,0 +1,109 @@
+"""#517 盘中快照读失败的有界重试与 API 降级。
+
+盘中实时刷新 (默认 ~6s) 每轮换新 generation; factor 快照路径此前 attempts=1
+读后校验失败直接抛、API 裸 500。锁定三层行为:
+① service 层有界重试 (撞发布 → 重读新世代重跑, 耗尽上抛);
+② API 层把耗尽的 EnrichedGenerationUnavailableError 映射为 400 中文指引。
+"""
+from __future__ import annotations
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from app.api import backtest as backtest_api
+from app.backtest.factor import FactorBacktestService
+from app.enriched_generation import EnrichedGenerationUnavailableError
+from app.services import preferences
+
+
+class _FakeEngine:
+    """只做占位; _run 由 monkeypatch 接管。"""
+
+
+@pytest.fixture(autouse=True)
+def _fast_retry(monkeypatch):
+    monkeypatch.setattr(FactorBacktestService, "_SNAPSHOT_RETRY_SLEEP_S", 0.0)
+
+
+def test_run_retries_once_on_generation_change(monkeypatch):
+    svc = FactorBacktestService(_FakeEngine())
+    calls = []
+
+    def _fake_run(config, *, regime_by_date=None):
+        calls.append(1)
+        if len(calls) == 1:
+            raise EnrichedGenerationUnavailableError("changed while reading")
+        return "ok"
+
+    monkeypatch.setattr(svc, "_run", _fake_run)
+    assert svc.run(object()) == "ok"
+    assert len(calls) == 2
+
+
+def test_run_raises_after_attempts_exhausted(monkeypatch):
+    svc = FactorBacktestService(_FakeEngine())
+    calls = []
+
+    def _fake_run(config, *, regime_by_date=None):
+        calls.append(1)
+        raise EnrichedGenerationUnavailableError("still changing")
+
+    monkeypatch.setattr(svc, "_run", _fake_run)
+    with pytest.raises(EnrichedGenerationUnavailableError):
+        svc.run(object())
+    assert len(calls) == FactorBacktestService._SNAPSHOT_MAX_ATTEMPTS
+
+
+def test_run_batch_retries_on_generation_change(monkeypatch):
+    svc = FactorBacktestService(_FakeEngine())
+    calls = []
+
+    def _fake_batch(config, *, regime_by_date=None):
+        calls.append(1)
+        if len(calls) < 3:
+            raise EnrichedGenerationUnavailableError("changed")
+        return "ok"
+
+    monkeypatch.setattr(svc, "_run_batch", _fake_batch)
+    assert svc.run_batch(object()) == "ok"
+    assert len(calls) == 3
+
+
+def _client(monkeypatch, tmp_path) -> TestClient:
+    from app.config import settings
+
+    monkeypatch.setattr(settings, "data_dir", tmp_path)
+    monkeypatch.setattr(preferences, "load", lambda: {})
+    app = FastAPI()
+    app.include_router(backtest_api.router)
+    app.state.repo = None
+    return TestClient(app)
+
+
+def test_factor_run_api_maps_generation_error_to_400(monkeypatch, tmp_path):
+    client = _client(monkeypatch, tmp_path)
+
+    def _boom(self, cfg):
+        raise EnrichedGenerationUnavailableError("exhausted")
+
+    monkeypatch.setattr(FactorBacktestService, "run", _boom)
+    r = client.post("/api/backtest/factor/run", json={
+        "factor_name": "momentum_5d", "start": "2026-01-01", "end": "2026-06-30",
+    })
+    assert r.status_code == 400
+    assert "正在更新" in r.json()["detail"]
+
+
+def test_factor_batch_api_maps_generation_error_to_400(monkeypatch, tmp_path):
+    client = _client(monkeypatch, tmp_path)
+
+    def _boom(self, cfg):
+        raise EnrichedGenerationUnavailableError("exhausted")
+
+    monkeypatch.setattr(FactorBacktestService, "run_batch", _boom)
+    r = client.post("/api/backtest/factor/batch", json={
+        "factor_names": ["momentum_5d"], "start": "2026-01-01", "end": "2026-06-30",
+    })
+    assert r.status_code == 400
+    assert "正在更新" in r.json()["detail"]