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

Merged
sakibcc merged 1 commits from develop into main 2026-09-07 14:46:27 +08:00
13 changed files with 1083 additions and 69 deletions
@@ -0,0 +1,3 @@
{"file": ".trellis/spec/backend/quality-guidelines.md", "reason": "后端必要检查与测试形状"}
{"file": ".trellis/spec/backend/http-api-contracts.md", "reason": "HTTP 兼容性"}
{"file": ".trellis/tasks/09-07-api-performance-diagnosis/design.md", "reason": "已批准设计与性能证据"}
@@ -0,0 +1,44 @@
# 性能诊断与建议设计(已批准实施)
## 证据与边界
2026-09-07,低频公网 GET:示例 detail 四次 3.045 / 3.059 / 3.074 / 3.442 秒,200,10750 字节;显式绕过本机代理复测 3.312557 秒,TLS 完成 0.443439 秒,首字节 3.312484 秒。直连 healthz 0.810132 秒、dates 0.815043 秒、ratio 榜单 2.570251 秒。默认网络路径 history 1.834179 秒、2020-01-01 无数据 detail 0.746555 秒。请求成功验证示例为 SPD概念,成员 14 个。
这些是客户端端到端耗时,不是服务端或 SQL 独立耗时;样本不足以推断 P95、并发容量或全站所有接口。生产代码版本、CPU/IO/锁等待及 SQL 执行计划未验证。
## 当前调用链
- presentation/http.py:355 -> ReadRadarDetails.detail -> history -> history_data。
- application/details.py:145 的 history_data 加载最多 30 个发布日的全部排名;为了取得 calendar,调用 snapshots 加载整个批次全部原始输入。
- application/details.py:294 的 detail 再次调用 snapshots,并在 Python 中重建股票基础信息、全体成员关系、行情和资金索引,再计算相似板块。
- infrastructure/postgres.py:194 的 load_publication_sources 按 publication_id 连接来源表和快照表,SELECT 包括完整 JSONB payload,没有 source_group 或证券过滤;:1112 还把每个 payload 行复制为 dict。
- application/details.py:182 的 ranking_extras 具有相同的重复读取模式;read.py:285 在 amount/ratio 榜单调用它。
- presentation/http.py:258 已按进程缓存仓储;postgres.py:48 默认连接池上限 4。当前部署 Dockerfile:48 没有显式指定 worker 数量,但生产环境覆盖与并发压力未知,不能据此确诊排队。
本地使用 tests/unit/sector_radar/test_read.py 的内存仓储夹具,仅代理计数真实应用层调用,不改产品代码:detail 为 get_successful_publication=1, load_history_publications=1, load_publication_rankings=1, load_publication_sources=2;普通金额榜单另有 load_rankings=1。此实验确认调用次数,不测量生产 SQL 成本。
## 已批准的优化次序
1. 取得服务端分段计时与只读查询计划,分别测连接池等待、SQL 执行与取数、JSON 转换、Python 组装;先对最重的快照读取确认行数和字节数。
2. 日历只读取 calendar 来源;同一请求避免重复读取同一发布输入,先减少明显多余工作。
3. 详情优先读取既有 publication 归属的事实和聚合投影;成员及股票只取所需范围,相似板块考虑在发布阶段预计算。旧批次投影缺失必须保持当前缺失语义,不能换用全局最新成员或直接读 Tushare。
4. 历史请求保留同类同版本的真实排名池大小、名次、百分位和缺失状态,SQL 只返回所需板块结果与分组统计,避免每次构造全量排名对象并重复扫描。
5. 仅在查询计划显示需要时提出索引;当前 publication/source/ranking 已有主键及索引,不能笼统归因为缺索引。
6. 优化冷请求后,依据重复访问与并发数据决定是否加入有界进程缓存或 Redis。
## Redis 取舍与一致性
Redis 可缓存最终响应/紧凑投影,适合读多写少的已发布收盘数据,尤其多进程/多实例需要共享结果时。它不是当前诊断的必要前提;只安装服务并不加速,必须接入读取、写入和失效逻辑,未命中仍走原查询。
缓存键至少含响应 schema 版本、请求参数、当前 publication_id;含历史曲线的响应还依赖此前各日选中的 publication_id/source_version/metric_version,历史补录或重建也必须改变键或触发失效,不能仅使用目标日期或当前批次 ID。缓存只保存成功且版本明确的投影,设置容量上限、TTL、并发回填保护和故障回源。TTL 不替代明确的发布版本语义。
官方资料:https://redis.io/docs/latest/develop/use-cases/cache-aside/ ,已经 Context7 与官方网页核验通用 cache-aside、TTL 和显式失效机制;当前项目未发现 Redis 依赖,未选择版本。
## 兼容与回滚
不改变 HTTP 字段、精度、历史缺失语义、最后有效发布规则。缓存层应可关闭回源;如后续需要新增投影或迁移,应先独立评审与授权。用户已批准本地实现与验证;线上发布由用户负责。
## 本轮落实的读取设计
- 新增 publication-scoped 原始行投影读取接口,明确 sources、trade_date、ts_codes 过滤,不伪造带原快照哈希的裁剪快照。保留 source_order 和快照内行顺序,保证重复键覆盖行为不变。
- 历史只读 calendar;榜单额外只读指数;详情先读取指数/股票基础/成员用于当前上市池和重合率,再按目标日与成员读取 daily/moneyflow_dc/moneyflow。
- 新增按所需板块读取历史排名的查询,完整排名池分组统计在过滤目标板块前完成;不存在板块也保留该池大小。应用层一次建立按板块、指标、版本的查找表,避免循环扫描。
- 既有 stock_fact 非 available 会清除 net_amount 且没有 publication_id,旧字段也可能空,不能直接无损替代独立来源读数;本轮不迁移、不重建、不引入全局缓存。相似度在请求内基于必要成员集计算。
- 性能目标用同一隔离 PostgreSQL 数据集前后对比与传输范围断言验证,实际公网改善由用户发布后验证。
## 验证结论
实现只改四个 sector_radar 后端文件,新增读取回归测试文件。无迁移、依赖、HTTP 字段或全局缓存改动。真实数据库结果、完整响应一致性及前后耗时记录在 research/performance.json。详情仍需要相似板块和上市过滤使用的完整成员/基础信息;不兼容来源版本的少量池统计仍在查询后丢弃,这两点保留为后续测量候选。
@@ -0,0 +1,3 @@
{"file": ".trellis/spec/backend/quality-guidelines.md", "reason": "后端必要检查与测试形状"}
{"file": ".trellis/spec/backend/http-api-contracts.md", "reason": "HTTP 兼容性"}
{"file": ".trellis/tasks/09-07-api-performance-diagnosis/design.md", "reason": "已批准设计与性能证据"}
@@ -0,0 +1,30 @@
# 后续执行建议(已批准实施)
## 当前已完成
- [x] 公网低频请求复现与分段计时,包含显式直连对照。
- [x] 检查路由、应用层、仓储、既有迁移和部署配置。
- [x] 用现有内存仓储夹具追踪重复读取,未修改产品代码。
- [x] Redis 取舍与发布/历史依赖失效边界分析。
## 实施顺序
1. 核对生产版本并收集脱敏的 server/request 分段耗时、池等待、查询行数与字节数;针对 SELECT 使用只读执行计划并设超时,不做生产压力测试。
2. 审阅需修改文件全文并加载 backend 规格,明确最终性能目标;补齐真实 implement/check 上下文清单后再 task.py start。
3. 优先修复完整快照重复读取和日历过量读取,增加能约束调用次数、过滤范围及历史一致性的回归测试。
4. 根据实测再决定缩小排名结果、使用既有投影或新增预计算。每项范围变化都更新设计;不以增加连接数或 worker 数替代测量。
5. 运行受影响测试、后端规定的 Ruff、Pyright 和 pytest;固定样本比较输出语义与耗时。在获批的环境验证冷/热请求与必要并发,不凭公网少量样本宣称 P95。
6. 仅在确认缓存需求后评审 Redis;验证重建、历史补录、并发回填、容量淘汰和 Redis 不可用时回源。
## 风险与授权
用户已明确批准按方案实施本地优化和验证。隔离本地 PostgreSQL 用于查询正确性与对比,禁止连接生产执行写入或压测。用户随后明确要求提交到本地 develop;不推送、不部署、不更新共享规范。
## 本轮实现与验证结果
- [x] publication 行投影查询按 source_group、目标日、ts_code 过滤,source_order/ordinality 稳定,无原始快照重复取数。
- [x] 历史排名在 SQL 中统计完整池后过滤板块;应用层索引查找;缺失板块、旧指标版本、不兼容来源版本、请求期间重建钉住均覆盖。
- [x] 真实 PostgreSQL 隔离 schema 验证,旧版仅原始快照数据仍可读取独立缺失值;未排行和行业池不会污染概念池。
- [x] 独立只读审查 /root/read_path_review 完成,无阻塞发现;主代理负责修改与执行验证。
- [x] 本地同一数据集三端点各 5 次 before/after 比较,全量响应一致。详情中位 1.7107→0.4841s;history 0.8580→0.0262s;rankings 1.2415→0.0645s。详见 research/performance.json,不能作为生产 SLA。
- [x] Ruff format/check 通过;受改文件 Pyright 验证;全仓 Pyright 14 个 selection 错误与修改前文件/行/内容逐项完全一致。
- [x] 后端全套以 unit→HTTP→integration 顺序执行:226 passed / 1 failed。剩余市场数据集成测试在修改前复现 Connection.executemany AttributeError;不扩大修改范围。默认集成测试优先顺序另有既存 Alembic fileConfig 污染 caplog 的问题。
- [ ] 用户自行发布后复测公网 detail/history/rankings;无 Redis、无迁移、不要求历史重跑。
本地实现已完成。质量门禁存在明确的既有阻塞,未宣称全仓全绿;用户已授权将实现、回归测试和性能证据提交到本地 develop;任务待用户发布验证,暂不归档。
@@ -0,0 +1,27 @@
# 接口性能诊断与缓存方案评估
## 目标
解释用户观察到的多个接口约 3 秒延迟,以概念板块 BK1147.DC 在 2026-09-04 的详情接口为切入点,在不引入 Redis 的情况下优化冷请求读取与计算,供用户自行发布后评估效果。
## 已确认事实
- 用户已在诊断与方案回顾后明确批准按优化顺序开始实施,由用户自行发布;Redis 留待上线效果验证后评估。
- 公网 GET 四次均返回 HTTP 200,总耗时 3.045–3.442 秒,响应体 10750 字节;主要等待发生在首字节之前。
- 排查开始时本地 develop 分支工作区干净;生产是否与当前提交一致尚未确认。
- 静态检查及本地调用追踪确认:详情与金额榜单每次调用两次 load_publication_sources;来源查询读取完整 payload(application/details.py:145、:182、:294;infrastructure/postgres.py:194)。
## 范围与要求
- 低频只读测量示例请求,分析本地路由、查询、连接管理与部署配置。
- 区分观测事实、代码风险和待生产证据验证的假设。
- 实施资金雷达详情、历史和榜单的最小充分读取优化;不修改外部系统。
## 验收标准
- 详情/榜单/历史不再加载完整发布快照,按需读取来源组、目标日与成员;历史只传输目标板块排名及完整池统计。
- 保留全部 HTTP 字段、Decimal 精度、旧批次独立缺失值、历史发布选择与相似板块口径。
- 通过真实 PostgreSQL 查询验证、语义回归测试和后端质量门禁,给出同一数据集的前后性能比较;不承诺未上线的公网秒数。
- 记录可复现请求的分段耗时,避免把总耗时直接等同于 SQL 耗时。
- 为关键判断提供代码位置或实测依据。
- 说明 Redis 是否必要、适用条件及失效策略边界。
- 明确未验证事项及下一步建议。
## 不在范围内
生产压测、部署、迁移、创建索引、接入 Redis、修改共享规范、推送远端。用户已另行授权提交本地 develop。
@@ -0,0 +1,126 @@
"""Task-local benchmark: isolated localhost PostgreSQL only; no production writes.
Run from zhixing-server with DATABASE_URL pointing to a disposable test database.
Set PYTHONPATH to the before/after source tree to compare the same persisted data.
Use --seed once, then --run before.json / --run after.json (five reads per endpoint).
"""
import argparse
import hashlib
import json
import os
import runpy
import statistics
import time
from dataclasses import replace
from datetime import timedelta
from decimal import Decimal
from pathlib import Path
from urllib.parse import urlsplit
from unittest.mock import patch
import psycopg
from alembic import command
from alembic.config import Config
from zhixing_server.bootstrap.config import Settings, sqlalchemy_database_url
from zhixing_server.modules.sector_radar.application.details import ReadRadarDetails
from zhixing_server.modules.sector_radar.application.read import ReadSectorRadar, RadarQuery
from zhixing_server.modules.sector_radar.domain.models import PublicationStatus, SectorType, MetricKind, MetricUnit
from zhixing_server.modules.sector_radar.domain.metrics import AmountNetStrategy, RatioTurnoverStrategy, SwingEqualThreeToTenStrategy
from zhixing_server.modules.sector_radar.domain.persistence import PublicationSourceGroup as Group, PublicationSourceRecord, RankingRecord
from zhixing_server.modules.sector_radar.domain.ranking import rank_metric_observations
from zhixing_server.modules.sector_radar.domain.source import build_source_snapshot
from zhixing_server.modules.sector_radar.infrastructure.postgres import PostgresSectorRadarRepository
from zhixing_server.modules.sector_radar.presentation.http import _detail_response, _history_response, _rankings_response
parser = argparse.ArgumentParser()
parser.add_argument('--seed', action='store_true')
parser.add_argument('--run', type=Path)
args = parser.parse_args()
url = os.environ['DATABASE_URL']
assert urlsplit(url).hostname in {'localhost', '127.0.0.1'}, 'disposable local database only'
samples = runpy.run_path('tests/unit/sector_radar/test_detail_reads.py')
target, now, code = samples['TARGET'], samples['NOW'], samples['CODE']
repo = PostgresSectorRadarRepository(url)
if args.seed:
config = Config('alembic.ini')
config.set_main_option('sqlalchemy.url', sqlalchemy_database_url(url).replace('%', '%%'))
config.config_file_name = None
with patch('zhixing_server.bootstrap.config.get_settings', return_value=Settings(database_url=url)):
command.upgrade(config, 'head')
samples['seed_detail'](repo)
current = repo.get_successful_publication(target)
template = repo.load_rankings(current.publication_id)[0].observation
stocks = [f'{i:06d}.SZ' for i in range(10, 3010)]
sectors = [f'PERF{i:04d}.DC' for i in range(500)]
def save(group, rows, order, day=target):
snapshot = build_source_snapshot(api_name=group.value, params={'batch': str(order)}, rows=rows,
target_trade_date=day, observed_at=now)
repo.save_source_snapshots((snapshot,))
repo.save_publication_sources((PublicationSourceRecord(current.publication_id, group, order, snapshot),))
save(Group.STOCK_BASICS, tuple({'ts_code': s, 'symbol': s[:6], 'name': s, 'exchange': 'SZSE',
'list_status': 'L', 'list_date': '20200101'} for s in stocks), 1)
save(Group.CONCEPT_INDICES, tuple({'ts_code': s, 'name': s, 'trade_date': str(target),
'pct_change': '1.1234'} for s in sectors), 1)
save(Group.MEMBERS, tuple({'ts_code': sector, 'con_code': stocks[(i*7+j)%len(stocks)],
'name': stocks[(i*7+j)%len(stocks)], 'trade_date': str(target)}
for i, sector in enumerate(sectors) for j in range(100)), 1)
for offset in range(10):
day = target-timedelta(days=offset)
for group in (Group.DAILY, Group.MONEYFLOW_DC, Group.MONEYFLOW):
rows = tuple({'ts_code': stock, 'trade_date': str(day), 'name': stock,
'pct_chg': str(Decimal(i % 123)/100), 'amount': str(i*12),
'net_amount': str(i-1500), 'net_mf_amount': str(i-500),
'close': str(Decimal(i%1000)/10+1), 'vol': str(i*33)}
for i, stock in enumerate(stocks))
save(group, rows, offset+1, day)
versions = ((MetricKind.AMOUNT, AmountNetStrategy.metric_version, MetricUnit.CNY_100M),
(MetricKind.RATIO, RatioTurnoverStrategy.metric_version, MetricUnit.RATIO),
(MetricKind.SWING, SwingEqualThreeToTenStrategy.metric_version, MetricUnit.RATIO))
for offset in range(30):
day = target-timedelta(days=offset)
if offset < 2:
publication = repo.get_successful_publication(day)
else:
running = replace(current, publication_id=f'bench-{offset}', target_trade_date=day,
status=PublicationStatus.RUNNING, input_hash=None, finished_at=None)
repo.create_publication(running)
publication = replace(running, status=PublicationStatus.SUCCESS, input_hash='a'*64,
finished_at=now+timedelta(seconds=1))
repo.finish_publication(publication)
rankings = []
for kind, version, unit in versions:
observations = tuple(replace(template, trade_date=day, sector_code=sector, sector_name=sector,
metric_kind=kind, metric_version=version, unit=unit,
value=Decimal(i+1)) for i, sector in enumerate(sectors))
rankings.extend(RankingRecord(publication.publication_id, row)
for row in rank_metric_observations(observations))
repo.save_rankings(rankings)
with psycopg.connect(url) as connection:
connection.execute('ANALYZE')
print('dataset', connection.execute('SELECT count(*), sum(row_count), sum(octet_length(payload::text)) FROM sector_radar_source_snapshot').fetchone(),
'rankings', connection.execute('SELECT count(*) FROM sector_radar_ranking').fetchone())
if args.run:
repo.open()
reads = {
'detail': lambda: _detail_response(ReadRadarDetails(repo).detail(target, SectorType.CONCEPT, code)),
'history': lambda: _history_response(ReadRadarDetails(repo).history(target, SectorType.CONCEPT, code)),
'rankings': lambda: _rankings_response(ReadSectorRadar(repo).query(RadarQuery(trade_date=target, page_size=20))),
}
report = {}
for label, read in reads.items():
timings = []
responses = []
for _ in range(5):
start = time.perf_counter()
response = read().model_dump(mode='json')
timings.append(time.perf_counter()-start)
responses.append(response)
assert all(item == responses[0] for item in responses)
fingerprint = hashlib.sha256(json.dumps(responses[0], sort_keys=True).encode()).hexdigest()
report[label] = {'seconds': timings, 'median_seconds': statistics.median(timings),
'response_sha256': fingerprint, 'response': responses[0]}
print(label, 'median', report[label]['median_seconds'], 'sha256', fingerprint, flush=True)
args.run.write_text(json.dumps(report, ensure_ascii=False, indent=2))
repo.close()
@@ -0,0 +1,100 @@
{
"environment": "isolated local PostgreSQL 16-alpine, Python 3.12.11, psycopg 3.3.4; same data, sequential requests, warm database/connection pool, no application cache",
"baseline_commit": "e567e5f",
"dataset": {
"snapshot_count": 41,
"source_rows": 143521,
"source_json_bytes": 22785765,
"ranking_rows": 45003
},
"samples_per_endpoint": 5,
"measurements": {
"detail": {
"before_seconds": [
1.7919143750004878,
1.7288594170004217,
1.7106771250000747,
1.7057415830004174,
1.6912810410012753
],
"after_seconds": [
0.6212627920012892,
0.4883682920008141,
0.44150824999996985,
0.43969708400072705,
0.4841277909999917
],
"before_median_seconds": 1.7106771250000747,
"after_median_seconds": 0.4841277909999917,
"reduction_percent": 71.7,
"responses_identical": true,
"response_sha256": "d2d5693152e949767a2db8e807852fe6fd2cb6ec175664a5dab9f615cdda4467"
},
"history": {
"before_seconds": [
0.8467425830003776,
0.8580366249989311,
0.8756502500000352,
0.8701953330000833,
0.8388115420002578
],
"after_seconds": [
0.026230833000226994,
0.024217250000219792,
0.025365250001414097,
0.028609290999156656,
0.026671499999793014
],
"before_median_seconds": 0.8580366249989311,
"after_median_seconds": 0.026230833000226994,
"reduction_percent": 96.9,
"responses_identical": true,
"response_sha256": "edf098e1f124a6d5526d9f202b67c6e97986095a6cb18fad3d0d6ba07d10d387"
},
"rankings": {
"before_seconds": [
1.3065002089988411,
1.2318050410012802,
1.2226990420003858,
1.3149193330009439,
1.2414632910004002
],
"after_seconds": [
0.06522883400066348,
0.06445087500105728,
0.06437025000013818,
0.06701250000151049,
0.06281666699942434
],
"before_median_seconds": 1.2414632910004002,
"after_median_seconds": 0.06445087500105728,
"reduction_percent": 94.8,
"responses_identical": true,
"response_sha256": "dbe90217c0086e22a570990062e08583c44d72ca1b4cf87b39208b033ce12e62"
}
},
"query_plan_samples": [
{
"query": "source_rows",
"execution_ms": 31.641,
"planning_ms": 1.534,
"rows": 3
},
{
"query": "rank_history",
"execution_ms": 23.911,
"planning_ms": 0.664,
"rows": 90
}
],
"validation": {
"pytest_passed": 226,
"pytest_failed_baseline": 1,
"baseline_failure": "test_market_data_repository_pool.py:65 Connection.executemany AttributeError, reproduced before edits",
"pyright_baseline_errors": 14,
"pyright_new_errors": 0,
"ruff": "passed",
"review_agent": "/root/read_path_review complete: no blocking SQL/data correctness findings",
"caveat": "Default integration-first ordering also disables caplog loggers through pre-existing Alembic fileConfig. Full tests were run unit/HTTP/integration order with both DB environment variables set; one baseline integration failure remained."
}
}
@@ -0,0 +1,26 @@
{
"id": "api-performance-diagnosis",
"name": "api-performance-diagnosis",
"title": "接口性能诊断与缓存方案评估",
"description": "优化资金雷达详情、历史与榜单读取;无 Redis、无迁移,待用户发布验证。",
"status": "in_progress",
"dev_type": null,
"scope": null,
"package": null,
"priority": "P2",
"creator": "yuxuanhui",
"assignee": "yuxuanhui",
"createdAt": "2026-09-07",
"completedAt": null,
"branch": "develop",
"base_branch": "develop",
"worktree_path": null,
"commit": null,
"pr_url": null,
"subtasks": [],
"children": [],
"parent": null,
"relatedFiles": [],
"notes": "用户已授权提交本地 develop。226 项测试通过;1 个既有集成测试失败和 14 个既有类型错误均在修改前复现。",
"meta": {}
}
@@ -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
@@ -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]: ...
@@ -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]:
@@ -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]:
@@ -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"]