my-home/zhixing-system
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf(sector_radar): narrow publication reads and index rank history

Browse Source
This commit is contained in:
yuxuanhui
2026-09-07 12:16:25 +08:00
parent e567e5f717
commit d7586b27f6
13 changed files with 1083 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zhixing-server/src/zhixing_server/modules/sector_radar/application/details.py
+104 -66
View File
@@ -10,14 +10,13 @@ from decimal import Decimal
from ..domain.metrics import AmountNetStrategy, RatioTurnoverStrategy, SwingEqualThreeToTenStrategy
from ..domain.models import MetricKind, RadarPublication, RankedMetric, RankSide, SectorType
from ..domain.normalize import is_current_listed_stock
from ..domain.persistence import PublicationSourceGroup, SectorRadarRepository
from ..domain.persistence import HistoricalRanking, PublicationSourceGroup, SectorRadarRepository
from ..domain.source import (
DailyRow,
MoneyflowDcRow,
MoneyflowRow,
SectorIndexRow,
SectorMemberRow,
SourceSnapshot,
StockBasicRow,
TradeCalendarRow,
)
@@ -132,6 +131,39 @@ class SectorDetail:
similar_sectors: tuple[SimilarSector, ...]
class _RankHistory:
"""Index one day's selected ranks once, retaining complete versioned pool counts."""
def __init__(self, records: Sequence[HistoricalRanking]) -> None:
self.pools: dict[tuple[SectorType, MetricKind, str], int] = {}
self.rankings: dict[tuple[SectorType, str, MetricKind, str], RankedMetric] = {}
self.names: dict[tuple[SectorType, str], str] = {}
for record in records:
self.pools[(record.sector_type, record.metric_kind, record.metric_version)] = (
record.pool_size
)
if record.ranking is not None:
observation = record.ranking.observation
self.rankings.setdefault(
(
record.sector_type,
observation.sector_code,
record.metric_kind,
record.metric_version,
),
record.ranking,
)
self.names.setdefault(
(record.sector_type, observation.sector_code), observation.sector_name
)
def metric(self, sector_type: SectorType, sector_code: str, kind: MetricKind) -> HistoryMetric:
version = _METRIC_VERSIONS[kind]
size = self.pools.get((sector_type, kind, version), 0)
row = self.rankings.get((sector_type, sector_code, kind, version))
return _metric_from_ranking(row, size)
class ReadRadarDetails:
"""Reuse batched history and exact publication raw snapshots across read views."""
@@ -139,8 +171,8 @@ class ReadRadarDetails:
self.repository = repository
def history_data(
self, publication: RadarPublication
) -> tuple[dict[date, RadarPublication], dict[date, Sequence[RankedMetric]], tuple[date, ...]]:
self, publication: RadarPublication, sector_codes: Sequence[str]
) -> tuple[dict[date, RadarPublication], dict[date, _RankHistory], tuple[date, ...]]:
"""Load history in bounded batches; calendar holes remain explicit missing points."""
publications = {
item.target_trade_date: item
@@ -149,22 +181,23 @@ class ReadRadarDetails:
# Pin the current date to the response's chosen publication if a rebuild finishes
# during this request. All history rows are then fetched by these exact IDs.
publications[publication.target_trade_date] = publication
by_id = dict(
self.repository.load_publication_rankings(
tuple(item.publication_id for item in publications.values())
)
)
by_id: dict[str, list[HistoricalRanking]] = {}
for record in self.repository.load_ranked_history(
tuple(item.publication_id for item in publications.values()), sector_codes
):
by_id.setdefault(record.publication_id, []).append(record)
rows = {
day: by_id.get(item.publication_id, ())
day: _RankHistory(by_id.get(item.publication_id, ()))
if item.source_version == publication.source_version
else ()
else _RankHistory(())
for day, item in publications.items()
}
snapshots = self.snapshots(publication)
sources = self.repository.load_publication_rows(
publication.publication_id, (PublicationSourceGroup.CALENDAR,)
)
calendar = [
TradeCalendarRow.from_mapping(row)
for snapshot in snapshots.get(PublicationSourceGroup.CALENDAR, ())
for row in snapshot.rows
for row in sources.get(PublicationSourceGroup.CALENDAR, ())
]
dates = tuple(
sorted(
@@ -176,7 +209,7 @@ class ReadRadarDetails:
| set(publications)
)[-30:]
)
return publications, rows, dates
return publications, {day: rows.get(day, _RankHistory(())) for day in dates}, dates
def ranking_extras(
self, publication: RadarPublication, rankings: Sequence[RankedMetric], side: RankSide
@@ -184,16 +217,22 @@ class ReadRadarDetails:
"""Enrich one page from one batched thirty-session history, with no per-sector IO."""
if not rankings:
return {}
_, history, dates = self.history_data(publication)
snapshots = self.snapshots(publication)
_, history, dates = self.history_data(
publication, tuple(row.observation.sector_code for row in rankings)
)
sources = self.repository.load_publication_rows(
publication.publication_id,
(PublicationSourceGroup.CONCEPT_INDICES, PublicationSourceGroup.INDUSTRY_INDICES),
trade_date=publication.target_trade_date,
ts_codes=tuple(row.observation.sector_code for row in rankings),
)
indices = {
(item.sector_type, item.sector_code): item
for group, kind in (
(PublicationSourceGroup.CONCEPT_INDICES, SectorType.CONCEPT),
(PublicationSourceGroup.INDUSTRY_INDICES, SectorType.INDUSTRY),
)
for snapshot in snapshots.get(group, ())
for row in snapshot.rows
for row in sources.get(group, ())
for item in (SectorIndexRow.from_mapping(row, kind),)
if item.trade_date == publication.target_trade_date
}
@@ -202,12 +241,10 @@ class ReadRadarDetails:
observation = ranking.observation
key = (observation.sector_type, observation.sector_code)
index = indices.get(key)
current = history.get(publication.target_trade_date, ())
amount = metric_at(current, *key, MetricKind.AMOUNT).metric_value
ratio = metric_at(current, *key, MetricKind.RATIO).metric_value
points = [
metric_at(history.get(day, ()), *key, observation.metric_kind) for day in dates
]
current = history[publication.target_trade_date]
amount = current.metric(*key, MetricKind.AMOUNT).metric_value
ratio = current.metric(*key, MetricKind.RATIO).metric_value
points = [history[day].metric(*key, observation.metric_kind) for day in dates]
available_days = sum(not point.missing for point in points)
on_list_count = None
if available_days and side is not RankSide.ALL:
@@ -223,15 +260,6 @@ class ReadRadarDetails:
)
return result
def snapshots(
self, publication: RadarPublication
) -> dict[PublicationSourceGroup, list[SourceSnapshot]]:
"""Load exact source revisions, never global latest membership or provider data."""
grouped: dict[PublicationSourceGroup, list[SourceSnapshot]] = {}
for record in self.repository.load_publication_sources(publication.publication_id):
grouped.setdefault(record.source_group, []).append(record.snapshot)
return grouped
def history(self, target: date, sector_type: SectorType, sector_code: str) -> SectorHistory:
"""Return an exact-date publication and its compatible, past-only rank trajectory."""
publication = self.repository.get_successful_publication(target)
@@ -239,24 +267,15 @@ class ReadRadarDetails:
return SectorHistory(
"no_data", target, None, None, sector_type, sector_code, None, (), 0
)
publications, rows, dates = self.history_data(publication)
current_rows = rows.get(target, ())
name = next(
(
row.observation.sector_name
for row in current_rows
if row.observation.sector_type is sector_type
and row.observation.sector_code == sector_code
),
None,
)
publications, rows, dates = self.history_data(publication, (sector_code,))
name = rows[target].names.get((sector_type, sector_code))
points = tuple(
HistoryPoint(
day,
publications[day].publication_id if day in publications else None,
metric_at(rows.get(day, ()), sector_type, sector_code, MetricKind.AMOUNT),
metric_at(rows.get(day, ()), sector_type, sector_code, MetricKind.RATIO),
metric_at(rows.get(day, ()), sector_type, sector_code, MetricKind.SWING),
rows[day].metric(sector_type, sector_code, MetricKind.AMOUNT),
rows[day].metric(sector_type, sector_code, MetricKind.RATIO),
rows[day].metric(sector_type, sector_code, MetricKind.SWING),
)
for day in dates
)
@@ -286,15 +305,22 @@ class ReadRadarDetails:
latest = next((point for point in history.points if point.trade_date == target), None)
if latest:
summary = {kind.value: getattr(latest, kind.value) for kind in MetricKind}
snapshots = self.snapshots(history.publication)
sources = self.repository.load_publication_rows(
history.publication.publication_id,
(
PublicationSourceGroup.CONCEPT_INDICES,
PublicationSourceGroup.INDUSTRY_INDICES,
PublicationSourceGroup.STOCK_BASICS,
PublicationSourceGroup.MEMBERS,
),
)
indices = [
SectorIndexRow.from_mapping(row, kind)
for group, kind in (
(PublicationSourceGroup.CONCEPT_INDICES, SectorType.CONCEPT),
(PublicationSourceGroup.INDUSTRY_INDICES, SectorType.INDUSTRY),
)
for snapshot in snapshots.get(group, ())
for row in snapshot.rows
for row in sources.get(group, ())
]
index = next(
(
@@ -308,37 +334,44 @@ class ReadRadarDetails:
)
basics = {
item.ts_code: item
for snapshot in snapshots.get(PublicationSourceGroup.STOCK_BASICS, ())
for row in snapshot.rows
for row in sources.get(PublicationSourceGroup.STOCK_BASICS, ())
for item in (StockBasicRow.from_mapping(row),)
if is_current_listed_stock(item, target)
}
memberships: dict[str, dict[str, str]] = {}
for snapshot in snapshots.get(PublicationSourceGroup.MEMBERS, ()):
for row in snapshot.rows:
member = SectorMemberRow.from_mapping(row)
if member.trade_date == target and member.stock_code in basics:
memberships.setdefault(member.sector_code, {})[member.stock_code] = (
member.stock_name
)
for row in sources.get(PublicationSourceGroup.MEMBERS, ()):
member = SectorMemberRow.from_mapping(row)
if member.trade_date == target and member.stock_code in basics:
memberships.setdefault(member.sector_code, {})[member.stock_code] = (
member.stock_name
)
# Only the chosen sector's current-day facts are needed. Keep independent
# raw fields: normalized stock facts can suppress amounts when daily is missing.
sources = self.repository.load_publication_rows(
history.publication.publication_id,
(
PublicationSourceGroup.DAILY,
PublicationSourceGroup.MONEYFLOW_DC,
PublicationSourceGroup.MONEYFLOW,
),
trade_date=target,
ts_codes=tuple(sorted(memberships.get(sector_code, {}))),
)
daily = {
item.ts_code: item
for snapshot in snapshots.get(PublicationSourceGroup.DAILY, ())
for row in snapshot.rows
for row in sources.get(PublicationSourceGroup.DAILY, ())
for item in (DailyRow.from_mapping(row),)
if item.trade_date == target
}
main = {
item.ts_code: item
for snapshot in snapshots.get(PublicationSourceGroup.MONEYFLOW_DC, ())
for row in snapshot.rows
for row in sources.get(PublicationSourceGroup.MONEYFLOW_DC, ())
for item in (MoneyflowDcRow.from_mapping(row),)
if item.trade_date == target
}
active = {
item.ts_code: item
for snapshot in snapshots.get(PublicationSourceGroup.MONEYFLOW, ())
for row in snapshot.rows
for row in sources.get(PublicationSourceGroup.MONEYFLOW, ())
for item in (MoneyflowRow.from_mapping(row),)
if item.trade_date == target
}
@@ -414,6 +447,11 @@ def metric_at(
]
size = sum(row.rank_position is not None for row in pool)
row = next((row for row in pool if row.observation.sector_code == sector_code), None)
return _metric_from_ranking(row, size)
def _metric_from_ranking(row: RankedMetric | None, size: int) -> HistoryMetric:
"""Preserve missing metrics, stored percentiles and pool size independently."""
if row is None:
return HistoryMetric(pool_size=size)
percentile = row.rank_percentile
zhixing-server/src/zhixing_server/modules/sector_radar/domain/persistence.py
+40 -1
View File
@@ -12,13 +12,14 @@ from typing import Protocol
from .models import (
MembershipStatus,
MetricKind,
RadarPublication,
RankedMetric,
SectorDailyAggregate,
SectorType,
StockFactStatus,
)
from .source import SourceSnapshot
from .source import SourceScalar, SourceSnapshot
def _validate_digest(value: str, field_name: str) -> None:
@@ -151,6 +152,22 @@ class RankingRecord:
raise ValueError("publication_id must not be empty")
@dataclass(frozen=True, slots=True)
class HistoricalRanking:
"""A requested ranking plus its unfiltered publication/type/version pool size.
A pool without a requested sector still returns one entry with ranking=None,
so missing history retains the actual pool size instead of inventing zero.
"""
publication_id: str
sector_type: SectorType
metric_kind: MetricKind
metric_version: str
pool_size: int
ranking: RankedMetric | None
@dataclass(frozen=True, slots=True)
class DailyAggregateRecord:
"""One exact daily strategy input and optional source detail owned by a publication."""
@@ -230,6 +247,22 @@ class SectorRadarRepository(Protocol):
self, publication_id: str
) -> Sequence[PublicationSourceRecord]: ...
def load_publication_rows(
self,
publication_id: str,
source_groups: Sequence[PublicationSourceGroup],
*,
trade_date: date | None = None,
ts_codes: Sequence[str] | None = None,
) -> dict[PublicationSourceGroup, tuple[dict[str, SourceScalar], ...]]:
"""Read ordered source rows, optionally matching row trade_date and ts_code.
None means unfiltered; empty groups/codes return no rows. Projections do
not claim the source snapshot's row count or content hash. All reads
remain bound to the exact publication, including legacy publications.
"""
...
def mark_publication_sources_for_retry(
self,
publication_id: str,
@@ -307,6 +340,12 @@ class SectorRadarRepository(Protocol):
self, publication_ids: Sequence[str]
) -> Sequence[tuple[str, Sequence[RankedMetric]]]: ...
def load_ranked_history(
self, publication_ids: Sequence[str], sector_codes: Sequence[str]
) -> Sequence[HistoricalRanking]:
"""Read requested sectors and full pool counts from exact revisions."""
...
def load_daily_aggregate_history(
self, target_trade_date: date, *, limit_dates: int
) -> Sequence[SectorDailyAggregate]: ...
zhixing-server/src/zhixing_server/modules/sector_radar/infrastructure/memory.py
+58 -1
View File
@@ -8,6 +8,7 @@ from dataclasses import replace
from datetime import date, datetime
from ..domain.models import (
MetricKind,
PublicationStatus,
RadarPublication,
RankedMetric,
@@ -16,6 +17,7 @@ from ..domain.models import (
)
from ..domain.persistence import (
DailyAggregateRecord,
HistoricalRanking,
MembershipRecord,
PublicationSourceGroup,
PublicationSourceRecord,
@@ -25,7 +27,7 @@ from ..domain.persistence import (
StockMembershipEntry,
WriteCounts,
)
from ..domain.source import SourceSnapshot
from ..domain.source import SourceScalar, SourceSnapshot
class InMemorySectorRadarRepository:
@@ -98,6 +100,36 @@ class InMemorySectorRadarRepository:
)
)
def load_publication_rows(
self,
publication_id: str,
source_groups: Sequence[PublicationSourceGroup],
*,
trade_date: date | None = None,
ts_codes: Sequence[str] | None = None,
) -> dict[PublicationSourceGroup, tuple[dict[str, SourceScalar], ...]]:
"""Project exact source rows with the same filtering and order as PostgreSQL."""
groups = set(source_groups)
codes = None if ts_codes is None else set(ts_codes)
target = None if trade_date is None else trade_date.strftime("%Y%m%d")
result: dict[PublicationSourceGroup, list[dict[str, SourceScalar]]] = {}
for record in sorted(
self.publication_sources.values(),
key=lambda item: (item.source_group.value, item.source_order),
):
if record.publication_id != publication_id or record.source_group not in groups:
continue
for row in record.snapshot.rows:
if (
target is not None
and str(row.get("trade_date", "")).strip().replace("-", "") != target
):
continue
if codes is not None and str(row.get("ts_code", "")).strip() not in codes:
continue
result.setdefault(record.source_group, []).append(dict(row))
return {group: tuple(rows) for group, rows in result.items()}
def mark_publication_sources_for_retry(
self,
publication_id: str,
@@ -469,6 +501,31 @@ class InMemorySectorRadarRepository:
"""Read exact immutable revisions selected by the history reader."""
return tuple((key, self.load_rankings(key)) for key in publication_ids)
def load_ranked_history(
self, publication_ids: Sequence[str], sector_codes: Sequence[str]
) -> Sequence[HistoricalRanking]:
"""Keep unfiltered pool counts while returning only requested sector rows."""
if not publication_ids or not sector_codes:
return ()
wanted = set(sector_codes)
pools: dict[tuple[str, SectorType, MetricKind, str], list[RankedMetric]] = {}
for publication_id in publication_ids:
for row in self.load_rankings(publication_id):
observation = row.observation
key = (
publication_id,
observation.sector_type,
observation.metric_kind,
observation.metric_version,
)
pools.setdefault(key, []).append(row)
result: list[HistoricalRanking] = []
for key, rows in sorted(pools.items()):
size = sum(row.rank_position is not None for row in rows)
selected = [row for row in rows if row.observation.sector_code in wanted]
result.extend(HistoricalRanking(*key, size, row) for row in selected or [None])
return tuple(result)
def load_daily_aggregate_history(
self, target_trade_date: date, *, limit_dates: int
) -> Sequence[SectorDailyAggregate]:
zhixing-server/src/zhixing_server/modules/sector_radar/infrastructure/postgres.py
+94 -1
View File
@@ -26,6 +26,7 @@ from ..domain.models import (
)
from ..domain.persistence import (
DailyAggregateRecord,
HistoricalRanking,
MembershipRecord,
PublicationSourceGroup,
PublicationSourceRecord,
@@ -35,7 +36,7 @@ from ..domain.persistence import (
StockMembershipEntry,
WriteCounts,
)
from ..domain.source import SourceSnapshot
from ..domain.source import SourceScalar, SourceSnapshot
class SectorRadarRepositoryError(RuntimeError):
@@ -223,6 +224,46 @@ class PostgresSectorRadarRepository:
for row in rows
)
def load_publication_rows(
self,
publication_id: str,
source_groups: Sequence[PublicationSourceGroup],
*,
trade_date: date | None = None,
ts_codes: Sequence[str] | None = None,
) -> dict[PublicationSourceGroup, tuple[dict[str, SourceScalar], ...]]:
"""Filter immutable source payloads before transport and Python decoding.
Keep original source/row order so duplicate-key resolution is unchanged.
This is a row projection, not a snapshot with a misleading content hash.
Database failures are translated by the repository connection boundary.
"""
if not source_groups or ts_codes is not None and not ts_codes:
return {}
predicates = ["link.publication_id = %s", "link.source_group = ANY(%s)"]
parameters: list[object] = [publication_id, [group.value for group in source_groups]]
if trade_date is not None:
predicates.append("replace(btrim(item.value ->> 'trade_date'), '-', '') = %s")
parameters.append(trade_date.strftime("%Y%m%d"))
if ts_codes is not None:
predicates.append("btrim(item.value ->> 'ts_code') = ANY(%s)")
parameters.append(list(ts_codes))
query = (
"""
SELECT link.source_group,
jsonb_agg(item.value ORDER BY link.source_order, item.ordinality)
FROM sector_radar_publication_source AS link
JOIN sector_radar_source_snapshot AS snapshot ON snapshot.id = link.source_snapshot_id
CROSS JOIN LATERAL jsonb_array_elements(snapshot.payload)
WITH ORDINALITY AS item(value, ordinality)
WHERE """
+ " AND ".join(predicates)
+ " GROUP BY link.source_group"
)
with self._connection() as connection:
rows = connection.execute(query, tuple(parameters)).fetchall()
return {PublicationSourceGroup(row[0]): tuple(row[1]) for row in rows}
def mark_publication_sources_for_retry(
self,
publication_id: str,
@@ -870,6 +911,58 @@ class PostgresSectorRadarRepository:
grouped.setdefault(str(row[0]), []).append(self._ranking_from_row(row[1:]))
return tuple((key, tuple(grouped.get(key, ()))) for key in publication_ids)
def load_ranked_history(
self, publication_ids: Sequence[str], sector_codes: Sequence[str]
) -> Sequence[HistoricalRanking]:
"""Return requested rankings and complete pool sizes from immutable revisions.
Aggregate before filtering sectors: absent sectors must retain nonzero
pool sizes, and unavailable ranks must not count toward the pool.
"""
if not publication_ids or not sector_codes:
return ()
with self._connection() as connection:
rows = connection.execute(
"""
WITH pools AS (
SELECT publication_id, sector_type, metric_kind, metric_version,
COUNT(rank_position) AS pool_size
FROM sector_radar_ranking
WHERE publication_id = ANY(%s)
GROUP BY publication_id, sector_type, metric_kind, metric_version
)
SELECT pool.publication_id, pool.sector_type, pool.metric_kind,
pool.metric_version, pool.pool_size,
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 pools AS pool
LEFT JOIN sector_radar_ranking AS ranking
ON ranking.publication_id = pool.publication_id
AND ranking.sector_type = pool.sector_type
AND ranking.metric_kind = pool.metric_kind
AND ranking.metric_version = pool.metric_version
AND ranking.sector_code = ANY(%s)
ORDER BY pool.publication_id, pool.sector_type, pool.metric_kind,
ranking.rank_position NULLS LAST, ranking.sector_code
""",
(list(publication_ids), list(sector_codes)),
).fetchall()
return tuple(
HistoricalRanking(
str(row[0]),
SectorType(row[1]),
MetricKind(row[2]),
str(row[3]),
int(row[4]),
None if row[5] is None else self._ranking_from_row(row[5:]),
)
for row in rows
)
def load_daily_aggregate_history(
self, target_trade_date: date, *, limit_dates: int
) -> Sequence[SectorDailyAggregate]:
zhixing-server/tests/unit/sector_radar/test_detail_reads.py
+428
View File
@@ -0,0 +1,428 @@
"""Regression coverage for publication-scoped, bounded radar read paths."""
import os
from collections.abc import Generator, Sequence
from dataclasses import replace
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from pathlib import Path
from unittest.mock import patch
from uuid import uuid4
import psycopg
import pytest
from alembic import command
from alembic.config import Config
from fastapi.testclient import TestClient
from psycopg.conninfo import make_conninfo
from psycopg.sql import SQL, Identifier
from zhixing_server.bootstrap.app import create_app
from zhixing_server.bootstrap.config import Settings, sqlalchemy_database_url
from zhixing_server.modules.sector_radar.application.details import ReadRadarDetails, metric_at
from zhixing_server.modules.sector_radar.application.read import RadarQuery, ReadSectorRadar
from zhixing_server.modules.sector_radar.domain.metrics import AmountNetStrategy
from zhixing_server.modules.sector_radar.domain.models import (
MetricKind,
MetricObservation,
MetricQuality,
MetricUnit,
PublicationStatus,
RadarPublication,
RankSide,
SectorType,
)
from zhixing_server.modules.sector_radar.domain.persistence import (
PublicationSourceGroup,
PublicationSourceRecord,
RankingRecord,
SectorRadarRepository,
)
from zhixing_server.modules.sector_radar.domain.ranking import rank_metric_observations
from zhixing_server.modules.sector_radar.domain.source import SourceScalar, build_source_snapshot
from zhixing_server.modules.sector_radar.infrastructure.memory import InMemorySectorRadarRepository
from zhixing_server.modules.sector_radar.infrastructure.postgres import (
PostgresSectorRadarRepository,
)
from zhixing_server.modules.sector_radar.presentation.http import get_sector_radar_reader
TARGET = date(2026, 9, 4)
NOW = datetime(2026, 9, 4, 18, tzinfo=UTC)
CODE = "BK1147.DC"
def seed_detail(repository: SectorRadarRepository) -> None:
"""Seed raw-only legacy publications with missing facts and unrelated rows."""
for offset in (-1, 0):
day = TARGET + timedelta(days=offset)
publication = RadarPublication(
publication_id=f"detail-{offset}",
target_trade_date=day,
status=PublicationStatus.RUNNING,
source_version="tushare-pro-v1",
universe_version="test-v1",
metric_versions=(AmountNetStrategy.metric_version,),
input_hash=None,
coverage=Decimal(0),
started_at=NOW,
)
repository.create_publication(publication)
observations = tuple(
MetricObservation(
trade_date=day,
sector_type=SectorType.CONCEPT,
sector_code=code,
sector_name=code,
metric_kind=MetricKind.AMOUNT,
metric_version=AmountNetStrategy.metric_version,
implementation_kind="independent",
unit=MetricUnit.CNY_100M,
value=Decimal(20 - index),
quality=MetricQuality.AVAILABLE,
member_count=2,
valid_sample_count=2,
membership_coverage=Decimal(1),
moneyflow_coverage=Decimal(1),
)
for index, code in enumerate((CODE, "OTHER") if offset == 0 else ("OTHER",))
)
repository.save_rankings(
RankingRecord(publication.publication_id, row)
for row in rank_metric_observations(observations)
)
repository.finish_publication(
replace(
publication,
status=PublicationStatus.SUCCESS,
input_hash="a" * 64,
coverage=Decimal(1),
finished_at=NOW + timedelta(seconds=1),
)
)
groups: dict[PublicationSourceGroup, tuple[dict[str, SourceScalar], ...]] = {
PublicationSourceGroup.CALENDAR: tuple(
{"exchange": "SSE", "cal_date": (TARGET + timedelta(days=i)).isoformat(), "is_open": 1}
for i in (-2, -1, 0, 1)
),
PublicationSourceGroup.CONCEPT_INDICES: tuple(
{
"ts_code": code,
"name": code,
"trade_date": TARGET.isoformat(),
"pct_change": "1.25",
"leading_code": "000003.SZ",
}
for code in (CODE, "OTHER")
),
PublicationSourceGroup.STOCK_BASICS: tuple(
{
"ts_code": code,
"symbol": code[:6],
"name": code,
"exchange": "SZSE",
"list_status": status,
"list_date": "20200101",
}
for code, status in (
("000001.SZ", "L"),
("000002.SZ", "L"),
("000003.SZ", "L"),
("000004.SZ", "D"),
)
),
PublicationSourceGroup.MEMBERS: tuple(
{"ts_code": sector, "con_code": code, "name": code, "trade_date": "20260904"}
for sector, code in (
(CODE, "000001.SZ"),
(CODE, "000002.SZ"),
(CODE, "000004.SZ"),
("OTHER", "000002.SZ"),
("OTHER", "000003.SZ"),
)
),
PublicationSourceGroup.DAILY: (
{"ts_code": "000001.SZ", "trade_date": "20260904", "pct_chg": "2.50"},
{"ts_code": "000003.SZ", "trade_date": "20260904", "pct_chg": "5"},
{"ts_code": "000001.SZ", "trade_date": "20260903", "pct_chg": "99"},
),
PublicationSourceGroup.MONEYFLOW_DC: (
{
"ts_code": "000002.SZ",
"trade_date": "2026-09-04",
"net_amount": "1.2345",
"name": "000002.SZ",
},
),
PublicationSourceGroup.MONEYFLOW: (
{
"ts_code": "000001.SZ",
"trade_date": "20260904",
"net_mf_amount": "11",
"buy_lg_amount": "10",
"sell_lg_amount": "2",
"buy_elg_amount": "4",
"sell_elg_amount": "1",
},
),
PublicationSourceGroup.SUSPENSIONS: ({"unused": "must not be read"},),
}
for group, rows in groups.items():
snapshot = build_source_snapshot(
api_name=group.value,
params={},
rows=rows,
target_trade_date=TARGET,
observed_at=NOW,
)
repository.save_source_snapshots((snapshot,))
repository.save_publication_sources(
(PublicationSourceRecord("detail-0", group, 0, snapshot),)
)
class TrackingRepository(InMemorySectorRadarRepository):
def __init__(self) -> None:
super().__init__()
self.reads: list[
tuple[tuple[PublicationSourceGroup, ...], date | None, tuple[str, ...] | None]
] = []
def load_publication_sources(self, publication_id: str) -> Sequence[PublicationSourceRecord]:
raise AssertionError("HTTP read must not load complete audit snapshots")
def load_publication_rows(
self,
publication_id: str,
source_groups: Sequence[PublicationSourceGroup],
*,
trade_date: date | None = None,
ts_codes: Sequence[str] | None = None,
) -> dict[PublicationSourceGroup, tuple[dict[str, SourceScalar], ...]]:
self.reads.append(
(tuple(source_groups), trade_date, None if ts_codes is None else tuple(ts_codes))
)
return super().load_publication_rows(
publication_id,
source_groups,
trade_date=trade_date,
ts_codes=ts_codes,
)
def test_detail_http_reads_only_required_rows_and_preserves_independent_missing_values() -> None:
repository = TrackingRepository()
seed_detail(repository)
app = create_app()
app.dependency_overrides[get_sector_radar_reader] = lambda: ReadSectorRadar(repository)
with TestClient(app) as client:
response = client.get(
f"/api/v1/sector-radar/sectors/concept/{CODE}/detail?trade_date={TARGET}"
)
assert response.status_code == 200
body = response.json()
assert body["members"] == [
{
"ts_code": "000001.SZ",
"name": "000001.SZ",
"pct_change": "2.50",
"net_amount_yuan": None,
"active_buy_net_amount_yuan": "110000",
},
{
"ts_code": "000002.SZ",
"name": "000002.SZ",
"pct_change": None,
"net_amount_yuan": "12345.0000",
"active_buy_net_amount_yuan": None,
},
]
assert body["similar_sectors"][0]["intersection_count"] == 1
assert body["similar_sectors"][0]["union_count"] == 3
assert body["history"]["available_days"] == 1
assert [point["amount"]["pool_size"] for point in body["history"]["points"]] == [0, 1, 2]
assert len(repository.reads) == 3
assert repository.reads[0][0] == (PublicationSourceGroup.CALENDAR,)
assert repository.reads[-1][1:] == (TARGET, ("000001.SZ", "000002.SZ"))
assert all(PublicationSourceGroup.SUSPENSIONS not in read[0] for read in repository.reads)
def test_history_and_ranking_enrichment_do_not_read_members_or_stock_facts() -> None:
repository = TrackingRepository()
seed_detail(repository)
history = ReadRadarDetails(repository).history(TARGET, SectorType.CONCEPT, CODE)
assert history.status == "success"
assert [read[0] for read in repository.reads] == [(PublicationSourceGroup.CALENDAR,)]
repository.reads.clear()
page = ReadSectorRadar(repository).query(
RadarQuery(trade_date=TARGET, page_size=1, side=RankSide.TOP)
)
assert page.extras[CODE].on_list_count == 1
assert page.extras[CODE].pct_change == Decimal("1.25")
assert len(repository.reads) == 2
assert all(
set(read[0])
<= {
PublicationSourceGroup.CALENDAR,
PublicationSourceGroup.CONCEPT_INDICES,
PublicationSourceGroup.INDUSTRY_INDICES,
}
for read in repository.reads
)
def test_filtered_history_matches_full_pool_metrics_even_when_sector_is_absent() -> None:
repository = InMemorySectorRadarRepository()
seed_detail(repository)
records = repository.load_ranked_history(("detail--1", "detail-0"), (CODE,))
assert {record.ranking.observation.sector_code for record in records if record.ranking} == {
CODE
}
assert any(record.ranking is None and record.pool_size == 1 for record in records)
history = ReadRadarDetails(repository).history(TARGET, SectorType.CONCEPT, CODE)
for point in history.points:
rows = repository.load_rankings(point.publication_id) if point.publication_id else ()
assert point.amount == metric_at(rows, SectorType.CONCEPT, CODE, MetricKind.AMOUNT)
def test_history_keeps_selected_publication_when_rebuild_finishes_during_read() -> None:
repository = InMemorySectorRadarRepository()
seed_detail(repository)
selected = repository.get_successful_publication(TARGET)
assert selected is not None
replacement = replace(selected, publication_id="newer-revision")
with patch.object(repository, "load_history_publications", return_value=(replacement,)):
history = ReadRadarDetails(repository).history(TARGET, SectorType.CONCEPT, CODE)
assert history.publication == selected
assert history.points[-1].publication_id == selected.publication_id
assert history.points[-1].amount.pool_size == 2
def test_history_does_not_mix_source_versions_or_metric_versions() -> None:
repository = InMemorySectorRadarRepository()
seed_detail(repository)
prior = repository.publications["detail--1"]
repository.publications[prior.publication_id] = replace(prior, source_version="incompatible")
current = repository.load_rankings("detail-0")[0]
repository.save_rankings(
(
RankingRecord(
"detail-0",
replace(
current,
observation=replace(
current.observation, metric_version="old-amount", value=Decimal("999")
),
),
),
)
)
history = ReadRadarDetails(repository).history(TARGET, SectorType.CONCEPT, CODE)
assert history.points[-2].amount.pool_size == 0
assert history.points[-2].amount.missing
assert history.points[-1].amount.pool_size == 2
assert history.points[-1].amount.metric_value == Decimal(20)
@pytest.fixture
def postgres_read_repository() -> Generator[PostgresSectorRadarRepository]:
"""Use an isolated schema so read regression tests never overwrite other fixtures."""
database_url = os.getenv("ZHIXING_TEST_DATABASE_URL")
if not database_url:
pytest.skip("set ZHIXING_TEST_DATABASE_URL to run PostgreSQL integration tests")
schema = "radar_read_" + uuid4().hex
with psycopg.connect(database_url, autocommit=True) as connection:
connection.execute(SQL("CREATE SCHEMA {}").format(Identifier(schema)))
isolated_url = (
database_url + ("&" if "?" in database_url else "?") + f"options=-csearch_path%3D{schema}"
)
repository = PostgresSectorRadarRepository(
make_conninfo(database_url, options=f"-c search_path={schema}")
)
try:
config = Config(str(Path(__file__).parents[3] / "alembic.ini"))
config.set_main_option(
"sqlalchemy.url", sqlalchemy_database_url(isolated_url).replace("%", "%%")
)
config.config_file_name = None
with patch(
"zhixing_server.bootstrap.config.get_settings",
return_value=Settings(database_url=isolated_url),
):
command.upgrade(config, "head")
yield repository
finally:
repository.close()
with psycopg.connect(database_url, autocommit=True) as connection:
connection.execute(SQL("DROP SCHEMA {} CASCADE").format(Identifier(schema)))
@pytest.mark.integration
def test_postgres_bounded_reads_match_legacy_raw_only_data(
postgres_read_repository: PostgresSectorRadarRepository,
) -> None:
repository = postgres_read_repository
memory = InMemorySectorRadarRepository()
for repo in (repository, memory):
seed_detail(repo)
ranked = repo.load_rankings("detail-0")[0]
repo.save_rankings(
(
RankingRecord(
"detail-0",
replace(
ranked,
rank_position=None,
rank_percentile=None,
observation=replace(
ranked.observation,
sector_code="MISSING",
value=None,
quality=MetricQuality.UNAVAILABLE,
),
),
),
RankingRecord(
"detail-0",
replace(
ranked,
observation=replace(ranked.observation, sector_type=SectorType.INDUSTRY),
),
),
)
)
expected = ReadRadarDetails(memory).detail(TARGET, SectorType.CONCEPT, CODE)
actual = ReadRadarDetails(repository).detail(TARGET, SectorType.CONCEPT, CODE)
assert actual == expected
assert actual.summary["amount"].pool_size == 2
assert repository.load_ranked_history(
("detail--1", "detail-0"), (CODE,)
) == memory.load_ranked_history(("detail--1", "detail-0"), (CODE,))
groups = (PublicationSourceGroup.DAILY, PublicationSourceGroup.MONEYFLOW_DC)
assert repository.load_publication_rows(
"detail-0", groups, trade_date=TARGET, ts_codes=("000002.SZ",)
) == memory.load_publication_rows(
"detail-0", groups, trade_date=TARGET, ts_codes=("000002.SZ",)
)
assert repository.load_publication_rows("detail-0", groups, ts_codes=()) == {}
assert repository.load_publication_rows("detail-0", ()) == {}
assert repository.load_ranked_history((), (CODE,)) == ()
assert repository.load_ranked_history(("detail-0",), ()) == ()
assert repository.load_publication_rows("unknown-publication", groups) == {}
# Later snapshot rows win over earlier ones even for the same stock/date.
replacement = build_source_snapshot(
api_name="daily",
params={"retry": "1"},
rows=({"ts_code": "000001.SZ", "trade_date": "20260904", "pct_chg": "8.5"},),
target_trade_date=TARGET,
observed_at=NOW,
)
repository.save_source_snapshots((replacement,))
repository.save_publication_sources(
(PublicationSourceRecord("detail-0", PublicationSourceGroup.DAILY, 1, replacement),)
)
rows = repository.load_publication_rows(
"detail-0", (PublicationSourceGroup.DAILY,), trade_date=TARGET, ts_codes=("000001.SZ",)
)
assert [row["pct_chg"] for row in rows[PublicationSourceGroup.DAILY]] == ["2.50", "8.5"]