diff --git a/backend/app/api/mining.py b/backend/app/api/mining.py index b028f0e..c80777f 100644 --- a/backend/app/api/mining.py +++ b/backend/app/api/mining.py @@ -20,6 +20,7 @@ from app.backtest.mining import ( MAX_FINALISTS, evaluate_candidate_gate, ) +from app.enriched_generation import EnrichedGenerationUnavailableError from app.services import preferences from app.services.mining_jobs import ( RUN_STATUSES, @@ -208,6 +209,13 @@ def start_run(payload: MiningStartRequest, request: Request) -> dict[str, Any]: return projected except (MiningRunValidationError, ValueError) as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc + except EnrichedGenerationUnavailableError as exc: + # build_data_fingerprint 读世代时撞上正在进行的 enriched 发布 (如盘后更新): + # 映射为 400 带指引, 而不是 500 英文堆栈。 + raise HTTPException( + status_code=400, + detail="行情数据正在更新(enriched 发布中), 请等数据更新完成后再开始挖掘", + ) from exc except MiningRunStoreError as exc: raise HTTPException(status_code=500, detail="failed to persist mining run") from exc diff --git a/backend/app/backtest/mining_runtime.py b/backend/app/backtest/mining_runtime.py index a04759d..b05dc67 100644 --- a/backend/app/backtest/mining_runtime.py +++ b/backend/app/backtest/mining_runtime.py @@ -53,6 +53,10 @@ from app.backtest.strategy import ( _merge_resolved_feature_plans, build_matrix_cache_profile, ) +from app.enriched_generation import ( + EnrichedGenerationUnavailableError, + enriched_publication_incomplete, +) from app.services.mining_jobs import MiningRunStore from app.services.mining_preflight import enriched_partition_dates from app.services.mining_schedule import MINING_ALGORITHM_VERSION @@ -80,6 +84,11 @@ _REGIME_FILTERS: dict[str, dict[str, list[str]]] = { "weak": {"states": ["lean_weak", "weak"]}, } +# 面板/撮合矩阵读取期间 enriched 世代被并发发布打断时, 允许用新世代整体重读的次数 +# 与发布未完成时的等待秒数; 超过后以带指引的错误终止运行。 +_SNAPSHOT_MAX_ATTEMPTS = 3 +_SNAPSHOT_PUBLISH_WAIT_SECONDS = 5.0 + class MiningRuntimeCancelledError(RuntimeError): pass @@ -462,7 +471,6 @@ def run_mining_runtime( emit({"phase": "panel", "label": "加载因子面板", "done": 0, "total": 1}) start_phase() _raise_if_cancelled(cancel_check) - panel_started = time.perf_counter() factor_service = FactorBacktestService(service.engine) factor_config = FactorBatchConfig( factor_names=list(request.factor_names), @@ -474,80 +482,106 @@ def run_mining_runtime( stamp_tax_pct=request.stamp_tax_pct, slippage_bps=request.slippage_bps, ) - generation = factor_service._data_generation(request.asset_type) - if generation != expected_generation: - raise ValueError( - "mining data generation changed after the run was queued" - ) - source_panel = _load_compact_factor_panel( - factor_service, - factor_config, - request.factor_names, - expected_generation=generation, - cancel_check=cancel_check, - ) - if source_panel.is_empty(): - raise ValueError("mining date range contains no enriched data") - service.engine.clear_panel_cache() - trading_dates = enriched_partition_dates( - data_dir, - request.asset_type, - request.start, - request.end, - ) - factor_service._assert_data_generation(request.asset_type, generation) - panel = attach_single_forward_return( - source_panel, - start=request.start, - end=request.end, - horizon=request.forward_horizon, - trading_dates=trading_dates, - factor_names=request.factor_names, - target_column=request.mining_request.target_column, - assume_unique_symbol_date=True, - ) - del source_panel - if panel.is_empty(): - raise ValueError("mining panel contains no valid price rows") - phase_ms: dict[str, float] = { - "panel": round((time.perf_counter() - panel_started) * 1000.0, 3) - } - finish_phase("panel") - emit({ - "phase": "panel", - "label": "因子面板已准备", - "done": 1, - "total": 1, - "rows": panel.height, - "factors": len(request.factor_names), - }) + phase_ms: dict[str, float] = {} + panel: pl.DataFrame | None = None + base_market: np.ndarray | None = None + generation: str | None = None + for attempt in range(_SNAPSHOT_MAX_ATTEMPTS): + if attempt: + # 读取期间 enriched 发布了新世代, 面板可能新旧混合: 丢弃本轮, + # 用新世代整体重读。排队指纹只在首轮校验; 数据在运行中前进, + # 重试跟随新世代属于预期 (等价于"更新后立刻重跑")。 + emit({"phase": "panel", "label": "数据已更新, 重新读取快照", "done": 0, "total": 1}) + if enriched_publication_incomplete(data_dir): + time.sleep(_SNAPSHOT_PUBLISH_WAIT_SECONDS) + service.engine.clear_panel_cache() + panel_started = time.perf_counter() + try: + generation = factor_service._data_generation(request.asset_type) + if attempt == 0 and generation != expected_generation: + raise ValueError( + "mining data generation changed after the run was queued" + ) + source_panel = _load_compact_factor_panel( + factor_service, + factor_config, + request.factor_names, + expected_generation=generation, + cancel_check=cancel_check, + ) + if source_panel.is_empty(): + raise ValueError("mining date range contains no enriched data") + service.engine.clear_panel_cache() + trading_dates = enriched_partition_dates( + data_dir, + request.asset_type, + request.start, + request.end, + ) + factor_service._assert_data_generation(request.asset_type, generation) + panel = attach_single_forward_return( + source_panel, + start=request.start, + end=request.end, + horizon=request.forward_horizon, + trading_dates=trading_dates, + factor_names=request.factor_names, + target_column=request.mining_request.target_column, + assume_unique_symbol_date=True, + ) + del source_panel + if panel.is_empty(): + raise ValueError("mining panel contains no valid price rows") + phase_ms["panel"] = round( + (time.perf_counter() - panel_started) * 1000.0, 3 + ) + finish_phase("panel") + emit({ + "phase": "panel", + "label": "因子面板已准备", + "done": 1, + "total": 1, + "rows": panel.height, + "factors": len(request.factor_names), + }) - if request.require_regime: - emit({"phase": "panel", "label": "校验市场环境数据", "done": 0, "total": 1}) - _validate_regime_availability(panel, request, data_dir) + if request.require_regime: + emit({"phase": "panel", "label": "校验市场环境数据", "done": 0, "total": 1}) + _validate_regime_availability(panel, request, data_dir) - _raise_if_cancelled(cancel_check) - start_phase() - matrix_started = time.perf_counter() - emit({"phase": "matrix", "label": "准备共享撮合矩阵", "done": 0, "total": 1}) - base_market = _prepare_base_market( - service, - strategy_engine, - data_dir, - request, - expected_generation=generation, - cancel_check=cancel_check, - ) - factor_service._assert_data_generation(request.asset_type, generation) - phase_ms["matrix"] = round((time.perf_counter() - matrix_started) * 1000.0, 3) - finish_phase("matrix") - emit({ - "phase": "matrix", - "label": "共享撮合矩阵已准备", - "done": 1, - "total": 1, - "matrix_bytes": base_market.nbytes, - }) + _raise_if_cancelled(cancel_check) + start_phase() + matrix_started = time.perf_counter() + emit({"phase": "matrix", "label": "准备共享撮合矩阵", "done": 0, "total": 1}) + base_market = _prepare_base_market( + service, + strategy_engine, + data_dir, + request, + expected_generation=generation, + cancel_check=cancel_check, + ) + factor_service._assert_data_generation(request.asset_type, generation) + phase_ms["matrix"] = round( + (time.perf_counter() - matrix_started) * 1000.0, 3 + ) + finish_phase("matrix") + emit({ + "phase": "matrix", + "label": "共享撮合矩阵已准备", + "done": 1, + "total": 1, + "matrix_bytes": base_market.nbytes, + }) + break + except EnrichedGenerationUnavailableError as exc: + if attempt + 1 >= _SNAPSHOT_MAX_ATTEMPTS: + raise ValueError( + "行情数据正在更新: 挖掘读取期间 enriched 数据世代反复变化, " + "重试后仍拿不到稳定快照; 请等数据更新完成后再开始挖掘" + ) from exc + if panel is None or base_market is None or generation is None: + raise RuntimeError("mining data snapshot did not settle") metric_provider = TrainingMetricProvider(request.mining_request.target_column) evaluator = MatcherCandidateEvaluator( diff --git a/backend/tests/backtest/test_worker_process.py b/backend/tests/backtest/test_worker_process.py index 18cd68f..d45e32b 100644 --- a/backend/tests/backtest/test_worker_process.py +++ b/backend/tests/backtest/test_worker_process.py @@ -480,3 +480,126 @@ def test_spawn_walkforward_skips_folds_before_available_matrix_data(tmp_path): assert result["skipped"][0]["reason"] == "训练区间无可用行情数据" assert result["n_folds"] == 3 assert result["worker"]["worker_exitcode"] == 0 + + +def _mining_runtime_services(data_dir): + """按 worker._worker_entry 的方式在进程内构造挖掘运行时依赖 (便于 monkeypatch)。""" + from app.backtest.engine import BacktestEngine + from app.backtest.strategy import StrategyBacktestService + from app.strategy import config as strategy_config + from app.strategy.engine import StrategyEngine + from app.tickflow.repository import DataStore, KlineRepository + + store = DataStore(data_dir) + repo = KlineRepository(store) + strategy_engine = StrategyEngine( + strategy_dirs=worker_module._strategy_dirs(data_dir), + override_loader=lambda sid: strategy_config.load_override(data_dir, sid), + ) + service = StrategyBacktestService(BacktestEngine(repo), strategy_engine) + return service, strategy_engine + + +def _queue_mining_run(data_dir, start: date, run_id: str) -> dict: + store = MiningRunStore(data_dir) + manifest = store.create( + { + "factor_names": ["turnover_rate"], + "strategy_ids": [], + "symbols": None, + "asset_type": "stock", + "start": (start - timedelta(days=7)).isoformat(), + "end": (start + timedelta(days=225)).isoformat(), + "budget_profile": "exploratory", + "forward_horizon": 1, + "commission_pct": 0.0, + "stamp_tax_pct": 0.0, + "slippage_bps": 0.0, + "correlation_threshold": 0.75, + "max_combination_factors": 1, + "beam_width": 2, + "max_finalists": 2, + "require_regime": False, + }, + {"generation": get_enriched_generation(data_dir, "stock")}, + run_id=run_id, + ) + return { + "run_id": manifest["run_id"], + "request": manifest["request"], + "data_fingerprint": manifest["data_fingerprint"], + "source": "manual", + } + + +def _patch_generation_drift(monkeypatch, data_dir, *, always: bool) -> dict: + """让世代校验第一次(或每次)调用前先 bump 再抛错, 模拟读取期间并发发布完成。""" + from app.backtest.engine import BacktestEngine + from app.enriched_generation import EnrichedGenerationUnavailableError + + original = BacktestEngine.assert_data_generation + state = {"drifts": 0} + + def drift(engine_self, asset_type, expected): + if expected is not None and (always or state["drifts"] == 0): + state["drifts"] += 1 + bump_enriched_generation(data_dir, asset_type) + raise EnrichedGenerationUnavailableError( + "simulated concurrent publication" + ) + return original(engine_self, asset_type, expected) + + monkeypatch.setattr(BacktestEngine, "assert_data_generation", drift) + return state + + +def test_mining_rereads_snapshot_when_generation_commits_mid_read( + monkeypatch, tmp_path +): + start = date(2023, 1, 2) + data_dir = tmp_path / "data" + _write_mining_market_data(data_dir, start) + payload = _queue_mining_run(data_dir, start, "midread_drift_mining") + service, strategy_engine = _mining_runtime_services(data_dir) + state = _patch_generation_drift(monkeypatch, data_dir, always=False) + + from app.backtest.mining_runtime import run_mining_runtime + + events: list[dict] = [] + result = run_mining_runtime( + payload, + data_dir=data_dir, + service=service, + strategy_engine=strategy_engine, + progress_cb=events.append, + cancel_check=None, + ) + + assert result["status"] in {"succeeded", "succeeded_with_budget_exhausted"} + assert state["drifts"] == 1 + assert any( + event.get("label") == "数据已更新, 重新读取快照" for event in events + ) + + +def test_mining_fails_with_guidance_when_generation_keeps_drifting( + monkeypatch, tmp_path +): + start = date(2023, 1, 2) + data_dir = tmp_path / "data" + _write_mining_market_data(data_dir, start) + payload = _queue_mining_run(data_dir, start, "endless_drift_mining") + service, strategy_engine = _mining_runtime_services(data_dir) + _patch_generation_drift(monkeypatch, data_dir, always=True) + + from app.backtest.mining_runtime import run_mining_runtime + + with pytest.raises(ValueError, match="稳定快照"): + run_mining_runtime( + payload, + data_dir=data_dir, + service=service, + strategy_engine=strategy_engine, + progress_cb=None, + cancel_check=None, + ) diff --git a/backend/tests/test_mining_api.py b/backend/tests/test_mining_api.py index bd49847..160ed31 100644 --- a/backend/tests/test_mining_api.py +++ b/backend/tests/test_mining_api.py @@ -13,6 +13,7 @@ from fastapi.testclient import TestClient from app.api.mining import router from app.backtest.mining import compute_candidate_signature +from app.enriched_generation import EnrichedGenerationUnavailableError from app.services.mining_jobs import MiningRunStore from app.strategy.engine import StrategyEngine @@ -604,3 +605,33 @@ def test_config_patch_merges_current_values(tmp_path, monkeypatch): assert response.status_code == 200 assert saved == [(True, 4, "balanced")] assert client.patch("/api/backtest/mining/config", json={}).status_code == 400 + + +class _PublishingRepo(_Repo): + """模拟 enriched 发布进行中: 世代读取抛 EnrichedGenerationUnavailableError。""" + + @staticmethod + def get_matrix_data_generation(asset_type="stock"): + raise EnrichedGenerationUnavailableError( + "enriched data is being published; retry after the update finishes" + ) + + +def test_start_returns_400_with_guidance_while_enriched_publication_active( + tmp_path, +): + _write_enriched_dates(tmp_path, 219, first=date(2022, 8, 15)) + app = FastAPI() + app.include_router(router) + app.state.repo = _PublishingRepo(tmp_path) + app.state.mining_manager = _Manager(tmp_path) + app.state.strategy_engine = SimpleNamespace() + client = TestClient(app) + + response = client.post( + "/api/backtest/mining/runs", + json={"factor_names": ["turnover_rate"], "budget_profile": "exploratory"}, + ) + + assert response.status_code == 400 + assert "数据更新" in response.json()["detail"]