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

from dataclasses import replace
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal

import pytest
from fastapi.testclient import TestClient
from pydantic import ValidationError

from zhixing_server.bootstrap.app import create_app
from zhixing_server.modules.sector_radar.application.read import (
    RadarDateIndex,
    RadarMetricDefinition,
    RadarQuery,
    RadarView,
    RankingPage,
    SectorRef,
    StockSectorMembership,
    StockSectorQuery,
)
from zhixing_server.modules.sector_radar.domain.metrics import AmountNetStrategy
from zhixing_server.modules.sector_radar.domain.models import (
    MetricKind,
    MetricObservation,
    MetricQuality,
    MetricUnit,
    PublicationStatus,
    RadarPublication,
    RankChange,
    RankedMetric,
    RankSide,
    SectorType,
)
from zhixing_server.modules.sector_radar.infrastructure.postgres import (
    SectorRadarRepositoryError,
)
from zhixing_server.modules.sector_radar.presentation.http import (
    RadarRankingRowResponse,
    get_sector_radar_reader,
)

TARGET_DATE = date(2026, 8, 28)
NOW = datetime(2026, 8, 28, 17, 30, tzinfo=UTC)


def _publication(
    publication_id: str,
    status: PublicationStatus,
    *,
    trade_date: date = TARGET_DATE,
) -> RadarPublication:
    return RadarPublication(
        publication_id=publication_id,
        target_trade_date=trade_date,
        status=status,
        source_version="tushare-pro-v1",
        universe_version="eastmoney-dc-v1",
        metric_versions=(AmountNetStrategy.metric_version,),
        input_hash="a" * 64 if status is PublicationStatus.SUCCESS else None,
        coverage=Decimal(1) if status is PublicationStatus.SUCCESS else Decimal("0.8"),
        started_at=NOW,
        finished_at=None if status is PublicationStatus.RUNNING else NOW + timedelta(minutes=5),
        error_summary=None if status is PublicationStatus.SUCCESS else "safe_error",
    )


def _ranking() -> RankedMetric:
    return RankedMetric(
        observation=MetricObservation(
            trade_date=TARGET_DATE,
            sector_type=SectorType.CONCEPT,
            sector_code="BK0001.DC",
            sector_name="机器人",
            metric_kind=MetricKind.AMOUNT,
            metric_version=AmountNetStrategy.metric_version,
            implementation_kind="independent",
            unit=MetricUnit.CNY_100M,
            value=Decimal("12.5"),
            quality=MetricQuality.AVAILABLE,
            member_count=20,
            valid_sample_count=19,
            membership_coverage=Decimal(1),
            moneyflow_coverage=Decimal("0.95"),
        ),
        rank_position=1,
        rank_percentile=Decimal(100),
        rank_changes=(RankChange(days=5, value=3),),
    )


class FakeReader:
    def __init__(self, *, no_data: bool = False, fail: bool = False) -> None:
        self.fail = fail
        self.no_data = no_data
        self.last_query: RadarQuery | None = None
        self.last_membership_query: StockSectorQuery | None = None
        success = _publication("publication-success", PublicationStatus.SUCCESS)
        current = _publication(
            "publication-partial",
            PublicationStatus.PARTIAL,
            trade_date=TARGET_DATE + timedelta(days=1),
        )
        self.date_index = RadarDateIndex(
            available_dates=() if no_data else (TARGET_DATE,),
            current_attempt=None if no_data else current,
            last_good=None if no_data else success,
        )
        query = RadarQuery()
        self.page = RankingPage(
            status="no_data" if no_data else "success",
            query=query,
            publication=None if no_data else success,
            definition=RadarMetricDefinition(
                metric_kind=MetricKind.AMOUNT,
                metric_version=AmountNetStrategy.metric_version,
                label="主力净流入（知行独立实现）",
                unit=MetricUnit.CNY_100M,
            ),
            rows=() if no_data else (_ranking(),),
            total=0 if no_data else 1,
        )

    def list_dates(self) -> RadarDateIndex:
        if self.fail:
            raise SectorRadarRepositoryError("private database detail")
        return self.date_index

    def query(self, query: RadarQuery) -> RankingPage:
        if self.fail:
            raise SectorRadarRepositoryError("private database detail")
        self.last_query = query
        return replace(self.page, query=query)

    def stock_membership(self, query: StockSectorQuery) -> StockSectorMembership:
        if self.fail:
            raise SectorRadarRepositoryError("private database detail")
        self.last_membership_query = query
        if self.no_data:
            return StockSectorMembership(
                status="no_data",
                query=query,
                trade_date=None,
                industries=(),
                concepts=(),
                concept_total=0,
            )
        return StockSectorMembership(
            status="success",
            query=query,
            trade_date=TARGET_DATE,
            industries=(SectorRef(code="BK0901.DC", name="银行"),),
            concepts=(
                SectorRef(code="BK0001.DC", name="机器人"),
                SectorRef(code="BK0002.DC", name="人工智能"),
                SectorRef(code="BK0003.DC", name="数字经济"),
            ),
            concept_total=3,
        )


def _client(reader: FakeReader) -> TestClient:
    application = create_app()
    application.dependency_overrides[get_sector_radar_reader] = lambda: reader
    return TestClient(application)


def test_dates_exposes_partial_attempt_without_replacing_last_good() -> None:
    response = _client(FakeReader()).get("/api/v1/sector-radar/dates")

    assert response.status_code == 200
    payload = response.json()
    assert payload["status"] == "success"
    assert payload["available_dates"] == ["2026-08-28"]
    assert payload["current_attempt"]["status"] == "partial"
    assert payload["last_good"]["status"] == "success"
    assert payload["last_good"]["coverage"] == "1"


def test_rankings_maps_filters_and_independent_metric_contract() -> None:
    reader = FakeReader()
    response = _client(reader).get(
        "/api/v1/sector-radar/rankings",
        params={
            "trade_date": "2026-08-28",
            "sector_type": "concept",
            "view": "rank_change",
            "rank_change_metric": "amount",
            "rank_change_days": 5,
            "side": "top",
            "search": " 机器人 ",
            "page": 2,
            "page_size": 10,
        },
    )

    assert response.status_code == 200
    assert reader.last_query == RadarQuery(
        trade_date=TARGET_DATE,
        sector_type=SectorType.CONCEPT,
        view=RadarView.RANK_CHANGE,
        rank_change_metric=MetricKind.AMOUNT,
        rank_change_days=5,
        side=RankSide.TOP,
        search="机器人",
        page=2,
        page_size=10,
    )
    payload = response.json()
    assert payload["definition"]["metric_version"] == "zhixing_amount_net_bn_v1"
    assert payload["definition"]["implementation_kind"] == "independent"
    assert "知行独立实现" in payload["definition"]["disclaimer"]
    assert payload["rows"][0]["rank_change"] == 3
    assert payload["rows"][0]["unit"] == "CNY_100M"


def test_no_data_is_a_stable_200_response() -> None:
    client = _client(FakeReader(no_data=True))

    dates = client.get("/api/v1/sector-radar/dates")
    rankings = client.get("/api/v1/sector-radar/rankings")

    assert dates.status_code == 200
    assert dates.json()["status"] == "no_data"
    assert rankings.status_code == 200
    assert rankings.json()["status"] == "no_data"
    assert rankings.json()["rows"] == []


def test_weighted_score_and_all_rank_changes_cross_the_http_boundary() -> None:
    reader = FakeReader()
    ranking = _ranking()
    reader.page = replace(
        reader.page,
        rows=(
            replace(
                ranking,
                observation=replace(
                    ranking.observation, weighted_score=Decimal("712.345678901234")
                ),
            ),
        ),
        comparison_trade_date=date(2026, 8, 21),
        rank_change_values={
            "BK0001.DC": {MetricKind.AMOUNT: 3, MetricKind.RATIO: 0, MetricKind.SWING: None}
        },
    )
    response = _client(reader).get("/api/v1/sector-radar/rankings", params={"view": "rank_change"})
    assert response.status_code == 200
    assert (
        reader.last_query is not None and reader.last_query.rank_change_metric is MetricKind.SWING
    )
    payload = response.json()
    assert payload["comparison_trade_date"] == "2026-08-21"
    assert payload["rows"][0]["weighted_score"] == "712.345678901234"
    assert payload["rows"][0]["rank_change_values"] == {"amount": 3, "ratio": 0, "swing": None}


def test_http_contract_rejects_zero_rank_percentile() -> None:
    payload = _client(FakeReader()).get("/api/v1/sector-radar/rankings").json()["rows"][0]
    payload["rank_percentile"] = "0"

    with pytest.raises(ValidationError):
        RadarRankingRowResponse.model_validate(payload)


def test_invalid_query_values_return_422() -> None:
    client = _client(FakeReader())

    for params in (
        {"rank_change_days": 0},
        {"rank_change_days": 6},
        {"page": 0},
        {"page_size": 101},
        {"sector_type": "region"},
        {"view": "unknown"},
        {"side": "unknown"},
    ):
        assert client.get("/api/v1/sector-radar/rankings", params=params).status_code == 422


def test_repository_error_maps_to_redacted_503() -> None:
    response = _client(FakeReader(fail=True)).get("/api/v1/sector-radar/rankings")

    assert response.status_code == 503
    assert response.json() == {
        "detail": {
            "code": "sector_radar_storage_unavailable",
            "message": "sector radar storage is unavailable",
        }
    }
    assert "private database detail" not in response.text


def test_stock_membership_maps_industries_and_concepts() -> None:
    reader = FakeReader()
    response = _client(reader).get(
        "/api/v1/sector-radar/stocks/000001.SZ/membership",
        params={"trade_date": "2026-08-28"},
    )

    assert response.status_code == 200
    assert reader.last_membership_query == StockSectorQuery(
        trade_date=TARGET_DATE,
        ts_code="000001.SZ",
    )
    payload = response.json()
    assert payload["status"] == "success"
    assert payload["ts_code"] == "000001.SZ"
    assert payload["requested_trade_date"] == "2026-08-28"
    assert payload["trade_date"] == "2026-08-28"
    assert payload["industries"] == [{"code": "BK0901.DC", "name": "银行"}]
    assert payload["concept_total"] == 3
    assert payload["concept_limit"] == 20
    assert [concept["code"] for concept in payload["concepts"]] == [
        "BK0001.DC",
        "BK0002.DC",
        "BK0003.DC",
    ]


def test_stock_membership_no_data_is_a_stable_200_response() -> None:
    response = _client(FakeReader(no_data=True)).get(
        "/api/v1/sector-radar/stocks/000001.SZ/membership",
        params={"trade_date": "2026-08-28"},
    )

    assert response.status_code == 200
    payload = response.json()
    assert payload["status"] == "no_data"
    assert payload["trade_date"] is None
    assert payload["industries"] == []
    assert payload["concepts"] == []
    assert payload["concept_total"] == 0


def test_stock_membership_rejects_invalid_query_values() -> None:
    client = _client(FakeReader())

    missing_date = client.get("/api/v1/sector-radar/stocks/000001.SZ/membership")
    bad_limit = client.get(
        "/api/v1/sector-radar/stocks/000001.SZ/membership",
        params={"trade_date": "2026-08-28", "concept_limit": 0},
    )
    over_limit = client.get(
        "/api/v1/sector-radar/stocks/000001.SZ/membership",
        params={"trade_date": "2026-08-28", "concept_limit": 101},
    )
    empty_code = client.get(
        "/api/v1/sector-radar/stocks//membership", params={"trade_date": "2026-08-28"}
    )

    assert missing_date.status_code == 422
    assert bad_limit.status_code == 422
    assert over_limit.status_code == 422
    assert empty_code.status_code in (404, 422)


def test_stock_membership_repository_error_maps_to_redacted_503() -> None:
    response = _client(FakeReader(fail=True)).get(
        "/api/v1/sector-radar/stocks/000001.SZ/membership",
        params={"trade_date": "2026-08-28"},
    )

    assert response.status_code == 503
    assert response.json()["detail"]["code"] == "sector_radar_storage_unavailable"
    assert "private database detail" not in response.text