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 19:04:15 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
64ab9a0818626a768a6ecfaee231f0b643dbc639
tick-stock-panel/backend/tests/test_mining_api.py
T
shy3130 55b8e739c3 fix(mining): enriched 数据并发更新时重读快照而非整轮失败 
用户在数据更新进行中跑挖掘: 面板读取期间 enriched 发布提交新世代,
assert_data_generation 直接抛 EnrichedGenerationUnavailableError,
整轮运行作废且只留一句英文堆栈; 启动接口撞上发布中状态还会 500。

- mining_runtime: 面板+撮合矩阵读取包进有限重试环 (默认 3 次)。
  世代变化时丢弃半新半旧的读取, 用新世代整体重读; 发布未完成时先等
  5s; 耗尽后以带指引的中文错误终止。排队指纹仍只在首轮校验, 重试跟随
  新世代等价于"更新后立刻重跑"
- api/mining: build_data_fingerprint 撞上发布中状态映射为 400 中文
  提示 (原来未捕获直接 500), 前端 task.error 红字通路直接可见
- 测试: 读取中单次世代漂移重读后成功 / 持续漂移耗尽后带指引失败 /
  发布中启动返回 400; 全量挖掘回归 103 passed
2026-08-26 13:19:16 +08:00

638 lines
21 KiB
Python
Raw Blame History

from __future__ import annotations
import json
from datetime import date, timedelta
from pathlib import Path
from types import SimpleNamespace
from urllib.parse import quote
import polars as pl
import pytest
from fastapi import FastAPI
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
_FACTOR_DEFINITION = {
"kind": "factor_rank",
"factor_names": ["turnover_rate"],
"scoring": {"turnover_rate": 1.0},
"directions": {"turnover_rate": "high"},
}
_FACTOR_SIGNATURE = compute_candidate_signature(_FACTOR_DEFINITION)
class _Repo:
def __init__(self, data_dir) -> None:
self.store = SimpleNamespace(data_dir=data_dir)
@staticmethod
def get_matrix_data_generation(asset_type="stock"):
return f"generation-{asset_type}"
@staticmethod
def latest_enriched_date(asset_type="stock"):
del asset_type
return date(2026, 1, 9)
@staticmethod
def get_instruments_asset(asset_type):
del asset_type
return pl.DataFrame({"symbol": ["000001.SZ"], "name": ["测试"]})
class _Manager:
def __init__(self, data_dir) -> None:
self.store = MiningRunStore(data_dir)
def start(self, request, fingerprint, force=False, source="manual", run_id=None):
del force
manifest = self.store.create(request, fingerprint, run_id=run_id)
self.store.append_event(
manifest["run_id"],
"queued",
{"status": "queued", "source": source},
)
return manifest
def cancel(self, run_id):
manifest = self.store.get(run_id)
if manifest is None:
raise KeyError(run_id)
if manifest["status"] == "queued":
manifest = self.store.transition_status(run_id, "cancelled")
self.store.append_event(run_id, "cancelled", {"status": "cancelled"})
return manifest
def _write_enriched_dates(
data_dir: Path,
count: int,
*,
asset_type: str = "stock",
first: date = date(2020, 1, 1),
) -> list[date]:
dirname = "kline_etf_enriched" if asset_type == "etf" else "kline_daily_enriched"
values = [first + timedelta(days=index) for index in range(count)]
for value in values:
partition = data_dir / dirname / f"date={value.isoformat()}"
partition.mkdir(parents=True, exist_ok=True)
(partition / "part.parquet").touch()
return values
def _client(tmp_path):
_write_enriched_dates(tmp_path, 219, first=date(2022, 8, 15))
app = FastAPI()
app.include_router(router)
app.state.repo = _Repo(tmp_path)
app.state.mining_manager = _Manager(tmp_path)
app.state.strategy_engine = SimpleNamespace()
return TestClient(app), app.state.mining_manager.store
def _successful_run(store: MiningRunStore, run_id: str = "result-run"):
manifest = store.create(
{
"factor_names": ["turnover_rate"],
"strategy_ids": [],
"asset_type": "stock",
"budget_profile": "exploratory",
"correlation_threshold": 0.75,
},
{"generation": "test"},
run_id=run_id,
)
store.append_event(run_id, "queued", {"status": "queued", "source": "manual"})
store.transition_status(run_id, "running")
store.append_event(run_id, "running", {"status": "running"})
frames = {
"factors": pl.DataFrame({
"factor_name": ["turnover_rate"],
"label": ["换手率"],
"direction": [1],
"score": [1.2],
"ic_mean": [0.1],
"ir": [0.8],
"coverage": [1.0],
"turnover": [0.2],
"spread_return": [None],
"spread_sharpe": [None],
"selected": [True],
"excluded_reason": [None],
}),
"correlation": pl.DataFrame({
"factor_x": ["turnover_rate"],
"factor_y": ["turnover_rate"],
"rho": [1.0],
"pair_count": [3],
}),
"candidates": pl.DataFrame({
"signature": [_FACTOR_SIGNATURE],
"name": ["因子组合 · 换手率"],
"kind": ["factor_combination"],
"factor_names_json": ['["turnover_rate"]'],
"strategy_id": [None],
"definition_json": [json.dumps(_FACTOR_DEFINITION, sort_keys=True)],
"regime_state": ["overall"],
"score": [0.9],
"oos_return": [0.08],
"oos_sharpe": [0.9],
"oos_max_drawdown": [-0.12],
"oos_positive_fold_ratio": [1.0],
"oos_n_trades": [70],
"confidence": ["standard"],
"valid_folds": [3],
"skipped_folds": [0],
"promoted_candidate_id": [None],
"published_strategy_id": [None],
}, schema_overrides={
"strategy_id": pl.String,
"promoted_candidate_id": pl.String,
"published_strategy_id": pl.String,
}),
"folds": pl.DataFrame({
"candidate_signature": [_FACTOR_SIGNATURE, _FACTOR_SIGNATURE],
"fold": [0, 0],
"label": ["OOS 1", "OOS 1"],
"regime_state": ["overall", "strong"],
"n_dates": [3, 2],
"train_start": ["2025-01-01", "2025-01-01"],
"train_end": ["2025-12-31", "2025-12-31"],
"test_start": ["2026-01-05", "2026-01-05"],
"test_end": ["2026-01-09", "2026-01-09"],
"selected_factors_json": ['["turnover_rate"]'] * 2,
"total_return": [0.08, 0.1],
"sharpe": [0.9, 1.1],
"max_drawdown": [-0.12, -0.08],
"n_trades": [70, 40],
"skipped": [False, False],
"reason": [None, None],
}),
}
for name, frame in frames.items():
frame.write_parquet(store.artifact_path(run_id, name))
store.register_artifact(run_id, name)
store.write_summary(run_id, {
"status": "succeeded",
"factor_count": 1,
"selected_factor_count": 1,
"candidate_count": 1,
"valid_fold_count": 1,
"skipped_fold_count": 0,
"confidence": "low",
"budget_exhausted": False,
"elapsed_ms": 123.4,
"data_as_of": "2026-01-09",
"methodology_version": "factor_v2",
"algorithm_version": "mining-v1",
"panel_scans": 1,
"matrix_bytes": 4096,
"phase_ms": {"panel": 1.0, "total": 123.4},
"worker": {"peak_rss_bytes": 1000, "serialized_result_bytes": 2000},
})
store.transition_status(run_id, "succeeded")
store.append_event(run_id, "succeeded", {"status": "succeeded"})
return manifest
def test_start_is_strict_and_projects_server_signature(tmp_path):
client, store = _client(tmp_path)
payload = {
"factor_names": ["turnover_rate"],
"budget_profile": "exploratory",
"force": False,
}
response = client.post("/api/backtest/mining/runs", json=payload)
assert response.status_code == 200
body = response.json()
assert body["status"] == "queued"
assert body["signature"] == store.get(body["run_id"])["run_signature"]
assert body["request"]["factor_names"] == ["turnover_rate"]
assert "force" not in body["request"]
assert body["source"] == "manual"
assert client.post(
"/api/backtest/mining/runs",
json={**payload, "matching": "close_t"},
).status_code == 422
assert client.post(
"/api/backtest/mining/runs",
json={"factor_names": ["unknown_factor"]},
).status_code == 422
def test_start_accepts_iso_dates_and_persists_json_safe_request(tmp_path):
client, store = _client(tmp_path)
response = client.post(
"/api/backtest/mining/runs",
json={
"factor_names": ["turnover_rate"],
"budget_profile": "exploratory",
"start": "2022-08-15",
"end": "2026-08-15",
},
)
assert response.status_code == 200
run_id = response.json()["run_id"]
manifest = store.get(run_id)
assert manifest is not None
assert manifest["request"]["start"] == "2022-08-15"
assert manifest["request"]["end"] == "2026-08-15"
assert client.post(
"/api/backtest/mining/runs",
json={
"factor_names": ["turnover_rate"],
"start": "2022/08/15",
},
).status_code == 422
@pytest.mark.parametrize(
("profile", "required_bars"),
[("exploratory", 219), ("balanced", 786), ("strict", 1164)],
)
def test_availability_enforces_exact_profile_boundaries(
tmp_path,
profile,
required_bars,
):
client, _store = _client(tmp_path)
dates = _write_enriched_dates(
tmp_path,
required_bars,
first=date(2015, 1, 1),
)
eligible = client.get(
"/api/backtest/mining/availability",
params={
"asset_type": "stock",
"budget_profile": profile,
"start": dates[0].isoformat(),
"end": dates[-1].isoformat(),
},
)
insufficient = client.get(
"/api/backtest/mining/availability",
params={
"asset_type": "stock",
"budget_profile": profile,
"start": dates[1].isoformat(),
"end": dates[-1].isoformat(),
},
)
assert eligible.status_code == 200
assert eligible.json() == {
"asset_type": "stock",
"budget_profile": profile,
"trading_bars": required_bars,
"required_bars": required_bars,
"outer_folds": 1 if profile == "exploratory" else 3,
"required_outer_folds": 1 if profile == "exploratory" else 3,
"eligible": True,
"available_start": dates[0].isoformat(),
"available_end": "2023-03-21",
"effective_start": dates[0].isoformat(),
"effective_end": dates[-1].isoformat(),
"suggested_start": dates[0].isoformat(),
}
assert insufficient.status_code == 200
assert insufficient.json()["trading_bars"] == required_bars - 1
assert insufficient.json()["eligible"] is False
assert insufficient.json()["suggested_start"] == dates[0].isoformat()
def test_start_rejects_balanced_625_bar_range_before_creating_run(tmp_path):
client, store = _client(tmp_path)
dates = _write_enriched_dates(tmp_path, 625, first=date(2024, 1, 15))
response = client.post(
"/api/backtest/mining/runs",
json={
"factor_names": ["turnover_rate"],
"budget_profile": "balanced",
"start": dates[0].isoformat(),
"end": dates[-1].isoformat(),
"force": True,
},
)
assert response.status_code == 400
assert response.json()["detail"] == (
"balanced mining requires at least 786 enriched trading bars for 3 outer "
"folds; effective range 2024-01-15 to 2025-09-30 has 625"
)
assert store.list_runs() == []
def test_availability_uses_asset_specific_valid_partitions(tmp_path):
client, _store = _client(tmp_path)
etf_dates = _write_enriched_dates(
tmp_path,
219,
asset_type="etf",
first=date(2024, 1, 1),
)
malformed = tmp_path / "kline_etf_enriched" / "date=not-a-date"
malformed.mkdir(parents=True)
(malformed / "part.parquet").touch()
(tmp_path / "kline_etf_enriched" / "date=2023-12-31").mkdir(parents=True)
response = client.get(
"/api/backtest/mining/availability",
params={
"asset_type": "etf",
"budget_profile": "exploratory",
"start": etf_dates[0].isoformat(),
"end": etf_dates[-1].isoformat(),
},
)
assert response.status_code == 200
assert response.json()["trading_bars"] == 219
assert response.json()["eligible"] is True
def test_result_reconstructs_artifacts_without_exposing_definition(tmp_path):
client, store = _client(tmp_path)
_successful_run(store)
response = client.get("/api/backtest/mining/runs/result-run/result")
assert response.status_code == 200
body = response.json()
assert body["summary"]["peak_rss_bytes"] == 1000
assert body["correlation"]["matrix"] == [[1.0]]
assert body["candidates"][0]["factor_names"] == ["turnover_rate"]
assert "definition_json" not in body["candidates"][0]
assert body["candidates"][0]["folds"][0]["selected_factors"] == ["turnover_rate"]
assert body["candidates"][0]["gate"] == {"qualified": True, "reasons": []}
assert body["request_summary"] == {
"asset_type": "stock",
"budget_profile": "exploratory",
"start": None,
"end": None,
"factor_count": 1,
"strategy_count": 0,
"commission_pct": None,
"stamp_tax_pct": None,
"slippage_bps": None,
"correlation_threshold": 0.75,
}
regimes = {row["state"]: row for row in body["regimes"]}
assert regimes["overall"]["n_dates"] == 3
assert regimes["strong"]["n_dates"] == 2
assert regimes["range"]["total_return"] is None
assert body["telemetry"]["panel_scans"] == 1
def test_result_marks_legacy_fold_rows_without_evaluation_kind(tmp_path):
client, store = _client(tmp_path)
_successful_run(store)
response = client.get("/api/backtest/mining/runs/result-run/result")
assert response.status_code == 200
fold = response.json()["candidates"][0]["folds"][0]
assert fold["evaluation_kind"] is None
def test_publish_endpoint_rejects_candidate_below_evidence_gate(tmp_path):
client, store = _client(tmp_path)
_successful_run(store)
path = store.artifact_path("result-run", "candidates")
pl.read_parquet(path).with_columns(
pl.lit("low").alias("confidence"),
).write_parquet(path)
response = client.post(
f"/api/backtest/mining/runs/result-run/candidates/"
f"{quote(_FACTOR_SIGNATURE, safe='')}/publish",
)
assert response.status_code == 400
assert "exploratory results can only be saved" in response.json()["detail"]
def test_promote_endpoint_uses_persisted_definition_only(tmp_path):
client, store = _client(tmp_path)
_successful_run(store)
response = client.post(
f"/api/backtest/mining/runs/result-run/candidates/"
f"{quote(_FACTOR_SIGNATURE, safe='')}/promote"
)
assert response.status_code == 200
body = response.json()
assert body["status"] == "pending"
assert body["config"]["origin_run_id"] == "result-run"
assert body["config"]["candidate_signature"] == _FACTOR_SIGNATURE
assert body["config"]["factor_names"] == ["turnover_rate"]
assert "definition_json" not in body
persisted = pl.read_parquet(store.artifact_path("result-run", "candidates"))
assert persisted["promoted_candidate_id"][0] == body["id"]
def test_publish_endpoint_uses_persisted_definition_and_invalidates_runtime(
tmp_path,
) -> None:
client, store = _client(tmp_path)
_successful_run(store)
builtin_dir = Path(__file__).resolve().parents[1] / "app" / "strategy" / "builtin"
custom_dir = tmp_path / "strategies" / "custom"
engine = StrategyEngine(strategy_dirs=[builtin_dir, custom_dir])
invalidations: list[str] = []
client.app.state.strategy_engine = engine
client.app.state.monitor_engine = SimpleNamespace(
invalidate_strategy_state=lambda: invalidations.append("monitor")
)
response = client.post(
f"/api/backtest/mining/runs/result-run/candidates/"
f"{quote(_FACTOR_SIGNATURE, safe='')}/publish",
json={
"strategy_id": "caller-selected",
"scoring": {"rsi_14": 999.0},
},
)
assert response.status_code == 200
strategy_id = response.json()["strategy_id"]
strategy = engine.get(strategy_id)
assert strategy.meta["origin_run_id"] == "result-run"
assert strategy.meta["candidate_signature"] == _FACTOR_SIGNATURE
assert strategy.matrix_strategy._scoring == {"turnover_rate": 1.0}
assert strategy.matrix_strategy._directions == {"turnover_rate": "high"}
assert invalidations == ["monitor"]
persisted = pl.read_parquet(store.artifact_path("result-run", "candidates"))
assert persisted["published_strategy_id"][0] == strategy_id
def test_result_fails_closed_when_artifact_is_missing(tmp_path):
client, store = _client(tmp_path)
_successful_run(store)
store.artifact_path("result-run", "folds").unlink()
response = client.get("/api/backtest/mining/runs/result-run/result")
assert response.status_code == 500
assert response.json()["detail"] == "mining result artifacts are unavailable"
def test_sse_maps_failed_event_and_honors_last_event_id(tmp_path):
client, store = _client(tmp_path)
store.create(
{"factor_names": ["turnover_rate"]},
{"generation": "test"},
run_id="failed-run",
)
queued = store.append_event(
"failed-run", "queued", {"status": "queued", "source": "manual"}
)
store.transition_status("failed-run", "failed", error="worker failed")
failed = store.append_event(
"failed-run", "error", {"status": "failed", "message": "worker failed"}
)
response = client.get(
"/api/backtest/mining/runs/failed-run/events",
headers={"Last-Event-ID": str(queued["id"])},
)
assert response.status_code == 200
assert f"id: {failed['id']}" in response.text
assert "event: failed" in response.text
assert "event: error" not in response.text
assert "worker failed" in response.text
def test_sse_recovers_progress_snapshot_when_history_is_truncated(tmp_path):
client, store = _client(tmp_path)
store.create(
{"factor_names": ["turnover_rate"]},
{"generation": "test"},
run_id="truncated-run",
)
store.write_summary(
"truncated-run",
{"progress": {"phase": "search", "done": 7, "total": 10}},
)
for index in range(260):
store.append_event(
"truncated-run",
"progress",
{"phase": "search", "done": index, "total": 260},
)
store.transition_status("truncated-run", "failed", error="worker failed")
store.append_event(
"truncated-run",
"error",
{"status": "failed", "message": "worker failed"},
)
response = client.get("/api/backtest/mining/runs/truncated-run/events")
assert response.status_code == 200
assert '"done": 7' in response.text
assert "event: failed" in response.text
def test_start_rejects_incompatible_strategy_before_creating_run(tmp_path):
client, store = _client(tmp_path)
client.app.state.strategy_engine = SimpleNamespace(
get=lambda _strategy_id: SimpleNamespace(
meta={
"research_only": True,
"asset_types": ["stock"],
"timeframes": ["1d"],
},
execution_backend="matrix_native",
)
)
response = client.post(
"/api/backtest/mining/runs",
json={
"factor_names": ["turnover_rate"],
"strategy_ids": ["factor_rank_research"],
"budget_profile": "exploratory",
},
)
assert response.status_code == 400
assert store.list_runs() == []
def test_config_patch_merges_current_values(tmp_path, monkeypatch):
client, _store = _client(tmp_path)
current = {
"mining_schedule_enabled": False,
"mining_schedule_weekday": 4,
"mining_budget_profile": "balanced",
}
saved = []
monkeypatch.setattr(
"app.api.mining.preferences.get_mining_schedule",
lambda: dict(current),
)
def set_schedule(enabled, weekday, profile):
saved.append((enabled, weekday, profile))
return {
"mining_schedule_enabled": enabled,
"mining_schedule_weekday": weekday,
"mining_budget_profile": profile,
}
monkeypatch.setattr("app.api.mining.preferences.set_mining_schedule", set_schedule)
response = client.patch(
"/api/backtest/mining/config",
json={"mining_schedule_enabled": True},
)
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"]
Reference in New Issue View Git Blame Copy Permalink