zhixing-system/zhixing-server/tests/test_selection_http.py

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