"""HTTP contracts for triggering and querying persisted selection runs.""" from dataclasses import replace from datetime import date from decimal import Decimal import pytest from fastapi.testclient import TestClient import zhixing_server.modules.selection.presentation.http as selection_http from zhixing_server.bootstrap.app import create_app from zhixing_server.bootstrap.config import Settings from zhixing_server.modules.selection.application.chart import ( SelectionChart, SelectionChartNotFound, SelectionChartPoint, ) from zhixing_server.modules.selection.application.run import PreparedSelectionRun from zhixing_server.modules.selection.domain.models import SelectionSignal, ZhixingB1Category from zhixing_server.modules.selection.domain.pattern_scoring import ( PATTERN_SCORING_VERSION, ZHIXING_B1_PATTERN_CASES, PatternScore, PatternScoreBreakdown, ) from zhixing_server.modules.selection.domain.ports import MarketDataReaderError from zhixing_server.modules.selection.domain.runs import ( SelectionExecutionSource, SelectionRerunRequired, SelectionResultQuery, SelectionRun, SelectionRunInProgress, SelectionStock, ) from zhixing_server.modules.selection.infrastructure.postgres_reader import ( SelectionMarketDataNotReady, ) from zhixing_server.modules.selection.presentation.http import ( get_selection_chart_service, get_selection_postgres_pool, get_selection_service, ) TARGET = date(2026, 8, 8) class FakeSelectionService: def __init__(self, run: SelectionRun | None = None) -> None: self.run = run self.executed = False self.mode = "ok" self.last_query: SelectionResultQuery | None = None def prepare( self, strategy: str, target_trade_date: date, *, rerun: bool, ) -> PreparedSelectionRun: if self.mode == "in_progress": raise SelectionRunInProgress("already running") if self.mode == "rerun_required": raise SelectionRerunRequired("confirm rerun") if self.mode == "market_data_not_ready": raise SelectionMarketDataNotReady("market data is not ready") run = self.run or _run("run-http", "running") return PreparedSelectionRun( run=run, source=SelectionExecutionSource( market_sync_batch_id="market-run-1", target_trade_date=target_trade_date, target_count=1, valid_count=1, coverage=Decimal("1"), stocks=(SelectionStock("000001.SZ", "平安银行"),), ), ) def execute(self, prepared: PreparedSelectionRun) -> None: self.executed = True def get_run( self, run_id: str, *, query: SelectionResultQuery | None = None, ) -> SelectionRun | None: self.last_query = query return self.run if self.run and self.run.id == run_id else None def get_latest( self, strategy: str, target_trade_date: date | None = None, *, query: SelectionResultQuery | None = None, ) -> SelectionRun | None: self.last_query = query if self.run is None: return None if target_trade_date is not None and self.run.target_trade_date != target_trade_date: return None return self.run class FakeChartService: """Return or fail one deterministic chart response.""" def __init__(self) -> None: self.mode = "ok" self.request: tuple[str, date] | None = None def execute(self, ts_code: str, target_trade_date: date) -> SelectionChart: self.request = (ts_code, target_trade_date) if self.mode == "missing": raise SelectionChartNotFound("chart unavailable") if self.mode == "storage_error": raise MarketDataReaderError("chart storage unavailable") return SelectionChart( ts_code=ts_code, name="平安银行", target_trade_date=target_trade_date, source_adj="qfq", points=( SelectionChartPoint( trade_date=target_trade_date, open=10.0, high=11.0, low=9.5, close=10.5, volume=1000.0, k=52.0, d=48.0, j=60.0, trend_white=10.2, trend_yellow=10.4, ), ), ) def _run(run_id: str, status: str) -> SelectionRun: original_signal = SelectionSignal( ts_code="000001.SZ", name="平安银行", target_trade_date=TARGET, strategy="zhixing_b1", category=ZhixingB1Category.ORIGINAL_B1, close=10.5, details={"j": 12.0}, ) pullback_signal = SelectionSignal( ts_code="000001.SZ", name="平安银行", target_trade_date=TARGET, strategy="zhixing_b1", category=ZhixingB1Category.PULLBACK_WHITE, close=10.5, details={"j": 13.0, "rsi": 20.0}, ) from zhixing_server.modules.selection.domain.runs import SelectionRunItem return SelectionRun( id=run_id, strategy="zhixing_b1", target_trade_date=TARGET, market_sync_batch_id="market-run-1", status=status, # type: ignore[arg-type] target_count=1, eligible_count=1, evaluated_count=1, selected_stock_count=1, signal_count=2, failed_count=0, coverage=Decimal("1"), items=( SelectionRunItem( ts_code="000001.SZ", name="平安银行", status="selected", signal_count=2, pattern_score=PatternScore( status="matched", value=86.4, threshold=60.0, version=PATTERN_SCORING_VERSION, case=ZHIXING_B1_PATTERN_CASES[0], breakdown=PatternScoreBreakdown(71.2, 83.0, 88.0, 90.1), ), signals=(original_signal, pullback_signal), ), ), signals=(original_signal, pullback_signal), stocks_total=1, ) def _client( service: FakeSelectionService, chart_service: FakeChartService | None = None, ) -> TestClient: app = create_app() app.dependency_overrides[get_selection_service] = lambda: service if chart_service is not None: app.dependency_overrides[get_selection_chart_service] = lambda: chart_service return TestClient(app) def test_selection_dependency_reuses_one_bounded_pool(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(selection_http, "_SELECTION_POOL_CACHE", {}) settings = Settings(database_url="postgresql://test", selection_max_workers=4) first = get_selection_postgres_pool(settings) second = get_selection_postgres_pool(settings) assert first is second assert first.max_connections == 6 def test_trigger_returns_accepted_run_and_schedules_execution() -> None: service = FakeSelectionService() response = _client(service).post( "/api/v1/selection/runs", json={ "strategy": "zhixing_b1", "target_trade_date": "2026-08-08", "rerun": False, }, ) assert response.status_code == 202 assert response.json()["status"] == "running" assert response.json()["target_trade_date"] == "2026-08-08" assert service.executed is True def test_trigger_requires_explicit_rerun_confirmation() -> None: service = FakeSelectionService() service.mode = "rerun_required" response = _client(service).post( "/api/v1/selection/runs", json={"strategy": "zhixing_b1", "target_trade_date": "2026-08-08"}, ) assert response.status_code == 409 assert response.json()["detail"]["code"] == "rerun_confirmation_required" def test_trigger_rejects_a_duplicate_running_request() -> None: service = FakeSelectionService() service.mode = "in_progress" response = _client(service).post( "/api/v1/selection/runs", json={"strategy": "zhixing_b1", "target_trade_date": "2026-08-08"}, ) assert response.status_code == 409 assert response.json()["detail"]["code"] == "run_in_progress" def test_trigger_rejects_unqualified_market_data() -> None: service = FakeSelectionService() service.mode = "market_data_not_ready" response = _client(service).post( "/api/v1/selection/runs", json={"strategy": "zhixing_b1", "target_trade_date": "2026-08-08"}, ) assert response.status_code == 422 assert response.json()["detail"]["code"] == "market_data_not_ready" def test_query_returns_no_data_without_fabricating_a_result() -> None: response = _client(FakeSelectionService()).get( "/api/v1/selection/results?strategy=zhixing_b1&target_trade_date=2026-08-08" ) assert response.status_code == 200 assert response.json()["status"] == "no_data" assert response.json()["stocks"] == [] def test_query_returns_persisted_signal_details() -> None: response = _client(FakeSelectionService(_run("run-http", "success"))).get( "/api/v1/selection/results?strategy=zhixing_b1&target_trade_date=2026-08-08" ) assert response.status_code == 200 body = response.json() assert body["run_id"] == "run-http" assert body["signal_count"] == 2 assert body["page"] == 1 assert body["page_size"] == 10 assert body["stocks_total"] == 1 assert "signals_total" not in body assert "signals" not in body assert len(body["stocks"]) == 1 assert body["stocks"][0]["ts_code"] == "000001.SZ" assert body["stocks"][0]["score"] == { "status": "matched", "value": 86.4, "threshold": 60.0, "version": PATTERN_SCORING_VERSION, "case": { "id": "case_001", "name": "华纳药厂", "breakout_date": "2025-05-12", }, "breakdown": { "trend_structure": 71.2, "kdj_state": 83.0, "volume_pattern": 88.0, "price_shape": 90.1, }, "reason": None, } assert [signal["category"] for signal in body["stocks"][0]["signals"]] == [ "zhixing_b1_original_b1", "zhixing_b1_pullback_white", ] assert [signal["details"] for signal in body["stocks"][0]["signals"]] == [ {"j": 12.0}, {"j": 13.0, "rsi": 20.0}, ] @pytest.mark.parametrize( ("pattern_score", "expected_score"), [ ( PatternScore( status="below_threshold", value=42.5, threshold=60.0, version=PATTERN_SCORING_VERSION, case=ZHIXING_B1_PATTERN_CASES[0], breakdown=PatternScoreBreakdown(40.0, 42.0, 43.0, 44.0), ), { "status": "below_threshold", "value": 42.5, "threshold": 60.0, "version": PATTERN_SCORING_VERSION, "case": { "id": "case_001", "name": "华纳药厂", "breakout_date": "2025-05-12", }, "breakdown": { "trend_structure": 40.0, "kdj_state": 42.0, "volume_pattern": 43.0, "price_shape": 44.0, }, "reason": None, }, ), ( PatternScore.failed("FastDTW unavailable"), { "status": "failed", "value": None, "threshold": None, "version": None, "case": None, "breakdown": None, "reason": "FastDTW unavailable", }, ), (PatternScore(), None), ], ) def test_query_preserves_signals_for_every_pattern_score_state( pattern_score: PatternScore, expected_score: dict[str, object] | None, ) -> None: run = _run("run-http", "success") run = replace(run, items=(replace(run.items[0], pattern_score=pattern_score),)) response = _client(FakeSelectionService(run)).get("/api/v1/selection/results") assert response.status_code == 200 stock = response.json()["stocks"][0] assert stock["score"] == expected_score assert len(stock["signals"]) == 2 assert response.json()["failures"] == [] def test_query_forwards_pagination_and_filters() -> None: service = FakeSelectionService(_run("run-http", "success")) response = _client(service).get( "/api/v1/selection/results", params={ "strategy": "zhixing_b1", "page": 2, "page_size": 5, "search": " 平安银行 ", "category": "original", "sort": "score_desc", }, ) assert response.status_code == 200 assert service.last_query == SelectionResultQuery( page=2, page_size=5, search="平安银行", category="original", sort="score_desc", ) assert response.json()["page"] == 2 assert response.json()["page_size"] == 5 def test_query_rejects_invalid_page_size() -> None: response = _client(FakeSelectionService()).get( "/api/v1/selection/results", params={"strategy": "zhixing_b1", "page_size": 101}, ) assert response.status_code == 422 def test_chart_returns_bounded_qfq_contract() -> None: chart_service = FakeChartService() response = _client(FakeSelectionService(), chart_service).get( "/api/v1/selection/stocks/000001.SZ/chart", params={"target_trade_date": "2026-08-08"}, ) assert response.status_code == 200 assert chart_service.request == ("000001.SZ", TARGET) assert response.json() == { "ts_code": "000001.SZ", "name": "平安银行", "target_trade_date": "2026-08-08", "source_adj": "qfq", "points": [ { "trade_date": "2026-08-08", "open": 10.0, "high": 11.0, "low": 9.5, "close": 10.5, "volume": 1000.0, "k": 52.0, "d": 48.0, "j": 60.0, "trend_white": 10.2, "trend_yellow": 10.4, } ], } @pytest.mark.parametrize( ("mode", "status_code", "error_code"), [ ("missing", 404, "chart_data_not_found"), ("storage_error", 503, "selection_storage_unavailable"), ], ) def test_chart_maps_application_errors( mode: str, status_code: int, error_code: str, ) -> None: chart_service = FakeChartService() chart_service.mode = mode response = _client(FakeSelectionService(), chart_service).get( "/api/v1/selection/stocks/000001.SZ/chart", params={"target_trade_date": "2026-08-08"}, ) assert response.status_code == status_code assert response.json()["detail"]["code"] == error_code def test_chart_requires_a_valid_target_date() -> None: response = _client(FakeSelectionService(), FakeChartService()).get( "/api/v1/selection/stocks/000001.SZ/chart", params={"target_trade_date": "not-a-date"}, ) assert response.status_code == 422 def test_query_forwards_score_ascending_sort() -> None: service = FakeSelectionService(_run("run-http", "success")) response = _client(service).get( "/api/v1/selection/results", params={"strategy": "zhixing_b1", "sort": "score_asc"}, ) assert response.status_code == 200 assert service.last_query == SelectionResultQuery(sort="score_asc") def test_run_polling_returns_the_persisted_terminal_result() -> None: response = _client(FakeSelectionService(_run("run-http", "success"))).get( "/api/v1/selection/runs/run-http" ) assert response.status_code == 200 assert response.json()["status"] == "success" assert response.json()["stocks"][0]["signals"][0]["category"] == ("zhixing_b1_original_b1")