Merge remote-tracking branch 'origin/main' into feat/capability-routing

# Conflicts:
#	README.md
#	backend/app/backtest/worker.py

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
 

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,76 +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)
+    phase_ms: dict[str, float] = {}
-    if generation != expected_generation:
+    panel: pl.DataFrame | None = None
-        raise ValueError(
+    base_market: np.ndarray | None = None
-            "mining data generation changed after the run was queued"
+    generation: str | None = None
-        )
+    for attempt in range(_SNAPSHOT_MAX_ATTEMPTS):
-    source_panel = _load_compact_factor_panel(
+        if attempt:
-        factor_service,
+            # 读取期间 enriched 发布了新世代, 面板可能新旧混合: 丢弃本轮,
-        factor_config,
+            # 用新世代整体重读。排队指纹只在首轮校验; 数据在运行中前进,
-        request.factor_names,
+            # 重试跟随新世代属于预期 (等价于"更新后立刻重跑")。
-        expected_generation=generation,
+            emit({"phase": "panel", "label": "数据已更新, 重新读取快照", "done": 0, "total": 1})
-        cancel_check=cancel_check,
+            if enriched_publication_incomplete(data_dir):
-    )
+                time.sleep(_SNAPSHOT_PUBLISH_WAIT_SECONDS)
-    if source_panel.is_empty():
+            service.engine.clear_panel_cache()
-        raise ValueError("mining date range contains no enriched data")
+        panel_started = time.perf_counter()
-    service.engine.clear_panel_cache()
+        try:
-    trading_dates = enriched_partition_dates(
+            generation = factor_service._data_generation(request.asset_type)
-        data_dir,
+            if attempt == 0 and generation != expected_generation:
-        request.asset_type,
+                raise ValueError(
-        request.start,
+                    "mining data generation changed after the run was queued"
-        request.end,
+                )
-    )
+            source_panel = _load_compact_factor_panel(
-    factor_service._assert_data_generation(request.asset_type, generation)
+                factor_service,
-    panel = attach_single_forward_return(
+                factor_config,
-        source_panel,
+                request.factor_names,
-        start=request.start,
+                expected_generation=generation,
-        end=request.end,
+                cancel_check=cancel_check,
-        horizon=request.forward_horizon,
+            )
-        trading_dates=trading_dates,
+            if source_panel.is_empty():
-        factor_names=request.factor_names,
+                raise ValueError("mining date range contains no enriched data")
-        target_column=request.mining_request.target_column,
+            service.engine.clear_panel_cache()
-        assume_unique_symbol_date=True,
+            trading_dates = enriched_partition_dates(
-    )
+                data_dir,
-    del source_panel
+                request.asset_type,
-    if panel.is_empty():
+                request.start,
-        raise ValueError("mining panel contains no valid price rows")
+                request.end,
-    phase_ms: dict[str, float] = {
+            )
-        "panel": round((time.perf_counter() - panel_started) * 1000.0, 3)
+            factor_service._assert_data_generation(request.asset_type, generation)
-    }
+            panel = attach_single_forward_return(
-    finish_phase("panel")
+                source_panel,
-    emit({
+                start=request.start,
-        "phase": "panel",
+                end=request.end,
-        "label": "因子面板已准备",
+                horizon=request.forward_horizon,
-        "done": 1,
+                trading_dates=trading_dates,
-        "total": 1,
+                factor_names=request.factor_names,
-        "rows": panel.height,
+                target_column=request.mining_request.target_column,
-        "factors": len(request.factor_names),
+                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),
+            })
 
-    _raise_if_cancelled(cancel_check)
+            if request.require_regime:
-    start_phase()
+                emit({"phase": "panel", "label": "校验市场环境数据", "done": 0, "total": 1})
-    matrix_started = time.perf_counter()
+                _validate_regime_availability(panel, request, data_dir)
-    emit({"phase": "matrix", "label": "准备共享撮合矩阵", "done": 0, "total": 1})
+
-    base_market = _prepare_base_market(
+            _raise_if_cancelled(cancel_check)
-        service,
+            start_phase()
-        strategy_engine,
+            matrix_started = time.perf_counter()
-        data_dir,
+            emit({"phase": "matrix", "label": "准备共享撮合矩阵", "done": 0, "total": 1})
-        request,
+            base_market = _prepare_base_market(
-        expected_generation=generation,
+                service,
-        cancel_check=cancel_check,
+                strategy_engine,
-    )
+                data_dir,
-    factor_service._assert_data_generation(request.asset_type, generation)
+                request,
-    phase_ms["matrix"] = round((time.perf_counter() - matrix_started) * 1000.0, 3)
+                expected_generation=generation,
-    finish_phase("matrix")
+                cancel_check=cancel_check,
-    emit({
+            )
-        "phase": "matrix",
+            factor_service._assert_data_generation(request.asset_type, generation)
-        "label": "共享撮合矩阵已准备",
+            phase_ms["matrix"] = round(
-        "done": 1,
+                (time.perf_counter() - matrix_started) * 1000.0, 3
-        "total": 1,
+            )
-        "matrix_bytes": base_market.nbytes,
+            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(
@@ -1309,6 +1347,32 @@ def _fold_row(
     }
 
 
+def _validate_regime_availability(
+    panel: pl.DataFrame,
+    request: RuntimeRequest,
+    data_dir: Path,
+) -> None:
+    """搜索开始前 fail-fast 校验市场环境数据。
+
+    环境分组评估 (strong/range/weak) 依赖 T-1 市场环境序列; 数据为空或覆盖不足时,
+    若拖到 artifacts 阶段才在 _regime_date_count 中抛错, 整轮嵌套搜索的计算全部浪费。
+    这里用与 artifacts 相同的 fold 口径 (panel 标签) 预先构建一次 mask:
+    required 区间取所有外层 fold 测试窗的并集, 覆盖后续每个 fold 单独调用的要求,
+    任何 ValueError (数据为空 / 覆盖不完整) 立即带指引消息终止运行。
+    """
+    labels = _date_labels(panel)
+    nested = generate_nested_folds(labels, request.mining_request.validation)
+    if not nested:
+        return
+    StrategyBacktestService._build_regime_mask(
+        labels,
+        _REGIME_FILTERS["strong"],
+        data_dir,
+        required_start=date.fromisoformat(nested[0].outer.test_start),
+        required_end=date.fromisoformat(nested[-1].outer.test_end),
+    )
+
+
 def _regime_date_count(
     panel: pl.DataFrame,
     validation_fold,

backend/app/backtest/worker.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import multiprocessing as mp
 import os
 import queue
@@ -17,6 +18,8 @@ from typing import Any
 
 import psutil
 
+logger = logging.getLogger(__name__)
+
 
 class BacktestWorkerError(RuntimeError):
     """Raised when a spawned worker fails before returning a task result."""
@@ -254,12 +257,14 @@ def _worker_entry(task: dict[str, Any], event_queue, cancel_event) -> None:
         if store is not None:
             with suppress(Exception):
                 store.db.close()
-        # 保证结果消息在进程退出前完整刷入管道: put 只是入队,
+        # 终态消息已入队: 显式冲刷队列后立即退出。put 只是入队, 实际写管道的
-        # 实际写管道的是后台 feeder 线程; 不 join 的话主线程先退出,
+        # 是后台 feeder 线程, close+join_thread 保证消息完整落管 (否则父进程误判
-        # feeder 随进程销毁, 消息尾部丢失 → 父进程误判 "exited without result"。
+        # "exited without result"); 大数据量任务再跳过解释器 teardown (GC、DuckDB
+        # 线程 join、DLL 卸载), 否则收尾可达数十秒, 撞上父进程 10s 退出预算。
         with suppress(Exception):
             event_queue.close()
             event_queue.join_thread()
+        os._exit(0)
 
 
 def run_worker_task(
@@ -332,10 +337,19 @@ def run_worker_task(
                     failure = message
 
         process.join(timeout=10.0)
+        worker_exit_forcibly = False
         if process.is_alive():
+            # 终态消息 (result/error) 已完整送达, 子进程只是退出收尾慢:
+            # 强制结束并继续走结果/错误处理, 不把已送达的成功结果当失败丢弃。
             process.terminate()
             process.join(timeout=5.0)
-            raise BacktestWorkerError("backtest worker returned but did not exit within 10 seconds")
+            worker_exit_forcibly = True
+            logger.warning(
+                "%s worker delivered its terminal message but did not exit within "
+                "10s; terminated forcibly (exitcode=%s)",
+                task["kind"],
+                process.exitcode,
+            )
         if failure is not None:
             raise BacktestWorkerError(
                 f"{failure.get('message', 'worker failed')}\n{failure.get('traceback', '')}".rstrip()
@@ -350,6 +364,7 @@ def run_worker_task(
             "parent_rss_before_bytes": parent_rss_before,
             "parent_rss_after_worker_exit_bytes": _rss_bytes(),
             "worker_exitcode": process.exitcode,
+            "worker_exit_forcibly": worker_exit_forcibly,
         }
         kind = task["kind"]
         if kind == "backtest":

backend/tests/backtest/test_mining_runtime.py

@@ -6,7 +6,11 @@ from types import SimpleNamespace
 import polars as pl
 import pytest
 
-from app.backtest.mining import MiningCandidate
+from app.backtest.mining import (
+    MiningCandidate,
+    NestedValidationConfig,
+    generate_nested_folds,
+)
 from app.backtest.mining_runtime import (
     TrainingMetricProvider,
     _decode_runtime_request,
@@ -14,6 +18,7 @@ from app.backtest.mining_runtime import (
     _prepare_base_market,
     _rank_artifact_candidates,
     _regime_date_count,
+    _validate_regime_availability,
     attach_single_forward_return,
 )
 from app.services import regime_builder
@@ -312,3 +317,76 @@ def test_regime_date_count_uses_t_minus_one_market_labels(tmp_path) -> None:
 
     assert _regime_date_count(panel, fold, "strong", tmp_path) == 2
     assert _regime_date_count(panel, fold, "weak", tmp_path) == 1
+
+
+def _small_validation() -> NestedValidationConfig:
+    return NestedValidationConfig(
+        outer_train_bars=10,
+        outer_test_bars=3,
+        outer_step_bars=5,
+        inner_train_bars=5,
+        inner_test_bars=2,
+        inner_step_bars=3,
+        purge_bars=1,
+        embargo_bars=1,
+        min_train_bars=3,
+    )
+
+
+def _regime_panel(n: int = 20) -> tuple[list[date], pl.DataFrame]:
+    labels = [date(2024, 1, 2) + timedelta(days=offset) for offset in range(n)]
+    return labels, pl.DataFrame({"date": labels})
+
+
+def _upsert_regime(tmp_path, labels, states: list[str]) -> None:
+    regime_builder.upsert_regime_history(tmp_path, pl.DataFrame({
+        "date": labels,
+        "state": states,
+        "score": [50] * len(labels),
+    }))
+
+
+def test_validate_regime_availability_fails_fast_when_regime_data_missing(
+    tmp_path,
+) -> None:
+    labels, panel = _regime_panel()
+    request = SimpleNamespace(
+        mining_request=SimpleNamespace(validation=_small_validation()),
+    )
+
+    with pytest.raises(ValueError, match="市场环境数据为空"):
+        _validate_regime_availability(panel, request, tmp_path)
+
+
+def test_validate_regime_availability_passes_when_regime_covers_fold_windows(
+    tmp_path,
+) -> None:
+    labels, panel = _regime_panel()
+    _upsert_regime(
+        tmp_path,
+        labels[:-1],
+        ["range"] * (len(labels) - 1),
+    )
+    request = SimpleNamespace(
+        mining_request=SimpleNamespace(validation=_small_validation()),
+    )
+
+    _validate_regime_availability(panel, request, tmp_path)
+
+
+def test_validate_regime_availability_reports_coverage_gaps_in_fold_windows(
+    tmp_path,
+) -> None:
+    labels, panel = _regime_panel()
+    nested = generate_nested_folds(
+        [value.isoformat() for value in labels], _small_validation()
+    )
+    gap_date = date.fromisoformat(nested[0].outer.test_start) + timedelta(days=1)
+    covered = [value for value in labels if value != gap_date]
+    _upsert_regime(tmp_path, covered, ["range"] * len(covered))
+    request = SimpleNamespace(
+        mining_request=SimpleNamespace(validation=_small_validation()),
+    )
+
+    with pytest.raises(ValueError, match="市场环境数据覆盖不完整"):
+        _validate_regime_availability(panel, request, tmp_path)

backend/tests/backtest/test_worker_process.py

@@ -448,6 +448,62 @@ def test_worker_terminates_child_after_cancel_grace(monkeypatch, tmp_path):
     assert process.exitcode == -15
 
 
+def test_worker_accepts_delivered_result_when_child_exit_is_slow(monkeypatch, tmp_path):
+    """终态消息已送达但子进程退出收尾超时: 应强杀后采纳结果, 而非丢弃报错。"""
+
+    class FakeQueue:
+        def __init__(self):
+            self._messages = [{"type": "result", "payload": {"status": "ok"}}]
+
+        def get(self, timeout):
+            if self._messages:
+                return self._messages.pop(0)
+            raise queue.Empty
+
+        def close(self):
+            pass
+
+        def join_thread(self):
+            pass
+
+    class FakeEvent:
+        def set(self):
+            pass
+
+    class FakeProcess:
+        def __init__(self):
+            self.alive = True
+            self.exitcode = None
+
+        def start(self):
+            pass
+
+        def is_alive(self):
+            return self.alive
+
+        def join(self, timeout=None):
+            pass
+
+        def terminate(self):
+            self.alive = False
+            self.exitcode = -15
+
+    process = FakeProcess()
+    context = SimpleNamespace(
+        Queue=FakeQueue,
+        Event=FakeEvent,
+        Process=lambda **_kwargs: process,
+    )
+    monkeypatch.setattr(worker_module.mp, "get_context", lambda _method: context)
+
+    result = run_worker_task({"kind": "mining", "data_dir": str(tmp_path), "config": {}})
+
+    assert result["status"] == "ok"
+    assert result["worker"]["worker_exit_forcibly"] is True
+    assert result["worker"]["worker_exitcode"] == -15
+    assert process.exitcode == -15
+
+
 def test_spawn_walkforward_skips_folds_before_available_matrix_data(tmp_path):
     configured_start = date(2024, 1, 1)
     market_start = configured_start + timedelta(days=4)
@@ -480,3 +536,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,
+        )

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"]

frontend/src/pages/backtest/MiningWorkbench.tsx

@@ -1,6 +1,6 @@
 import { useEffect, useMemo, useRef, useState } from 'react'
 import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'
-import { useSearchParams } from 'react-router-dom'
+import { Link, useSearchParams } from 'react-router-dom'
 import {
   AlertTriangle,
   CheckCircle2,
@@ -303,6 +303,7 @@ export function MiningWorkbench() {
     enabled: validDateRange,
     staleTime: 30_000,
   })
+  const regimeLatestQuery = useQuery({ queryKey: QK.regimeLatest, queryFn: api.regimeLatest, staleTime: 60_000 })
   const configQuery = useQuery({ queryKey: QK.miningConfig, queryFn: api.miningConfig })
 
   useEffect(() => {
@@ -454,6 +455,18 @@ export function MiningWorkbench() {
       )
       return
     }
+    if (regimeLatestQuery.isPending || regimeLatestQuery.isFetching) {
+      toast('正在核验市场环境数据，请稍候', 'error')
+      return
+    }
+    if (regimeLatestQuery.isError || !regimeLatestQuery.data) {
+      toast('无法核验市场环境数据，请检查数据状态后重试', 'error')
+      return
+    }
+    if (!regimeLatestQuery.data.row) {
+      toast('尚未计算市场环境数据：挖掘含市场环境分组评估，请先在数据页完成市场环境计算', 'error')
+      return
+    }
     const commissionBps = parseBoundedNumber(draft.commissionBps, '佣金', 0, 500)
     const stampTaxBps = parseBoundedNumber(draft.stampTaxBps, '印花税', 0, 500)
     const slippageBps = parseBoundedNumber(draft.slippageBps, '滑点', 0, 1000)
@@ -567,6 +580,12 @@ export function MiningWorkbench() {
             )}
           </section>
 
+          {regimeLatestQuery.data && !regimeLatestQuery.data.row && (
+            <Link to="/data" className="block rounded-btn border border-warning/40 bg-warning/5 px-2 py-1.5 text-[9px] leading-4 text-warning transition-colors hover:border-warning/70">
+              尚未计算市场环境数据 — 挖掘含市场环境分组评估，缺少时启动即校验失败。<span className="underline underline-offset-2">前往数据页完成市场环境计算 →</span>
+            </Link>
+          )}
+
           <section className="border-t border-border pt-3">
             <div className="mb-2 flex items-center justify-between">
               <span className="flex items-center gap-1.5 text-[10px] font-semibold text-secondary"><FlaskConical className="h-3 w-3" />因子目录</span>