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 f351ccd..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,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)
- 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),
+ })
- _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,
- })
+ 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,
+ })
+ 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,
diff --git a/backend/app/backtest/worker.py b/backend/app/backtest/worker.py
index a84c295..6b4ca9d 100644
--- a/backend/app/backtest/worker.py
+++ b/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 只是入队,
- # 实际写管道的是后台 feeder 线程; 不 join 的话主线程先退出,
- # feeder 随进程销毁, 消息尾部丢失 → 父进程误判 "exited without result"。
+ # 终态消息已入队: 显式冲刷队列后立即退出。put 只是入队, 实际写管道的
+ # 是后台 feeder 线程, close+join_thread 保证消息完整落管 (否则父进程误判
+ # "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":
diff --git a/backend/tests/backtest/test_mining_runtime.py b/backend/tests/backtest/test_mining_runtime.py
index b982700..7c13449 100644
--- a/backend/tests/backtest/test_mining_runtime.py
+++ b/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)
diff --git a/backend/tests/backtest/test_worker_process.py b/backend/tests/backtest/test_worker_process.py
index 18cd68f..e76193f 100644
--- a/backend/tests/backtest/test_worker_process.py
+++ b/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,
+ )
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"]
diff --git a/frontend/src/pages/backtest/MiningWorkbench.tsx b/frontend/src/pages/backtest/MiningWorkbench.tsx
index d76f4be..c217ce5 100644
--- a/frontend/src/pages/backtest/MiningWorkbench.tsx
+++ b/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() {
)}
+ {regimeLatestQuery.data && !regimeLatestQuery.data.row && (
+
+ 尚未计算市场环境数据 — 挖掘含市场环境分组评估，缺少时启动即校验失败。前往数据页完成市场环境计算 →
+
+ )}
+