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

from datetime import UTC, date, datetime
from decimal import Decimal

import pytest

from zhixing_server.modules.sector_radar.domain.facts import aggregate_sector_snapshot
from zhixing_server.modules.sector_radar.domain.models import (
    MembershipStatus,
    PublicationStatus,
    RadarPublication,
    SectorMembershipSnapshot,
    SectorType,
    StockDailyFact,
    StockFactStatus,
)

TARGET_DATE = date(2026, 8, 28)


def test_point_in_time_aggregation_distinguishes_suspension_missing_and_zero() -> None:
    snapshot = SectorMembershipSnapshot(
        trade_date=TARGET_DATE,
        sector_type=SectorType.CONCEPT,
        sector_code="BK0001.DC",
        sector_name="示例概念",
        member_codes=("000001.SZ", "000002.SZ", "000003.SZ", "000004.SZ"),
        status=MembershipStatus.AVAILABLE,
        source_version="dc-member-20260828-a",
    )
    facts = (
        StockDailyFact(
            trade_date=TARGET_DATE,
            ts_code="000001.SZ",
            status=StockFactStatus.AVAILABLE,
            turnover_yuan=Decimal("1000"),
            net_amount_yuan=Decimal("100"),
        ),
        StockDailyFact(
            trade_date=TARGET_DATE,
            ts_code="000002.SZ",
            status=StockFactStatus.AVAILABLE,
            turnover_yuan=Decimal("2000"),
            net_amount_yuan=Decimal("0"),
        ),
        StockDailyFact(
            trade_date=TARGET_DATE,
            ts_code="000003.SZ",
            status=StockFactStatus.SUSPENDED,
        ),
        StockDailyFact(
            trade_date=TARGET_DATE,
            ts_code="000004.SZ",
            status=StockFactStatus.MISSING,
        ),
    )

    aggregate = aggregate_sector_snapshot(snapshot, facts)

    assert aggregate.member_count == 4
    assert aggregate.valid_sample_count == 2
    assert aggregate.net_amount_yuan == Decimal("100")
    assert aggregate.turnover_yuan == Decimal("3000")
    assert aggregate.membership_coverage == Decimal("1")
    assert aggregate.moneyflow_coverage == Decimal("2") / Decimal("3")


def test_unknown_membership_never_falls_back_to_available_stock_facts() -> None:
    snapshot = SectorMembershipSnapshot(
        trade_date=TARGET_DATE,
        sector_type=SectorType.INDUSTRY,
        sector_code="BK1001.DC",
        sector_name="示例行业",
        member_codes=(),
        status=MembershipStatus.UNKNOWN,
        source_version="dc-member-missing",
    )
    fact = StockDailyFact(
        trade_date=TARGET_DATE,
        ts_code="000001.SZ",
        status=StockFactStatus.AVAILABLE,
        turnover_yuan=Decimal("1000"),
        net_amount_yuan=Decimal("100"),
    )

    aggregate = aggregate_sector_snapshot(snapshot, (fact,))

    assert aggregate.member_count == 0
    assert aggregate.net_amount_yuan is None
    assert aggregate.turnover_yuan is None
    assert aggregate.membership_coverage == Decimal("0")


def test_stock_fact_rejects_non_finite_values_and_invalid_status_payloads() -> None:
    with pytest.raises(ValueError, match="finite"):
        StockDailyFact(
            trade_date=TARGET_DATE,
            ts_code="000001.SZ",
            status=StockFactStatus.AVAILABLE,
            turnover_yuan=Decimal("Infinity"),
            net_amount_yuan=Decimal("1"),
        )

    with pytest.raises(ValueError, match="must not expose amounts"):
        StockDailyFact(
            trade_date=TARGET_DATE,
            ts_code="000001.SZ",
            status=StockFactStatus.SUSPENDED,
            turnover_yuan=Decimal("0"),
        )


def test_publication_requires_terminal_completion_and_replay_identity() -> None:
    started_at = datetime(2026, 8, 28, 17, 30, tzinfo=UTC)

    publication = RadarPublication(
        publication_id="radar-20260828-a",
        target_trade_date=TARGET_DATE,
        status=PublicationStatus.SUCCESS,
        source_version="tushare-pro-v1",
        universe_version="eastmoney-dc-20260828-a",
        metric_versions=(
            "zhixing_amount_net_bn_v1",
            "zhixing_ratio_turnover_v1",
            "zhixing_swing_equal_3_10_v1",
        ),
        input_hash="a" * 64,
        coverage=Decimal("0.995"),
        started_at=started_at,
        finished_at=datetime(2026, 8, 28, 17, 35, tzinfo=UTC),
    )

    assert publication.status is PublicationStatus.SUCCESS

    with pytest.raises(ValueError, match="finished_at"):
        RadarPublication(
            publication_id="radar-20260828-running",
            target_trade_date=TARGET_DATE,
            status=PublicationStatus.RUNNING,
            source_version="tushare-pro-v1",
            universe_version="eastmoney-dc-20260828-a",
            metric_versions=("zhixing_amount_net_bn_v1",),
            input_hash=None,
            coverage=Decimal("0"),
            started_at=started_at,
            finished_at=started_at,
        )