diff --git a/.trellis/tasks/09-07-api-performance-diagnosis/check.jsonl b/.trellis/tasks/09-07-api-performance-diagnosis/check.jsonl new file mode 100644 index 0000000..9c79292 --- /dev/null +++ b/.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": "已批准设计与性能证据"} diff --git a/.trellis/tasks/09-07-api-performance-diagnosis/design.md b/.trellis/tasks/09-07-api-performance-diagnosis/design.md new file mode 100644 index 0000000..30cb5a5 --- /dev/null +++ b/.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。详情仍需要相似板块和上市过滤使用的完整成员/基础信息；不兼容来源版本的少量池统计仍在查询后丢弃，这两点保留为后续测量候选。 diff --git a/.trellis/tasks/09-07-api-performance-diagnosis/implement.jsonl b/.trellis/tasks/09-07-api-performance-diagnosis/implement.jsonl new file mode 100644 index 0000000..9c79292 --- /dev/null +++ b/.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": "已批准设计与性能证据"} diff --git a/.trellis/tasks/09-07-api-performance-diagnosis/implement.md b/.trellis/tasks/09-07-api-performance-diagnosis/implement.md new file mode 100644 index 0000000..4c8978c --- /dev/null +++ b/.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；任务待用户发布验证，暂不归档。 diff --git a/.trellis/tasks/09-07-api-performance-diagnosis/prd.md b/.trellis/tasks/09-07-api-performance-diagnosis/prd.md new file mode 100644 index 0000000..f39985b --- /dev/null +++ b/.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。 diff --git a/.trellis/tasks/09-07-api-performance-diagnosis/research/benchmark.py b/.trellis/tasks/09-07-api-performance-diagnosis/research/benchmark.py new file mode 100644 index 0000000..32f1cb3 --- /dev/null +++ b/.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() diff --git a/.trellis/tasks/09-07-api-performance-diagnosis/research/performance.json b/.trellis/tasks/09-07-api-performance-diagnosis/research/performance.json new file mode 100644 index 0000000..04c48d1 --- /dev/null +++ b/.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." + } +} diff --git a/.trellis/tasks/09-07-api-performance-diagnosis/task.json b/.trellis/tasks/09-07-api-performance-diagnosis/task.json new file mode 100644 index 0000000..5a50d91 --- /dev/null +++ b/.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": {} +} diff --git a/zhixing-server/src/zhixing_server/modules/sector_radar/application/details.py b/zhixing-server/src/zhixing_server/modules/sector_radar/application/details.py index e55f05f..f8c9e7f 100644 --- a/zhixing-server/src/zhixing_server/modules/sector_radar/application/details.py +++ b/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: - 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 diff --git a/zhixing-server/src/zhixing_server/modules/sector_radar/domain/persistence.py b/zhixing-server/src/zhixing_server/modules/sector_radar/domain/persistence.py index c26fac0..cde8b2c 100644 --- a/zhixing-server/src/zhixing_server/modules/sector_radar/domain/persistence.py +++ b/zhixing-server/src/zhixing_server/modules/sector_radar/domain/persistence.py @@ -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]: ... diff --git a/zhixing-server/src/zhixing_server/modules/sector_radar/infrastructure/memory.py b/zhixing-server/src/zhixing_server/modules/sector_radar/infrastructure/memory.py index fb9f75d..4d7278e 100644 --- a/zhixing-server/src/zhixing_server/modules/sector_radar/infrastructure/memory.py +++ b/zhixing-server/src/zhixing_server/modules/sector_radar/infrastructure/memory.py @@ -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]: diff --git a/zhixing-server/src/zhixing_server/modules/sector_radar/infrastructure/postgres.py b/zhixing-server/src/zhixing_server/modules/sector_radar/infrastructure/postgres.py index 1d47312..8525d5b 100644 --- a/zhixing-server/src/zhixing_server/modules/sector_radar/infrastructure/postgres.py +++ b/zhixing-server/src/zhixing_server/modules/sector_radar/infrastructure/postgres.py @@ -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]: diff --git a/zhixing-server/tests/unit/sector_radar/test_detail_reads.py b/zhixing-server/tests/unit/sector_radar/test_detail_reads.py new file mode 100644 index 0000000..b43fe34 --- /dev/null +++ b/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"]