diff --git a/.env.example b/.env.example index 3d1199c..5a170bc 100644 --- a/.env.example +++ b/.env.example @@ -23,6 +23,11 @@ ZHIXING_MARKET_DATA_MAX_RETRIES=3 ZHIXING_MARKET_DATA_RETRY_BACKOFF_SECONDS=1.0 ZHIXING_MARKET_DATA_REQUEST_INTERVAL_SECONDS=0.2 ZHIXING_MARKET_DATA_ADVISORY_LOCK_KEY=7380521 +ZHIXING_SECTOR_RADAR_COVERAGE_THRESHOLD=0.99 +ZHIXING_SECTOR_RADAR_MAX_RETRIES=3 +ZHIXING_SECTOR_RADAR_RETRY_BACKOFF_SECONDS=1.0 +ZHIXING_SECTOR_RADAR_REQUEST_INTERVAL_SECONDS=0.2 +ZHIXING_SECTOR_RADAR_ADVISORY_LOCK_KEY=7380522 ZHIXING_SELECTION_MAX_WORKERS=4 ZHIXING_SELECTION_BATCH_SIZE=200 API_UPSTREAM=http://server:8000 diff --git a/.trellis/tasks/08-28-sector-capital-radar/implement.md b/.trellis/tasks/08-28-sector-capital-radar/implement.md index 75d3f38..1afd9d2 100644 --- a/.trellis/tasks/08-28-sector-capital-radar/implement.md +++ b/.trellis/tasks/08-28-sector-capital-radar/implement.md @@ -30,10 +30,12 @@ ## 3. 构建 Job -- [ ] 实现 `BuildSectorRadar.execute` 的单日与日期区间编排、质量屏障、publication 状态和失败保留 last-good。 -- [ ] 新增 `sector-radar-build` CLI 及退出码；支持目标日、回填区间和失败 publication 重试。 -- [ ] 增加 Compose job service，但不启用生产定时；更新运行文档与无凭据示例。 -- [ ] 用 fake/golden 验证完整成功、部分数据、截断响应、重复运行、输入修订、并发锁和失败降级。 +- [x] 实现 `BuildSectorRadar.execute` 的单日与日期区间编排、质量屏障、publication 状态和失败保留 last-good。 +- [x] 新增 `sector-radar-build` CLI 及退出码；支持目标日、回填区间和失败 publication 重试。 +- [x] 增加 Compose job service，但不启用生产定时；更新运行文档与无凭据示例。 +- [x] 用 fake/golden 验证完整成功、部分数据、截断响应、重复运行、输入修订、并发锁和失败降级。 + +阶段结果（2026-08-29）：单日/区间构建、上海时区最近已收盘日、provisional publication、同日锁、遗留 running 接管、内容 hash 去重、严格 last-good 与来源组检查点均已落地。failed 重试只补未完成来源组，partial 只刷新显式覆盖缺口；规范事实、日聚合、排名与 terminal publication 由 PostgreSQL 单事务完成。CLI、开发/生产 Compose entrypoint 和运行文档已提供。完整后端门禁为 122 passed、3 skipped；迁移头与离线升级 SQL、四种 Compose config 和 CLI help 已通过。当前未设置 `ZHIXING_TEST_DATABASE_URL`，三项真实 PostgreSQL 集成测试未执行；真实 Tushare capability 与数据到达时点也未在本阶段宣称通过。 ## 4. HTTP 读取链 diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 605f5a8..ba70033 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -36,6 +36,11 @@ services: ZHIXING_MARKET_DATA_RETRY_BACKOFF_SECONDS: ${ZHIXING_MARKET_DATA_RETRY_BACKOFF_SECONDS:-1.0} ZHIXING_MARKET_DATA_REQUEST_INTERVAL_SECONDS: ${ZHIXING_MARKET_DATA_REQUEST_INTERVAL_SECONDS:-0.2} ZHIXING_MARKET_DATA_ADVISORY_LOCK_KEY: ${ZHIXING_MARKET_DATA_ADVISORY_LOCK_KEY:-7380521} + ZHIXING_SECTOR_RADAR_COVERAGE_THRESHOLD: ${ZHIXING_SECTOR_RADAR_COVERAGE_THRESHOLD:-0.99} + ZHIXING_SECTOR_RADAR_MAX_RETRIES: ${ZHIXING_SECTOR_RADAR_MAX_RETRIES:-3} + ZHIXING_SECTOR_RADAR_RETRY_BACKOFF_SECONDS: ${ZHIXING_SECTOR_RADAR_RETRY_BACKOFF_SECONDS:-1.0} + ZHIXING_SECTOR_RADAR_REQUEST_INTERVAL_SECONDS: ${ZHIXING_SECTOR_RADAR_REQUEST_INTERVAL_SECONDS:-0.2} + ZHIXING_SECTOR_RADAR_ADVISORY_LOCK_KEY: ${ZHIXING_SECTOR_RADAR_ADVISORY_LOCK_KEY:-7380522} ZHIXING_SELECTION_MAX_WORKERS: ${ZHIXING_SELECTION_MAX_WORKERS:-4} ZHIXING_SELECTION_BATCH_SIZE: ${ZHIXING_SELECTION_BATCH_SIZE:-200} ZHIXING_TUSHARE_TOKEN: ${ZHIXING_TUSHARE_TOKEN:-} @@ -112,6 +117,28 @@ services: - market-data:/app/data/market-data - server-venv:/app/.venv + sector-radar-build: + profiles: ["jobs"] + build: + context: ./zhixing-server + target: development + entrypoint: ["uv", "run", "sector-radar-build"] + command: [] + depends_on: + migrate: + condition: service_completed_successfully + environment: + ZHIXING_DATABASE_URL: ${ZHIXING_DATABASE_URL:-postgresql://zhixing:zhixing@postgres:5432/zhixing} + ZHIXING_SECTOR_RADAR_COVERAGE_THRESHOLD: ${ZHIXING_SECTOR_RADAR_COVERAGE_THRESHOLD:-0.99} + ZHIXING_SECTOR_RADAR_MAX_RETRIES: ${ZHIXING_SECTOR_RADAR_MAX_RETRIES:-3} + ZHIXING_SECTOR_RADAR_RETRY_BACKOFF_SECONDS: ${ZHIXING_SECTOR_RADAR_RETRY_BACKOFF_SECONDS:-1.0} + ZHIXING_SECTOR_RADAR_REQUEST_INTERVAL_SECONDS: ${ZHIXING_SECTOR_RADAR_REQUEST_INTERVAL_SECONDS:-0.2} + ZHIXING_SECTOR_RADAR_ADVISORY_LOCK_KEY: ${ZHIXING_SECTOR_RADAR_ADVISORY_LOCK_KEY:-7380522} + ZHIXING_TUSHARE_TOKEN: ${ZHIXING_TUSHARE_TOKEN:-} + volumes: + - ./zhixing-server:/app + - server-venv:/app/.venv + volumes: market-data: postgres-data: diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 01d77a4..6557ec0 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -18,6 +18,11 @@ services: ZHIXING_MARKET_DATA_RETRY_BACKOFF_SECONDS: ${ZHIXING_MARKET_DATA_RETRY_BACKOFF_SECONDS:-1.0} ZHIXING_MARKET_DATA_REQUEST_INTERVAL_SECONDS: ${ZHIXING_MARKET_DATA_REQUEST_INTERVAL_SECONDS:-0.2} ZHIXING_MARKET_DATA_ADVISORY_LOCK_KEY: ${ZHIXING_MARKET_DATA_ADVISORY_LOCK_KEY:-7380521} + ZHIXING_SECTOR_RADAR_COVERAGE_THRESHOLD: ${ZHIXING_SECTOR_RADAR_COVERAGE_THRESHOLD:-0.99} + ZHIXING_SECTOR_RADAR_MAX_RETRIES: ${ZHIXING_SECTOR_RADAR_MAX_RETRIES:-3} + ZHIXING_SECTOR_RADAR_RETRY_BACKOFF_SECONDS: ${ZHIXING_SECTOR_RADAR_RETRY_BACKOFF_SECONDS:-1.0} + ZHIXING_SECTOR_RADAR_REQUEST_INTERVAL_SECONDS: ${ZHIXING_SECTOR_RADAR_REQUEST_INTERVAL_SECONDS:-0.2} + ZHIXING_SECTOR_RADAR_ADVISORY_LOCK_KEY: ${ZHIXING_SECTOR_RADAR_ADVISORY_LOCK_KEY:-7380522} ZHIXING_SELECTION_MAX_WORKERS: ${ZHIXING_SELECTION_MAX_WORKERS:-4} ZHIXING_SELECTION_BATCH_SIZE: ${ZHIXING_SELECTION_BATCH_SIZE:-200} ZHIXING_TUSHARE_TOKEN: ${ZHIXING_TUSHARE_TOKEN:-} @@ -111,6 +116,28 @@ services: networks: - 1panel-network + sector-radar-build: + profiles: ["jobs"] + build: + context: ./zhixing-server + target: production + entrypoint: ["sector-radar-build"] + command: [] + depends_on: + migrate: + condition: service_completed_successfully + environment: + TZ: Asia/Shanghai + ZHIXING_DATABASE_URL: ${ZHIXING_DATABASE_URL:?Set ZHIXING_DATABASE_URL to the 1Panel PostgreSQL URL} + ZHIXING_SECTOR_RADAR_COVERAGE_THRESHOLD: ${ZHIXING_SECTOR_RADAR_COVERAGE_THRESHOLD:-0.99} + ZHIXING_SECTOR_RADAR_MAX_RETRIES: ${ZHIXING_SECTOR_RADAR_MAX_RETRIES:-3} + ZHIXING_SECTOR_RADAR_RETRY_BACKOFF_SECONDS: ${ZHIXING_SECTOR_RADAR_RETRY_BACKOFF_SECONDS:-1.0} + ZHIXING_SECTOR_RADAR_REQUEST_INTERVAL_SECONDS: ${ZHIXING_SECTOR_RADAR_REQUEST_INTERVAL_SECONDS:-0.2} + ZHIXING_SECTOR_RADAR_ADVISORY_LOCK_KEY: ${ZHIXING_SECTOR_RADAR_ADVISORY_LOCK_KEY:-7380522} + ZHIXING_TUSHARE_TOKEN: ${ZHIXING_TUSHARE_TOKEN:-} + networks: + - 1panel-network + volumes: market-data: diff --git a/docs/market-data-sync.md b/docs/market-data-sync.md index 7cb162b..6567d9c 100644 --- a/docs/market-data-sync.md +++ b/docs/market-data-sync.md @@ -95,3 +95,29 @@ docker compose -f docker-compose.prod.yml --profile jobs config ``` 真实 PostgreSQL 迁移和批量 upsert 集成测试使用 `ZHIXING_TEST_DATABASE_URL` 显式开启；普通单元测试不会访问网络、Tushare 或数据库。 + +## 板块资金雷达 Job + +`sector-radar-build` 同样是外部调度器触发的一次性任务，FastAPI 不会在进程内启动定时器。它只读取 Tushare 的 `trade_cal`、`dc_index`、`dc_member`、`stock_basic`、`suspend_d`、`daily` 和 `moneyflow_dc`，保存 point-in-time 原始快照与规范化事实，再生成明确标注为“知行独立实现”的版本化指标。生产运行时不请求 OneChartLab。 + +开发环境没有 token 时可以检查命令契约，但不能执行真实构建： + +```bash +cd zhixing-server +uv run sector-radar-build --help +``` + +提供 `ZHIXING_TUSHARE_TOKEN` 并完成迁移后，可构建单日、按交易日顺序回填区间，或从一个 `partial`/`failed` publication 的来源检查点继续重试。失败 publication 会复用此前已成功保存的来源组；覆盖率不足的 partial 只刷新被标记为缺口的 `daily` 或 `moneyflow_dc`，不会全量重采： + +```bash +docker compose -f docker-compose.prod.yml --profile jobs run --rm sector-radar-build \ + --trade-date 2026-08-28 + +docker compose -f docker-compose.prod.yml --profile jobs run --rm sector-radar-build \ + --start-date 2026-08-18 --end-date 2026-08-28 + +docker compose -f docker-compose.prod.yml --profile jobs run --rm sector-radar-build \ + --retry-publication-id +``` + +重复输入通过内容 hash 复用已有成功发布，不产生无意义修订；同一目标日由 PostgreSQL advisory lock 阻止并发构建。`success` 或 `unchanged` 返回 0，覆盖率不足的 `partial` 返回 2，输入、上游、锁或基础设施失败返回 1。`partial`/`failed` 会保留审计，但读取端只选择 `success` 作为 last-good。当前版本只提供手工和外部调度入口，不新增生产 Cron；待真实账号 capability、到达时点和首轮回填验证完成后再单独启用调度。 diff --git a/zhixing-server/migrations/versions/0005_radar_daily_aggregate.py b/zhixing-server/migrations/versions/0005_radar_daily_aggregate.py new file mode 100644 index 0000000..bca446a --- /dev/null +++ b/zhixing-server/migrations/versions/0005_radar_daily_aggregate.py @@ -0,0 +1,155 @@ +"""Persist publication-owned sector daily aggregates for exact metric replay.""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "0005_radar_daily_aggregate" +down_revision: str | None = "0004_sector_radar" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Create source recovery links and exact multi-day metric inputs.""" + + op.create_table( + "sector_radar_publication_source", + sa.Column( + "publication_id", + sa.String(64), + sa.ForeignKey("sector_radar_publication.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column("source_group", sa.String(32), nullable=False), + sa.Column("source_order", sa.Integer(), nullable=False), + sa.Column( + "refresh_on_retry", + sa.Boolean(), + nullable=False, + server_default=sa.false(), + ), + sa.Column( + "source_snapshot_id", + sa.String(64), + sa.ForeignKey("sector_radar_source_snapshot.id", ondelete="RESTRICT"), + nullable=False, + ), + sa.PrimaryKeyConstraint("publication_id", "source_group", "source_order"), + sa.UniqueConstraint( + "publication_id", + "source_group", + "source_snapshot_id", + name="uq_sector_radar_publication_source_snapshot", + ), + sa.CheckConstraint( + "source_group IN ('calendar', 'concept_indices', 'industry_indices', " + "'members', 'stock_basics', 'suspensions', 'daily', 'moneyflow_dc')", + name="ck_sector_radar_publication_source_group", + ), + sa.CheckConstraint( + "source_order >= 0", + name="ck_sector_radar_publication_source_order", + ), + ) + op.create_index( + "ix_sector_radar_publication_source_snapshot", + "sector_radar_publication_source", + ["source_snapshot_id"], + ) + + op.create_unique_constraint( + "uq_sector_radar_publication_id_date", + "sector_radar_publication", + ["id", "target_trade_date"], + ) + op.create_foreign_key( + "fk_sector_radar_ranking_publication_date", + "sector_radar_ranking", + "sector_radar_publication", + ["publication_id", "trade_date"], + ["id", "target_trade_date"], + ondelete="CASCADE", + ) + + op.create_table( + "sector_radar_daily_aggregate", + sa.Column("publication_id", sa.String(64), nullable=False), + sa.Column("trade_date", sa.Date(), nullable=False), + sa.Column("sector_type", sa.String(16), nullable=False), + sa.Column("sector_code", sa.String(16), nullable=False), + sa.Column("sector_name", sa.String(128), nullable=False), + sa.Column("member_count", sa.Integer(), nullable=False), + sa.Column("valid_sample_count", sa.Integer(), nullable=False), + sa.Column("net_amount_yuan", sa.Numeric(28, 6)), + sa.Column("turnover_yuan", sa.Numeric(28, 6)), + sa.Column("membership_coverage", sa.Numeric(8, 6), nullable=False), + sa.Column("moneyflow_coverage", sa.Numeric(8, 6), nullable=False), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("now()"), + ), + sa.PrimaryKeyConstraint("publication_id", "sector_type", "sector_code"), + sa.ForeignKeyConstraint( + ["publication_id", "trade_date"], + ["sector_radar_publication.id", "sector_radar_publication.target_trade_date"], + name="fk_sector_radar_daily_aggregate_publication_date", + ondelete="CASCADE", + ), + sa.CheckConstraint( + "sector_type IN ('concept', 'industry')", + name="ck_sector_radar_daily_aggregate_type", + ), + sa.CheckConstraint( + "member_count >= 0 AND valid_sample_count >= 0 AND valid_sample_count <= member_count", + name="ck_sector_radar_daily_aggregate_counts", + ), + sa.CheckConstraint( + "membership_coverage >= 0 AND membership_coverage <= 1 " + "AND moneyflow_coverage >= 0 AND moneyflow_coverage <= 1", + name="ck_sector_radar_daily_aggregate_coverage", + ), + sa.CheckConstraint( + "turnover_yuan IS NULL OR (turnover_yuan >= 0 AND " + "turnover_yuan NOT IN ('NaN'::numeric, 'Infinity'::numeric))", + name="ck_sector_radar_daily_aggregate_turnover", + ), + sa.CheckConstraint( + "net_amount_yuan IS NULL OR net_amount_yuan NOT IN " + "('NaN'::numeric, 'Infinity'::numeric, '-Infinity'::numeric)", + name="ck_sector_radar_daily_aggregate_net_amount", + ), + ) + op.create_index( + "ix_sector_radar_daily_aggregate_history", + "sector_radar_daily_aggregate", + ["trade_date", "sector_type", "sector_code"], + ) + + +def downgrade() -> None: + """Drop only the replay aggregate extension.""" + + op.drop_index( + "ix_sector_radar_daily_aggregate_history", + table_name="sector_radar_daily_aggregate", + ) + op.drop_table("sector_radar_daily_aggregate") + op.drop_constraint( + "fk_sector_radar_ranking_publication_date", + "sector_radar_ranking", + type_="foreignkey", + ) + op.drop_constraint( + "uq_sector_radar_publication_id_date", + "sector_radar_publication", + type_="unique", + ) + op.drop_index( + "ix_sector_radar_publication_source_snapshot", + table_name="sector_radar_publication_source", + ) + op.drop_table("sector_radar_publication_source") diff --git a/zhixing-server/pyproject.toml b/zhixing-server/pyproject.toml index 024b63b..84ccb87 100644 --- a/zhixing-server/pyproject.toml +++ b/zhixing-server/pyproject.toml @@ -35,6 +35,7 @@ packages = ["src/zhixing_server"] [project.scripts] market-data-sync = "zhixing_server.modules.market_data.presentation.cli:main" +sector-radar-build = "zhixing_server.modules.sector_radar.presentation.cli:main" [tool.pytest.ini_options] addopts = "-ra --strict-config --strict-markers" diff --git a/zhixing-server/src/zhixing_server/modules/sector_radar/application/__init__.py b/zhixing-server/src/zhixing_server/modules/sector_radar/application/__init__.py new file mode 100644 index 0000000..698aef6 --- /dev/null +++ b/zhixing-server/src/zhixing_server/modules/sector_radar/application/__init__.py @@ -0,0 +1 @@ +"""Application use cases for sector radar production and reads.""" diff --git a/zhixing-server/src/zhixing_server/modules/sector_radar/application/build.py b/zhixing-server/src/zhixing_server/modules/sector_radar/application/build.py new file mode 100644 index 0000000..5f5c406 --- /dev/null +++ b/zhixing-server/src/zhixing_server/modules/sector_radar/application/build.py @@ -0,0 +1,696 @@ +"""One-shot, idempotent sector radar publication orchestration.""" + +from __future__ import annotations + +import hashlib +import json +from collections import defaultdict +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass, replace +from datetime import UTC, date, datetime, time, timedelta +from decimal import Decimal +from typing import Literal +from uuid import uuid4 +from zoneinfo import ZoneInfo + +from ..domain.facts import aggregate_sector_snapshot +from ..domain.metrics import ( + AmountNetStrategy, + MetricStrategy, + RatioTurnoverStrategy, + SwingEqualThreeToTenStrategy, +) +from ..domain.models import ( + MembershipStatus, + MetricObservation, + PublicationStatus, + RadarPublication, + RankedMetric, + SectorDailyAggregate, + SectorMembershipSnapshot, + SectorType, + StockDailyFact, + StockFactStatus, +) +from ..domain.normalize import normalize_memberships, normalize_stock_facts +from ..domain.persistence import ( + DailyAggregateRecord, + MembershipRecord, + PublicationSourceGroup, + PublicationSourceRecord, + RankingRecord, + SectorRadarRepository, + StockFactRecord, +) +from ..domain.ports import SectorRadarSource +from ..domain.ranking import rank_metric_observations, with_rank_changes +from ..domain.source import ( + DailyRow, + MoneyflowDcRow, + SectorIndexRow, + SectorMemberRow, + SourceResult, + SourceScalar, + SourceSnapshot, + StockBasicRow, + SuspendRow, + TradeCalendarRow, +) + +BuildOutcomeStatus = Literal["success", "partial", "failed", "locked", "unchanged"] +SHANGHAI = ZoneInfo("Asia/Shanghai") +MARKET_DATA_READY_TIME = time(15, 30) + + +@dataclass(frozen=True, slots=True) +class BuildSectorRadarCommand: + """Select one date, an inclusive range, or a failed publication retry.""" + + trade_date: date | None = None + start_date: date | None = None + end_date: date | None = None + retry_publication_id: str | None = None + + def __post_init__(self) -> None: + """Reject ambiguous build modes before any source or database call.""" + + has_range = self.start_date is not None or self.end_date is not None + modes = sum( + ( + self.trade_date is not None, + has_range, + self.retry_publication_id is not None, + ) + ) + if modes > 1: + raise ValueError("trade date, date range, and retry publication are mutually exclusive") + if has_range and (self.start_date is None or self.end_date is None): + raise ValueError("date range requires both start_date and end_date") + if ( + self.start_date is not None + and self.end_date is not None + and self.end_date < self.start_date + ): + raise ValueError("end_date must not precede start_date") + if self.retry_publication_id is not None and not self.retry_publication_id.strip(): + raise ValueError("retry_publication_id must not be empty") + + +@dataclass(frozen=True, slots=True) +class BuildDateOutcome: + """Redacted result for one target trade date.""" + + target_trade_date: date + status: BuildOutcomeStatus + publication_id: str | None + coverage: Decimal + sector_count: int + ranking_count: int + error_type: str | None = None + error_message: str | None = None + + def as_dict(self) -> dict[str, object]: + """Serialize without raw payloads, credentials, or provider exception text.""" + + return { + "target_trade_date": self.target_trade_date.isoformat(), + "status": self.status, + "publication_id": self.publication_id, + "coverage": str(self.coverage), + "sector_count": self.sector_count, + "ranking_count": self.ranking_count, + "error_type": self.error_type, + "error_message": self.error_message, + } + + +@dataclass(frozen=True, slots=True) +class BuildSummary: + """Cron-friendly aggregate result for one CLI invocation.""" + + outcomes: tuple[BuildDateOutcome, ...] + + @property + def status(self) -> str: + """Return the worst invocation state.""" + + if not self.outcomes or any(item.status in {"failed", "locked"} for item in self.outcomes): + return "failed" + if any(item.status == "partial" for item in self.outcomes): + return "partial" + if all(item.status == "unchanged" for item in self.outcomes): + return "unchanged" + return "success" + + @property + def exit_code(self) -> int: + """Return 0 for usable success, 2 for incomplete input, and 1 for failure.""" + + if self.status == "failed": + return 1 + if self.status == "partial": + return 2 + return 0 + + def as_dict(self) -> dict[str, object]: + """Serialize the invocation summary for external schedulers.""" + + return { + "status": self.status, + "exit_code": self.exit_code, + "outcomes": [item.as_dict() for item in self.outcomes], + } + + +class BuildSectorRadar: + """Hide target resolution, source replay, metrics, ranking, and publication switching.""" + + source_version = "tushare-pro-v1" + + def __init__( + self, + source: SectorRadarSource, + repository: SectorRadarRepository, + *, + coverage_threshold: Decimal = Decimal("0.99"), + today: date | None = None, + now_fn: Callable[[], datetime] = lambda: datetime.now(UTC), + strategies: Sequence[MetricStrategy] | None = None, + ) -> None: + if not Decimal(0) <= coverage_threshold <= Decimal(1): + raise ValueError("coverage_threshold must be between 0 and 1") + self.now_fn = now_fn + self.source = source + self.repository = repository + self.coverage_threshold = coverage_threshold + self.today = today or self.now_fn().astimezone(SHANGHAI).date() + self.strategies = tuple( + strategies + or ( + AmountNetStrategy(), + RatioTurnoverStrategy(), + SwingEqualThreeToTenStrategy(), + ) + ) + + def execute(self, command: BuildSectorRadarCommand | None = None) -> BuildSummary: + """Build each selected trade date sequentially for deterministic history.""" + + command = command or BuildSectorRadarCommand() + try: + targets = self._resolve_targets(command) + except Exception as exc: + target = command.trade_date or command.start_date or self.today + error_type, message = self._safe_failure(exc) + return BuildSummary( + ( + BuildDateOutcome( + target, + "failed", + None, + Decimal(0), + 0, + 0, + error_type, + message, + ), + ) + ) + return BuildSummary(tuple(self._build_target(target) for target in targets)) + + def _resolve_targets(self, command: BuildSectorRadarCommand) -> tuple[_BuildTarget, ...]: + if command.retry_publication_id is not None: + publication = self.repository.get_publication(command.retry_publication_id) + if publication is None: + raise ValueError("retry publication does not exist") + if publication.status not in {PublicationStatus.PARTIAL, PublicationStatus.FAILED}: + raise ValueError("only partial or failed publications can be retried") + return (_BuildTarget(publication.target_trade_date, publication.publication_id),) + + if command.trade_date is not None: + start = end = command.trade_date + elif command.start_date is not None and command.end_date is not None: + start, end = command.start_date, command.end_date + else: + end = self._default_calendar_end() + start = end - timedelta(days=14) + calendar = self.source.fetch_trade_calendar(start, end) + targets = tuple(sorted({row.cal_date for row in calendar.rows if row.is_open})) + if command.trade_date is not None and command.trade_date not in targets: + raise ValueError("target date is not an open trading day") + if not targets: + raise ValueError("no open trading date found") + selected = targets if command.start_date is not None else (targets[-1],) + return tuple(_BuildTarget(target) for target in selected) + + def _default_calendar_end(self) -> date: + """Exclude today's session until Tushare closing facts are expected to be ready.""" + + local_now = self.now_fn().astimezone(SHANGHAI) + if self.today == local_now.date() and local_now.time() < MARKET_DATA_READY_TIME: + return self.today - timedelta(days=1) + return self.today + + def _build_target(self, target: _BuildTarget) -> BuildDateOutcome: + try: + with self.repository.advisory_lock(target.trade_date) as acquired: + if not acquired: + return BuildDateOutcome( + target.trade_date, + "locked", + None, + Decimal(0), + 0, + 0, + "build_locked", + "another sector radar build is running for this date", + ) + return self._build_locked(target) + except Exception as exc: + error_type, message = self._safe_failure(exc) + return BuildDateOutcome( + target.trade_date, + "failed", + None, + Decimal(0), + 0, + 0, + error_type, + message, + ) + + def _build_locked(self, target: _BuildTarget) -> BuildDateOutcome: + started_at = self.now_fn() + publication: RadarPublication | None = None + publication_created = False + try: + self.repository.recover_running_publications( + target.trade_date, + finished_at=started_at, + ) + publication_id = self._running_id(target.trade_date) + publication = RadarPublication( + publication_id=publication_id, + target_trade_date=target.trade_date, + status=PublicationStatus.RUNNING, + source_version=self.source_version, + universe_version="pending", + metric_versions=tuple(strategy.metric_version for strategy in self.strategies), + input_hash=None, + coverage=Decimal(0), + started_at=started_at, + ) + self.repository.create_publication(publication) + publication_created = True + reusable = self._reusable_sources(target.retry_publication_id) + collected = self._collect(target.trade_date, publication_id, reusable) + input_hash = self._input_hash(collected.snapshots) + existing = self.repository.find_reusable_publication(target.trade_date, input_hash) + if existing is not None: + self.repository.discard_running_publication(publication_id) + publication_created = False + is_success = existing.status is PublicationStatus.SUCCESS + return BuildDateOutcome( + target.trade_date, + "unchanged" if is_success else "partial", + existing.publication_id, + existing.coverage, + 0, + 0, + None if is_success else "duplicate_input", + None + if is_success + else "input is unchanged from an existing partial publication", + ) + + publication = replace( + publication, + universe_version=self._universe_version(collected.membership_snapshots), + input_hash=input_hash, + ) + aggregates = self._aggregate(collected) + rankings = self._rank(target.trade_date, aggregates) + coverage = self._coverage(collected.stock_facts) + terminal = ( + PublicationStatus.SUCCESS + if coverage >= self.coverage_threshold + else PublicationStatus.PARTIAL + ) + finished = RadarPublication( + publication_id=publication.publication_id, + target_trade_date=target.trade_date, + status=terminal, + source_version=publication.source_version, + universe_version=publication.universe_version, + metric_versions=publication.metric_versions, + input_hash=input_hash, + coverage=coverage, + started_at=started_at, + finished_at=self.now_fn(), + error_summary=( + None if terminal is PublicationStatus.SUCCESS else "coverage_below_threshold" + ), + ) + self.repository.finalize_publication( + finished, + memberships=collected.memberships, + stock_facts=collected.stock_facts, + daily_aggregates=( + DailyAggregateRecord(publication_id, aggregate) for aggregate in aggregates + ), + rankings=(RankingRecord(publication_id, ranking) for ranking in rankings), + retry_source_groups=( + self._retry_source_groups(collected.stock_facts) + if terminal is PublicationStatus.PARTIAL + else () + ), + ) + return BuildDateOutcome( + target.trade_date, + "success" if terminal is PublicationStatus.SUCCESS else "partial", + publication_id, + coverage, + len(aggregates), + len(rankings), + ) + except Exception as exc: + error_type, message = self._safe_failure(exc) + failed_id = ( + publication.publication_id + if publication is not None + else self._failure_id(target.trade_date) + ) + if publication is not None and publication_created: + self.repository.finish_publication( + RadarPublication( + publication_id=failed_id, + target_trade_date=target.trade_date, + status=PublicationStatus.FAILED, + source_version=publication.source_version, + universe_version=publication.universe_version, + metric_versions=publication.metric_versions, + input_hash=publication.input_hash, + coverage=Decimal(0), + started_at=started_at, + finished_at=self.now_fn(), + error_summary=f"{error_type}:{message}", + ) + ) + return BuildDateOutcome( + target.trade_date, + "failed", + failed_id, + Decimal(0), + 0, + 0, + error_type, + message, + ) + + def _reusable_sources( + self, publication_id: str | None + ) -> dict[PublicationSourceGroup, tuple[SourceSnapshot, ...]]: + """Load successful checkpoints while forcing incomplete coverage facts to refresh.""" + + if publication_id is None: + return {} + publication = self.repository.get_publication(publication_id) + if publication is None: + raise ValueError("retry publication does not exist") + grouped: defaultdict[PublicationSourceGroup, list[PublicationSourceRecord]] = defaultdict( + list + ) + for record in self.repository.load_publication_sources(publication_id): + if not record.refresh_on_retry: + grouped[record.source_group].append(record) + result: dict[PublicationSourceGroup, tuple[SourceSnapshot, ...]] = {} + for group, records in grouped.items(): + ordered = sorted(records, key=lambda item: item.source_order) + if [item.source_order for item in ordered] != list(range(len(ordered))): + raise ValueError("publication source checkpoint order is incomplete") + result[group] = tuple(item.snapshot for item in ordered) + return result + + def _fetch_group[T]( + self, + publication_id: str, + source_group: PublicationSourceGroup, + reusable: Mapping[PublicationSourceGroup, tuple[SourceSnapshot, ...]], + fetch: Callable[[], SourceResult[T]], + parser: Callable[[Mapping[str, SourceScalar]], T], + ) -> SourceResult[T]: + """Replay a completed group or fetch and checkpoint it immediately.""" + + snapshots = reusable.get(source_group) + if snapshots is None: + result = fetch() + else: + result = SourceResult( + snapshots=snapshots, + rows=tuple(parser(row) for snapshot in snapshots for row in snapshot.rows), + ) + if not result.snapshots: + raise ValueError("source group must include at least one replay snapshot") + snapshot_ids = [snapshot.snapshot_id for snapshot in result.snapshots] + if len(snapshot_ids) != len(set(snapshot_ids)): + raise ValueError("source group contains duplicate snapshots") + self.repository.save_source_snapshots(result.snapshots) + self.repository.save_publication_sources( + PublicationSourceRecord(publication_id, source_group, order, snapshot) + for order, snapshot in enumerate(result.snapshots) + ) + return result + + def _collect( + self, + target: date, + publication_id: str, + reusable: Mapping[PublicationSourceGroup, tuple[SourceSnapshot, ...]], + ) -> _CollectedInputs: + calendar = self._fetch_group( + publication_id, + PublicationSourceGroup.CALENDAR, + reusable, + lambda: self.source.fetch_trade_calendar(target, target), + TradeCalendarRow.from_mapping, + ) + if target not in {row.cal_date for row in calendar.rows if row.is_open}: + raise ValueError("target date is not an open trading day") + concepts = self._fetch_group( + publication_id, + PublicationSourceGroup.CONCEPT_INDICES, + reusable, + lambda: self.source.fetch_sector_indices(target, SectorType.CONCEPT), + lambda row: SectorIndexRow.from_mapping(row, SectorType.CONCEPT), + ) + industries = self._fetch_group( + publication_id, + PublicationSourceGroup.INDUSTRY_INDICES, + reusable, + lambda: self.source.fetch_sector_indices(target, SectorType.INDUSTRY), + lambda row: SectorIndexRow.from_mapping(row, SectorType.INDUSTRY), + ) + indices = concepts.rows + industries.rows + sector_codes = tuple(row.sector_code for row in indices) + members = self._fetch_group( + publication_id, + PublicationSourceGroup.MEMBERS, + reusable, + lambda: self.source.fetch_sector_members(target, sector_codes), + SectorMemberRow.from_mapping, + ) + stock_basics = self._fetch_group( + publication_id, + PublicationSourceGroup.STOCK_BASICS, + reusable, + self.source.fetch_stock_basics, + StockBasicRow.from_mapping, + ) + suspensions = self._fetch_group( + publication_id, + PublicationSourceGroup.SUSPENSIONS, + reusable, + lambda: self.source.fetch_suspensions(target), + SuspendRow.from_mapping, + ) + daily = self._fetch_group( + publication_id, + PublicationSourceGroup.DAILY, + reusable, + lambda: self.source.fetch_daily(target), + DailyRow.from_mapping, + ) + moneyflow = self._fetch_group( + publication_id, + PublicationSourceGroup.MONEYFLOW_DC, + reusable, + lambda: self.source.fetch_moneyflow_dc(target), + MoneyflowDcRow.from_mapping, + ) + + memberships = normalize_memberships(indices, members) + candidate_codes = tuple(sorted({item.stock_code for item in memberships})) + stock_facts = normalize_stock_facts( + target_trade_date=target, + candidate_codes=candidate_codes, + stock_basics=stock_basics, + suspensions=suspensions, + daily=daily, + moneyflow=moneyflow, + ) + snapshots = ( + calendar.snapshots + + concepts.snapshots + + industries.snapshots + + members.snapshots + + stock_basics.snapshots + + suspensions.snapshots + + daily.snapshots + + moneyflow.snapshots + ) + return _CollectedInputs( + target_trade_date=target, + snapshots=snapshots, + membership_snapshots=members.snapshots, + memberships=memberships, + stock_facts=stock_facts, + ) + + def _aggregate(self, inputs: _CollectedInputs) -> tuple[SectorDailyAggregate, ...]: + facts = tuple( + StockDailyFact( + trade_date=item.trade_date, + ts_code=item.ts_code, + status=item.status, + turnover_yuan=item.turnover_yuan, + net_amount_yuan=item.net_amount_yuan, + ) + for item in inputs.stock_facts + ) + grouped: defaultdict[tuple[SectorType, str, str], list[str]] = defaultdict(list) + for member in inputs.memberships: + grouped[(member.sector_type, member.sector_code, member.sector_name)].append( + member.stock_code + ) + aggregates = tuple( + aggregate_sector_snapshot( + SectorMembershipSnapshot( + trade_date=inputs.target_trade_date, + sector_type=sector_type, + sector_code=sector_code, + sector_name=sector_name, + member_codes=tuple(sorted(member_codes)), + status=MembershipStatus.AVAILABLE, + source_version=self._universe_version(inputs.membership_snapshots), + ), + facts, + ) + for (sector_type, sector_code, sector_name), member_codes in sorted( + grouped.items(), key=lambda item: (str(item[0][0]), item[0][1]) + ) + ) + if not aggregates: + raise ValueError("sector universe produced no aggregates") + return aggregates + + def _rank( + self, target: date, aggregates: Sequence[SectorDailyAggregate] + ) -> tuple[RankedMetric, ...]: + history = tuple(self.repository.load_daily_aggregate_history(target, limit_dates=9)) + observations: list[MetricObservation] = [] + for current in aggregates: + sector_history = tuple( + item + for item in history + if (item.sector_type, item.sector_code) + == (current.sector_type, current.sector_code) + ) + (current,) + observations.extend( + strategy.evaluate(sector_history, target) for strategy in self.strategies + ) + current_rankings = rank_metric_observations(observations) + previous = self.repository.load_previous_rankings(target, limit_dates=5) + history_by_days = {days: rankings for days, (_, rankings) in enumerate(previous, start=1)} + return with_rank_changes(current_rankings, history_by_days) + + @staticmethod + def _coverage(stock_facts: Sequence[StockFactRecord]) -> Decimal: + expected_statuses = { + "available", + "missing", + "missing_daily", + "missing_moneyflow", + "null_daily_amount", + "null_moneyflow", + "low_liquidity", + } + expected = sum(item.status.value in expected_statuses for item in stock_facts) + covered_statuses = {StockFactStatus.AVAILABLE, StockFactStatus.LOW_LIQUIDITY} + covered = sum(item.status in covered_statuses for item in stock_facts) + return Decimal(covered) / Decimal(expected) if expected else Decimal(0) + + @staticmethod + def _retry_source_groups( + stock_facts: Sequence[StockFactRecord], + ) -> tuple[PublicationSourceGroup, ...]: + groups: list[PublicationSourceGroup] = [] + statuses = {item.status for item in stock_facts} + if statuses & { + StockFactStatus.MISSING, + StockFactStatus.MISSING_DAILY, + StockFactStatus.NULL_DAILY_AMOUNT, + }: + groups.append(PublicationSourceGroup.DAILY) + if statuses & { + StockFactStatus.MISSING, + StockFactStatus.MISSING_MONEYFLOW, + StockFactStatus.NULL_MONEYFLOW, + }: + groups.append(PublicationSourceGroup.MONEYFLOW_DC) + return tuple(groups) + + def _input_hash(self, snapshots: Sequence[SourceSnapshot]) -> str: + payload = json.dumps( + { + "snapshot_ids": sorted(snapshot.snapshot_id for snapshot in snapshots), + "metric_versions": sorted(strategy.metric_version for strategy in self.strategies), + "normalizer": "zhixing_stock_fact_v1", + }, + sort_keys=True, + separators=(",", ":"), + ) + return hashlib.sha256(payload.encode()).hexdigest() + + @staticmethod + def _universe_version(snapshots: Sequence[SourceSnapshot]) -> str: + payload = "\n".join(sorted(snapshot.snapshot_id for snapshot in snapshots)) + return f"eastmoney-dc-{hashlib.sha256(payload.encode()).hexdigest()[:32]}" + + @staticmethod + def _failure_id(target: date) -> str: + return f"radar-{target:%Y%m%d}-failed-{uuid4().hex[:24]}" + + @staticmethod + def _running_id(target: date) -> str: + return f"radar-{target:%Y%m%d}-running-{uuid4().hex[:23]}" + + @staticmethod + def _safe_failure(error: BaseException) -> tuple[str, str]: + if isinstance(error, ValueError): + return type(error).__name__, "input or source contract validation failed" + return type(error).__name__, "sector radar build failed" + + +@dataclass(frozen=True, slots=True) +class _CollectedInputs: + target_trade_date: date + snapshots: tuple[SourceSnapshot, ...] + membership_snapshots: tuple[SourceSnapshot, ...] + memberships: tuple[MembershipRecord, ...] + stock_facts: tuple[StockFactRecord, ...] + + +@dataclass(frozen=True, slots=True) +class _BuildTarget: + trade_date: date + retry_publication_id: str | None = None diff --git a/zhixing-server/src/zhixing_server/modules/sector_radar/domain/persistence.py b/zhixing-server/src/zhixing_server/modules/sector_radar/domain/persistence.py index 2b9b2d4..b5ab806 100644 --- a/zhixing-server/src/zhixing_server/modules/sector_radar/domain/persistence.py +++ b/zhixing-server/src/zhixing_server/modules/sector_radar/domain/persistence.py @@ -5,14 +5,16 @@ from __future__ import annotations from collections.abc import Iterable, Sequence from contextlib import AbstractContextManager from dataclasses import dataclass -from datetime import date +from datetime import date, datetime from decimal import Decimal +from enum import StrEnum from typing import Protocol from .models import ( MembershipStatus, RadarPublication, RankedMetric, + SectorDailyAggregate, SectorType, StockFactStatus, ) @@ -104,6 +106,52 @@ class RankingRecord: raise ValueError("publication_id must not be empty") +@dataclass(frozen=True, slots=True) +class DailyAggregateRecord: + """One exact daily strategy input owned by a publication revision.""" + + publication_id: str + aggregate: SectorDailyAggregate + + def __post_init__(self) -> None: + """Validate the publication foreign identity.""" + + if not self.publication_id.strip(): + raise ValueError("publication_id must not be empty") + + +class PublicationSourceGroup(StrEnum): + """Stable source checkpoints that can be retried independently.""" + + CALENDAR = "calendar" + CONCEPT_INDICES = "concept_indices" + INDUSTRY_INDICES = "industry_indices" + MEMBERS = "members" + STOCK_BASICS = "stock_basics" + SUSPENSIONS = "suspensions" + DAILY = "daily" + MONEYFLOW_DC = "moneyflow_dc" + + +@dataclass(frozen=True, slots=True) +class PublicationSourceRecord: + """One ordered raw snapshot checkpoint attached to a build attempt.""" + + publication_id: str + source_group: PublicationSourceGroup + source_order: int + snapshot: SourceSnapshot + refresh_on_retry: bool = False + + def __post_init__(self) -> None: + """Validate the publication identity and deterministic group ordering.""" + + if not self.publication_id.strip(): + raise ValueError("publication_id must not be empty") + if self.source_order < 0: + raise ValueError("source_order must not be negative") + + @dataclass(frozen=True, slots=True) class WriteCounts: """Idempotent persistence outcome.""" @@ -125,16 +173,53 @@ class SectorRadarRepository(Protocol): def save_source_snapshots(self, snapshots: Iterable[SourceSnapshot]) -> WriteCounts: ... + def save_publication_sources( + self, records: Iterable[PublicationSourceRecord] + ) -> WriteCounts: ... + + def load_publication_sources( + self, publication_id: str + ) -> Sequence[PublicationSourceRecord]: ... + + def mark_publication_sources_for_retry( + self, + publication_id: str, + source_groups: Sequence[PublicationSourceGroup], + ) -> None: ... + def save_memberships(self, records: Iterable[MembershipRecord]) -> WriteCounts: ... def save_stock_facts(self, records: Iterable[StockFactRecord]) -> WriteCounts: ... + def save_daily_aggregates(self, records: Iterable[DailyAggregateRecord]) -> WriteCounts: ... + def create_publication(self, publication: RadarPublication) -> WriteCounts: ... def finish_publication(self, publication: RadarPublication) -> None: ... + def finalize_publication( + self, + publication: RadarPublication, + *, + memberships: Iterable[MembershipRecord], + stock_facts: Iterable[StockFactRecord], + daily_aggregates: Iterable[DailyAggregateRecord], + rankings: Iterable[RankingRecord], + retry_source_groups: Sequence[PublicationSourceGroup] = (), + ) -> None: ... + + def recover_running_publications( + self, target_trade_date: date, *, finished_at: datetime + ) -> Sequence[str]: ... + + def discard_running_publication(self, publication_id: str) -> None: ... + def save_rankings(self, records: Iterable[RankingRecord]) -> WriteCounts: ... + def find_reusable_publication( + self, target_trade_date: date, input_hash: str + ) -> RadarPublication | None: ... + def get_publication(self, publication_id: str) -> RadarPublication | None: ... def get_last_good_publication( @@ -142,3 +227,11 @@ class SectorRadarRepository(Protocol): ) -> RadarPublication | None: ... def list_successful_dates(self) -> Sequence[date]: ... + + def load_daily_aggregate_history( + self, target_trade_date: date, *, limit_dates: int + ) -> Sequence[SectorDailyAggregate]: ... + + def load_previous_rankings( + self, target_trade_date: date, *, limit_dates: int + ) -> Sequence[tuple[date, Sequence[RankedMetric]]]: ... diff --git a/zhixing-server/src/zhixing_server/modules/sector_radar/infrastructure/memory.py b/zhixing-server/src/zhixing_server/modules/sector_radar/infrastructure/memory.py index e735c9f..5c63d2b 100644 --- a/zhixing-server/src/zhixing_server/modules/sector_radar/infrastructure/memory.py +++ b/zhixing-server/src/zhixing_server/modules/sector_radar/infrastructure/memory.py @@ -4,11 +4,15 @@ from __future__ import annotations from collections.abc import Callable, Generator, Iterable, Sequence from contextlib import contextmanager -from datetime import date +from dataclasses import replace +from datetime import date, datetime -from ..domain.models import PublicationStatus, RadarPublication +from ..domain.models import PublicationStatus, RadarPublication, RankedMetric, SectorDailyAggregate from ..domain.persistence import ( + DailyAggregateRecord, MembershipRecord, + PublicationSourceGroup, + PublicationSourceRecord, RankingRecord, StockFactRecord, WriteCounts, @@ -21,9 +25,11 @@ class InMemorySectorRadarRepository: def __init__(self) -> None: self.source_snapshots: dict[str, SourceSnapshot] = {} + self.publication_sources: dict[tuple[str, str, int], PublicationSourceRecord] = {} self.memberships: dict[tuple[str, str, str], MembershipRecord] = {} self.stock_facts: dict[tuple[str, str], StockFactRecord] = {} self.publications: dict[str, RadarPublication] = {} + self.daily_aggregates: dict[tuple[str, str, str], DailyAggregateRecord] = {} self.rankings: dict[tuple[str, str, str, str], RankingRecord] = {} self.lock_available = True @@ -51,6 +57,51 @@ class InMemorySectorRadarRepository: inserted += 1 return WriteCounts(inserted, unchanged) + def save_publication_sources(self, records: Iterable[PublicationSourceRecord]) -> WriteCounts: + """Checkpoint completed source groups under one publication attempt.""" + + items = tuple(records) + for item in items: + if item.publication_id not in self.publications: + raise ValueError("publication source publication does not exist") + if item.snapshot.snapshot_id not in self.source_snapshots: + raise ValueError("publication source snapshot does not exist") + return self._insert_immutable( + self.publication_sources, + items, + key=lambda item: ( + item.publication_id, + item.source_group.value, + item.source_order, + ), + ) + + def load_publication_sources(self, publication_id: str) -> Sequence[PublicationSourceRecord]: + """Load source checkpoints in stable group and request order.""" + + return tuple( + sorted( + ( + item + for item in self.publication_sources.values() + if item.publication_id == publication_id + ), + key=lambda item: (item.source_group.value, item.source_order), + ) + ) + + def mark_publication_sources_for_retry( + self, + publication_id: str, + source_groups: Sequence[PublicationSourceGroup], + ) -> None: + """Mark only incomplete source groups for a future partial retry.""" + + requested = set(source_groups) + for key, record in tuple(self.publication_sources.items()): + if record.publication_id == publication_id and record.source_group in requested: + self.publication_sources[key] = replace(record, refresh_on_retry=True) + def save_memberships(self, records: Iterable[MembershipRecord]) -> WriteCounts: """Insert membership rows without overwriting an earlier source revision.""" @@ -69,6 +120,23 @@ class InMemorySectorRadarRepository: key=lambda item: (item.fact_revision, item.ts_code), ) + def save_daily_aggregates(self, records: Iterable[DailyAggregateRecord]) -> WriteCounts: + """Insert publication-owned exact strategy inputs idempotently.""" + + items = tuple(records) + for item in items: + if item.publication_id not in self.publications: + raise ValueError("daily aggregate publication does not exist") + return self._insert_immutable( + self.daily_aggregates, + items, + key=lambda item: ( + item.publication_id, + item.aggregate.sector_type.value, + item.aggregate.sector_code, + ), + ) + def create_publication(self, publication: RadarPublication) -> WriteCounts: """Create one running publication without replacing an existing identity.""" @@ -99,6 +167,90 @@ class InMemorySectorRadarRepository: raise ValueError("publication target_trade_date cannot change") self.publications[publication.publication_id] = publication + def finalize_publication( + self, + publication: RadarPublication, + *, + memberships: Iterable[MembershipRecord], + stock_facts: Iterable[StockFactRecord], + daily_aggregates: Iterable[DailyAggregateRecord], + rankings: Iterable[RankingRecord], + retry_source_groups: Sequence[PublicationSourceGroup] = (), + ) -> None: + """Atomically expose all derived rows and the terminal publication in tests.""" + + previous = ( + self.memberships.copy(), + self.stock_facts.copy(), + self.daily_aggregates.copy(), + self.rankings.copy(), + self.publication_sources.copy(), + self.publications.copy(), + ) + try: + self.save_memberships(memberships) + self.save_stock_facts(stock_facts) + self.save_daily_aggregates(daily_aggregates) + self.save_rankings(rankings) + self.mark_publication_sources_for_retry( + publication.publication_id, + retry_source_groups, + ) + self.finish_publication(publication) + except Exception: + ( + self.memberships, + self.stock_facts, + self.daily_aggregates, + self.rankings, + self.publication_sources, + self.publications, + ) = previous + raise + + def recover_running_publications( + self, target_trade_date: date, *, finished_at: datetime + ) -> Sequence[str]: + """Fail orphaned attempts after the caller has acquired the date lock.""" + + recovered: list[str] = [] + for publication_id, publication in tuple(self.publications.items()): + if ( + publication.target_trade_date == target_trade_date + and publication.status is PublicationStatus.RUNNING + ): + self.publications[publication_id] = replace( + publication, + status=PublicationStatus.FAILED, + finished_at=finished_at, + error_summary="recovered_stale_running", + ) + recovered.append(publication_id) + return tuple(sorted(recovered)) + + def discard_running_publication(self, publication_id: str) -> None: + """Remove only a provisional duplicate attempt and its owned projections.""" + + publication = self.publications.get(publication_id) + if publication is None or publication.status is not PublicationStatus.RUNNING: + raise ValueError("discarded publication must exist in running status") + del self.publications[publication_id] + self.publication_sources = { + key: item + for key, item in self.publication_sources.items() + if item.publication_id != publication_id + } + self.daily_aggregates = { + key: item + for key, item in self.daily_aggregates.items() + if item.publication_id != publication_id + } + self.rankings = { + key: item + for key, item in self.rankings.items() + if item.publication_id != publication_id + } + def save_rankings(self, records: Iterable[RankingRecord]) -> WriteCounts: """Insert publication-owned rankings idempotently.""" @@ -122,6 +274,22 @@ class InMemorySectorRadarRepository: return self.publications.get(publication_id) + def find_reusable_publication( + self, target_trade_date: date, input_hash: str + ) -> RadarPublication | None: + """Find an identical success or partial revision without hiding failures.""" + + return next( + ( + item + for item in self.publications.values() + if item.status in {PublicationStatus.SUCCESS, PublicationStatus.PARTIAL} + and item.target_trade_date == target_trade_date + and item.input_hash == input_hash + ), + None, + ) + def get_last_good_publication( self, target_trade_date: date | None = None ) -> RadarPublication | None: @@ -153,6 +321,66 @@ class InMemorySectorRadarRepository: ) ) + def load_daily_aggregate_history( + self, target_trade_date: date, *, limit_dates: int + ) -> Sequence[SectorDailyAggregate]: + """Load aggregates from the latest successful revision of prior dates.""" + + dates = self._previous_successful_dates(target_trade_date, limit_dates) + selected_publications = { + self._latest_success_for_date(item).publication_id for item in dates + } + return tuple( + record.aggregate + for record in self.daily_aggregates.values() + if record.publication_id in selected_publications + ) + + def load_previous_rankings( + self, target_trade_date: date, *, limit_dates: int + ) -> Sequence[tuple[date, Sequence[RankedMetric]]]: + """Load prior successful rankings newest first for rank-change attachment.""" + + result: list[tuple[date, Sequence[RankedMetric]]] = [] + for trade_date in self._previous_successful_dates(target_trade_date, limit_dates): + publication_id = self._latest_success_for_date(trade_date).publication_id + result.append( + ( + trade_date, + tuple( + record.ranking + for record in self.rankings.values() + if record.publication_id == publication_id + ), + ) + ) + return tuple(result) + + def _previous_successful_dates(self, target_trade_date: date, limit: int) -> tuple[date, ...]: + if limit < 1: + raise ValueError("limit_dates must be positive") + return tuple( + sorted( + { + item.target_trade_date + for item in self.publications.values() + if item.status is PublicationStatus.SUCCESS + and item.target_trade_date < target_trade_date + }, + reverse=True, + )[:limit] + ) + + def _latest_success_for_date(self, trade_date: date) -> RadarPublication: + return max( + ( + item + for item in self.publications.values() + if item.status is PublicationStatus.SUCCESS and item.target_trade_date == trade_date + ), + key=lambda item: item.finished_at or item.started_at, + ) + @staticmethod def _insert_immutable[K, V]( target: dict[K, V], diff --git a/zhixing-server/src/zhixing_server/modules/sector_radar/infrastructure/postgres.py b/zhixing-server/src/zhixing_server/modules/sector_radar/infrastructure/postgres.py index 79b30dc..639fb0b 100644 --- a/zhixing-server/src/zhixing_server/modules/sector_radar/infrastructure/postgres.py +++ b/zhixing-server/src/zhixing_server/modules/sector_radar/infrastructure/postgres.py @@ -5,16 +5,30 @@ from __future__ import annotations import threading from collections.abc import Callable, Generator, Iterable, Sequence from contextlib import contextmanager -from datetime import date +from datetime import date, datetime from decimal import Decimal from typing import Any from psycopg.types.json import Jsonb from psycopg_pool import ConnectionPool -from ..domain.models import PublicationStatus, RadarPublication +from ..domain.models import ( + MetricKind, + MetricObservation, + MetricQuality, + MetricUnit, + PublicationStatus, + RadarPublication, + RankChange, + RankedMetric, + SectorDailyAggregate, + SectorType, +) from ..domain.persistence import ( + DailyAggregateRecord, MembershipRecord, + PublicationSourceGroup, + PublicationSourceRecord, RankingRecord, StockFactRecord, WriteCounts, @@ -145,6 +159,87 @@ class PostgresSectorRadarRepository: unchanged = len(existing) return WriteCounts(inserted=len(items) - unchanged, unchanged=unchanged) + def save_publication_sources(self, records: Iterable[PublicationSourceRecord]) -> WriteCounts: + """Checkpoint ordered source snapshots under a running publication.""" + + items = tuple(records) + self._require_unique( + items, + lambda item: (item.publication_id, item.source_group, item.source_order), + ) + return self._copy_immutable( + "sector_radar_publication_source", + ( + "publication_id", + "source_group", + "source_order", + "source_snapshot_id", + "refresh_on_retry", + ), + ("publication_id", "source_group", "source_order"), + tuple( + ( + item.publication_id, + item.source_group.value, + item.source_order, + item.snapshot.snapshot_id, + item.refresh_on_retry, + ) + for item in items + ), + ) + + def load_publication_sources(self, publication_id: str) -> Sequence[PublicationSourceRecord]: + """Load source checkpoints with their sanitized raw snapshots.""" + + with self._connection() as connection: + rows = connection.execute( + """ + SELECT link.publication_id, link.source_group, link.source_order, + snapshot.id, snapshot.api_name, snapshot.normalized_params, + snapshot.target_trade_date, snapshot.partition_key, + snapshot.observed_at, snapshot.payload, snapshot.row_count, + snapshot.returned_fields, snapshot.content_sha256, + snapshot.row_limit, snapshot.limit_reached, + link.refresh_on_retry + FROM sector_radar_publication_source AS link + JOIN sector_radar_source_snapshot AS snapshot + ON snapshot.id = link.source_snapshot_id + WHERE link.publication_id = %s + ORDER BY link.source_group, link.source_order + """, + (publication_id,), + ).fetchall() + return tuple( + PublicationSourceRecord( + publication_id=str(row[0]), + source_group=PublicationSourceGroup(str(row[1])), + source_order=int(row[2]), + snapshot=self._snapshot_from_row(row[3:]), + refresh_on_retry=bool(row[15]), + ) + for row in rows + ) + + def mark_publication_sources_for_retry( + self, + publication_id: str, + source_groups: Sequence[PublicationSourceGroup], + ) -> None: + """Mark only incomplete source groups for a future partial retry.""" + + if not source_groups: + return + with self._connection() as connection, connection.transaction(): + connection.execute( + """ + UPDATE sector_radar_publication_source + SET refresh_on_retry = TRUE + WHERE publication_id = %s AND source_group = ANY(%s) + """, + (publication_id, [group.value for group in source_groups]), + ) + def save_memberships(self, records: Iterable[MembershipRecord]) -> WriteCounts: """COPY point-in-time members into an immutable revision key.""" @@ -214,6 +309,53 @@ class PostgresSectorRadarRepository: rows, ) + def save_daily_aggregates(self, records: Iterable[DailyAggregateRecord]) -> WriteCounts: + """COPY exact daily metric inputs under their publication revision.""" + + items = tuple(records) + self._require_unique( + items, + lambda item: ( + item.publication_id, + item.aggregate.sector_type, + item.aggregate.sector_code, + ), + ) + rows = tuple( + ( + item.publication_id, + item.aggregate.trade_date, + item.aggregate.sector_type.value, + item.aggregate.sector_code, + item.aggregate.sector_name, + item.aggregate.member_count, + item.aggregate.valid_sample_count, + item.aggregate.net_amount_yuan, + item.aggregate.turnover_yuan, + item.aggregate.membership_coverage, + item.aggregate.moneyflow_coverage, + ) + for item in items + ) + return self._copy_immutable( + "sector_radar_daily_aggregate", + ( + "publication_id", + "trade_date", + "sector_type", + "sector_code", + "sector_name", + "member_count", + "valid_sample_count", + "net_amount_yuan", + "turnover_yuan", + "membership_coverage", + "moneyflow_coverage", + ), + ("publication_id", "sector_type", "sector_code"), + rows, + ) + def create_publication(self, publication: RadarPublication) -> WriteCounts: """Insert a new running publication identity idempotently.""" @@ -248,26 +390,199 @@ class PostgresSectorRadarRepository: if publication.status is PublicationStatus.RUNNING: raise ValueError("finished publication must use a terminal status") with self._connection() as connection, connection.transaction(): - result = connection.execute( + self._finish_publication_on_connection(connection, publication) + + def finalize_publication( + self, + publication: RadarPublication, + *, + memberships: Iterable[MembershipRecord], + stock_facts: Iterable[StockFactRecord], + daily_aggregates: Iterable[DailyAggregateRecord], + rankings: Iterable[RankingRecord], + retry_source_groups: Sequence[PublicationSourceGroup] = (), + ) -> None: + """Commit all derived projections and the terminal state in one transaction.""" + + if publication.status is PublicationStatus.RUNNING: + raise ValueError("finalized publication must use a terminal status") + membership_items = tuple(memberships) + stock_items = tuple(stock_facts) + aggregate_items = tuple(daily_aggregates) + ranking_items = tuple(rankings) + self._require_unique( + membership_items, + lambda item: (item.source_snapshot_id, item.sector_code, item.stock_code), + ) + self._require_unique(stock_items, lambda item: (item.fact_revision, item.ts_code)) + self._require_unique( + aggregate_items, + lambda item: ( + item.publication_id, + item.aggregate.sector_type, + item.aggregate.sector_code, + ), + ) + self._require_unique( + ranking_items, + lambda item: ( + item.publication_id, + item.ranking.observation.sector_type, + item.ranking.observation.sector_code, + item.ranking.observation.metric_version, + ), + ) + with self._connection() as connection, connection.transaction(): + self._copy_immutable_on_connection( + connection, + "sector_radar_membership", + ( + "source_snapshot_id", + "trade_date", + "sector_type", + "sector_code", + "sector_name", + "stock_code", + "stock_name", + "membership_status", + ), + ("source_snapshot_id", "sector_code", "stock_code"), + tuple( + ( + item.source_snapshot_id, + item.trade_date, + item.sector_type.value, + item.sector_code, + item.sector_name, + item.stock_code, + item.stock_name, + item.status.value, + ) + for item in membership_items + ), + ) + self._copy_immutable_on_connection( + connection, + "sector_radar_stock_fact", + ( + "fact_revision", + "trade_date", + "ts_code", + "source_snapshot_ids", + "status", + "turnover_yuan", + "net_amount_yuan", + ), + ("fact_revision", "ts_code"), + tuple( + ( + item.fact_revision, + item.trade_date, + item.ts_code, + Jsonb(list(item.source_snapshot_ids)), + item.status.value, + item.turnover_yuan, + item.net_amount_yuan, + ) + for item in stock_items + ), + ) + self._copy_immutable_on_connection( + connection, + "sector_radar_daily_aggregate", + ( + "publication_id", + "trade_date", + "sector_type", + "sector_code", + "sector_name", + "member_count", + "valid_sample_count", + "net_amount_yuan", + "turnover_yuan", + "membership_coverage", + "moneyflow_coverage", + ), + ("publication_id", "sector_type", "sector_code"), + tuple( + ( + item.publication_id, + item.aggregate.trade_date, + item.aggregate.sector_type.value, + item.aggregate.sector_code, + item.aggregate.sector_name, + item.aggregate.member_count, + item.aggregate.valid_sample_count, + item.aggregate.net_amount_yuan, + item.aggregate.turnover_yuan, + item.aggregate.membership_coverage, + item.aggregate.moneyflow_coverage, + ) + for item in aggregate_items + ), + ) + self._copy_immutable_on_connection( + connection, + "sector_radar_ranking", + ( + "publication_id", + "trade_date", + "sector_type", + "sector_code", + "sector_name", + "metric_kind", + "metric_version", + "implementation_kind", + "unit", + "metric_value", + "quality", + "member_count", + "valid_sample_count", + "membership_coverage", + "moneyflow_coverage", + "rank_position", + "rank_percentile", + "rank_changes", + ), + ("publication_id", "sector_type", "sector_code", "metric_version"), + self._ranking_rows(ranking_items), + ) + if retry_source_groups: + connection.execute( + "UPDATE sector_radar_publication_source SET refresh_on_retry = TRUE " + "WHERE publication_id = %s AND source_group = ANY(%s)", + ( + publication.publication_id, + [group.value for group in retry_source_groups], + ), + ) + self._finish_publication_on_connection(connection, publication) + + def recover_running_publications( + self, target_trade_date: date, *, finished_at: datetime + ) -> Sequence[str]: + """Fail orphaned attempts after the caller has acquired the date lock.""" + + with self._connection() as connection, connection.transaction(): + rows = connection.execute( """ UPDATE sector_radar_publication - SET status = %s, source_version = %s, universe_version = %s, - metric_versions = %s, input_hash = %s, coverage = %s, - finished_at = %s, error_summary = %s - WHERE id = %s AND target_trade_date = %s AND status = 'running' + SET status = 'failed', finished_at = %s, + error_summary = 'recovered_stale_running' + WHERE target_trade_date = %s AND status = 'running' + RETURNING id """, - ( - publication.status.value, - publication.source_version, - publication.universe_version, - Jsonb(list(publication.metric_versions)), - publication.input_hash, - publication.coverage, - publication.finished_at, - self._safe_error(publication.error_summary), - publication.publication_id, - publication.target_trade_date, - ), + (finished_at, target_trade_date), + ).fetchall() + return tuple(sorted(str(row[0]) for row in rows)) + + def discard_running_publication(self, publication_id: str) -> None: + """Delete only a provisional duplicate publication and owned projections.""" + + with self._connection() as connection, connection.transaction(): + result = connection.execute( + "DELETE FROM sector_radar_publication WHERE id = %s AND status = 'running'", + (publication_id,), ) if result.rowcount != 1: raise SectorRadarRepositoryError("publication is not in running status") @@ -285,32 +600,6 @@ class PostgresSectorRadarRepository: item.ranking.observation.metric_version, ), ) - rows: list[tuple[object, ...]] = [] - for item in items: - ranking = item.ranking - observation = ranking.observation - rows.append( - ( - item.publication_id, - observation.trade_date, - observation.sector_type.value, - observation.sector_code, - observation.sector_name, - observation.metric_kind.value, - observation.metric_version, - observation.implementation_kind, - observation.unit.value, - observation.value, - observation.quality.value, - observation.member_count, - observation.valid_sample_count, - observation.membership_coverage, - observation.moneyflow_coverage, - ranking.rank_position, - ranking.rank_percentile, - Jsonb({str(change.days): change.value for change in ranking.rank_changes}), - ) - ) return self._copy_immutable( "sector_radar_ranking", ( @@ -334,7 +623,7 @@ class PostgresSectorRadarRepository: "rank_changes", ), ("publication_id", "sector_type", "sector_code", "metric_version"), - tuple(rows), + self._ranking_rows(items), ) def get_publication(self, publication_id: str) -> RadarPublication | None: @@ -347,6 +636,20 @@ class PostgresSectorRadarRepository: ).fetchone() return None if row is None else self._publication_from_row(row) + def find_reusable_publication( + self, target_trade_date: date, input_hash: str + ) -> RadarPublication | None: + """Find an identical success or partial input revision for idempotent reruns.""" + + with self._connection() as connection: + row = connection.execute( + self._publication_select() + " WHERE target_trade_date = %s AND input_hash = %s " + "AND status IN ('success', 'partial') " + "ORDER BY finished_at DESC, created_at DESC, id DESC LIMIT 1", + (target_trade_date, input_hash), + ).fetchone() + return None if row is None else self._publication_from_row(row) + def get_last_good_publication( self, target_trade_date: date | None = None ) -> RadarPublication | None: @@ -380,6 +683,75 @@ class PostgresSectorRadarRepository: ).fetchall() return tuple(row[0] for row in rows) + def load_daily_aggregate_history( + self, target_trade_date: date, *, limit_dates: int + ) -> Sequence[SectorDailyAggregate]: + """Load prior aggregates from the latest successful revision per date.""" + + if limit_dates < 1: + raise ValueError("limit_dates must be positive") + with self._connection() as connection: + rows = connection.execute( + """ + WITH selected AS ( + SELECT DISTINCT ON (target_trade_date) id, target_trade_date + FROM sector_radar_publication + WHERE status = 'success' AND target_trade_date < %s + ORDER BY target_trade_date DESC, finished_at DESC, created_at DESC, id DESC + LIMIT %s + ) + SELECT aggregate.trade_date, aggregate.sector_type, aggregate.sector_code, + aggregate.sector_name, aggregate.member_count, + aggregate.valid_sample_count, aggregate.net_amount_yuan, + aggregate.turnover_yuan, aggregate.membership_coverage, + aggregate.moneyflow_coverage + FROM sector_radar_daily_aggregate AS aggregate + JOIN selected ON selected.id = aggregate.publication_id + ORDER BY aggregate.trade_date, aggregate.sector_type, aggregate.sector_code + """, + (target_trade_date, limit_dates), + ).fetchall() + return tuple(self._aggregate_from_row(row) for row in rows) + + def load_previous_rankings( + self, target_trade_date: date, *, limit_dates: int + ) -> Sequence[tuple[date, Sequence[RankedMetric]]]: + """Load prior successful rankings newest-first for rank-change calculation.""" + + if limit_dates < 1: + raise ValueError("limit_dates must be positive") + with self._connection() as connection: + rows = connection.execute( + """ + WITH selected AS ( + SELECT DISTINCT ON (target_trade_date) id, target_trade_date + FROM sector_radar_publication + WHERE status = 'success' AND target_trade_date < %s + ORDER BY target_trade_date DESC, finished_at DESC, created_at DESC, id DESC + LIMIT %s + ) + SELECT selected.target_trade_date, ranking.trade_date, ranking.sector_type, + ranking.sector_code, ranking.sector_name, ranking.metric_kind, + ranking.metric_version, ranking.implementation_kind, ranking.unit, + ranking.metric_value, ranking.quality, ranking.member_count, + ranking.valid_sample_count, ranking.membership_coverage, + ranking.moneyflow_coverage, ranking.rank_position, + ranking.rank_percentile, ranking.rank_changes + FROM sector_radar_ranking AS ranking + JOIN selected ON selected.id = ranking.publication_id + ORDER BY selected.target_trade_date DESC, ranking.sector_type, + ranking.metric_version, ranking.rank_position NULLS LAST, + ranking.sector_code + """, + (target_trade_date, limit_dates), + ).fetchall() + grouped: dict[date, list[RankedMetric]] = {} + for row in rows: + grouped.setdefault(row[0], []).append(self._ranking_from_row(row[1:])) + return tuple( + (trade_date, tuple(grouped[trade_date])) for trade_date in sorted(grouped, reverse=True) + ) + def _copy_immutable( self, table: str, @@ -390,26 +762,46 @@ class PostgresSectorRadarRepository: if not rows: return WriteCounts(0, 0) if table not in { + "sector_radar_publication_source", "sector_radar_membership", "sector_radar_stock_fact", + "sector_radar_daily_aggregate", "sector_radar_ranking", }: raise ValueError("unsupported radar staging table") + with self._connection() as connection, connection.transaction(): + return self._copy_immutable_on_connection( + connection, + table, + columns, + conflict_columns, + rows, + ) + + @staticmethod + def _copy_immutable_on_connection( + connection: Any, + table: str, + columns: tuple[str, ...], + conflict_columns: tuple[str, ...], + rows: tuple[tuple[object, ...], ...], + ) -> WriteCounts: + if not rows: + return WriteCounts(0, 0) stage = f"{table}_stage" column_sql = ", ".join(columns) conflict_sql = ", ".join(conflict_columns) - with self._connection() as connection, connection.transaction(): - cursor = connection.cursor() - cursor.execute( - f"CREATE TEMP TABLE {stage} (LIKE {table} INCLUDING DEFAULTS) ON COMMIT DROP" - ) - with cursor.copy(f"COPY {stage} ({column_sql}) FROM STDIN") as copy: - for row in rows: - copy.write_row(row) - inserted_rows = cursor.execute( - f"INSERT INTO {table} ({column_sql}) SELECT {column_sql} FROM {stage} " - f"ON CONFLICT ({conflict_sql}) DO NOTHING RETURNING 1" - ).fetchall() + cursor = connection.cursor() + cursor.execute( + f"CREATE TEMP TABLE {stage} (LIKE {table} INCLUDING DEFAULTS) ON COMMIT DROP" + ) + with cursor.copy(f"COPY {stage} ({column_sql}) FROM STDIN") as copy: + for row in rows: + copy.write_row(row) + inserted_rows = cursor.execute( + f"INSERT INTO {table} ({column_sql}) SELECT {column_sql} FROM {stage} " + f"ON CONFLICT ({conflict_sql}) DO NOTHING RETURNING 1" + ).fetchall() inserted = len(inserted_rows) return WriteCounts(inserted=inserted, unchanged=len(rows) - inserted) @@ -470,6 +862,130 @@ class PostgresSectorRadarRepository: error_summary=None if row[10] is None else str(row[10]), ) + @staticmethod + def _ranking_rows(items: Sequence[RankingRecord]) -> tuple[tuple[object, ...], ...]: + rows: list[tuple[object, ...]] = [] + for item in items: + ranking = item.ranking + observation = ranking.observation + rows.append( + ( + item.publication_id, + observation.trade_date, + observation.sector_type.value, + observation.sector_code, + observation.sector_name, + observation.metric_kind.value, + observation.metric_version, + observation.implementation_kind, + observation.unit.value, + observation.value, + observation.quality.value, + observation.member_count, + observation.valid_sample_count, + observation.membership_coverage, + observation.moneyflow_coverage, + ranking.rank_position, + ranking.rank_percentile, + Jsonb({str(change.days): change.value for change in ranking.rank_changes}), + ) + ) + return tuple(rows) + + @staticmethod + def _finish_publication_on_connection( + connection: Any, + publication: RadarPublication, + ) -> None: + result = connection.execute( + """ + UPDATE sector_radar_publication + SET status = %s, source_version = %s, universe_version = %s, + metric_versions = %s, input_hash = %s, coverage = %s, + finished_at = %s, error_summary = %s + WHERE id = %s AND target_trade_date = %s AND status = 'running' + """, + ( + publication.status.value, + publication.source_version, + publication.universe_version, + Jsonb(list(publication.metric_versions)), + publication.input_hash, + publication.coverage, + publication.finished_at, + PostgresSectorRadarRepository._safe_error(publication.error_summary), + publication.publication_id, + publication.target_trade_date, + ), + ) + if result.rowcount != 1: + raise SectorRadarRepositoryError("publication is not in running status") + + @staticmethod + def _snapshot_from_row(row: tuple[Any, ...]) -> SourceSnapshot: + normalized_params = row[2] + return SourceSnapshot( + snapshot_id=str(row[0]), + api_name=str(row[1]), + normalized_params=tuple( + sorted((str(key), str(value)) for key, value in normalized_params.items()) + ), + target_trade_date=row[3], + partition_key=None if row[4] is None else str(row[4]), + observed_at=row[5], + rows=tuple(dict(item) for item in row[6]), + row_count=int(row[7]), + returned_fields=tuple(str(value) for value in row[8]), + content_sha256=str(row[9]), + row_limit=None if row[10] is None else int(row[10]), + limit_reached=bool(row[11]), + ) + + @staticmethod + def _aggregate_from_row(row: tuple[Any, ...]) -> SectorDailyAggregate: + return SectorDailyAggregate( + trade_date=row[0], + sector_type=SectorType(str(row[1])), + sector_code=str(row[2]), + sector_name=str(row[3]), + member_count=int(row[4]), + valid_sample_count=int(row[5]), + net_amount_yuan=None if row[6] is None else Decimal(str(row[6])), + turnover_yuan=None if row[7] is None else Decimal(str(row[7])), + membership_coverage=Decimal(str(row[8])), + moneyflow_coverage=Decimal(str(row[9])), + ) + + @staticmethod + def _ranking_from_row(row: tuple[Any, ...]) -> RankedMetric: + raw_changes = row[16] + changes = tuple( + RankChange(days=int(days), value=None if value is None else int(value)) + for days, value in sorted(raw_changes.items(), key=lambda item: int(item[0])) + ) + observation = MetricObservation( + trade_date=row[0], + sector_type=SectorType(str(row[1])), + sector_code=str(row[2]), + sector_name=str(row[3]), + metric_kind=MetricKind(str(row[4])), + metric_version=str(row[5]), + implementation_kind="independent", + unit=MetricUnit(str(row[7])), + value=None if row[8] is None else Decimal(str(row[8])), + quality=MetricQuality(str(row[9])), + member_count=int(row[10]), + valid_sample_count=int(row[11]), + membership_coverage=Decimal(str(row[12])), + moneyflow_coverage=Decimal(str(row[13])), + ) + return RankedMetric( + observation=observation, + rank_position=None if row[14] is None else int(row[14]), + rank_percentile=None if row[15] is None else Decimal(str(row[15])), + rank_changes=changes, + ) + @staticmethod def _safe_error(message: str | None) -> str | None: return None if message is None else " ".join(message.split())[:500] diff --git a/zhixing-server/src/zhixing_server/modules/sector_radar/presentation/__init__.py b/zhixing-server/src/zhixing_server/modules/sector_radar/presentation/__init__.py new file mode 100644 index 0000000..69da197 --- /dev/null +++ b/zhixing-server/src/zhixing_server/modules/sector_radar/presentation/__init__.py @@ -0,0 +1 @@ +"""Delivery adapters for sector radar build and query use cases.""" diff --git a/zhixing-server/src/zhixing_server/modules/sector_radar/presentation/cli.py b/zhixing-server/src/zhixing_server/modules/sector_radar/presentation/cli.py new file mode 100644 index 0000000..4a8c3a0 --- /dev/null +++ b/zhixing-server/src/zhixing_server/modules/sector_radar/presentation/cli.py @@ -0,0 +1,108 @@ +"""One-shot ``sector-radar-build`` command for external schedulers.""" + +from __future__ import annotations + +import argparse +import json +import logging +from collections.abc import Sequence +from datetime import date + +from ....bootstrap.config import get_settings +from ..application.build import BuildSectorRadar, BuildSectorRadarCommand +from ..infrastructure.postgres import PostgresSectorRadarRepository +from ..infrastructure.tushare import TushareSectorRadarAdapter + +logger = logging.getLogger(__name__) + + +def build_parser() -> argparse.ArgumentParser: + """Build mutually exclusive single-date, range, and retry modes.""" + + parser = argparse.ArgumentParser( + description="Build independent Tushare sector radar publications" + ) + mode = parser.add_mutually_exclusive_group() + mode.add_argument("--trade-date", type=_parse_date, help="target date in YYYY-MM-DD") + mode.add_argument( + "--retry-publication-id", + help="resume failed source groups from a partial or failed publication", + ) + mode.add_argument("--start-date", type=_parse_date, help="inclusive backfill start date") + parser.add_argument("--end-date", type=_parse_date, help="inclusive backfill end date") + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + """Execute one build invocation and print a redacted JSON summary.""" + + args = build_parser().parse_args(argv) + if (args.start_date is None) != (args.end_date is None): + raise SystemExit("--start-date and --end-date must be provided together") + command = BuildSectorRadarCommand( + trade_date=args.trade_date, + start_date=args.start_date, + end_date=args.end_date, + retry_publication_id=args.retry_publication_id, + ) + try: + settings = get_settings() + logging.basicConfig( + level=settings.log_level.upper(), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + force=True, + ) + logger.info( + "sector_radar_build_cli trade_date=%s start_date=%s end_date=%s retry=%s", + command.trade_date or "auto", + command.start_date or "none", + command.end_date or "none", + bool(command.retry_publication_id), + ) + source = TushareSectorRadarAdapter.from_token( + settings.tushare_token, + max_retries=settings.sector_radar_max_retries, + backoff_seconds=settings.sector_radar_retry_backoff_seconds, + request_interval_seconds=settings.sector_radar_request_interval_seconds, + ) + repository = PostgresSectorRadarRepository( + settings.database_url, + advisory_lock_key=settings.sector_radar_advisory_lock_key, + ) + try: + summary = BuildSectorRadar( + source, + repository, + coverage_threshold=settings.sector_radar_coverage_threshold, + ).execute(command) + finally: + repository.close() + except Exception as exc: # noqa: BLE001 - CLI boundary returns a redacted scheduler result + logger.error("sector_radar_build_initialization_failed error_type=%s", type(exc).__name__) + print( + json.dumps( + { + "status": "failed", + "exit_code": 1, + "outcomes": [], + "error_type": type(exc).__name__, + "error_message": "sector radar build initialization failed", + }, + ensure_ascii=False, + sort_keys=True, + ) + ) + return 1 + print(json.dumps(summary.as_dict(), ensure_ascii=False, sort_keys=True)) + return summary.exit_code + + +def _parse_date(value: str) -> date: + try: + return date.fromisoformat(value) + except ValueError as exc: + raise argparse.ArgumentTypeError("date must use YYYY-MM-DD") from exc + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/zhixing-server/tests/integration/test_market_data_migration.py b/zhixing-server/tests/integration/test_market_data_migration.py index cc6954a..0e047b0 100644 --- a/zhixing-server/tests/integration/test_market_data_migration.py +++ b/zhixing-server/tests/integration/test_market_data_migration.py @@ -43,6 +43,8 @@ def test_postgres_migration_creates_market_data_contract( "sector_radar_stock_fact", "sector_radar_publication", "sector_radar_ranking", + "sector_radar_daily_aggregate", + "sector_radar_publication_source", } <= tables finally: engine.dispose() diff --git a/zhixing-server/tests/unit/sector_radar/test_build.py b/zhixing-server/tests/unit/sector_radar/test_build.py new file mode 100644 index 0000000..2ebd473 --- /dev/null +++ b/zhixing-server/tests/unit/sector_radar/test_build.py @@ -0,0 +1,435 @@ +from collections.abc import Sequence +from datetime import UTC, date, datetime, timedelta +from decimal import Decimal + +from zhixing_server.modules.sector_radar.application.build import ( + BuildSectorRadar, + BuildSectorRadarCommand, +) +from zhixing_server.modules.sector_radar.domain.models import ( + PublicationStatus, + RadarPublication, + SectorType, +) +from zhixing_server.modules.sector_radar.domain.persistence import PublicationSourceGroup +from zhixing_server.modules.sector_radar.domain.source import ( + CapabilityProbeResult, + DailyRow, + MoneyflowDcRow, + SectorIndexRow, + SectorMemberRow, + SourceResult, + StockBasicRow, + SuspendRow, + TradeCalendarRow, + build_source_snapshot, +) +from zhixing_server.modules.sector_radar.infrastructure.memory import ( + InMemorySectorRadarRepository, +) + +TARGET_DATE = date(2026, 8, 28) +NOW = datetime(2026, 8, 28, 17, 30, tzinfo=UTC) + + +class FakeRadarSource: + def __init__( + self, + *, + missing_moneyflow: bool = False, + net_scale: Decimal = Decimal(1), + ) -> None: + self.missing_moneyflow = missing_moneyflow + self.net_scale = net_scale + self.fail_daily = False + self.calls: list[str] = [] + + def _result[T]( + self, api_name: str, target: date | None, rows: tuple[T, ...] + ) -> SourceResult[T]: + snapshot = build_source_snapshot( + api_name=api_name, + params={ + "trade_date": target.isoformat() if target is not None else "all", + "fixture_fingerprint": repr(rows), + }, + rows=tuple(self._raw_row(row) for row in rows), + target_trade_date=target, + partition_key="all" if api_name == "dc_member" else None, + observed_at=NOW, + ) + return SourceResult((snapshot,), rows) + + @staticmethod + def _raw_row(row: object) -> dict[str, object]: + if isinstance(row, TradeCalendarRow): + return { + "exchange": row.exchange, + "cal_date": row.cal_date, + "is_open": int(row.is_open), + "pretrade_date": row.pretrade_date, + } + if isinstance(row, SectorIndexRow): + return { + "trade_date": row.trade_date, + "ts_code": row.sector_code, + "name": row.name, + "level": row.level, + "pct_change": row.pct_change, + "leading_code": row.leading_code, + } + if isinstance(row, SectorMemberRow): + return { + "trade_date": row.trade_date, + "ts_code": row.sector_code, + "con_code": row.stock_code, + "name": row.stock_name, + } + if isinstance(row, StockBasicRow): + return { + "ts_code": row.ts_code, + "symbol": row.symbol, + "name": row.name, + "market": row.market, + "exchange": row.exchange, + "list_status": row.list_status, + "list_date": row.list_date, + "delist_date": row.delist_date, + } + if isinstance(row, SuspendRow): + return { + "ts_code": row.ts_code, + "trade_date": row.trade_date, + "suspend_timing": row.suspend_timing, + "suspend_type": row.suspend_type, + } + if isinstance(row, DailyRow): + return { + "ts_code": row.ts_code, + "trade_date": row.trade_date, + "close": row.close, + "pre_close": row.pre_close, + "pct_chg": row.pct_chg, + "vol": row.volume, + "amount": row.amount_thousand_yuan, + } + if isinstance(row, MoneyflowDcRow): + return { + "trade_date": row.trade_date, + "ts_code": row.ts_code, + "name": row.name, + "net_amount": row.net_amount_ten_thousand_yuan, + "net_amount_rate": row.net_amount_rate, + "pct_change": row.pct_change, + "close": row.close, + } + raise TypeError(f"unsupported fake source row: {type(row).__name__}") + + def fetch_trade_calendar(self, start: date, end: date) -> SourceResult[TradeCalendarRow]: + self.calls.append("calendar") + rows = tuple( + TradeCalendarRow("SSE", start + timedelta(days=offset), True, None) + for offset in range((end - start).days + 1) + ) + return self._result("trade_cal", end, rows) + + def fetch_sector_indices( + self, trade_date: date, sector_type: SectorType + ) -> SourceResult[SectorIndexRow]: + self.calls.append(f"{sector_type.value}_indices") + prefix = "BK0" if sector_type is SectorType.CONCEPT else "BK1" + row = SectorIndexRow( + trade_date, + sector_type, + f"{prefix}001.DC", + "示例概念" if sector_type is SectorType.CONCEPT else "示例行业", + "一级", + Decimal(1), + "000001.SZ", + ) + return self._result(f"dc_index_{sector_type.value}", trade_date, (row,)) + + def fetch_sector_members( + self, trade_date: date, sector_codes: Sequence[str] + ) -> SourceResult[SectorMemberRow]: + self.calls.append("members") + rows = tuple( + SectorMemberRow(trade_date, sector_code, f"00000{index}.SZ", f"股票{index}") + for sector_code in sector_codes + for index in range(1, 6) + ) + return self._result("dc_member", trade_date, rows) + + def fetch_stock_basics(self) -> SourceResult[StockBasicRow]: + self.calls.append("stock_basics") + rows = tuple( + StockBasicRow( + f"00000{index}.SZ", + f"00000{index}", + f"股票{index}", + "主板", + "SZSE", + "L", + date(2020, 1, 1), + None, + ) + for index in range(1, 6) + ) + return self._result("stock_basic", None, rows) + + def fetch_suspensions(self, trade_date: date) -> SourceResult[SuspendRow]: + self.calls.append("suspensions") + return self._result("suspend_d", trade_date, ()) + + def fetch_daily(self, trade_date: date) -> SourceResult[DailyRow]: + self.calls.append("daily") + if self.fail_daily: + raise RuntimeError("private provider detail") + rows = tuple( + DailyRow( + f"00000{index}.SZ", + trade_date, + Decimal(10), + Decimal(10), + Decimal(0), + Decimal(100), + Decimal(1000), + ) + for index in range(1, 6) + ) + return self._result("daily", trade_date, rows) + + def fetch_moneyflow_dc(self, trade_date: date) -> SourceResult[MoneyflowDcRow]: + self.calls.append("moneyflow_dc") + count = 4 if self.missing_moneyflow else 5 + rows = tuple( + MoneyflowDcRow( + trade_date, + f"00000{index}.SZ", + f"股票{index}", + Decimal(index) * self.net_scale, + Decimal(0), + Decimal(0), + Decimal(10), + ) + for index in range(1, count + 1) + ) + return self._result("moneyflow_dc", trade_date, rows) + + def probe(self, trade_date: date) -> CapabilityProbeResult: + return CapabilityProbeResult(NOW, ()) + + +def test_successful_build_is_idempotent_and_failed_retry_preserves_last_good() -> None: + source = FakeRadarSource() + repository = InMemorySectorRadarRepository() + use_case = BuildSectorRadar( + source, + repository, + today=TARGET_DATE, + now_fn=lambda: NOW, + ) + + first = use_case.execute(BuildSectorRadarCommand(trade_date=TARGET_DATE)) + repeated = use_case.execute(BuildSectorRadarCommand(trade_date=TARGET_DATE)) + source.fail_daily = True + failed = use_case.execute(BuildSectorRadarCommand(trade_date=TARGET_DATE)) + + assert first.status == "success" + assert first.exit_code == 0 + assert first.outcomes[0].ranking_count == 6 + assert repeated.status == "unchanged" + assert repeated.outcomes[0].publication_id == first.outcomes[0].publication_id + assert failed.status == "failed" + assert "private provider detail" not in str(failed.as_dict()) + last_good = repository.get_last_good_publication() + assert last_good is not None + assert last_good.publication_id == first.outcomes[0].publication_id + assert any(item.status is PublicationStatus.FAILED for item in repository.publications.values()) + + +def test_partial_coverage_and_lock_have_distinct_exit_codes() -> None: + repository = InMemorySectorRadarRepository() + partial = BuildSectorRadar( + FakeRadarSource(missing_moneyflow=True), + repository, + today=TARGET_DATE, + now_fn=lambda: NOW, + ).execute(BuildSectorRadarCommand(trade_date=TARGET_DATE)) + publication_count = len(repository.publications) + partial_repeated = BuildSectorRadar( + FakeRadarSource(missing_moneyflow=True), + repository, + today=TARGET_DATE, + now_fn=lambda: NOW, + ).execute(BuildSectorRadarCommand(trade_date=TARGET_DATE)) + + repository.lock_available = False + locked = BuildSectorRadar( + FakeRadarSource(), + repository, + today=TARGET_DATE, + now_fn=lambda: NOW, + ).execute(BuildSectorRadarCommand(trade_date=TARGET_DATE)) + + assert partial.status == "partial" + assert partial.exit_code == 2 + assert partial.outcomes[0].coverage == Decimal("0.8") + assert partial_repeated.status == "partial" + assert len(repository.publications) == publication_count + assert repository.get_last_good_publication() is None + assert { + record.source_group + for record in repository.load_publication_sources(partial.outcomes[0].publication_id or "") + if record.refresh_on_retry + } == {PublicationSourceGroup.MONEYFLOW_DC} + assert locked.status == "failed" + assert locked.exit_code == 1 + assert locked.outcomes[0].status == "locked" + + +def test_range_builds_dates_in_order_and_retry_uses_old_target() -> None: + repository = InMemorySectorRadarRepository() + source = FakeRadarSource(missing_moneyflow=True) + use_case = BuildSectorRadar(source, repository, now_fn=lambda: NOW) + end = TARGET_DATE + timedelta(days=1) + + summary = use_case.execute(BuildSectorRadarCommand(start_date=TARGET_DATE, end_date=end)) + partial_id = summary.outcomes[0].publication_id + assert partial_id is not None + source.missing_moneyflow = False + source.calls.clear() + retried = use_case.execute(BuildSectorRadarCommand(retry_publication_id=partial_id)) + + assert [item.target_trade_date for item in summary.outcomes] == [TARGET_DATE, end] + assert all(item.status == "partial" for item in summary.outcomes) + assert retried.outcomes[0].target_trade_date == TARGET_DATE + assert retried.status == "success" + assert source.calls == ["moneyflow_dc"] + + +def test_failed_retry_reuses_every_completed_source_group() -> None: + repository = InMemorySectorRadarRepository() + source = FakeRadarSource() + source.fail_daily = True + use_case = BuildSectorRadar(source, repository, now_fn=lambda: NOW) + + failed = use_case.execute(BuildSectorRadarCommand(trade_date=TARGET_DATE)) + failed_id = failed.outcomes[0].publication_id + assert failed_id is not None + source.fail_daily = False + source.calls.clear() + + retried = use_case.execute(BuildSectorRadarCommand(retry_publication_id=failed_id)) + + assert retried.status == "success" + assert source.calls == ["daily", "moneyflow_dc"] + old_groups = {record.source_group for record in repository.load_publication_sources(failed_id)} + assert len(old_groups) == 6 + + +def test_date_lock_recovers_an_orphaned_running_publication() -> None: + repository = InMemorySectorRadarRepository() + stale = RadarPublication( + publication_id="stale-running", + target_trade_date=TARGET_DATE, + status=PublicationStatus.RUNNING, + source_version="tushare-pro-v1", + universe_version="pending", + metric_versions=("zhixing_amount_net_bn_v1",), + input_hash=None, + coverage=Decimal(0), + started_at=NOW - timedelta(hours=1), + ) + repository.create_publication(stale) + + summary = BuildSectorRadar(FakeRadarSource(), repository, now_fn=lambda: NOW).execute( + BuildSectorRadarCommand(trade_date=TARGET_DATE) + ) + + recovered = repository.get_publication("stale-running") + assert summary.status == "success" + assert recovered is not None + assert recovered.status is PublicationStatus.FAILED + assert recovered.error_summary == "recovered_stale_running" + + +def test_default_target_excludes_today_before_closing_data_is_ready() -> None: + before_close = datetime(2026, 8, 28, 6, 0, tzinfo=UTC) + after_close = datetime(2026, 8, 28, 8, 0, tzinfo=UTC) + + before = BuildSectorRadar( + FakeRadarSource(), + InMemorySectorRadarRepository(), + today=TARGET_DATE, + now_fn=lambda: before_close, + ).execute() + after = BuildSectorRadar( + FakeRadarSource(), + InMemorySectorRadarRepository(), + today=TARGET_DATE, + now_fn=lambda: after_close, + ).execute() + + assert before.outcomes[0].target_trade_date == TARGET_DATE - timedelta(days=1) + assert after.outcomes[0].target_trade_date == TARGET_DATE + + +def test_tenth_trading_day_publishes_swing_and_five_rank_changes() -> None: + repository = InMemorySectorRadarRepository() + end = TARGET_DATE + timedelta(days=9) + summary = BuildSectorRadar(FakeRadarSource(), repository, now_fn=lambda: NOW).execute( + BuildSectorRadarCommand(start_date=TARGET_DATE, end_date=end) + ) + + publication = repository.get_last_good_publication(end) + assert summary.status == "success" + assert publication is not None + current = tuple( + record.ranking + for record in repository.rankings.values() + if record.publication_id == publication.publication_id + ) + swing = tuple( + ranking + for ranking in current + if ranking.observation.metric_version == "zhixing_swing_equal_3_10_v1" + ) + assert len(swing) == 2 + assert all(ranking.observation.value == Decimal("0.03") for ranking in swing) + assert all( + tuple(change.value for change in ranking.rank_changes) == (None, None, None, None, None) + for ranking in swing + ) + amount = tuple( + ranking + for ranking in current + if ranking.observation.metric_version == "zhixing_amount_net_bn_v1" + ) + assert all( + tuple(change.value for change in ranking.rank_changes) == (0, 0, 0, 0, 0) + for ranking in amount + ) + + +def test_history_uses_latest_successful_input_revision_for_a_date() -> None: + repository = InMemorySectorRadarRepository() + clock = [NOW] + first_source = FakeRadarSource(net_scale=Decimal(1)) + second_source = FakeRadarSource(net_scale=Decimal(2)) + + first = BuildSectorRadar(first_source, repository, now_fn=lambda: clock[0]).execute( + BuildSectorRadarCommand(trade_date=TARGET_DATE) + ) + clock[0] = NOW + timedelta(minutes=5) + second = BuildSectorRadar(second_source, repository, now_fn=lambda: clock[0]).execute( + BuildSectorRadarCommand(trade_date=TARGET_DATE) + ) + + history = repository.load_daily_aggregate_history( + TARGET_DATE + timedelta(days=1), limit_dates=1 + ) + assert first.status == "success" + assert second.status == "success" + assert len(history) == 2 + assert all(item.net_amount_yuan == Decimal(300_000) for item in history) diff --git a/zhixing-server/tests/unit/sector_radar/test_cli.py b/zhixing-server/tests/unit/sector_radar/test_cli.py new file mode 100644 index 0000000..9820748 --- /dev/null +++ b/zhixing-server/tests/unit/sector_radar/test_cli.py @@ -0,0 +1,127 @@ +import json +from datetime import date +from decimal import Decimal + +import pytest + +from zhixing_server.modules.sector_radar.application.build import ( + BuildDateOutcome, + BuildOutcomeStatus, + BuildSummary, +) +from zhixing_server.modules.sector_radar.presentation import cli +from zhixing_server.modules.sector_radar.presentation.cli import build_parser + + +def test_sector_radar_cli_parses_single_range_and_retry_modes() -> None: + parser = build_parser() + + single = parser.parse_args(["--trade-date", "2026-08-28"]) + date_range = parser.parse_args(["--start-date", "2026-08-18", "--end-date", "2026-08-28"]) + retry = parser.parse_args(["--retry-publication-id", "publication-a"]) + + assert single.trade_date == date(2026, 8, 28) + assert date_range.start_date == date(2026, 8, 18) + assert date_range.end_date == date(2026, 8, 28) + assert retry.retry_publication_id == "publication-a" + + +class FakeSettings: + log_level = "INFO" + tushare_token = "secret-token" + database_url = "postgresql://unused" + sector_radar_max_retries = 3 + sector_radar_retry_backoff_seconds = 1.0 + sector_radar_request_interval_seconds = 0.2 + sector_radar_advisory_lock_key = 7_380_522 + sector_radar_coverage_threshold = Decimal("0.99") + + +class FakeRepository: + closed = False + + def __init__(self, database_url: str, *, advisory_lock_key: int) -> None: + assert database_url == "postgresql://unused" + assert advisory_lock_key == 7_380_522 + + def close(self) -> None: + self.closed = True + + +@pytest.mark.parametrize( + ("outcome_status", "coverage", "expected_code"), + (("success", Decimal(1), 0), ("partial", Decimal("0.8"), 2), ("failed", Decimal(0), 1)), +) +def test_cli_main_returns_summary_exit_code_and_json( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + outcome_status: BuildOutcomeStatus, + coverage: Decimal, + expected_code: int, +) -> None: + summary = BuildSummary( + ( + BuildDateOutcome( + date(2026, 8, 28), + outcome_status, + "publication-a", + coverage, + 2, + 6, + ), + ) + ) + + class FakeSourceFactory: + @staticmethod + def from_token(token: str, **kwargs: object) -> object: + assert token == "secret-token" + assert kwargs + return object() + + class FakeBuild: + def __init__(self, source: object, repository: object, **kwargs: object) -> None: + assert source is not None + assert repository is not None + assert kwargs + + def execute(self, command: object) -> BuildSummary: + assert command is not None + return summary + + monkeypatch.setattr(cli, "get_settings", FakeSettings) + monkeypatch.setattr(cli, "TushareSectorRadarAdapter", FakeSourceFactory) + monkeypatch.setattr(cli, "PostgresSectorRadarRepository", FakeRepository) + monkeypatch.setattr(cli, "BuildSectorRadar", FakeBuild) + + exit_code = cli.main(["--trade-date", "2026-08-28"]) + output = json.loads(capsys.readouterr().out) + + assert exit_code == expected_code + assert output["status"] == summary.status + assert output["exit_code"] == expected_code + assert "secret-token" not in str(output) + + +def test_cli_initialization_failure_is_redacted( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + class FailingSourceFactory: + @staticmethod + def from_token(token: str, **kwargs: object) -> object: + del token, kwargs + raise RuntimeError("private provider detail secret-token") + + monkeypatch.setattr(cli, "get_settings", FakeSettings) + monkeypatch.setattr(cli, "TushareSectorRadarAdapter", FailingSourceFactory) + + exit_code = cli.main(["--trade-date", "2026-08-28"]) + captured = capsys.readouterr() + output = json.loads(captured.out) + + assert exit_code == 1 + assert output["status"] == "failed" + assert output["error_type"] == "RuntimeError" + assert "private provider detail" not in captured.out + assert "secret-token" not in captured.out