zhixing-system/.trellis/tasks/archive/2026-08/08-11-market-data-sync-performance-audit/design.md

# 市场数据同步性能与完整性检查设计

## 1. 目标与边界

本设计保留现有“逐股获取六年 qfq 完整快照”的业务口径，通过 8 路受控并发和 PostgreSQL
连接/写入批量化缩短日常同步。另提供不访问 Tushare、不修改市场事实的手动完整性检查。

不引入按交易日增量下载、自动完整检查、自动修复或通用任务队列。选股仍只消费已完成且
覆盖率达标的市场同步批次。

## 2. 任务拆分与依赖

父任务只负责需求、跨子任务契约和最终集成，不直接实现产品代码。

1. `08-11-market-sync-concurrency-storage`
   - 交付 8 worker、共享频控、连接池、批量 item 写入和集合覆盖率。
   - 无子任务依赖，必须最先完成。
2. `08-11-market-integrity-check-api`
   - 依赖第 1 项提供的池化 PostgreSQL 基础设施和 advisory lock 语义。
   - 交付检查领域模型、迁移、只读比较、HTTP 触发和轮询。
3. `08-11-market-integrity-check-web`
   - 依赖第 2 项冻结 HTTP 字段和状态值。
   - 交付 `/sync` 页面、导航、触发、轮询和报告展示。

## 3. 同步执行模块

### 3.1 外部接口

`SyncMarketData.execute(command) -> SyncBatchSummary` 保持不变。并发数、请求协调器和仓储
由组合根注入，CLI 调用方不需要了解线程、频控或连接池细节。

### 3.2 单股 worker

把当前会修改共享 `failures` / `totals` 的 `_process_bar` 改为返回不可变
`SyncItemOutcome`：股票代码、状态、`WriteResult`、fingerprint 和可选安全错误。

每个 worker 内仍严格执行：

1. 获取该股票六年 qfq；
2. 读取旧正式 CSV 并比较重叠指纹；
3. 写临时 CSV；
4. 在独立 PostgreSQL 事务中幂等 upsert；
5. 数据库提交后原子发布 CSV；
6. 返回 outcome。

主线程通过 `as_completed` 聚合 outcome、更新进度和每 100 项批量写入同步审计。单股异常只
生成失败 outcome；未发布的临时文件必须清理。结果顺序不影响计数或最终状态。

### 3.3 请求协调器

请求协调器是 Tushare 适配器内部的深模块，接口只暴露“执行一个真实供应商调用”。实现隐藏：

- 正常请求保持最多 8 路并发，不用一个全局固定间隔把真实调用重新串行化；
- 保留现有可配置的单股请求间隔作为温和节流；
- “访问频繁/请稍后/超过频率/too many requests/429/403”等频控分类；
- 频控时共享 `cooldown_until`，默认按 60、120、180 秒增长并设置上限；
- 非频控临时错误沿用有界重试和普通退避；
- 注入 monotonic clock / wait 实现确定性单元测试；
- 日志只包含方法名、尝试次数和等待时间，不包含 token 或供应商原始响应。

为保留 Tushare qfq 计算，继续调用 `ts.pro_bar(api=coordinated_client, adj="qfq", ...)`。
`coordinated_client.daily` 与 `coordinated_client.adj_factor` 在原始异常仍可见的位置经过请求
协调器；`pro_bar` 内部重试降为一次，避免隐藏重试绕开全局频控。

### 3.4 PostgreSQL

- 增加 `psycopg_pool.ConnectionPool` 依赖，池最大连接数与 worker 数匹配并留出控制连接。
- CLI 用上下文管理器打开/关闭池；HTTP 进程使用按数据库 URL 缓存的池并在进程退出时关闭。
- worker 的 bar upsert 从池中借一个连接并保持单股票事务，不在线程间共享 connection。
- 主线程调用 `record_items(batch_id, outcomes)` 分批 upsert `market_sync_item`。
- 仓储新增 `count_valid_stocks(target_trade_date)`，用一条集合 SQL 计算 active 股票中同时存在
  bar 和 daily_basic 的数量；删除同步编排对逐股 `has_*` 的依赖。
- advisory lock 的连接在整个批次期间保持借出，不和 worker 连接混用。

## 4. 完整性检查模块

### 4.1 语义

“完整性检查”验证 PostgreSQL 事实与已发布 CSV 快照是否一致，并验证 CSV 可以按现有领域规则
解析。它不判断 Tushare 是否已发布某日数据，也不能发现 PostgreSQL 和 CSV 同时缺少但供应商
实际存在的数据。

检查范围：

- 当前 active 股票主数据与 `stock-basic/current.csv`；
- 最近成功同步窗口内的每股 qfq bar 与 `bars/<ts_code>.csv`；
- 同一窗口内 PostgreSQL daily_basic 与 `daily-basic/<YYYY>/<YYYYMMDD>.csv`；
- 缺失、多余、无法解析、重复键、字段内容不一致和窗口越界。

### 4.2 独立持久化模型

新增表而不复用 `market_sync_batch`：

- `market_integrity_check`
  - `id`, `status`, `window_start`, `window_end`
  - `target_count`, `checked_count`, `issue_count`
  - `error_type`, `error_message`, `created_at`, `finished_at`
- `market_integrity_issue`
  - `check_id`, `item_kind`, `item_key`, `issue_type`, `message`, `created_at`
  - 复合主键包含可稳定区分同一对象多个问题的序号或 issue key。

状态固定为：`running`、`passed`、`issues_found`、`failed`。`issues_found` 表示检查完整执行但数据
存在问题，不等于检查任务执行失败。

### 4.3 只读比较

`RunMarketIntegrityCheck` 是外部应用接口：

- `prepare() -> IntegrityCheckRun` 原子创建 running 记录；已有 running 时返回冲突。
- `execute(check_id) -> None` 获取与同步相同的 advisory lock，执行比较并收敛终态。
- `get(check_id, page, page_size)` 和 `get_latest(...)` 提供轮询读模型。

PostgreSQL adapter 使用有序 server-side cursor，按股票或交易日流式产生一组记录；CSV adapter
按同样 key 顺序读取。应用模块做 merge comparison，一次只保留一个股票或一个交易日的数据，
不把约 700 万行全量装入内存。

检查持有与同步相同的 advisory lock，保证 DB 与正式 CSV 在比较期间不会被同步修改。它只向
检查表写进度和问题；市场事实表和正式 CSV 不发生写入。

### 4.4 HTTP

路由归属 market_data presentation，并由 `/api/v1/market-data` 挂载：

- `POST /integrity-checks` → `202 Accepted`，返回 check id、status、窗口；
- `GET /integrity-checks/latest` → 最近一次检查或 `no_data`；
- `GET /integrity-checks/{check_id}?page=&page_size=` → 进度、终态和分页问题。

HTTP 使用仓库已有的 FastAPI `BackgroundTasks` 模式。后台任务出现未捕获异常时必须把检查记录
收敛为 `failed`。进程重启可能中断 in-process task；下一次触发允许显式把失去 advisory lock
且超过超时阈值的 running 记录标记为 failed，避免永久阻塞。该恢复只修改检查元数据。

## 5. Web 模块

启用现有“同步任务”导航并新增 `/sync` 路由，使用独立 `features/sync` 垂直切片：

- API 类型与函数；
- 最新检查 query、单次检查 polling query、触发 mutation；
- 页面展示检查说明、报告只读提示、最近窗口、进度、状态和分页问题；
- running 时禁用重复触发并按固定间隔轮询；终态停止轮询并刷新 latest；
- `409` 显示已有检查正在运行，`503` 显示存储不可用，其他错误保留可重试入口。

页面不把服务器状态复制到 Zustand。问题明细最少显示对象类型、key、问题类型和安全说明。

## 6. 兼容、部署与回滚

- 新迁移为 `0003_market_integrity_checks`，只新增检查表/索引，不修改事实表。
- 配置默认 worker 从当前未生效的 4 调整为 8；新增频控退避配置时同步 `.env.example`、
  Compose 和市场数据运维文档。
- CLI 参数和退出码保持兼容；旧批次和旧 CSV 无需迁移。
- 回滚应用版本时新增检查表可暂留；数据库 downgrade 只删除检查表，不触碰市场事实。
- 若并发上线后供应商频控持续恶化，可通过环境变量把 worker 降为 1，无需回滚代码。

## 7. 验证策略

- 确定性并发测试证明 active worker 不超过配置值，结果汇总不依赖完成顺序。
- fake clock/condition 测试证明频控触发共享冷却，普通错误不冻结其他 worker。
- PostgreSQL 集成测试验证池化并发 upsert、批量 item、集合覆盖率和事务回滚。
- 完整性检查 fixture 覆盖 passed、缺 CSV、缺 DB、内容不一致、非法 CSV、检查冲突和批次异常。
- HTTP 测试锁定 202、409、分页与状态契约；Web 测试锁定触发、轮询停止、报告只读提示和错误态。
- 无网络 5002 股票调度基准锁定线程与仓储调用数量；生产约 30 分钟目标以部署日志验收。