Merge pull request 'perf(sector_radar): narrow publication reads and index rank history' (#33) from develop into main

Reviewed-on: sakibcc/zhixing-system#33

.trellis/tasks/09-07-api-performance-diagnosis/check.jsonl

@@ -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": "已批准设计与性能证据"}

.trellis/tasks/09-07-api-performance-diagnosis/design.md

@@ -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。详情仍需要相似板块和上市过滤使用的完整成员/基础信息；不兼容来源版本的少量池统计仍在查询后丢弃，这两点保留为后续测量候选。

.trellis/tasks/09-07-api-performance-diagnosis/implement.jsonl

@@ -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": "已批准设计与性能证据"}

.trellis/tasks/09-07-api-performance-diagnosis/implement.md

@@ -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；任务待用户发布验证，暂不归档。

.trellis/tasks/09-07-api-performance-diagnosis/prd.md

@@ -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。

.trellis/tasks/09-07-api-performance-diagnosis/research/benchmark.py

@@ -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()

.trellis/tasks/09-07-api-performance-diagnosis/research/performance.json

@@ -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."
+  }
+}

.trellis/tasks/09-07-api-performance-diagnosis/task.json

@@ -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": {}
+}

zhixing-server/src/zhixing_server/modules/sector_radar/application/details.py

@@ -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:
+        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

@@ -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

@@ -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

@@ -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

@@ -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"]