yuxuanhui
148 lines
4.8 KiB
Python
148 lines
4.8 KiB
Python
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",
|
|
]
|