diff --git a/backend/app/api/backtest.py b/backend/app/api/backtest.py index dada9725..a746af93 100644 --- a/backend/app/api/backtest.py +++ b/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: diff --git a/backend/app/backtest/factor.py b/backend/app/backtest/factor.py index fd9acc7e..2b51d12c 100644 --- a/backend/app/backtest/factor.py +++ b/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, diff --git a/backend/tests/test_factor_snapshot_retry.py b/backend/tests/test_factor_snapshot_retry.py new file mode 100644 index 00000000..993a2c9f --- /dev/null +++ b/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"]