yuxuanhui
113 lines
3.8 KiB
Python
113 lines
3.8 KiB
Python
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
|