From 3789008ea61e830ad8df301d3ed49cdad61e2ba7 Mon Sep 17 00:00:00 2001 From: yuxuanhui Date: Sat, 29 Aug 2026 16:52:45 +0800 Subject: [PATCH] =?UTF-8?q?feat(sector-radar):=20=E5=BB=BA=E7=AB=8B?= =?UTF-8?q?=E7=8B=AC=E7=AB=8B=E6=8C=87=E6=A0=87=E4=B8=8E=E6=8E=92=E5=90=8D?= =?UTF-8?q?=E9=A2=86=E5=9F=9F=E6=A8=A1=E5=9E=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../08-28-sector-capital-radar/implement.md | 12 +- .../08-28-sector-capital-radar/task.json | 4 +- .../modules/sector_radar/__init__.py | 1 + .../modules/sector_radar/domain/__init__.py | 1 + .../modules/sector_radar/domain/facts.py | 97 ++++++ .../modules/sector_radar/domain/metrics.py | 204 ++++++++++++ .../modules/sector_radar/domain/models.py | 315 ++++++++++++++++++ .../modules/sector_radar/domain/ranking.py | 216 ++++++++++++ .../tests/unit/sector_radar/test_facts.py | 146 ++++++++ .../tests/unit/sector_radar/test_metrics.py | 112 +++++++ .../tests/unit/sector_radar/test_ranking.py | 147 ++++++++ 11 files changed, 1248 insertions(+), 7 deletions(-) create mode 100644 zhixing-server/src/zhixing_server/modules/sector_radar/__init__.py create mode 100644 zhixing-server/src/zhixing_server/modules/sector_radar/domain/__init__.py create mode 100644 zhixing-server/src/zhixing_server/modules/sector_radar/domain/facts.py create mode 100644 zhixing-server/src/zhixing_server/modules/sector_radar/domain/metrics.py create mode 100644 zhixing-server/src/zhixing_server/modules/sector_radar/domain/models.py create mode 100644 zhixing-server/src/zhixing_server/modules/sector_radar/domain/ranking.py create mode 100644 zhixing-server/tests/unit/sector_radar/test_facts.py create mode 100644 zhixing-server/tests/unit/sector_radar/test_metrics.py create mode 100644 zhixing-server/tests/unit/sector_radar/test_ranking.py diff --git a/.trellis/tasks/08-28-sector-capital-radar/implement.md b/.trellis/tasks/08-28-sector-capital-radar/implement.md index af39553..e031c97 100644 --- a/.trellis/tasks/08-28-sector-capital-radar/implement.md +++ b/.trellis/tasks/08-28-sector-capital-radar/implement.md @@ -8,11 +8,13 @@ ## 1. 纯领域安全里程碑 -- [ ] 在新的 `sector_radar` bounded context 定义板块类型、成员快照、股票事实、指标观察、发布与排名模型。 -- [ ] 先写固定人工样本测试，再实现 `zhixing_amount_net_bn_v1`、`zhixing_ratio_turnover_v1`、`zhixing_swing_equal_3_10_v1`。 -- [ ] 实现概念/行业分池、稳定并列键、1 基排名、百分位、TOP/BOTTOM 和 1—5 日排名变化。 -- [ ] 覆盖乱序输入、NULL/0、非有限数、空池、单元素、并列、历史缺失、停牌和 point-in-time 成员变化。 -- [ ] 运行 `uv run --directory zhixing-server pytest tests/unit/sector_radar`、Ruff 与 Pyright。此步绿灯是第一个可回滚安全点。 +- [x] 在新的 `sector_radar` bounded context 定义板块类型、成员快照、股票事实、指标观察、发布与排名模型。 +- [x] 先写固定人工样本测试，再实现 `zhixing_amount_net_bn_v1`、`zhixing_ratio_turnover_v1`、`zhixing_swing_equal_3_10_v1`。 +- [x] 实现概念/行业分池、稳定并列键、1 基排名、百分位、TOP/BOTTOM 和 1—5 日排名变化。 +- [x] 覆盖乱序输入、NULL/0、非有限数、空池、单元素、并列、历史缺失、停牌和 point-in-time 成员变化。 +- [x] 运行 `uv run --directory zhixing-server pytest tests/unit/sector_radar`、Ruff 与 Pyright。此步绿灯是第一个可回滚安全点。 + +阶段结果（2026-08-29）：三个透明指标策略、point-in-time 事实聚合、发布生命周期和横截面排名 seam 均已实现；13 个板块雷达领域测试通过。完整后端门禁为 96 passed、2 skipped，两个跳过项均为需要 `ZHIXING_TEST_DATABASE_URL` 的既有 PostgreSQL 集成测试。 ## 2. Tushare 输入与持久化 diff --git a/.trellis/tasks/08-28-sector-capital-radar/task.json b/.trellis/tasks/08-28-sector-capital-radar/task.json index b7b6a52..55e43bc 100644 --- a/.trellis/tasks/08-28-sector-capital-radar/task.json +++ b/.trellis/tasks/08-28-sector-capital-radar/task.json @@ -3,7 +3,7 @@ "name": "sector-capital-radar", "title": "板块资金雷达模块", "description": "基于 Tushare point-in-time 事实独立生产收盘后板块资金排名、版本化指标、last-good API 与前端页面。", - "status": "planning", + "status": "in_progress", "dev_type": null, "scope": "fullstack", "package": null, @@ -12,7 +12,7 @@ "assignee": "yuxuanhui", "createdAt": "2026-08-28", "completedAt": null, - "branch": "codex/sector-capital-radar", + "branch": "codex/zijin", "base_branch": "develop", "worktree_path": null, "commit": null, diff --git a/zhixing-server/src/zhixing_server/modules/sector_radar/__init__.py b/zhixing-server/src/zhixing_server/modules/sector_radar/__init__.py new file mode 100644 index 0000000..04a13bb --- /dev/null +++ b/zhixing-server/src/zhixing_server/modules/sector_radar/__init__.py @@ -0,0 +1 @@ +"""Independent post-close sector capital radar bounded context.""" diff --git a/zhixing-server/src/zhixing_server/modules/sector_radar/domain/__init__.py b/zhixing-server/src/zhixing_server/modules/sector_radar/domain/__init__.py new file mode 100644 index 0000000..2a78e0a --- /dev/null +++ b/zhixing-server/src/zhixing_server/modules/sector_radar/domain/__init__.py @@ -0,0 +1 @@ +"""Storage-independent sector radar models and calculation rules.""" diff --git a/zhixing-server/src/zhixing_server/modules/sector_radar/domain/facts.py b/zhixing-server/src/zhixing_server/modules/sector_radar/domain/facts.py new file mode 100644 index 0000000..0f4104b --- /dev/null +++ b/zhixing-server/src/zhixing_server/modules/sector_radar/domain/facts.py @@ -0,0 +1,97 @@ +"""Point-in-time stock fact aggregation for sector radar metrics.""" + +from __future__ import annotations + +from collections.abc import Iterable +from decimal import Decimal + +from .models import ( + MembershipStatus, + SectorDailyAggregate, + SectorMembershipSnapshot, + StockDailyFact, + StockFactStatus, +) + + +def aggregate_sector_snapshot( + snapshot: SectorMembershipSnapshot, + stock_facts: Iterable[StockDailyFact], +) -> SectorDailyAggregate: + """Aggregate only the members recorded in one dated membership snapshot. + + Unknown membership returns an unavailable aggregate and deliberately + ignores any supplied stock facts. For known membership, suspended and + lifecycle-invalid members are excluded from the expected moneyflow + denominator; missing facts remain expected and reduce coverage. + + Args: + snapshot: Dated sector identity and point-in-time member codes. + stock_facts: Normalized facts that may contain records outside the sector. + + Returns: + A yuan-denominated aggregate with explicit membership and moneyflow coverage. + + Raises: + ValueError: If member facts have a date mismatch or duplicate stock code. + """ + + if snapshot.status is MembershipStatus.UNKNOWN: + return SectorDailyAggregate( + trade_date=snapshot.trade_date, + sector_type=snapshot.sector_type, + sector_code=snapshot.sector_code, + sector_name=snapshot.sector_name, + member_count=0, + valid_sample_count=0, + net_amount_yuan=None, + turnover_yuan=None, + membership_coverage=Decimal(0), + moneyflow_coverage=Decimal(0), + ) + + members = set(snapshot.member_codes) + facts_by_code: dict[str, StockDailyFact] = {} + for fact in stock_facts: + if fact.ts_code not in members: + continue + if fact.trade_date != snapshot.trade_date: + raise ValueError("member stock facts must match the snapshot trade_date") + if fact.ts_code in facts_by_code: + raise ValueError("member stock facts must have unique ts_code values") + facts_by_code[fact.ts_code] = fact + + net_amount_total = Decimal(0) + turnover_total = Decimal(0) + valid_count = 0 + expected_count = 0 + for member_code in snapshot.member_codes: + fact = facts_by_code.get(member_code) + if fact is None or fact.status is StockFactStatus.MISSING: + expected_count += 1 + elif fact.status is StockFactStatus.AVAILABLE: + expected_count += 1 + net_amount = fact.net_amount_yuan + turnover = fact.turnover_yuan + if net_amount is None or turnover is None: + raise ValueError("available stock facts require both amounts") + net_amount_total += net_amount + turnover_total += turnover + valid_count += 1 + + moneyflow_coverage = ( + Decimal(valid_count) / Decimal(expected_count) if expected_count else Decimal(1) + ) + + return SectorDailyAggregate( + trade_date=snapshot.trade_date, + sector_type=snapshot.sector_type, + sector_code=snapshot.sector_code, + sector_name=snapshot.sector_name, + member_count=len(snapshot.member_codes), + valid_sample_count=valid_count, + net_amount_yuan=net_amount_total if valid_count else None, + turnover_yuan=turnover_total if valid_count else None, + membership_coverage=Decimal(1), + moneyflow_coverage=moneyflow_coverage, + ) diff --git a/zhixing-server/src/zhixing_server/modules/sector_radar/domain/metrics.py b/zhixing-server/src/zhixing_server/modules/sector_radar/domain/metrics.py new file mode 100644 index 0000000..c665322 --- /dev/null +++ b/zhixing-server/src/zhixing_server/modules/sector_radar/domain/metrics.py @@ -0,0 +1,204 @@ +"""Transparent, versioned metric strategies for the independent radar.""" + +from __future__ import annotations + +from collections.abc import Iterable +from datetime import date +from decimal import Decimal +from typing import Protocol + +from .models import ( + MetricKind, + MetricObservation, + MetricQuality, + MetricUnit, + SectorDailyAggregate, +) + + +class MetricStrategy(Protocol): + """Calculate one named metric from a sector's point-in-time daily history.""" + + metric_kind: MetricKind + metric_version: str + unit: MetricUnit + + def evaluate( + self, + history: Iterable[SectorDailyAggregate], + target_trade_date: date, + ) -> MetricObservation: + """Return the target date observation without inventing missing inputs.""" + + ... + + +def _target_aggregate( + history: Iterable[SectorDailyAggregate], target_trade_date: date +) -> SectorDailyAggregate: + matches = tuple(row for row in history if row.trade_date == target_trade_date) + if len(matches) != 1: + raise ValueError("history must contain exactly one target-date aggregate") + return matches[0] + + +def _quality(row: SectorDailyAggregate) -> MetricQuality: + if row.valid_sample_count < 5 or row.membership_coverage < 1 or row.moneyflow_coverage < 1: + return MetricQuality.AVAILABLE_LIMITED_SAMPLE + return MetricQuality.AVAILABLE + + +def _observation( + row: SectorDailyAggregate, + *, + metric_kind: MetricKind, + metric_version: str, + unit: MetricUnit, + value: Decimal | None, + quality: MetricQuality | None = None, +) -> MetricObservation: + return MetricObservation( + trade_date=row.trade_date, + sector_type=row.sector_type, + sector_code=row.sector_code, + sector_name=row.sector_name, + metric_kind=metric_kind, + metric_version=metric_version, + implementation_kind="independent", + unit=unit, + value=value, + quality=( + MetricQuality.UNAVAILABLE + if value is None + else quality + if quality is not None + else _quality(row) + ), + member_count=row.member_count, + valid_sample_count=row.valid_sample_count, + membership_coverage=row.membership_coverage, + moneyflow_coverage=row.moneyflow_coverage, + ) + + +class AmountNetStrategy: + """Aggregate main net amount and expose it in hundred-million yuan.""" + + metric_kind = MetricKind.AMOUNT + metric_version = "zhixing_amount_net_bn_v1" + unit = MetricUnit.CNY_100M + + def evaluate( + self, + history: Iterable[SectorDailyAggregate], + target_trade_date: date, + ) -> MetricObservation: + """Return the target net amount; missing moneyflow remains unavailable.""" + + row = _target_aggregate(history, target_trade_date) + value = None if row.net_amount_yuan is None else row.net_amount_yuan / Decimal("100000000") + return _observation( + row, + metric_kind=self.metric_kind, + metric_version=self.metric_version, + unit=self.unit, + value=value, + ) + + +class RatioTurnoverStrategy: + """Divide aggregated main net amount by aggregated daily turnover.""" + + metric_kind = MetricKind.RATIO + metric_version = "zhixing_ratio_turnover_v1" + unit = MetricUnit.RATIO + + def evaluate( + self, + history: Iterable[SectorDailyAggregate], + target_trade_date: date, + ) -> MetricObservation: + """Return a ratio only when numerator and positive denominator exist.""" + + row = _target_aggregate(history, target_trade_date) + value = None + if ( + row.net_amount_yuan is not None + and row.turnover_yuan is not None + and row.turnover_yuan > 0 + ): + value = row.net_amount_yuan / row.turnover_yuan + return _observation( + row, + metric_kind=self.metric_kind, + metric_version=self.metric_version, + unit=self.unit, + value=value, + ) + + +class SwingEqualThreeToTenStrategy: + """Average transparent 3-to-10-day aggregate turnover ratios equally. + + This strategy is deliberately named as a Zhixing implementation. It does + not reproduce or imply OneChartLab's unpublished window weights or score. + """ + + metric_kind = MetricKind.SWING + metric_version = "zhixing_swing_equal_3_10_v1" + unit = MetricUnit.RATIO + + def evaluate( + self, + history: Iterable[SectorDailyAggregate], + target_trade_date: date, + ) -> MetricObservation: + """Calculate eight complete trading-day windows ending at the target.""" + + rows = tuple(sorted(history, key=lambda row: row.trade_date)) + target = _target_aggregate(rows, target_trade_date) + eligible = tuple(row for row in rows if row.trade_date <= target_trade_date) + if any( + (row.sector_type, row.sector_code) != (target.sector_type, target.sector_code) + for row in eligible + ): + raise ValueError("history must contain exactly one sector identity") + if len({row.trade_date for row in eligible}) != len(eligible): + raise ValueError("history must not contain duplicate trade dates") + + value: Decimal | None = None + quality: MetricQuality | None = None + if len(eligible) >= 10: + latest = eligible[-10:] + window_ratios: list[Decimal] = [] + for window_size in range(3, 11): + window = latest[-window_size:] + net_amount = Decimal(0) + turnover = Decimal(0) + for row in window: + if row.net_amount_yuan is None or row.turnover_yuan is None: + break + net_amount += row.net_amount_yuan + turnover += row.turnover_yuan + else: + if turnover <= 0: + break + window_ratios.append(net_amount / turnover) + continue + break + if len(window_ratios) == 8: + value = sum(window_ratios, start=Decimal(0)) / Decimal(8) + quality = ( + MetricQuality.AVAILABLE_LIMITED_SAMPLE + if any(_quality(row) is not MetricQuality.AVAILABLE for row in latest) + else MetricQuality.AVAILABLE + ) + + return _observation( + target, + metric_kind=self.metric_kind, + metric_version=self.metric_version, + unit=self.unit, + value=value, + quality=quality, + ) diff --git a/zhixing-server/src/zhixing_server/modules/sector_radar/domain/models.py b/zhixing-server/src/zhixing_server/modules/sector_radar/domain/models.py new file mode 100644 index 0000000..7ebf4c1 --- /dev/null +++ b/zhixing-server/src/zhixing_server/modules/sector_radar/domain/models.py @@ -0,0 +1,315 @@ +"""Stable domain values for independently produced sector radar metrics.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date, datetime +from decimal import Decimal +from enum import StrEnum +from typing import Literal + + +class SectorType(StrEnum): + """Independent ranking pools supported by the first radar release.""" + + CONCEPT = "concept" + INDUSTRY = "industry" + + +class MembershipStatus(StrEnum): + """Availability of a point-in-time sector membership snapshot.""" + + AVAILABLE = "available" + UNKNOWN = "membership_unknown" + + +class StockFactStatus(StrEnum): + """Why one member does or does not contribute to a daily aggregate.""" + + AVAILABLE = "available" + SUSPENDED = "suspended" + MISSING = "missing" + LIFECYCLE_INVALID = "lifecycle_invalid" + LOW_LIQUIDITY = "low_liquidity" + + +class PublicationStatus(StrEnum): + """Immutable build states retained for audit and last-good selection.""" + + RUNNING = "running" + SUCCESS = "success" + PARTIAL = "partial" + FAILED = "failed" + + +class MetricKind(StrEnum): + """User-facing metric families without borrowing private score names.""" + + AMOUNT = "amount" + RATIO = "ratio" + SWING = "swing" + + +class MetricQuality(StrEnum): + """Whether a metric is usable and whether its sample needs a warning.""" + + AVAILABLE = "available" + AVAILABLE_LIMITED_SAMPLE = "available_limited_sample" + UNAVAILABLE = "unavailable" + + +class MetricUnit(StrEnum): + """Units exposed by independent metric strategies.""" + + CNY_100M = "CNY_100M" + RATIO = "ratio" + + +class RankSide(StrEnum): + """Ordinary percentile views exposed by the ranking read model.""" + + TOP = "top" + BOTTOM = "bottom" + ALL = "all" + + +def _validate_finite_decimal(value: Decimal | None, field_name: str) -> None: + """Reject non-finite domain values while preserving missing values.""" + + if value is not None and not value.is_finite(): + raise ValueError(f"{field_name} must be finite or None") + + +def _validate_coverage(value: Decimal, field_name: str) -> None: + """Require a finite fraction in the inclusive zero-to-one range.""" + + _validate_finite_decimal(value, field_name) + if value < 0 or value > 1: + raise ValueError(f"{field_name} must be between 0 and 1") + + +@dataclass(frozen=True, slots=True) +class SectorMembershipSnapshot: + """One sector's membership as observed for exactly one trade date. + + ``UNKNOWN`` is an explicit fact: callers must not substitute a current + member list when the historical snapshot is unavailable. + """ + + trade_date: date + sector_type: SectorType + sector_code: str + sector_name: str + member_codes: tuple[str, ...] + status: MembershipStatus + source_version: str + + def __post_init__(self) -> None: + """Validate identity, deterministic membership, and unknown semantics.""" + + if not self.sector_code.strip(): + raise ValueError("sector_code must not be empty") + if not self.sector_name.strip(): + raise ValueError("sector_name must not be empty") + if not self.source_version.strip(): + raise ValueError("source_version must not be empty") + if any(not code.strip() for code in self.member_codes): + raise ValueError("member_codes must not contain empty values") + if len(self.member_codes) != len(set(self.member_codes)): + raise ValueError("member_codes must be unique") + if self.status is MembershipStatus.UNKNOWN and self.member_codes: + raise ValueError("unknown membership must not expose member_codes") + + +@dataclass(frozen=True, slots=True) +class StockDailyFact: + """Normalized daily turnover and moneyflow for one member. + + Amounts are expressed in yuan. Available facts require both source + values, including an observed zero. Non-available statuses cannot carry + amounts because doing so would blur missing, suspended, and lifecycle + semantics at the metric boundary. + """ + + trade_date: date + ts_code: str + status: StockFactStatus + turnover_yuan: Decimal | None = None + net_amount_yuan: Decimal | None = None + + def __post_init__(self) -> None: + """Reject incomplete available facts and hidden non-finite values.""" + + if not self.ts_code.strip(): + raise ValueError("ts_code must not be empty") + _validate_finite_decimal(self.turnover_yuan, "turnover_yuan") + _validate_finite_decimal(self.net_amount_yuan, "net_amount_yuan") + if self.status is StockFactStatus.AVAILABLE: + if self.turnover_yuan is None or self.net_amount_yuan is None: + raise ValueError("available stock facts require both amounts") + if self.turnover_yuan < 0: + raise ValueError("turnover_yuan must not be negative") + elif self.turnover_yuan is not None or self.net_amount_yuan is not None: + raise ValueError("non-available stock facts must not expose amounts") + + +@dataclass(frozen=True, slots=True) +class RadarPublication: + """Traceable identity and lifecycle of one immutable radar build revision.""" + + publication_id: str + target_trade_date: date + status: PublicationStatus + source_version: str + universe_version: str + metric_versions: tuple[str, ...] + input_hash: str | None + coverage: Decimal + started_at: datetime + finished_at: datetime | None = None + error_summary: str | None = None + + def __post_init__(self) -> None: + """Keep running and terminal lifecycle timestamps internally consistent.""" + + if not self.publication_id.strip(): + raise ValueError("publication_id must not be empty") + if not self.source_version.strip() or not self.universe_version.strip(): + raise ValueError("publication source versions must not be empty") + if not self.metric_versions or any(not value.strip() for value in self.metric_versions): + raise ValueError("metric_versions must contain named strategies") + if len(self.metric_versions) != len(set(self.metric_versions)): + raise ValueError("metric_versions must be unique") + _validate_coverage(self.coverage, "coverage") + if self.started_at.tzinfo is None: + raise ValueError("started_at must be timezone-aware") + is_running = self.status is PublicationStatus.RUNNING + if is_running != (self.finished_at is None): + raise ValueError("finished_at must be absent only while publication is running") + if self.finished_at is not None: + if self.finished_at.tzinfo is None: + raise ValueError("finished_at must be timezone-aware") + if self.finished_at < self.started_at: + raise ValueError("finished_at must not precede started_at") + if self.status is PublicationStatus.SUCCESS and self.input_hash is None: + raise ValueError("successful publication requires input_hash") + if self.input_hash is not None and ( + len(self.input_hash) != 64 + or any(character not in "0123456789abcdef" for character in self.input_hash) + ): + raise ValueError("input_hash must be a lowercase SHA-256 hex digest") + + +@dataclass(frozen=True, slots=True) +class SectorDailyAggregate: + """One sector's point-in-time daily facts after source normalization. + + Amounts use yuan so strategies cannot accidentally mix Tushare's + ``moneyflow_dc.net_amount`` (ten-thousand yuan) with ``daily.amount`` + (thousand yuan). ``None`` means missing source data; zero remains an + observed value. + """ + + trade_date: date + sector_type: SectorType + sector_code: str + sector_name: str + member_count: int + valid_sample_count: int + net_amount_yuan: Decimal | None + turnover_yuan: Decimal | None + membership_coverage: Decimal + moneyflow_coverage: Decimal + + def __post_init__(self) -> None: + """Validate counts, coverage, and finite normalized values.""" + + if not self.sector_code.strip(): + raise ValueError("sector_code must not be empty") + if not self.sector_name.strip(): + raise ValueError("sector_name must not be empty") + if self.member_count < 0: + raise ValueError("member_count must not be negative") + if not 0 <= self.valid_sample_count <= self.member_count: + raise ValueError("valid_sample_count must be within member_count") + _validate_finite_decimal(self.net_amount_yuan, "net_amount_yuan") + _validate_finite_decimal(self.turnover_yuan, "turnover_yuan") + _validate_coverage(self.membership_coverage, "membership_coverage") + _validate_coverage(self.moneyflow_coverage, "moneyflow_coverage") + + +@dataclass(frozen=True, slots=True) +class MetricObservation: + """One versioned independent metric value ready for cross-sectional ranking.""" + + trade_date: date + sector_type: SectorType + sector_code: str + sector_name: str + metric_kind: MetricKind + metric_version: str + implementation_kind: Literal["independent"] + unit: MetricUnit + value: Decimal | None + quality: MetricQuality + member_count: int + valid_sample_count: int + membership_coverage: Decimal + moneyflow_coverage: Decimal + + def __post_init__(self) -> None: + """Keep unavailable and finite-value states internally consistent.""" + + _validate_finite_decimal(self.value, "value") + if self.value is None and self.quality is not MetricQuality.UNAVAILABLE: + raise ValueError("a missing metric value must be unavailable") + if self.value is not None and self.quality is MetricQuality.UNAVAILABLE: + raise ValueError("an unavailable metric must not expose a value") + + +@dataclass(frozen=True, slots=True) +class RankChange: + """One previous-publication rank delta using past minus current rank.""" + + days: int + value: int | None + + def __post_init__(self) -> None: + """Limit the public comparison window to one through five days.""" + + if not 1 <= self.days <= 5: + raise ValueError("rank change days must be between 1 and 5") + + +@dataclass(frozen=True, slots=True) +class RankedMetric: + """A metric observation with its position inside one independent pool.""" + + observation: MetricObservation + rank_position: int | None + rank_percentile: Decimal | None + rank_changes: tuple[RankChange, ...] = () + + def __post_init__(self) -> None: + """Require rank position and percentile to be present or absent together.""" + + if (self.rank_position is None) != (self.rank_percentile is None): + raise ValueError("rank_position and rank_percentile must be paired") + if self.rank_position is not None and self.rank_position < 1: + raise ValueError("rank_position must be positive") + _validate_finite_decimal(self.rank_percentile, "rank_percentile") + if self.rank_percentile is not None and not 0 < self.rank_percentile <= 100: + raise ValueError("rank_percentile must be within (0, 100]") + days = [change.days for change in self.rank_changes] + if len(days) != len(set(days)): + raise ValueError("rank change days must be unique") + + def rank_change(self, days: int) -> int | None: + """Return one configured rank delta, or ``None`` when history is absent.""" + + if not 1 <= days <= 5: + raise ValueError("rank change days must be between 1 and 5") + return next( + (change.value for change in self.rank_changes if change.days == days), + None, + ) diff --git a/zhixing-server/src/zhixing_server/modules/sector_radar/domain/ranking.py b/zhixing-server/src/zhixing_server/modules/sector_radar/domain/ranking.py new file mode 100644 index 0000000..d39bdc6 --- /dev/null +++ b/zhixing-server/src/zhixing_server/modules/sector_radar/domain/ranking.py @@ -0,0 +1,216 @@ +"""Deterministic cross-sectional ranking for independent sector pools.""" + +from __future__ import annotations + +from collections import defaultdict +from collections.abc import Iterable, Mapping +from dataclasses import replace +from datetime import date +from decimal import Decimal + +from .models import ( + MetricKind, + MetricObservation, + RankChange, + RankedMetric, + RankSide, + SectorType, +) + +PoolKey = tuple[date, SectorType, MetricKind, str] +SectorMetricKey = tuple[SectorType, str, MetricKind, str] + + +def _pool_key(observation: MetricObservation) -> PoolKey: + return ( + observation.trade_date, + observation.sector_type, + observation.metric_kind, + observation.metric_version, + ) + + +def _sector_metric_key(observation: MetricObservation) -> SectorMetricKey: + return ( + observation.sector_type, + observation.sector_code, + observation.metric_kind, + observation.metric_version, + ) + + +def _available_sort_key(observation: MetricObservation) -> tuple[Decimal, str]: + if observation.value is None: + raise ValueError("unavailable observations cannot use the ranking sort key") + return (-observation.value, observation.sector_code) + + +def rank_metric_observations( + observations: Iterable[MetricObservation], +) -> tuple[RankedMetric, ...]: + """Rank observations by value within date, type, metric, and version. + + Concept and industry observations never share a pool. Equal metric values + use ascending sector code as the documented Zhixing tie-breaker. Missing + values remain visible but do not consume a rank. + """ + + pools: defaultdict[PoolKey, list[MetricObservation]] = defaultdict(list) + for observation in observations: + pools[_pool_key(observation)].append(observation) + + result: list[RankedMetric] = [] + for pool_key in sorted( + pools, + key=lambda key: (key[0], key[1].value, key[2].value, key[3]), + ): + pool = pools[pool_key] + codes = [observation.sector_code for observation in pool] + if len(codes) != len(set(codes)): + raise ValueError("a ranking pool must not contain duplicate sector codes") + + available = sorted( + (observation for observation in pool if observation.value is not None), + key=_available_sort_key, + ) + pool_size = len(available) + for rank_position, observation in enumerate(available, start=1): + rank_percentile = ( + Decimal(100) * Decimal(pool_size - rank_position + 1) / Decimal(pool_size) + ) + result.append( + RankedMetric( + observation=observation, + rank_position=rank_position, + rank_percentile=rank_percentile, + ) + ) + + result.extend( + RankedMetric( + observation=observation, + rank_position=None, + rank_percentile=None, + ) + for observation in sorted( + (observation for observation in pool if observation.value is None), + key=lambda observation: observation.sector_code, + ) + ) + return tuple(result) + + +def select_percentile_side( + rankings: Iterable[RankedMetric], side: RankSide +) -> tuple[RankedMetric, ...]: + """Select confirmed inclusive percentile sides without fixed row counts.""" + + rows = tuple(rankings) + if side is RankSide.ALL: + return rows + + threshold_rows = tuple( + row + for row in rows + if row.rank_percentile is not None + and ( + row.rank_percentile >= Decimal(90) + if side is RankSide.TOP + else row.rank_percentile <= Decimal(10) + ) + ) + if side is RankSide.TOP: + return threshold_rows + + pools: defaultdict[PoolKey, list[RankedMetric]] = defaultdict(list) + for row in threshold_rows: + pools[_pool_key(row.observation)].append(row) + result: list[RankedMetric] = [] + for pool_key in sorted( + pools, + key=lambda key: (key[0], key[1].value, key[2].value, key[3]), + ): + result.extend( + sorted( + pools[pool_key], + key=lambda row: ( + row.observation.value if row.observation.value is not None else Decimal(0), + row.observation.sector_code, + ), + ) + ) + return tuple(result) + + +def with_rank_changes( + current_rankings: Iterable[RankedMetric], + history_by_days: Mapping[int, Iterable[RankedMetric]], +) -> tuple[RankedMetric, ...]: + """Attach 1-to-5-day deltas without turning missing history into zero.""" + + history_indexes: dict[int, dict[SectorMetricKey, int | None]] = {} + for days, historical_rankings in history_by_days.items(): + if not 1 <= days <= 5: + raise ValueError("rank change days must be between 1 and 5") + index: dict[SectorMetricKey, int | None] = {} + for row in historical_rankings: + key = _sector_metric_key(row.observation) + if key in index: + raise ValueError("historical rankings must have unique sector metrics") + index[key] = row.rank_position + history_indexes[days] = index + + result: list[RankedMetric] = [] + for row in current_rankings: + key = _sector_metric_key(row.observation) + changes: list[RankChange] = [] + for days in sorted(history_indexes): + past_rank = history_indexes[days].get(key) + value = ( + past_rank - row.rank_position + if past_rank is not None and row.rank_position is not None + else None + ) + changes.append(RankChange(days=days, value=value)) + result.append(replace(row, rank_changes=tuple(changes))) + return tuple(result) + + +def select_rank_change_side( + rankings: Iterable[RankedMetric], + *, + days: int, + side: RankSide, +) -> tuple[RankedMetric, ...]: + """Select the strongest or weakest ceiling-ten-percent rank changes per pool.""" + + if not 1 <= days <= 5: + raise ValueError("rank change days must be between 1 and 5") + pools: defaultdict[PoolKey, list[RankedMetric]] = defaultdict(list) + for row in rankings: + pools[_pool_key(row.observation)].append(row) + + result: list[RankedMetric] = [] + for pool_key in sorted( + pools, + key=lambda key: (key[0], key[1].value, key[2].value, key[3]), + ): + pool = pools[pool_key] + pool_size = sum(row.rank_position is not None for row in pool) + take_count = max(1, (pool_size + 9) // 10) if pool_size else 0 + candidates = tuple( + (change, row) for row in pool if (change := row.rank_change(days)) is not None + ) + if side is RankSide.BOTTOM: + ordered = sorted( + candidates, + key=lambda item: (item[0], item[1].observation.sector_code), + ) + else: + ordered = sorted( + candidates, + key=lambda item: (-item[0], item[1].observation.sector_code), + ) + selected = ordered if side is RankSide.ALL else ordered[:take_count] + result.extend(row for _, row in selected) + return tuple(result) diff --git a/zhixing-server/tests/unit/sector_radar/test_facts.py b/zhixing-server/tests/unit/sector_radar/test_facts.py new file mode 100644 index 0000000..6e88ab9 --- /dev/null +++ b/zhixing-server/tests/unit/sector_radar/test_facts.py @@ -0,0 +1,146 @@ +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, + ) diff --git a/zhixing-server/tests/unit/sector_radar/test_metrics.py b/zhixing-server/tests/unit/sector_radar/test_metrics.py new file mode 100644 index 0000000..8d49fca --- /dev/null +++ b/zhixing-server/tests/unit/sector_radar/test_metrics.py @@ -0,0 +1,112 @@ +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 diff --git a/zhixing-server/tests/unit/sector_radar/test_ranking.py b/zhixing-server/tests/unit/sector_radar/test_ranking.py new file mode 100644 index 0000000..5bda433 --- /dev/null +++ b/zhixing-server/tests/unit/sector_radar/test_ranking.py @@ -0,0 +1,147 @@ +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", + ]