zhixing-system/zhixing-server/tests/unit/sector_radar/test_ranking.py

from datetime import date
from decimal import Decimal

from zhixing_server.modules.sector_radar.domain.models import (
    MetricKind,
    MetricObservation,
    MetricQuality,
    MetricUnit,
    RankSide,
    SectorType,
)
from zhixing_server.modules.sector_radar.domain.ranking import (
    rank_metric_observations,
    select_percentile_side,
    select_rank_change_side,
    with_rank_changes,
)

TARGET_DATE = date(2026, 8, 28)


def make_observation(
    sector_code: str,
    sector_type: SectorType,
    value: str | None,
    *,
    trade_date: date = TARGET_DATE,
) -> MetricObservation:
    metric_value = Decimal(value) if value is not None else None
    return MetricObservation(
        trade_date=trade_date,
        sector_type=sector_type,
        sector_code=sector_code,
        sector_name=sector_code,
        metric_kind=MetricKind.AMOUNT,
        metric_version="zhixing_amount_net_bn_v1",
        implementation_kind="independent",
        unit=MetricUnit.CNY_100M,
        value=metric_value,
        quality=(
            MetricQuality.AVAILABLE if metric_value is not None else MetricQuality.UNAVAILABLE
        ),
        member_count=10,
        valid_sample_count=10 if metric_value is not None else 0,
        membership_coverage=Decimal("1"),
        moneyflow_coverage=Decimal("1"),
    )


def test_ranking_separates_types_and_uses_code_as_stable_tie_breaker() -> None:
    observations = (
        make_observation("BK2002.DC", SectorType.INDUSTRY, "20"),
        make_observation("BK1002.DC", SectorType.CONCEPT, "30"),
        make_observation("BK2001.DC", SectorType.INDUSTRY, "20"),
        make_observation("BK1001.DC", SectorType.CONCEPT, "10"),
    )

    ranked = rank_metric_observations(tuple(reversed(observations)))
    by_code = {row.observation.sector_code: row for row in ranked}

    assert by_code["BK1002.DC"].rank_position == 1
    assert by_code["BK1002.DC"].rank_percentile == Decimal("100")
    assert by_code["BK1001.DC"].rank_position == 2
    assert by_code["BK1001.DC"].rank_percentile == Decimal("50")

    assert by_code["BK2001.DC"].rank_position == 1
    assert by_code["BK2002.DC"].rank_position == 2


def test_ranking_handles_empty_and_single_element_pools() -> None:
    assert rank_metric_observations(()) == ()

    [single] = rank_metric_observations((make_observation("BK0001.DC", SectorType.CONCEPT, "0"),))

    assert single.rank_position == 1
    assert single.rank_percentile == Decimal("100")


def test_percentile_sides_use_confirmed_inclusive_thresholds() -> None:
    ranked = rank_metric_observations(
        make_observation(f"BK{position:04d}.DC", SectorType.CONCEPT, str(11 - position))
        for position in range(1, 11)
    )

    top = select_percentile_side(ranked, RankSide.TOP)
    bottom = select_percentile_side(ranked, RankSide.BOTTOM)

    assert [row.observation.sector_code for row in top] == ["BK0001.DC", "BK0002.DC"]
    assert [row.observation.sector_code for row in bottom] == ["BK0010.DC"]


def test_rank_change_is_past_rank_minus_current_and_preserves_missing_history() -> None:
    current = rank_metric_observations(
        (
            make_observation("BK0001.DC", SectorType.CONCEPT, "30"),
            make_observation("BK0002.DC", SectorType.CONCEPT, "20"),
        )
    )
    previous = rank_metric_observations(
        (
            make_observation(
                "BK0001.DC",
                SectorType.CONCEPT,
                "10",
                trade_date=date(2026, 8, 27),
            ),
            make_observation(
                "BK0002.DC",
                SectorType.CONCEPT,
                "40",
                trade_date=date(2026, 8, 27),
            ),
        )
    )

    changed = with_rank_changes(current, {1: previous, 5: ()})
    by_code = {row.observation.sector_code: row for row in changed}

    assert by_code["BK0001.DC"].rank_change(1) == 1
    assert by_code["BK0002.DC"].rank_change(1) == -1
    assert by_code["BK0001.DC"].rank_change(5) is None


def test_rank_change_sides_take_ceiling_ten_percent_per_pool() -> None:
    current = rank_metric_observations(
        make_observation(f"BK{position:04d}.DC", SectorType.CONCEPT, str(12 - position))
        for position in range(1, 12)
    )
    previous = rank_metric_observations(
        make_observation(
            f"BK{position:04d}.DC",
            SectorType.CONCEPT,
            str(position),
            trade_date=date(2026, 8, 27),
        )
        for position in range(1, 12)
    )
    changed = with_rank_changes(current, {1: previous})

    top = select_rank_change_side(changed, days=1, side=RankSide.TOP)
    bottom = select_rank_change_side(changed, days=1, side=RankSide.BOTTOM)

    assert [row.observation.sector_code for row in top] == ["BK0001.DC", "BK0002.DC"]
    assert [row.observation.sector_code for row in bottom] == [
        "BK0011.DC",
        "BK0010.DC",
    ]