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

from datetime import date
from decimal import Decimal

from zhixing_server.modules.sector_radar.domain.metrics import (
    AmountNetStrategy,
    RatioTurnoverStrategy,
    SwingEqualThreeToTenStrategy,
)
from zhixing_server.modules.sector_radar.domain.models import (
    MetricQuality,
    SectorDailyAggregate,
    SectorType,
)

TARGET_DATE = date(2026, 8, 28)


def make_aggregate(
    *,
    net_amount_yuan: Decimal | None = Decimal("125000000"),
    turnover_yuan: Decimal | None = Decimal("5000000000"),
) -> SectorDailyAggregate:
    return SectorDailyAggregate(
        trade_date=TARGET_DATE,
        sector_type=SectorType.CONCEPT,
        sector_code="BK0001.DC",
        sector_name="示例概念",
        member_count=10,
        valid_sample_count=10,
        net_amount_yuan=net_amount_yuan,
        turnover_yuan=turnover_yuan,
        membership_coverage=Decimal("1"),
        moneyflow_coverage=Decimal("1"),
    )


def test_amount_and_ratio_strategies_expose_independent_versioned_values() -> None:
    aggregate = make_aggregate()

    amount = AmountNetStrategy().evaluate((aggregate,), TARGET_DATE)
    ratio = RatioTurnoverStrategy().evaluate((aggregate,), TARGET_DATE)

    assert amount.value == Decimal("1.25")
    assert amount.metric_version == "zhixing_amount_net_bn_v1"
    assert amount.implementation_kind == "independent"
    assert amount.unit == "CNY_100M"
    assert amount.quality is MetricQuality.AVAILABLE

    assert ratio.value == Decimal("0.025")
    assert ratio.metric_version == "zhixing_ratio_turnover_v1"
    assert ratio.implementation_kind == "independent"
    assert ratio.unit == "ratio"


def test_missing_moneyflow_is_unavailable_but_zero_remains_a_real_value() -> None:
    missing = AmountNetStrategy().evaluate((make_aggregate(net_amount_yuan=None),), TARGET_DATE)
    zero = AmountNetStrategy().evaluate(
        (make_aggregate(net_amount_yuan=Decimal("0")),), TARGET_DATE
    )

    assert missing.value is None
    assert missing.quality is MetricQuality.UNAVAILABLE
    assert zero.value == Decimal("0")
    assert zero.quality is MetricQuality.AVAILABLE


def test_swing_strategy_uses_each_days_point_in_time_aggregate() -> None:
    history = tuple(
        SectorDailyAggregate(
            trade_date=date(2026, 8, 18 + offset),
            sector_type=SectorType.CONCEPT,
            sector_code="BK0001.DC",
            sector_name="示例概念",
            member_count=6 + offset,
            valid_sample_count=6 + offset,
            net_amount_yuan=Decimal(str(offset + 1)),
            turnover_yuan=Decimal("100"),
            membership_coverage=Decimal("1"),
            moneyflow_coverage=Decimal("1"),
        )
        for offset in range(10)
    )

    result = SwingEqualThreeToTenStrategy().evaluate(history, date(2026, 8, 27))

    # The worked 3..10-day window ratios average to exactly 0.0725.
    assert result.value == Decimal("0.0725")
    assert result.metric_version == "zhixing_swing_equal_3_10_v1"
    assert result.member_count == 15


def test_swing_strategy_carries_forward_limited_historical_sample_quality() -> None:
    history = tuple(
        SectorDailyAggregate(
            trade_date=date(2026, 8, 18 + offset),
            sector_type=SectorType.INDUSTRY,
            sector_code="BK1001.DC",
            sector_name="示例行业",
            member_count=10,
            valid_sample_count=4 if offset == 0 else 10,
            net_amount_yuan=Decimal("10"),
            turnover_yuan=Decimal("100"),
            membership_coverage=Decimal("1"),
            moneyflow_coverage=Decimal("1"),
        )
        for offset in range(10)
    )

    result = SwingEqualThreeToTenStrategy().evaluate(history, date(2026, 8, 27))

    assert result.value == Decimal("0.1")
    assert result.quality is MetricQuality.AVAILABLE_LIMITED_SAMPLE