aeroxw/tick-stock-panel
1
0
Fork 0
mirror of https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git synced 2026-09-12 15:34:16 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(mining): enriched 数据并发更新时重读快照而非整轮失败

Browse Source 
用户在数据更新进行中跑挖掘: 面板读取期间 enriched 发布提交新世代,
assert_data_generation 直接抛 EnrichedGenerationUnavailableError,
整轮运行作废且只留一句英文堆栈; 启动接口撞上发布中状态还会 500。

- mining_runtime: 面板+撮合矩阵读取包进有限重试环 (默认 3 次)。
  世代变化时丢弃半新半旧的读取, 用新世代整体重读; 发布未完成时先等
  5s; 耗尽后以带指引的中文错误终止。排队指纹仍只在首轮校验, 重试跟随
  新世代等价于"更新后立刻重跑"
- api/mining: build_data_fingerprint 撞上发布中状态映射为 400 中文
  提示 (原来未捕获直接 500), 前端 task.error 红字通路直接可见
- 测试: 读取中单次世代漂移重读后成功 / 持续漂移耗尽后带指引失败 /
  发布中启动返回 400; 全量挖掘回归 103 passed
This commit is contained in:
shy3130
2026-08-26 13:19:16 +08:00
parent de51929a44
commit 55b8e739c3
4 changed files with 269 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/mining.py
+8
View File
@@ -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
+107 -73
View File
@@ -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(
backend/tests/backtest/test_worker_process.py
+123
View File
@@ -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,
)
backend/tests/test_mining_api.py
+31
View File
@@ -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"]