diff --git a/AGENTS.md b/AGENTS.md index b65463b..1754bfc 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -166,6 +166,19 @@ data → universe → factors(特征) → alpha(合成/预测) → portfolio(+ri - secret scan：缓存 parquet + ledger 0 处 token / `.config.json`；ledger 列只有端点元数据。 - **缓存命中直接可见（review follow-up）**：`TushareFeed.cache_stats()` 暴露各端点 gap-fetch 计数，`_load_panel` 经 run-scoped logger 打 `data cache: market_daily_gap_fetches=N adj_factor_gap_fetches=M`——冷跑非零、暖跑 0/0（warm rerun 实证 run_phase2_baseline.log 含 `0/0`，secret scan 0）。 - **不变量守住**：factor/alpha/portfolio/execution/OOS 切片/report 全不动；`artifacts/data/{output_name}.parquet` 不当 SoT。范围克制：P4-1 只 market_daily+adj_factor，其余端点（index_weight/daily_basic/fina_indicator/...）P4-2/P4-3 再缓存。 -- ✅ 质量门：`pytest` **411 passed**（P0=97 / P1=78 / P2-1=22 / P2-2=22 / P2-3=14 / P2-4=8 / P3-1=10 / P3-2=18 / P3-3=16 / P3-4=15 / P3-5=22 / P3-6=27 / P3-7=25+1 throttle / P3-8=8 / P4-1=28）；`ruff` clean；`validate-config`（全部 12 配置）+ `run-phase0`（demo）均 OK。 +- ✅ **Phase 4-2 持久化 Tushare universe + tradability 缓存**（**分支 `data-cache-universe-tradability`**，index_weight + suspend_d + namechange + stk_limit + stock_basic）：把 P4-1 端点级 raw 缓存扩到 universe/可交易性端点——真实 run 不再每次重抓成分股/停牌/ST/涨跌停/上市日。**默认仍 disabled（向后兼容，旧配置行为一字不变）**；opt-in 后这五端点走 read-through。 + - **三种规划形态共用一引擎**：① dense per-symbol 日期区间（`suspend_d`/`stk_limit`，复用 P4-1 gap + recent-tail）；② index_code 维日期区间（`index_weight`，coverage key=index_code，gap 内仍 **90 天分页**，raw 快照入库）；③ snapshot 维度（`namechange` per-symbol、`stock_basic` 全局 sentinel），用 `refresh_dimension_days`（默认 30）staleness + force_refresh（`CoverageLedger.snapshot_fetched_at` 给新鲜度判定）。 + - **语义全保**：`index_weight` 仍 PIT/as-of（370 天 pre-start lookback 进缓存请求区间，latest-snapshot-on-or-before 仍在 feed/universe）；`stk_limit` 仍 **raw price**（限价检查在 front-adjust 前，不碰 qfq）；`stock_basic` 只取 list_date 供 `min_listing_days`（缺失仍 kept+披露），current-tag `industry` 绝不入缓存/中性化；`suspend_d`/`namechange`/ST 区间形状不变；**缓存只存 raw 端点事实，不存派生 flag 作 SoT**。 + - **三 feed 接线**：`IndexConstituentsFeed`/`TushareFlagsFeed`/`TushareCovariatesFeed` 各加 `cache=None` 注入；cache present → read-through + **共享 finalizer**（cached==direct，限价 frame `assert_frame_equal`、ST 区间集合相等、suspend set / listing dict 相等单测锁定）；`cache=None` → 直抓路径**逐字不变**（旧 feed 测试原样过）。per-symbol 限流/重试仍在各 feed 的 `_call` 闭包里（缓存 transport-agnostic）。 + - **coverage ledger 复用 + 扩展**：11 列不变；**empty 算覆盖、failed 不算**（fetch 抛错则不记 coverage、留待重试，测试锁定）；ledger 列只有端点元数据，无 token/secret。 + - **单一共享 cache 贯穿 4 runner**（run_phase0/phase2/oos/subset）：`_build_cache(cfg)` 建一个实例线穿 `_build_universe`/`_load_panel`/`_enrich_tradability`/`_maybe_enrich_listing`，跑完所有缓存端点后 `_log_run_cache_stats` 打**一行** 7 端点统计；P4-1 的 market 行前缀保留（旧统计测试过）。 + - **配置**：`data.cache.refresh_dimension_days`（默认 30，>=0 校验）；旧配置全 validate（含未知键拒绝）。 + - **真实 smoke（phase2 baseline，fresh temp root，cold→warm；fresh root 不动已 merge 的 v1 缓存）**： + - **cold 行**：`market_daily=68 adj_factor=68 index_weight=9 suspend_d=68 namechange=68 stk_limit=68 stock_basic=1`（全非零）；coverage = market/adj/namechange/stk_limit 各 68 ok + index_weight **1 ok**（整 gap 一行，内部 9 窗分页）+ stock_basic 1 ok + **suspend_d 68 empty**（SSE50 大盘股该窗口无停牌 → 空返回算覆盖，warm 不重抓）。 + - **warm 行：7 端点全 0**；coverage ledger **零新行**；report 指标与 cold/P4-1 cached baseline **逐数一致**（IC 0.0083 / annual −10.19% / maxDD −16.52% / vol 16.59% / sharpe −0.5703 / turnover 1.0818 / cost 1.19%）——cached==direct 二度端到端实证。 + - wall：cold **960s** / warm **366s**（暖跑省 ~594s = universe+tradability+market 抓取；`daily_basic`(market_cap)/`index_member_all`(pit_sw) 仍 live——P4-2 不缓存这俩，P4-3 再说，诚实标注）。 + - secret scan：缓存 parquet + ledger + 日志 + 报告 0 处 token 值 / `.config.json`；ledger 无 token 列。 + - **不变量守住**：factor/alpha/portfolio/execution/OOS 切片/report/`front_adjust` 全不动；`artifacts/data/*.parquet` 不当 SoT。范围克制：`daily_basic`/`fina_indicator`/`index_member_all` 仍留 P4-3。 +- ✅ 质量门：`pytest` **428 passed**（P0=97 / P1=78 / P2-1=22 / P2-2=22 / P2-3=14 / P2-4=8 / P3-1=10 / P3-2=18 / P3-3=16 / P3-4=15 / P3-5=22 / P3-6=27 / P3-7=25+1 throttle / P3-8=8 / P4-1=28 / P4-2=17）；`ruff` clean；`validate-config`（全部 12 配置）+ `run-phase0`（demo）均 OK。 - ⚠️ 剩余（已显式披露）：日线 only、demo 路径非真数据、旧三因子无信号（P3-3/P3-4 实证;但其组合在 2024-2026 holdout 上 SSE50/CSI300/CSI500 全正、CSI500 高达 +17.8%——小样本/regime 翻转的持续例证）;value/低波信号获得**独立样本符号级确认**（P3-7 SSE50/CSI300 量级衰减;P3-8 CSI500 泛化成立且更强），组合级盈利能力仍未确立（排名跨 cell 翻转）;subset 报告文件名已可配置（P3-8 起不再互覆盖）。 -- 路线图下一步：**P4-2 universe/tradability 缓存**（index_weight/suspend_d/namechange/stk_limit/stock_basic）→ **P4-3 因子支撑端点缓存**（fina_indicator/daily_basic/index_member_all）；研究侧：更长 holdout 滚动复检 / 中证1000 / 成本模型细化,或分钟级（architecture.html §11）。 +- 路线图下一步：**P4-3 因子支撑端点缓存**（fina_indicator/daily_basic/index_member_all）——暖跑剩余 live 抓取即这三端点；研究侧：更长 holdout 滚动复检 / 中证1000 / 成本模型细化,或分钟级（architecture.html §11）。 diff --git a/CLAUDE.md b/CLAUDE.md index b65463b..1754bfc 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -166,6 +166,19 @@ data → universe → factors(特征) → alpha(合成/预测) → portfolio(+ri - secret scan：缓存 parquet + ledger 0 处 token / `.config.json`；ledger 列只有端点元数据。 - **缓存命中直接可见（review follow-up）**：`TushareFeed.cache_stats()` 暴露各端点 gap-fetch 计数，`_load_panel` 经 run-scoped logger 打 `data cache: market_daily_gap_fetches=N adj_factor_gap_fetches=M`——冷跑非零、暖跑 0/0（warm rerun 实证 run_phase2_baseline.log 含 `0/0`，secret scan 0）。 - **不变量守住**：factor/alpha/portfolio/execution/OOS 切片/report 全不动；`artifacts/data/{output_name}.parquet` 不当 SoT。范围克制：P4-1 只 market_daily+adj_factor，其余端点（index_weight/daily_basic/fina_indicator/...）P4-2/P4-3 再缓存。 -- ✅ 质量门：`pytest` **411 passed**（P0=97 / P1=78 / P2-1=22 / P2-2=22 / P2-3=14 / P2-4=8 / P3-1=10 / P3-2=18 / P3-3=16 / P3-4=15 / P3-5=22 / P3-6=27 / P3-7=25+1 throttle / P3-8=8 / P4-1=28）；`ruff` clean；`validate-config`（全部 12 配置）+ `run-phase0`（demo）均 OK。 +- ✅ **Phase 4-2 持久化 Tushare universe + tradability 缓存**（**分支 `data-cache-universe-tradability`**，index_weight + suspend_d + namechange + stk_limit + stock_basic）：把 P4-1 端点级 raw 缓存扩到 universe/可交易性端点——真实 run 不再每次重抓成分股/停牌/ST/涨跌停/上市日。**默认仍 disabled（向后兼容，旧配置行为一字不变）**；opt-in 后这五端点走 read-through。 + - **三种规划形态共用一引擎**：① dense per-symbol 日期区间（`suspend_d`/`stk_limit`，复用 P4-1 gap + recent-tail）；② index_code 维日期区间（`index_weight`，coverage key=index_code，gap 内仍 **90 天分页**，raw 快照入库）；③ snapshot 维度（`namechange` per-symbol、`stock_basic` 全局 sentinel），用 `refresh_dimension_days`（默认 30）staleness + force_refresh（`CoverageLedger.snapshot_fetched_at` 给新鲜度判定）。 + - **语义全保**：`index_weight` 仍 PIT/as-of（370 天 pre-start lookback 进缓存请求区间，latest-snapshot-on-or-before 仍在 feed/universe）；`stk_limit` 仍 **raw price**（限价检查在 front-adjust 前，不碰 qfq）；`stock_basic` 只取 list_date 供 `min_listing_days`（缺失仍 kept+披露），current-tag `industry` 绝不入缓存/中性化；`suspend_d`/`namechange`/ST 区间形状不变；**缓存只存 raw 端点事实，不存派生 flag 作 SoT**。 + - **三 feed 接线**：`IndexConstituentsFeed`/`TushareFlagsFeed`/`TushareCovariatesFeed` 各加 `cache=None` 注入；cache present → read-through + **共享 finalizer**（cached==direct，限价 frame `assert_frame_equal`、ST 区间集合相等、suspend set / listing dict 相等单测锁定）；`cache=None` → 直抓路径**逐字不变**（旧 feed 测试原样过）。per-symbol 限流/重试仍在各 feed 的 `_call` 闭包里（缓存 transport-agnostic）。 + - **coverage ledger 复用 + 扩展**：11 列不变；**empty 算覆盖、failed 不算**（fetch 抛错则不记 coverage、留待重试，测试锁定）；ledger 列只有端点元数据，无 token/secret。 + - **单一共享 cache 贯穿 4 runner**（run_phase0/phase2/oos/subset）：`_build_cache(cfg)` 建一个实例线穿 `_build_universe`/`_load_panel`/`_enrich_tradability`/`_maybe_enrich_listing`，跑完所有缓存端点后 `_log_run_cache_stats` 打**一行** 7 端点统计；P4-1 的 market 行前缀保留（旧统计测试过）。 + - **配置**：`data.cache.refresh_dimension_days`（默认 30，>=0 校验）；旧配置全 validate（含未知键拒绝）。 + - **真实 smoke（phase2 baseline，fresh temp root，cold→warm；fresh root 不动已 merge 的 v1 缓存）**： + - **cold 行**：`market_daily=68 adj_factor=68 index_weight=9 suspend_d=68 namechange=68 stk_limit=68 stock_basic=1`（全非零）；coverage = market/adj/namechange/stk_limit 各 68 ok + index_weight **1 ok**（整 gap 一行，内部 9 窗分页）+ stock_basic 1 ok + **suspend_d 68 empty**（SSE50 大盘股该窗口无停牌 → 空返回算覆盖，warm 不重抓）。 + - **warm 行：7 端点全 0**；coverage ledger **零新行**；report 指标与 cold/P4-1 cached baseline **逐数一致**（IC 0.0083 / annual −10.19% / maxDD −16.52% / vol 16.59% / sharpe −0.5703 / turnover 1.0818 / cost 1.19%）——cached==direct 二度端到端实证。 + - wall：cold **960s** / warm **366s**（暖跑省 ~594s = universe+tradability+market 抓取；`daily_basic`(market_cap)/`index_member_all`(pit_sw) 仍 live——P4-2 不缓存这俩，P4-3 再说，诚实标注）。 + - secret scan：缓存 parquet + ledger + 日志 + 报告 0 处 token 值 / `.config.json`；ledger 无 token 列。 + - **不变量守住**：factor/alpha/portfolio/execution/OOS 切片/report/`front_adjust` 全不动；`artifacts/data/*.parquet` 不当 SoT。范围克制：`daily_basic`/`fina_indicator`/`index_member_all` 仍留 P4-3。 +- ✅ 质量门：`pytest` **428 passed**（P0=97 / P1=78 / P2-1=22 / P2-2=22 / P2-3=14 / P2-4=8 / P3-1=10 / P3-2=18 / P3-3=16 / P3-4=15 / P3-5=22 / P3-6=27 / P3-7=25+1 throttle / P3-8=8 / P4-1=28 / P4-2=17）；`ruff` clean；`validate-config`（全部 12 配置）+ `run-phase0`（demo）均 OK。 - ⚠️ 剩余（已显式披露）：日线 only、demo 路径非真数据、旧三因子无信号（P3-3/P3-4 实证;但其组合在 2024-2026 holdout 上 SSE50/CSI300/CSI500 全正、CSI500 高达 +17.8%——小样本/regime 翻转的持续例证）;value/低波信号获得**独立样本符号级确认**（P3-7 SSE50/CSI300 量级衰减;P3-8 CSI500 泛化成立且更强），组合级盈利能力仍未确立（排名跨 cell 翻转）;subset 报告文件名已可配置（P3-8 起不再互覆盖）。 -- 路线图下一步：**P4-2 universe/tradability 缓存**（index_weight/suspend_d/namechange/stk_limit/stock_basic）→ **P4-3 因子支撑端点缓存**（fina_indicator/daily_basic/index_member_all）；研究侧：更长 holdout 滚动复检 / 中证1000 / 成本模型细化,或分钟级（architecture.html §11）。 +- 路线图下一步：**P4-3 因子支撑端点缓存**（fina_indicator/daily_basic/index_member_all）——暖跑剩余 live 抓取即这三端点；研究侧：更长 holdout 滚动复检 / 中证1000 / 成本模型细化,或分钟级（architecture.html §11）。 diff --git a/TEST_REPORT.md b/TEST_REPORT.md index d8d78d0..88fd3c2 100644 --- a/TEST_REPORT.md +++ b/TEST_REPORT.md @@ -26,12 +26,12 @@ Run from the repo root with the project python (env `quant_mf`): | Gate | Command | Result | |---|---|---| -| Unit + integration | `pytest -q` | **411 passed, 0 failed** | +| Unit + integration | `pytest -q` | **428 passed, 0 failed** | | Lint | `ruff check .` | **All checks passed** | | Config validation | `validate-config` (demo + `example_tushare.yaml` + `phase2_real_baseline.yaml` + `phase3_real_multifactor.yaml` + `phase3_real_ic_weighted.yaml` + `phase3_real_oos_stability.yaml` + `phase3_real_robustness_matrix.yaml` + `phase3_real_factor_candidates.yaml` + `phase3_real_subset_costs.yaml` + `phase3_real_independent_validation.yaml` + `phase3_real_csi500_generalization.yaml` + `phase2_real_baseline_cached.yaml`) | exit `0`, prints `OK` | | End-to-end run | `run-phase0` (demo) | exit `0`, writes `artifacts/reports/phase0_summary.md` | -Counts below are the actual per-file `pytest` numbers (sum = 411). +Counts below are the actual per-file `pytest` numbers (sum = 428). ## Per-file breakdown — Phase 0 core (97) @@ -158,13 +158,19 @@ Counts below are the actual per-file `pytest` numbers (sum = 411). | `test_csi500_generalization.py` | 8 | CSI500 config validates with the expected cell roles (screened anchor SSE50|2022-2024; independent SSE50|2024-2026 + 000905.SH|2024-2026; CSI500|2022-2024 skipped+disclosed; same groups/scenarios/hypotheses as P3-7 — no tuning); sample classes labeled correctly; **`output.subset_report_name`** lets each subset-validation study own its report file (default None keeps the historical `phase3_subset_validation.md` bitwise — the P3-6/P3-7 configs are locked unchanged), so a P3-8 run never clobbers the accepted P3-7 artifact; **`output.subset_report_title`** config-drives the report H1 so the CSI500 study names itself ("Phase 3-8 — CSI500 Independent Generalization Check", asserted NOT to start with Phase 3-7) while the P3-6/P3-7 configs leave it unset and keep the renderer's sample-aware default title (regression-locked) | | **Total through P3-8** | **383** | | -## Per-file breakdown — Phase 4-1 persistent market cache (24) +## Per-file breakdown — Phase 4-1 persistent market cache (28) | Test file | Tests | Red-line / feature | |---|---|---| | `test_cache_config.py` | 17 | `data.cache` defaults disabled (backward compatible); fields + defaults; rejects negative refresh window / empty root / unknown key; **every existing config still validates** (parametrized over all `config/*.yaml`) | | `test_tushare_cache_market.py` | 11 | interval subtraction (full/partial/none); **full miss populates daily+adj cache, full hit = ZERO endpoint calls**; partial gap fetches only the missing tail; empty endpoint return still records coverage (no refetch); duplicate upsert keeps one row per (symbol,date); **cached panel == direct-fetch panel byte-for-byte after `front_adjust()`**; no token / secret-path in any cache file or ledger column; **`TushareFeed.cache_stats()` exposes cold {2,2}→warm {0,0} gap-fetch counts** and `pipeline._log_cache_stats` logs `data cache: market_daily_gap_fetches=N adj_factor_gap_fetches=M` through the run-scoped logger (review follow-up: warm-hit evidence is now directly visible in the run log) | -| **Total (P0 + P1 + P2-1..P2-4 + P3-1..P3-8 + P4-1)** | **411** | | + +## Per-file breakdown — Phase 4-2 universe + tradability cache (17) + +| Test file | Tests | Red-line / feature | +|---|---|---| +| `test_tushare_cache_universe.py` | 17 | for each new endpoint (`index_weight`/`suspend_d`/`namechange`/`stk_limit`/`stock_basic`): **cold miss populates, warm rerun on a fresh cache over the same root = ZERO endpoint calls**; **cached feed output == direct feed output** (index_weight `assert_frame_equal` + PIT cross-section, suspend set, ST-interval sets incl. open `None` end, stk_limit frame in RAW price terms, listing-date dict); empty endpoint/snapshot return records coverage (no refetch); **FAILED fetch records NO coverage** (a later good fetch retries) — dense + snapshot; duplicate upsert keeps one row per natural key (index_weight `(date,symbol)`, suspend_d `(date,symbol,suspend_type)`); `force_refresh` re-pulls a fresh dimension snapshot; **cache-disabled path never touches the cache tree**; `pipeline._format_cache_stats` / `_log_run_cache_stats` name all five new endpoints in one line (None → no line); no token / secret-path in any cache file or ledger column across all five endpoints | +| **Total (P0 + P1 + P2-1..P2-4 + P3-1..P3-8 + P4-1 + P4-2)** | **428** | | ## Real-data validation (manual, not in CI — TEST-002 keeps the suite network-free) @@ -195,6 +201,21 @@ Counts below are the actual per-file `pytest` numbers (sum = 411). ledger unchanged between cold and warm); cache files + ledger carry no token / secret. Market bars only — index/financial endpoints still fetch live (P4-2/3). Two earlier run attempts died on transient ConnectionError → default retry budget hardened 3→6 attempts. +- **P4-2** universe + tradability cache (real smoke = phase2 baseline on a fresh + temp cache root, cold → warm; the merged P4-1 `v1` cache left untouched): + **cold** run-log line `data cache: market_daily=68 adj_factor=68 + index_weight=9 suspend_d=68 namechange=68 stk_limit=68 stock_basic=1` (all + non-zero); coverage = 68 ok each for market/adj/namechange/stk_limit, 1 ok for + index_weight (one gap, paged in 9 windows) and stock_basic, and **68 empty for + suspend_d** (SSE50 large-caps had no suspensions in the window — empty-as- + coverage). **warm** line shows **all seven endpoints = 0**, coverage ledger + unchanged (zero new rows), and report metrics byte-identical to cold and to the + P4-1 cached baseline (IC 0.0083 / annual −10.19% / maxDD −16.52% / vol 16.59% / + sharpe −0.5703 / turnover 1.0818 / cost 1.19%). Wall: cold 960s → warm 366s + (the ~594 s saved is the universe + tradability + market fetches now served from + cache; `daily_basic` / `index_member_all` still fetch live — P4-3). Secret scan: + 0 token-value / `.config.json` occurrences across cache parquet, ledger, run + log, and report. ## Notes diff --git a/data/cache/coverage.py b/data/cache/coverage.py index 86f86d6..29d6933 100644 --- a/data/cache/coverage.py +++ b/data/cache/coverage.py @@ -92,6 +92,27 @@ def covered_intervals(self, endpoint: str, key: str) -> list[Interval]: for s, e in zip(sub["start_date"], sub["end_date"]) ] + def snapshot_fetched_at(self, endpoint: str, key: str) -> pd.Timestamp | None: + """Latest successful fetch time for a snapshot/dimension ``(endpoint, key)``. + + For the dimension endpoints (``stock_basic``, ``namechange``) coverage is + not a date range but a freshness timestamp: the most recent ``ok``/ + ``empty`` ``fetched_at``. ``None`` means never fetched (must fetch). The + caller compares against ``refresh_dimension_days`` to decide staleness. + """ + ledger = self.read() + if ledger.empty: + return None + mask = ( + (ledger["endpoint"] == endpoint) + & (ledger["key"] == str(key)) + & (ledger["status"].isin(_COVERING_STATUSES)) + ) + sub = ledger[mask] + if sub.empty: + return None + return pd.Timestamp(sub["fetched_at"].max()) + # -- write -------------------------------------------------------------- # def record( self, diff --git a/data/cache/tushare_cache.py b/data/cache/tushare_cache.py index 818643a..10cc531 100644 --- a/data/cache/tushare_cache.py +++ b/data/cache/tushare_cache.py @@ -1,21 +1,34 @@ -"""Read-through cache for tushare market endpoints (P4-1: daily + adj_factor). +"""Read-through cache for tushare endpoints (P4-1 market bars + P4-2 universe/tradability). -``TushareCache`` turns a per-symbol requested date range into ONLY the uncovered -gaps, fetches those via a caller-supplied ``fetch`` callable (the feed wraps its -own retry/throttle there, so the cache stays transport-agnostic), upserts the -raw rows, records coverage (including empty returns), then returns the full -requested range read back from the cache. +``TushareCache`` turns a requested range into ONLY the uncovered gaps (or a +stale snapshot), fetches those via a caller-supplied ``fetch`` callable (the +feed wraps its own retry/throttle there, so the cache stays transport-agnostic), +upserts the raw rows, records coverage (including empty returns), then returns +the full requested data read back from the cache. -Behaviour the P4-1 acceptance pins down: - * full cache miss -> fetch every symbol's full range, populate cache; +Three planning shapes share one engine: + * dense per-symbol date-range (``market_daily``, ``adj_factor``, ``suspend_d``, + ``stk_limit``): coverage key = symbol, gaps subtracted per symbol, a recent + tail refetched within ``refresh_recent_days``; + * index-keyed date-range (``index_weight``): coverage key = index_code, gaps + subtracted over the whole index, each uncovered gap paged in <=90-day + windows (tushare's per-call row cap), raw snapshots stored; + * snapshot / dimension (``namechange`` per-symbol, ``stock_basic`` global): + no date range — refetched only when never fetched, stale beyond + ``refresh_dimension_days``, or force-refreshed. + +Behaviour the acceptance pins down (each endpoint): + * full cache miss -> fetch the uncovered range, populate cache; * full cache hit -> ZERO fetch calls; * partial gap -> fetch only the missing sub-range; * empty return -> still recorded as coverage (no needless refetch); - * duplicate upsert -> one row per ``(symbol, date)``. + * failed fetch -> NOT recorded as coverage (a later run retries); + * duplicate upsert -> one row per the endpoint's natural key. Stored rows are RAW and canonical-shaped (``date`` as datetime, ``symbol`` as -str, native price/volume/amount or adj_factor). Joining + ``front_adjust`` stay -downstream, unchanged. No token or secret ever reaches this layer. +str, native prices / weights / limits). The PIT as-of membership, raw price +limit checks, and ``front_adjust`` all stay downstream, unchanged. No token or +secret ever reaches this layer; the ledger holds endpoint metadata only. """ from __future__ import annotations @@ -35,18 +48,54 @@ # endpoint identifiers (also the names accepted in data.cache.force_refresh). MARKET_DAILY = "market_daily" ADJ_FACTOR = "adj_factor" +INDEX_WEIGHT = "index_weight" +SUSPEND_D = "suspend_d" +NAMECHANGE = "namechange" +STK_LIMIT = "stk_limit" +STOCK_BASIC = "stock_basic" + +# every endpoint this cache knows (fetch_counts is seeded with all of them so a +# warm run reports an explicit 0 for an endpoint it never had to touch). +ALL_ENDPOINTS = ( + MARKET_DAILY, ADJ_FACTOR, INDEX_WEIGHT, SUSPEND_D, NAMECHANGE, + STK_LIMIT, STOCK_BASIC, +) + +# sentinel key for a global (whole-market) snapshot endpoint (stock_basic). +_GLOBAL_KEY = "__all__" -# canonical-raw column sets stored per endpoint. +# canonical-raw column sets + natural keys stored per endpoint. _DAILY_COLUMNS = ["date", "symbol", "open", "high", "low", "close", "volume", "amount"] _ADJ_COLUMNS = ["date", "symbol", "adj_factor"] _KEY_COLS = ["date", "symbol"] +_INDEX_WEIGHT_COLUMNS = ["index_code", "date", "symbol", "weight"] +_INDEX_WEIGHT_KEY = ["date", "symbol"] + +_SUSPEND_COLUMNS = ["date", "symbol", "suspend_type"] +_SUSPEND_KEY = ["date", "symbol", "suspend_type"] + +_STK_LIMIT_COLUMNS = ["date", "symbol", "up_limit", "down_limit"] +_STK_LIMIT_KEY = ["date", "symbol"] + +_NAMECHANGE_COLUMNS = ["symbol", "start_date", "end_date", "name"] +_NAMECHANGE_KEY = ["symbol", "start_date", "end_date", "name"] + +_STOCK_BASIC_COLUMNS = ["symbol", "list_date"] +_STOCK_BASIC_KEY = ["symbol"] + +# tushare per-call row cap forces index_weight to be paged in <=90-day windows. +_INDEX_WINDOW_DAYS = 90 + # tushare raw -> canonical name for the columns we keep. _DAILY_RENAME = {"ts_code": "symbol", "trade_date": "date", "vol": "volume"} _ADJ_RENAME = {"ts_code": "symbol", "trade_date": "date"} # A fetch callable: (symbol, start_compact, end_compact) -> raw tushare frame|None. FetchOne = Callable[[str, str, str], "pd.DataFrame | None"] +# A snapshot fetch: (symbol) -> raw frame|None (namechange) or () -> raw frame|None +# (stock_basic). Typed loosely; the cache only calls and parses the result. +FetchSnapshot = Callable[..., "pd.DataFrame | None"] def _fields_hash(columns: list[str]) -> str: @@ -81,8 +130,103 @@ def _parse_adj(raw: pd.DataFrame | None) -> pd.DataFrame: return df[_ADJ_COLUMNS] +def _parse_suspend(raw: pd.DataFrame | None) -> pd.DataFrame: + """tushare ``suspend_d`` frame -> canonical-raw rows (or empty). + + The feed queries with ``suspend_type='S'``; stored rows carry that type so + the suspended-set the feed builds from ``(date, symbol)`` is unchanged. + """ + if raw is None or len(raw) == 0: + return pd.DataFrame(columns=_SUSPEND_COLUMNS) + df = raw.rename(columns={"ts_code": "symbol", "trade_date": "date"}).copy() + df["date"] = pd.to_datetime(df["date"].astype(str), format="%Y%m%d") + df["symbol"] = df["symbol"].astype(str) + if "suspend_type" not in df.columns: + df["suspend_type"] = "S" + df["suspend_type"] = df["suspend_type"].astype(str) + return df[_SUSPEND_COLUMNS] + + +def _parse_stk_limit(raw: pd.DataFrame | None) -> pd.DataFrame: + """tushare ``stk_limit`` frame -> canonical-raw rows (or empty). + + up_limit / down_limit stay RAW price terms (the limit checks run before + front-adjustment, as today); nothing here touches qfq. + """ + if raw is None or len(raw) == 0: + return pd.DataFrame(columns=_STK_LIMIT_COLUMNS) + df = raw.rename(columns={"ts_code": "symbol", "trade_date": "date"}).copy() + df["date"] = pd.to_datetime(df["date"].astype(str), format="%Y%m%d") + df["symbol"] = df["symbol"].astype(str) + for col in ("up_limit", "down_limit"): + if col not in df.columns: + df[col] = float("nan") + return df[_STK_LIMIT_COLUMNS] + + +def _parse_index_weight(raw: pd.DataFrame | None, index_code: str) -> pd.DataFrame: + """tushare ``index_weight`` frame -> canonical-raw rows (or empty). + + ``con_code`` -> symbol, ``trade_date`` -> date; ``index_code`` is fixed from + the queried index. Raw snapshots only — the latest-snapshot as-of membership + logic stays downstream (unchanged). + """ + if raw is None or len(raw) == 0: + return pd.DataFrame(columns=_INDEX_WEIGHT_COLUMNS) + df = raw.rename(columns={"con_code": "symbol", "trade_date": "date"}).copy() + df["date"] = pd.to_datetime(df["date"].astype(str), format="%Y%m%d") + df["symbol"] = df["symbol"].astype(str) + df["index_code"] = str(index_code) + if "weight" not in df.columns: + df["weight"] = float("nan") + return df[_INDEX_WEIGHT_COLUMNS] + + +def _parse_namechange(raw: pd.DataFrame | None, symbol: str) -> pd.DataFrame: + """tushare ``namechange`` frame -> canonical-raw rows (or empty). + + ``end_date`` is NaT for an active (open) name; the feed maps NaT back to + ``None`` when it builds the ST intervals, so the interval shape is unchanged. + """ + if raw is None or len(raw) == 0: + return pd.DataFrame(columns=_NAMECHANGE_COLUMNS) + df = raw.copy() + df["symbol"] = ( + df["ts_code"].astype(str) if "ts_code" in df.columns else str(symbol) + ) + df["start_date"] = pd.to_datetime( + df["start_date"].astype(str), format="%Y%m%d", errors="coerce" + ) + if "end_date" in df.columns: + df["end_date"] = pd.to_datetime( + df["end_date"].astype(str), format="%Y%m%d", errors="coerce" + ) + else: + df["end_date"] = pd.NaT + df["name"] = df["name"].astype(str) + return df[_NAMECHANGE_COLUMNS] + + +def _parse_stock_basic(raw: pd.DataFrame | None) -> pd.DataFrame: + """tushare ``stock_basic`` frame -> canonical-raw rows (or empty). + + Stores ``list_date`` as the raw compact string; the feed parses it to a + Timestamp exactly as the direct path does (for the ``min_listing_days`` + selection filter). The current-tag ``industry`` is NOT stored — it must + never re-enter neutralization (the PIT SW path replaced it). + """ + if raw is None or len(raw) == 0: + return pd.DataFrame(columns=_STOCK_BASIC_COLUMNS) + df = raw.rename(columns={"ts_code": "symbol"}).copy() + df["symbol"] = df["symbol"].astype(str) + if "list_date" not in df.columns: + df["list_date"] = None + df["list_date"] = df["list_date"].astype(str) + return df[_STOCK_BASIC_COLUMNS] + + class TushareCache: - """Endpoint-level read-through cache for ``market_daily`` and ``adj_factor``.""" + """Endpoint-level read-through cache (market bars + universe/tradability).""" def __init__( self, @@ -90,6 +234,7 @@ def __init__( ledger: CoverageLedger, *, refresh_recent_days: int = 14, + refresh_dimension_days: int = 30, force_refresh: tuple[str, ...] | list[str] = (), today: pd.Timestamp | None = None, clock: Callable[[], pd.Timestamp] | None = None, @@ -98,35 +243,140 @@ def __init__( self._store = store self._ledger = ledger self._refresh_recent_days = int(refresh_recent_days) + self._refresh_dimension_days = int(refresh_dimension_days) self._force_refresh = set(force_refresh or ()) self._today = pd.Timestamp(today).normalize() if today is not None else None self._clock = clock or pd.Timestamp.now self._source_version = source_version # per-instance endpoint fetch counters (cache stats; one increment per - # gap actually sent to the API). A fully-covered repeat run leaves these - # at zero — the read-through hit rate is observable from the run log. - self.fetch_counts: dict[str, int] = {MARKET_DAILY: 0, ADJ_FACTOR: 0} + # gap/window/snapshot actually sent to the API). A fully-covered repeat + # run leaves these at zero — the read-through hit rate is observable from + # the run log. Seeded with every known endpoint so a warm run reports an + # explicit 0 even for endpoints it never had to touch. + self.fetch_counts: dict[str, int] = {ep: 0 for ep in ALL_ENDPOINTS} def stats(self) -> dict[str, int]: """Endpoint -> number of gap fetches sent to the API this instance.""" return dict(self.fetch_counts) - # -- public endpoint readers ------------------------------------------- # + # -- dense per-symbol date-range endpoints ----------------------------- # def daily_bars( self, symbols: list[str], start: str, end: str, fetch: FetchOne ) -> pd.DataFrame: return self._read_through( - MARKET_DAILY, symbols, start, end, fetch, _parse_daily, _DAILY_COLUMNS + MARKET_DAILY, symbols, start, end, fetch, _parse_daily, + _DAILY_COLUMNS, _KEY_COLS, ) def adj_factor( self, symbols: list[str], start: str, end: str, fetch: FetchOne ) -> pd.DataFrame: return self._read_through( - ADJ_FACTOR, symbols, start, end, fetch, _parse_adj, _ADJ_COLUMNS + ADJ_FACTOR, symbols, start, end, fetch, _parse_adj, + _ADJ_COLUMNS, _KEY_COLS, + ) + + def suspend_d( + self, symbols: list[str], start: str, end: str, fetch: FetchOne + ) -> pd.DataFrame: + """Canonical-raw suspension rows [date, symbol, suspend_type] over [start, end].""" + return self._read_through( + SUSPEND_D, symbols, start, end, fetch, _parse_suspend, + _SUSPEND_COLUMNS, _SUSPEND_KEY, ) - # -- read-through engine ----------------------------------------------- # + def stk_limit( + self, symbols: list[str], start: str, end: str, fetch: FetchOne + ) -> pd.DataFrame: + """Canonical-raw price-limit rows [date, symbol, up_limit, down_limit].""" + return self._read_through( + STK_LIMIT, symbols, start, end, fetch, _parse_stk_limit, + _STK_LIMIT_COLUMNS, _STK_LIMIT_KEY, + ) + + # -- index-keyed date-range endpoint (index_weight, 90-day paged) ------- # + def index_weight( + self, index_code: str, start: str, end: str, fetch: FetchOne + ) -> pd.DataFrame: + """Canonical-raw index_weight snapshots [date, symbol, weight] over [start, end]. + + Coverage is keyed by ``index_code`` and planned by date range; each + uncovered gap is paged in <=90-day windows (tushare's per-call row cap). + The latest-snapshot as-of membership stays the feed's job — this returns + every stored snapshot in the window. + """ + req_start = pd.Timestamp(start).normalize() + req_end = pd.Timestamp(end).normalize() + forced = INDEX_WEIGHT in self._force_refresh + fields_hash = _fields_hash(_INDEX_WEIGHT_COLUMNS) + gaps = self._gaps_for(INDEX_WEIGHT, index_code, req_start, req_end, forced) + for gap_start, gap_end in gaps: + self._fetch_index_gap(index_code, gap_start, gap_end, fetch, fields_hash) + _LOGGER.info( + "cache %s: index=%s, %d gap-intervals (api calls=%d)", + INDEX_WEIGHT, index_code, len(gaps), self.fetch_counts[INDEX_WEIGHT], + ) + cached = self._store.read_symbol(INDEX_WEIGHT, index_code) + if cached.empty: + return pd.DataFrame(columns=["date", "symbol", "weight"]) + mask = (cached["date"] >= req_start) & (cached["date"] <= req_end) + return cached.loc[mask, ["date", "symbol", "weight"]].reset_index(drop=True) + + # -- snapshot / dimension endpoints ------------------------------------ # + def namechange( + self, symbols: list[str], fetch: FetchSnapshot + ) -> pd.DataFrame: + """Canonical-raw namechange rows for ``symbols`` (per-symbol dimension). + + Each symbol's snapshot is refetched only when never fetched, stale beyond + ``refresh_dimension_days``, or force-refreshed. Returns the stored rows + for the requested symbols (the ST-interval shaping stays the feed's job). + """ + forced = NAMECHANGE in self._force_refresh + fields_hash = _fields_hash(_NAMECHANGE_COLUMNS) + out: list[pd.DataFrame] = [] + for sym in symbols: + if forced or self._snapshot_stale(NAMECHANGE, sym): + self._fetch_snapshot( + NAMECHANGE, sym, "symbol", lambda s=sym: fetch(s), + lambda raw, s=sym: _parse_namechange(raw, s), + _NAMECHANGE_KEY, fields_hash, + ) + cached = self._store.read_symbol(NAMECHANGE, sym) + if not cached.empty: + out.append(cached[_NAMECHANGE_COLUMNS]) + _LOGGER.info( + "cache %s: %d symbols (api calls=%d)", + NAMECHANGE, len(symbols), self.fetch_counts[NAMECHANGE], + ) + if not out: + return pd.DataFrame(columns=_NAMECHANGE_COLUMNS) + return pd.concat(out, ignore_index=True) + + def stock_basic(self, fetch: FetchSnapshot) -> pd.DataFrame: + """Canonical-raw stock_basic rows [symbol, list_date] (global dimension). + + One whole-market snapshot keyed by a global sentinel; refetched only when + never fetched, stale beyond ``refresh_dimension_days``, or force-refreshed. + The feed filters to the symbols it needs (list_date only). + """ + forced = STOCK_BASIC in self._force_refresh + fields_hash = _fields_hash(_STOCK_BASIC_COLUMNS) + if forced or self._snapshot_stale(STOCK_BASIC, _GLOBAL_KEY): + self._fetch_snapshot( + STOCK_BASIC, _GLOBAL_KEY, "global", lambda: fetch(), + lambda raw: _parse_stock_basic(raw), _STOCK_BASIC_KEY, fields_hash, + ) + _LOGGER.info( + "cache %s: global snapshot (api calls=%d)", + STOCK_BASIC, self.fetch_counts[STOCK_BASIC], + ) + cached = self._store.read_symbol(STOCK_BASIC, _GLOBAL_KEY) + if cached.empty: + return pd.DataFrame(columns=_STOCK_BASIC_COLUMNS) + return cached[_STOCK_BASIC_COLUMNS].reset_index(drop=True) + + # -- read-through engine (dense per-symbol date-range) ----------------- # def _read_through( self, endpoint: str, @@ -136,6 +386,7 @@ def _read_through( fetch: FetchOne, parse: Callable[["pd.DataFrame | None"], pd.DataFrame], columns: list[str], + key_cols: list[str], ) -> pd.DataFrame: req_start = pd.Timestamp(start).normalize() req_end = pd.Timestamp(end).normalize() @@ -153,7 +404,8 @@ def _read_through( n_covered += 1 for gap_start, gap_end in gaps: self._fetch_gap( - endpoint, symbol, gap_start, gap_end, fetch, parse, fields_hash + endpoint, symbol, gap_start, gap_end, fetch, parse, + key_cols, fields_hash, ) cached = self._store.read_symbol(endpoint, symbol) if not cached.empty: @@ -167,7 +419,7 @@ def _read_through( ) if not out: return pd.DataFrame(columns=columns) - return pd.concat(out, ignore_index=True) + return pd.concat(out, ignore_index=True).reset_index(drop=True) def _gaps_for(self, endpoint, symbol, req_start, req_end, forced): """The uncovered sub-intervals to fetch (+ a forced recent tail).""" @@ -190,14 +442,16 @@ def _recent_tail(self, req_start, req_end): return None # whole request is safely historical return (max(req_start, threshold), req_end) - def _fetch_gap(self, endpoint, symbol, gap_start, gap_end, fetch, parse, fields_hash): + def _fetch_gap( + self, endpoint, symbol, gap_start, gap_end, fetch, parse, key_cols, fields_hash + ): """Fetch one gap, upsert raw rows, record coverage (incl. empty).""" raw = fetch(symbol, _compact(gap_start), _compact(gap_end)) self.fetch_counts[endpoint] = self.fetch_counts.get(endpoint, 0) + 1 parsed = parse(raw) row_count = len(parsed) if row_count: - self._store.upsert_symbol(endpoint, symbol, parsed, _KEY_COLS) + self._store.upsert_symbol(endpoint, symbol, parsed, key_cols) self._ledger.record( endpoint=endpoint, key_type="symbol", @@ -210,3 +464,77 @@ def _fetch_gap(self, endpoint, symbol, gap_start, gap_end, fetch, parse, fields_ fetched_at=self._clock(), source_version=self._source_version, ) + + # -- index_weight gap fetch (paged in <=90-day windows) ---------------- # + def _fetch_index_gap(self, index_code, gap_start, gap_end, fetch, fields_hash): + """Page one uncovered gap into <=90-day windows, upsert, record coverage. + + Coverage is recorded for the WHOLE gap (ok if any snapshot landed, else + empty) so a covered window never refetches; each window is one API call + (counted), mirroring the feed's historical paging. + """ + frames: list[pd.DataFrame] = [] + total_rows = 0 + win_start = gap_start + while win_start <= gap_end: + win_end = min( + win_start + pd.Timedelta(days=_INDEX_WINDOW_DAYS - 1), gap_end + ) + raw = fetch(index_code, _compact(win_start), _compact(win_end)) + self.fetch_counts[INDEX_WEIGHT] += 1 + parsed = _parse_index_weight(raw, index_code) + if not parsed.empty: + frames.append(parsed) + total_rows += len(parsed) + win_start = win_end + pd.Timedelta(days=1) + if frames: + combined = pd.concat(frames, ignore_index=True) + self._store.upsert_symbol( + INDEX_WEIGHT, index_code, combined, _INDEX_WEIGHT_KEY + ) + self._ledger.record( + endpoint=INDEX_WEIGHT, + key_type="index_code", + key=index_code, + start_date=gap_start, + end_date=gap_end, + fields_hash=fields_hash, + row_count=total_rows, + status="ok" if total_rows else "empty", + fetched_at=self._clock(), + source_version=self._source_version, + ) + + # -- snapshot / dimension fetch (staleness policy) --------------------- # + def _snapshot_stale(self, endpoint: str, key: str) -> bool: + """Whether a dimension snapshot must be (re)fetched (never fetched/stale).""" + last = self._ledger.snapshot_fetched_at(endpoint, key) + if last is None: + return True + if self._refresh_dimension_days <= 0: + return False # only force_refresh re-pulls once present + today = self._today if self._today is not None else self._clock().normalize() + return (today - last.normalize()).days >= self._refresh_dimension_days + + def _fetch_snapshot( + self, endpoint, key, key_type, fetch, parse, key_cols, fields_hash + ): + """Fetch one snapshot, upsert raw rows, record coverage (incl. empty).""" + raw = fetch() + self.fetch_counts[endpoint] = self.fetch_counts.get(endpoint, 0) + 1 + parsed = parse(raw) + row_count = len(parsed) + if row_count: + self._store.upsert_symbol(endpoint, key, parsed, key_cols) + self._ledger.record( + endpoint=endpoint, + key_type=key_type, + key=key, + start_date=None, + end_date=None, + fields_hash=fields_hash, + row_count=row_count, + status="ok" if row_count else "empty", + fetched_at=self._clock(), + source_version=self._source_version, + ) diff --git a/data/feed/index_feed.py b/data/feed/index_feed.py index cdbe334..820bdb6 100644 --- a/data/feed/index_feed.py +++ b/data/feed/index_feed.py @@ -35,12 +35,18 @@ def __init__( token_key: str = "tushare.token", rate_limit: int | None = None, max_retries: int = 6, + cache=None, ) -> None: self._secret_file = str(secret_file) self._token_key = token_key self._rate_limit = rate_limit self._max_retries = max(1, int(max_retries)) self._pro = None # lazily built + # P4-2: optional shared read-through cache. None keeps the historical + # direct-fetch (paged) behaviour EXACTLY; only an opted-in config injects + # one. The cache stores RAW snapshots; the as-of membership stays + # downstream, unchanged. + self._cache = cache # -- secret handling (token never logged) ------------------------------- # def _read_token(self) -> str: @@ -81,9 +87,18 @@ def get_constituents(self, index_code: str, start: str, end: str) -> pd.DataFram (date, symbol), de-duplicated across window boundaries. Empty (schema-shaped) frame if tushare returns nothing — not an error. """ + if self._cache is not None: + # Read-through: the cache plans gaps by index_code and pages each + # uncovered gap in <=90-day windows (same cap rule). It returns the + # canonical [date, symbol, weight] snapshots; the as-of/dedupe/sort + # finalizer below is shared with the direct path (cached == direct). + df = self._cache.index_weight( + index_code, start, end, self._index_weight_fetch() + ) + return self._finalize_constituents(df) + pro = self._client() start_ts, end_ts = pd.Timestamp(start), pd.Timestamp(end) - frames: list[pd.DataFrame] = [] win_start = start_ts while win_start <= end_ts: @@ -107,8 +122,40 @@ def get_constituents(self, index_code: str, start: str, end: str) -> pd.DataFram df = df.rename(columns={"con_code": "symbol", "trade_date": "date"}) df["date"] = pd.to_datetime(df["date"].astype(str), format="%Y%m%d") df["symbol"] = df["symbol"].astype(str) - df = df[list(CONSTITUENT_COLUMNS)].drop_duplicates(["date", "symbol"]) - return df.sort_values(["date", "symbol"]).reset_index(drop=True) + return self._finalize_constituents(df) + + def _index_weight_fetch(self): + """A ``(index_code, start_compact, end_compact) -> raw frame`` closure. + + The per-call retry/throttle stays HERE (the cache is transport-agnostic); + the cache calls this once per uncovered <=90-day window. The client is + built lazily inside the closure, so a fully-covered warm run reads no + token and constructs no client. + """ + + def fetch(index_code: str, start_compact: str, end_compact: str): + return request_with_retry( + self._client().index_weight, + max_retries=self._max_retries, + rate_limit=self._rate_limit, + index_code=index_code, + start_date=start_compact, + end_date=end_compact, + ) + + return fetch + + def _finalize_constituents(self, df: pd.DataFrame) -> pd.DataFrame: + """Canonicalize: select [date, symbol, weight], dedup, sort, reset index. + + Shared by the cache and direct paths so both produce a byte-identical + frame. ``df`` is already canonical ([date, symbol, weight] present); + an empty frame returns the schema-shaped empty. + """ + if df is None or df.empty: + return self._empty() + out = df[list(CONSTITUENT_COLUMNS)].drop_duplicates(["date", "symbol"]) + return out.sort_values(["date", "symbol"]).reset_index(drop=True) @staticmethod def _empty() -> pd.DataFrame: diff --git a/data/feed/tushare_covariates.py b/data/feed/tushare_covariates.py index c217742..1bc9abc 100644 --- a/data/feed/tushare_covariates.py +++ b/data/feed/tushare_covariates.py @@ -41,12 +41,18 @@ def __init__( token_key: str = "tushare.token", rate_limit: int | None = None, max_retries: int = 6, + cache=None, ) -> None: self._secret_file = str(secret_file) self._token_key = token_key self._rate_limit = rate_limit self._max_retries = max(1, int(max_retries)) self._pro = None + # P4-2: optional shared read-through cache. Only ``listing_dates`` + # (stock_basic.list_date, for min_listing_days) reads through it; the + # daily_basic / index_member_all covariates are P4-3 and stay direct. + # None keeps the historical direct fetch EXACTLY. + self._cache = cache def _client(self): if self._pro is None: @@ -77,6 +83,18 @@ def listing_dates(self, symbols: list[str]) -> dict[str, pd.Timestamp]: ``stock_basic`` simply does not appear in the map (the caller treats an unknown listing date as a disclosed data gap, never as a young name). """ + if self._cache is not None: + df = self._cache.stock_basic(self._stock_basic_fetch()) + if df.empty: + return {} + wanted = set(map(str, symbols)) + df = df[df["symbol"].astype(str).isin(wanted)] + out: dict[str, pd.Timestamp] = {} + for r in df.itertuples(): + out[str(r.symbol)] = pd.to_datetime( + str(r.list_date), format="%Y%m%d", errors="coerce" + ) + return out pro = self._client() df = self._call(pro.stock_basic, fields="ts_code,list_date") if df is None or len(df) == 0: @@ -90,6 +108,18 @@ def listing_dates(self, symbols: list[str]) -> dict[str, pd.Timestamp]: ) return out + def _stock_basic_fetch(self): + """`() -> raw stock_basic frame` (ts_code, list_date) — global snapshot. + + The client is built lazily inside the closure, so a fully-covered warm + run reads no token and constructs no client. + """ + + def fetch(): + return self._call(self._client().stock_basic, fields="ts_code,list_date") + + return fetch + _SW_LEVEL_COLUMN = {"L1": "l1_name", "L2": "l2_name", "L3": "l3_name"} def pit_sw_intervals( diff --git a/data/feed/tushare_flags.py b/data/feed/tushare_flags.py index 58ab7f2..db68602 100644 --- a/data/feed/tushare_flags.py +++ b/data/feed/tushare_flags.py @@ -31,12 +31,18 @@ def __init__( token_key: str = "tushare.token", rate_limit: int | None = None, max_retries: int = 6, + cache=None, ) -> None: self._secret_file = str(secret_file) self._token_key = token_key self._rate_limit = rate_limit self._max_retries = max(1, int(max_retries)) self._pro = None + # P4-2: optional shared read-through cache. None keeps the historical + # direct per-symbol fetches EXACTLY; only an opted-in config injects one. + # The cache stores RAW suspend_d / namechange / stk_limit rows — never a + # derived flag as source of truth; the flag derivation stays here. + self._cache = cache def _client(self): if self._pro is None: @@ -53,6 +59,12 @@ def _call(self, fn, **kwargs): # -- suspensions (停牌) -------------------------------------------------- # def suspended(self, symbols: list[str], start: str, end: str) -> set[tuple]: """Return the set of (Timestamp date, symbol) suspended over [start, end].""" + if self._cache is not None: + df = self._cache.suspend_d(symbols, start, end, self._suspend_fetch()) + return { + (pd.Timestamp(d), str(sym)) + for d, sym in zip(df["date"], df["symbol"]) + } pro = self._client() s = pd.Timestamp(start).strftime("%Y%m%d") e = pd.Timestamp(end).strftime("%Y%m%d") @@ -70,6 +82,9 @@ def suspended(self, symbols: list[str], start: str, end: str) -> set[tuple]: # -- ST status (namechange) -------------------------------------------- # def st_intervals(self, symbols: list[str]) -> dict[str, list[tuple]]: """Return {symbol: [(start_ts, end_ts|None, is_st_bool), ...]} from names.""" + if self._cache is not None: + frame = self._cache.namechange(symbols, self._namechange_fetch()) + return self._intervals_from_namechange(frame) pro = self._client() result: dict[str, list[tuple]] = {} for sym in symbols: @@ -98,6 +113,13 @@ def st_intervals(self, symbols: list[str]) -> dict[str, list[tuple]]: # -- price limits (涨跌停) ---------------------------------------------- # def limits(self, symbols: list[str], start: str, end: str) -> pd.DataFrame: """Return DataFrame[date, symbol, up_limit, down_limit] over [start, end].""" + if self._cache is not None: + df = self._cache.stk_limit(symbols, start, end, self._stk_limit_fetch()) + if df.empty: + return self._empty_limits() + df = df.copy() + df["symbol"] = df["symbol"].astype(str) + return df[["date", "symbol", "up_limit", "down_limit"]].reset_index(drop=True) pro = self._client() s = pd.Timestamp(start).strftime("%Y%m%d") e = pd.Timestamp(end).strftime("%Y%m%d") @@ -107,13 +129,76 @@ def limits(self, symbols: list[str], start: str, end: str) -> pd.DataFrame: if df is not None and len(df) > 0: frames.append(df) if not frames: - return pd.DataFrame( - {"date": pd.Series([], dtype="datetime64[ns]"), - "symbol": pd.Series([], dtype=object), - "up_limit": pd.Series([], dtype=float), - "down_limit": pd.Series([], dtype=float)} - ) + return self._empty_limits() df = pd.concat(frames, ignore_index=True).rename(columns={"ts_code": "symbol"}) df["date"] = pd.to_datetime(df["trade_date"].astype(str), format="%Y%m%d") df["symbol"] = df["symbol"].astype(str) return df[["date", "symbol", "up_limit", "down_limit"]] + + # -- cache fetch closures (per-symbol retry/throttle stays here) -------- # + # The tushare client is built lazily INSIDE each closure (on the first gap + # actually fetched), so a fully-covered warm cache run reads no token and + # constructs no client — the read-through path stays self-sufficient. + def _suspend_fetch(self): + """`(symbol, s_compact, e_compact) -> raw suspend_d frame` (type 'S').""" + + def fetch(symbol, start_compact, end_compact): + return self._call( + self._client().suspend_d, ts_code=symbol, start_date=start_compact, + end_date=end_compact, suspend_type="S", + ) + + return fetch + + def _stk_limit_fetch(self): + """`(symbol, s_compact, e_compact) -> raw stk_limit frame`.""" + + def fetch(symbol, start_compact, end_compact): + return self._call( + self._client().stk_limit, ts_code=symbol, start_date=start_compact, + end_date=end_compact, + ) + + return fetch + + def _namechange_fetch(self): + """`(symbol) -> raw namechange frame` (dimension; no date range).""" + + def fetch(symbol): + return self._call(self._client().namechange, ts_code=symbol) + + return fetch + + def _intervals_from_namechange(self, frame: pd.DataFrame) -> dict[str, list[tuple]]: + """Build {symbol: [(start, end|None, is_st), ...]} from cached raw rows. + + Same dedupe (by start/end/name) and ST rule as the direct path, so the + per-symbol interval SET is identical to a direct fetch (NaT end -> None). + """ + result: dict[str, list[tuple]] = {} + if frame is None or frame.empty: + return result + for sym, sub in frame.groupby(frame["symbol"].astype(str), sort=False): + seen: set[tuple] = set() + intervals: list[tuple] = [] + for _, row in sub.iterrows(): + start = pd.Timestamp(row["start_date"]) + end_raw = row["end_date"] + end = None if pd.isna(end_raw) else pd.Timestamp(end_raw) + name = str(row["name"]) + key = (start, end, name) + if key in seen: + continue + seen.add(key) + intervals.append((start, end, "ST" in name.upper())) + result[str(sym)] = intervals + return result + + @staticmethod + def _empty_limits() -> pd.DataFrame: + return pd.DataFrame( + {"date": pd.Series([], dtype="datetime64[ns]"), + "symbol": pd.Series([], dtype=object), + "up_limit": pd.Series([], dtype=float), + "down_limit": pd.Series([], dtype=float)} + ) diff --git a/qt/config.py b/qt/config.py index 9261ff4..10eecfc 100644 --- a/qt/config.py +++ b/qt/config.py @@ -36,22 +36,35 @@ class ProjectCfg(_Strict): class CacheCfg(_Strict): - """P4-1 persistent endpoint-level raw cache (market_daily + adj_factor). + """Persistent endpoint-level raw cache (P4-1 market bars + P4-2 universe/tradability). Disabled by DEFAULT for backward compatibility — an existing real config - runs exactly as before until it opts in. When enabled, ``TushareFeed`` reads - market bars through the cache (read-through: only uncovered date ranges hit - the API). The cache stores RAW endpoint facts (unadjusted OHLCV/amount and - raw adj_factor) only — never qfq prices, never any secret. ``front_adjust`` - still runs in memory downstream, unchanged. + runs exactly as before until it opts in. When enabled the tushare-backed + feeds read through the cache (read-through: only uncovered date ranges / + stale snapshots hit the API). The cache stores RAW endpoint facts only + (unadjusted OHLCV/amount, raw adj_factor, raw index_weight / suspend_d / + stk_limit / namechange / stock_basic rows) — never qfq prices, never a + derived tradability flag as source of truth, never any secret. The + downstream transforms (``front_adjust``, raw price-limit checks, PIT as-of + membership / industry / financials) still run in memory, unchanged. + + P4-2 caches: index_weight, suspend_d, namechange, stk_limit, stock_basic. + Not yet cached (P4-3): daily_basic, fina_indicator, index_member_all. """ enabled: bool = False root_dir: str = "artifacts/cache/tushare/v1" # Any requested range whose end is within this many days of "today" has its # recent tail refetched (recent rows can be corrected/delayed upstream). + # Applies to the dense date-range endpoints (market bars, suspend_d, + # stk_limit, index_weight). refresh_recent_days: int = 14 - # Endpoint names (e.g. "market_daily", "adj_factor") to always refetch in + # Snapshot / dimension endpoints (stock_basic, namechange) carry no date + # range; they are refetched once their last successful fetch is older than + # this many days (a slow-moving freshness policy). 0 disables staleness + # refresh (only force_refresh re-pulls them). + refresh_dimension_days: int = 30 + # Endpoint names (e.g. "market_daily", "index_weight") to always refetch in # full, ignoring coverage — for forcing a clean re-pull of one endpoint. force_refresh: list[str] = Field(default_factory=list) @@ -64,6 +77,15 @@ def _check_recent_days(cls, v: int) -> int: ) return v + @field_validator("refresh_dimension_days") + @classmethod + def _check_dimension_days(cls, v: int) -> int: + if v < 0: + raise ValueError( + f"data.cache.refresh_dimension_days must be >= 0; got {v}." + ) + return v + @field_validator("root_dir") @classmethod def _check_root_dir(cls, v: str) -> str: diff --git a/qt/oos_stability.py b/qt/oos_stability.py index de8f4aa..f92ab6b 100644 --- a/qt/oos_stability.py +++ b/qt/oos_stability.py @@ -41,12 +41,14 @@ _FrameScores, _alpha_disclosure, _build_alpha, + _build_cache, _build_factors, _build_scores, _build_universe, _collect_downgrades, _compute_factor_panel, _load_panel, + _log_run_cache_stats, _make_logger, _maybe_enrich_covariates, _maybe_enrich_financials, @@ -365,13 +367,15 @@ def _run_oos_cell( logger.info("phase3 oos start: project=%s split=%s", cfg.project.name, split.date()) # --- ONE shared data load / factor panel (identical inputs for both) ----- # - universe, symbols = _build_universe(cfg, logger) - panel = _load_panel(cfg, symbols, logger) + cache = _build_cache(cfg) + universe, symbols = _build_universe(cfg, logger, cache) + panel = _load_panel(cfg, symbols, logger, cache) factors = _build_factors(cfg) panel = _maybe_enrich_financials(cfg, panel, symbols, factors, logger) panel = _maybe_enrich_value(cfg, panel, symbols, factors, logger) panel = _maybe_enrich_covariates(cfg, panel, symbols, logger) - panel = _maybe_enrich_listing(cfg, panel, symbols, logger) + panel = _maybe_enrich_listing(cfg, panel, symbols, logger, cache) + _log_run_cache_stats(cache, logger) factor_panel = _compute_factor_panel(cfg, panel, factors, logger) processed = _process_factors(cfg, factor_panel, panel) diff --git a/qt/phase2_baseline.py b/qt/phase2_baseline.py index 5194b59..ca0f040 100644 --- a/qt/phase2_baseline.py +++ b/qt/phase2_baseline.py @@ -40,6 +40,7 @@ _alpha_disclosure, _alpha_forward_returns, _build_alpha, + _build_cache, _build_factors, _build_scores, _build_universe, @@ -48,6 +49,7 @@ _factor_analytics, _standard_analytics, _load_panel, + _log_run_cache_stats, _make_logger, _maybe_enrich_covariates, _maybe_enrich_financials, @@ -359,14 +361,17 @@ def run_phase2_baseline(config_path: str) -> Phase2Result: logger.info("phase2 baseline start: project=%s", cfg.project.name) # --- reuse the exact P0/P1 spine ------------------------------------- # - universe, symbols = _build_universe(cfg, logger) - panel = _load_panel(cfg, symbols, logger) + cache = _build_cache(cfg) + universe, symbols = _build_universe(cfg, logger, cache) + panel = _load_panel(cfg, symbols, logger, cache) factors = _build_factors(cfg) primary = factors[0] panel = _maybe_enrich_financials(cfg, panel, symbols, factors, logger) panel = _maybe_enrich_value(cfg, panel, symbols, factors, logger) panel = _maybe_enrich_covariates(cfg, panel, symbols, logger) - panel = _maybe_enrich_listing(cfg, panel, symbols, logger) + panel = _maybe_enrich_listing(cfg, panel, symbols, logger, cache) + # P4-1/P4-2: one cache-stats line after every cached endpoint has run. + _log_run_cache_stats(cache, logger) factor_panel = _compute_factor_panel(cfg, panel, factors, logger) processed = _process_factors(cfg, factor_panel, panel) alpha = _build_alpha(cfg) diff --git a/qt/pipeline.py b/qt/pipeline.py index 0963c5d..7ea8318 100644 --- a/qt/pipeline.py +++ b/qt/pipeline.py @@ -435,14 +435,18 @@ def run_phase0(config_path: str) -> Phase0Result: logger = _make_logger(log_path) logger.info("phase0 start: project=%s source=%s", cfg.project.name, cfg.data.source) - universe, symbols = _build_universe(cfg, logger) - panel = _load_panel(cfg, symbols, logger) + cache = _build_cache(cfg) + universe, symbols = _build_universe(cfg, logger, cache) + panel = _load_panel(cfg, symbols, logger, cache) factors = _build_factors(cfg) primary = factors[0] panel = _maybe_enrich_financials(cfg, panel, symbols, factors, logger) panel = _maybe_enrich_value(cfg, panel, symbols, factors, logger) panel = _maybe_enrich_covariates(cfg, panel, symbols, logger) - panel = _maybe_enrich_listing(cfg, panel, symbols, logger) + panel = _maybe_enrich_listing(cfg, panel, symbols, logger, cache) + # P4-1/P4-2: one concise cache-stats line after every cached endpoint has run + # (market bars + universe/tradability). A warm historical rerun shows all 0s. + _log_run_cache_stats(cache, logger) factor_panel = _compute_factor_panel(cfg, panel, factors, logger) processed = _process_factors(cfg, factor_panel, panel) @@ -514,13 +518,14 @@ def run_phase0(config_path: str) -> Phase0Result: # --------------------------------------------------------------------------- # # Step helpers (each small + single-purpose; orchestration stays linear above). # --------------------------------------------------------------------------- # -def _build_feed(cfg: RootConfig) -> DataFeed: +def _build_feed(cfg: RootConfig, cache=None) -> DataFeed: """Construct the DataFeed for the configured source (``demo`` | ``tushare``). Dispatches on ``cfg.data.source`` so a ``tushare`` config is NOT silently served demo data. The tushare branch requires ``external_secret_file`` (the token never lives in the repo); constructing the feed performs no network / - token read — that is deferred to first ``get_bars`` (SEC-001/004). + token read — that is deferred to first ``get_bars`` (SEC-001/004). The shared + read-through ``cache`` (or None) is injected by the runner. """ if cfg.data.source == "demo": return DemoFeed(calendar_start=cfg.data.start) @@ -535,19 +540,23 @@ def _build_feed(cfg: RootConfig) -> DataFeed: return TushareFeed( secret_file=secret_file, token_key=cfg.data.tushare_token_key, - cache=_build_market_cache(cfg), + cache=cache, ) raise ValueError( f"Unsupported data.source {cfg.data.source!r}; expected 'demo' or 'tushare'." ) -def _build_market_cache(cfg: RootConfig): - """Build the read-through market cache when ``data.cache.enabled`` (P4-1). +def _build_cache(cfg: RootConfig): + """Build the ONE shared read-through cache when ``data.cache.enabled``. - Returns ``None`` when caching is off — the feed then behaves EXACTLY as - before (backward compatible). The cache stores RAW market_daily + adj_factor - only; ``front_adjust`` still runs in memory downstream, unchanged. + Returns ``None`` when caching is off — every feed then behaves EXACTLY as + before (backward compatible). A single instance is threaded through every + tushare-backed feed in a run so coverage/store stay one consistent + source-of-truth and the per-endpoint gap-fetch counts aggregate into one + run-log line. The cache stores RAW endpoint facts only (P4-1 market bars + + P4-2 index_weight / suspend_d / namechange / stk_limit / stock_basic); + front_adjust, raw price-limit checks, and PIT as-of logic stay downstream. """ cache_cfg = cfg.data.cache if not cache_cfg.enabled: @@ -559,12 +568,13 @@ def _build_market_cache(cfg: RootConfig): CacheParquetStore(root), CoverageLedger(root), refresh_recent_days=cache_cfg.refresh_recent_days, + refresh_dimension_days=cache_cfg.refresh_dimension_days, force_refresh=tuple(cache_cfg.force_refresh), ) def _build_universe( - cfg: RootConfig, logger: logging.Logger + cfg: RootConfig, logger: logging.Logger, cache=None ) -> tuple[Universe, list[str]]: """Construct the universe and the symbol set whose market data must be loaded. @@ -597,7 +607,9 @@ def _build_universe( "(token for index_weight)." ) feed = IndexConstituentsFeed( - cfg.data.external_secret_file, token_key=cfg.data.tushare_token_key + cfg.data.external_secret_file, + token_key=cfg.data.tushare_token_key, + cache=cache, ) # As-of membership at cfg.data.start needs the latest snapshot *before* # the backtest window too. Pulling only [start, end] can leave early @@ -622,12 +634,28 @@ def _build_universe( raise ValueError(f"Unsupported universe.type {cfg.universe.type!r}.") +# Endpoints surfaced in the run-log cache-stats line, in a fixed order. Market +# bars (P4-1) come first so the historical "data cache: market_daily_gap_fetches= +# N adj_factor_gap_fetches=M ..." prefix is preserved; P4-2 endpoints follow. +_CACHED_ENDPOINTS: tuple[str, ...] = ( + "market_daily", "adj_factor", "index_weight", "suspend_d", + "namechange", "stk_limit", "stock_basic", +) + + +def _format_cache_stats(stats: dict) -> str: + """One concise line of per-endpoint gap-fetch counts (endpoint counts only).""" + return "data cache: " + " ".join( + f"{ep}_gap_fetches={int(stats.get(ep, 0))}" for ep in _CACHED_ENDPOINTS + ) + + def _log_cache_stats(feed, logger: logging.Logger) -> None: - """Log the market cache's per-endpoint gap-fetch counts (P4-1), if any. + """Log a feed's per-endpoint cache gap-fetch counts (P4-1/P4-2), if any. - No-op unless the feed exposes ``cache_stats()`` returning a dict (i.e. the - persistent cache is enabled). Emits a single concise line through the - run-scoped logger — endpoint counts only, never a token or secret path. + No-op unless ``feed`` exposes ``cache_stats()`` returning a dict (i.e. the + persistent cache is enabled). Emits one concise line through the run-scoped + logger — endpoint counts only, never a token, secret path, or symbol dump. """ getter = getattr(feed, "cache_stats", None) if getter is None: @@ -635,26 +663,33 @@ def _log_cache_stats(feed, logger: logging.Logger) -> None: stats = getter() if not stats: return - logger.info( - "data cache: market_daily_gap_fetches=%d adj_factor_gap_fetches=%d", - int(stats.get("market_daily", 0)), - int(stats.get("adj_factor", 0)), - ) + logger.info("%s", _format_cache_stats(stats)) + + +def _log_run_cache_stats(cache, logger: logging.Logger) -> None: + """Log the SHARED run cache's aggregated gap-fetch counts (P4-1/P4-2). + + Called once after all tushare-backed feeds in a run have used the shared + cache, so the counts cover every cached endpoint (market bars + universe / + tradability). No-op when caching is disabled (``cache`` is None). Endpoint + counts only — never a token, secret path, or per-symbol detail. + """ + if cache is None: + return + stats = cache.stats() + if not stats: + return + logger.info("%s", _format_cache_stats(stats)) def _load_panel( - cfg: RootConfig, symbols: list[str], logger: logging.Logger + cfg: RootConfig, symbols: list[str], logger: logging.Logger, cache=None ) -> pd.DataFrame: """Fetch the market panel for ``symbols``, persist it, and read it back.""" if not symbols: raise ValueError("No symbols to load; the universe produced an empty set.") - feed = _build_feed(cfg) + feed = _build_feed(cfg, cache) panel = feed.get_bars(symbols, cfg.data.start, cfg.data.end, freq=cfg.data.freq) - # P4-1: when the persistent market cache is on, surface its gap-fetch counts - # in the run log so the read-through hit rate is directly observable (a warm - # historical rerun shows 0/0). No-op when caching is off or the feed has no - # cache (DemoFeed); the stats carry endpoint counts only, never a secret. - _log_cache_stats(feed, logger) if panel.empty: raise ValueError( "No market data returned for the configured window " @@ -668,7 +703,7 @@ def _load_panel( # the UNADJUSTED close against the raw stk_limit (limits are quoted in raw # price terms). front-adjust below only scales OHLC and leaves the boolean # flags intact, so factors/backtest see qfq prices while limit flags stay right. - panel = _enrich_tradability(cfg, panel, symbols, logger) + panel = _enrich_tradability(cfg, panel, symbols, logger, cache) # Store stays RAW (+ adj_factor, incremental-safe); front-adjust in memory so # factors / backtest see continuous qfq prices. Identity for demo (adj=1.0). panel = front_adjust(panel) @@ -681,7 +716,11 @@ def _load_panel( def _enrich_tradability( - cfg: RootConfig, panel: pd.DataFrame, symbols: list[str], logger: logging.Logger + cfg: RootConfig, + panel: pd.DataFrame, + symbols: list[str], + logger: logging.Logger, + cache=None, ) -> pd.DataFrame: """Add suspended / ST / price-limit flags when filters need them (tushare only). @@ -693,7 +732,9 @@ def _enrich_tradability( if cfg.data.source != "tushare" or not (flt.suspended or flt.st or flt.limit_up_down): return panel feed = TushareFlagsFeed( - cfg.data.external_secret_file, token_key=cfg.data.tushare_token_key + cfg.data.external_secret_file, + token_key=cfg.data.tushare_token_key, + cache=cache, ) suspended = feed.suspended(symbols, cfg.data.start, cfg.data.end) if flt.suspended else None st = feed.st_intervals(symbols) if flt.st else None @@ -871,7 +912,11 @@ def _maybe_enrich_value( def _maybe_enrich_listing( - cfg: RootConfig, panel: pd.DataFrame, symbols: list[str], logger: logging.Logger + cfg: RootConfig, + panel: pd.DataFrame, + symbols: list[str], + logger: logging.Logger, + cache=None, ) -> pd.DataFrame: """Attach a per-symbol ``list_date`` column when min_listing_days is enforced. @@ -885,7 +930,9 @@ def _maybe_enrich_listing( if cfg.data.source != "tushare": return panel # demo: no listing dates -> filter is a disclosed no-op feed = TushareCovariatesFeed( - cfg.data.external_secret_file, token_key=cfg.data.tushare_token_key + cfg.data.external_secret_file, + token_key=cfg.data.tushare_token_key, + cache=cache, ) listing = feed.listing_dates(symbols) panel = enrich_listing(panel, listing) diff --git a/qt/subset_validation.py b/qt/subset_validation.py index b1a8c22..7f9a274 100644 --- a/qt/subset_validation.py +++ b/qt/subset_validation.py @@ -47,12 +47,14 @@ from qt.pipeline import ( _alpha_disclosure, _build_alpha, + _build_cache, _build_factors, _build_scores, _build_universe, _collect_downgrades, _compute_factor_panel, _load_panel, + _log_run_cache_stats, _make_logger, _maybe_enrich_covariates, _maybe_enrich_financials, @@ -459,13 +461,15 @@ def _run_subset_cell(cfg: RootConfig, logger) -> SubsetCellResult: logger.info("subset cell start: project=%s split=%s", cfg.project.name, split.date()) # --- ONE shared data load / raw factor panel (same sequence as _run_oos_cell) - universe, symbols = _build_universe(cfg, logger) - panel = _load_panel(cfg, symbols, logger) + cache = _build_cache(cfg) + universe, symbols = _build_universe(cfg, logger, cache) + panel = _load_panel(cfg, symbols, logger, cache) factors = _build_factors(cfg) panel = _maybe_enrich_financials(cfg, panel, symbols, factors, logger) panel = _maybe_enrich_value(cfg, panel, symbols, factors, logger) panel = _maybe_enrich_covariates(cfg, panel, symbols, logger) - panel = _maybe_enrich_listing(cfg, panel, symbols, logger) + panel = _maybe_enrich_listing(cfg, panel, symbols, logger, cache) + _log_run_cache_stats(cache, logger) factor_panel = _compute_factor_panel(cfg, panel, factors, logger) horizon = int(cfg.analytics.forward_return_periods[0]) diff --git a/tests/test_index_pipeline.py b/tests/test_index_pipeline.py index b8e8101..4985254 100644 --- a/tests/test_index_pipeline.py +++ b/tests/test_index_pipeline.py @@ -54,9 +54,10 @@ def test_index_universe_fetches_pre_start_snapshot_for_asof(monkeypatch): calls: list[tuple[str, str, str]] = [] class _FakeIndexFeed: - def __init__(self, secret_file, token_key="tushare.token"): + def __init__(self, secret_file, token_key="tushare.token", cache=None): self.secret_file = secret_file self.token_key = token_key + self.cache = cache def get_constituents(self, index_code: str, start: str, end: str) -> pd.DataFrame: calls.append((index_code, start, end)) diff --git a/tests/test_tushare_cache_market.py b/tests/test_tushare_cache_market.py index 05acabe..b0d86dc 100644 --- a/tests/test_tushare_cache_market.py +++ b/tests/test_tushare_cache_market.py @@ -172,9 +172,11 @@ def test_feed_exposes_cache_stats_cold_then_warm(tmp_path, monkeypatch): feed, root = _feed_with_cache(tmp_path, monkeypatch, pro) symbols = ["000001.SZ", "000002.SZ"] - # cold run: gap-fetches == one per symbol per endpoint. + # cold run: gap-fetches == one per symbol per endpoint. stats() seeds every + # known endpoint (P4-2) so the run-log line always shows explicit 0s. feed.get_bars(symbols, "2024-01-01", "2024-01-31") - assert feed.cache_stats() == {"market_daily": 2, "adj_factor": 2} + cold = feed.cache_stats() + assert cold["market_daily"] == 2 and cold["adj_factor"] == 2 # warm rerun on a fresh feed + fresh cache over the SAME root -> 0 gap-fetches. pro2 = FakePro() @@ -187,7 +189,8 @@ def test_feed_exposes_cache_stats_cold_then_warm(tmp_path, monkeypatch): warm_feed = TushareFeed(secret_file=str(_write_fake_config(tmp_path)), cache=warm_cache) warm_feed.get_bars(symbols, "2024-01-01", "2024-01-31") - assert warm_feed.cache_stats() == {"market_daily": 0, "adj_factor": 0} + warm = warm_feed.cache_stats() + assert warm["market_daily"] == 0 and warm["adj_factor"] == 0 assert pro2.daily_calls == 0 and pro2.adj_calls == 0 diff --git a/tests/test_tushare_cache_universe.py b/tests/test_tushare_cache_universe.py new file mode 100644 index 0000000..e885b69 --- /dev/null +++ b/tests/test_tushare_cache_universe.py @@ -0,0 +1,523 @@ +"""P4-2: universe + tradability read-through cache (network-free, fake clients). + +Pins the P4-2 acceptance for the five new endpoints (index_weight, suspend_d, +namechange, stk_limit, stock_basic): + * cold miss populates the cache; an identical warm rerun makes ZERO calls; + * cached feed output EQUALS the direct (uncached) feed output, so PIT as-of + membership / suspension / ST intervals / raw price limits / list_date are + all unchanged; + * an empty endpoint return still records coverage (no needless refetch); + * a FAILED fetch records NO coverage (a later run retries); + * a duplicate upsert keeps one row per the endpoint's natural key; + * the cache-disabled path is byte-for-byte the direct path; + * the run-log stats line names the new endpoints; + * cache files + ledger contain no token / secret-file content. + +No test hits the network or reads a real token: fake ``pro`` clients with call +counters drive both the cache path and the direct path. +""" + +from __future__ import annotations + +import pandas as pd +import pytest + +from data.cache import CacheParquetStore, CoverageLedger, TushareCache +from data.cache.tushare_cache import ( + INDEX_WEIGHT, + NAMECHANGE, + STK_LIMIT, + STOCK_BASIC, + SUSPEND_D, +) +from data.feed.index_feed import IndexConstituentsFeed +from data.feed.tushare_covariates import TushareCovariatesFeed +from data.feed.tushare_flags import TushareFlagsFeed + +FAKE_TOKEN = "FAKE_TUSHARE_TOKEN_do_not_leak_0123456789abcdef" +SECRET_PATH_MARKER = "/abs/path/to/.config.json" + + +# --------------------------------------------------------------------------- # +# Fake tushare clients (deterministic rows + per-endpoint call counters). +# --------------------------------------------------------------------------- # +class _FakeIndexPro: + """``index_weight`` with API-like window filtering + a call counter.""" + + _ROWS = [ + ("20240131", "000002.SZ", 1.5), + ("20240131", "000001.SZ", 2.5), + ("20240229", "000001.SZ", 2.0), + ] + + def __init__(self): + self.calls = 0 + + def index_weight(self, index_code, start_date, end_date, **_): + self.calls += 1 + s, e = pd.Timestamp(start_date), pd.Timestamp(end_date) + keep = [r for r in self._ROWS if s <= pd.Timestamp(r[0]) <= e] + if not keep: + return pd.DataFrame( + columns=["index_code", "con_code", "trade_date", "weight"] + ) + return pd.DataFrame( + { + "index_code": [index_code] * len(keep), + "con_code": [r[1] for r in keep], + "trade_date": [r[0] for r in keep], + "weight": [r[2] for r in keep], + } + ) + + +class _FakeFlagsPro: + """``suspend_d`` / ``namechange`` / ``stk_limit`` with call counters.""" + + def __init__(self): + self.suspend_calls = 0 + self.namechange_calls = 0 + self.stk_limit_calls = 0 + + def suspend_d(self, ts_code, start_date, end_date, suspend_type, **_): + self.suspend_calls += 1 + if ts_code == "000001.SZ": + return pd.DataFrame( + {"ts_code": ["000001.SZ"], "trade_date": ["20240110"], + "suspend_type": ["S"]} + ) + return pd.DataFrame() + + def namechange(self, ts_code, **_): + self.namechange_calls += 1 + return pd.DataFrame( + { + "ts_code": [ts_code, ts_code, ts_code], + "name": ["*ST X", "*ST X", "X"], + "start_date": ["20240301", "20240301", "20230101"], + "end_date": [None, None, "20240229"], + } + ) + + def stk_limit(self, ts_code, start_date, end_date, **_): + self.stk_limit_calls += 1 + return pd.DataFrame( + { + "trade_date": ["20240110", "20240111"], + "ts_code": [ts_code, ts_code], + "up_limit": [11.0, 11.5], + "down_limit": [9.0, 9.2], + } + ) + + +class _FakeBasicPro: + """``stock_basic`` (global snapshot) with a call counter.""" + + def __init__(self): + self.calls = 0 + + def stock_basic(self, **_): + self.calls += 1 + return pd.DataFrame( + { + "ts_code": ["000001.SZ", "000002.SZ", "000300.SH"], + "list_date": ["19910403", "19910101", "20050408"], + } + ) + + +# --------------------------------------------------------------------------- # +# helpers +# --------------------------------------------------------------------------- # +def _make_cache(root, *, today="2024-12-31", force_refresh=()): + """A cache with a frozen clock so dimension freshness is deterministic.""" + ts = pd.Timestamp(today) + return TushareCache( + CacheParquetStore(root), + CoverageLedger(root), + refresh_recent_days=14, + refresh_dimension_days=30, + force_refresh=force_refresh, + today=ts, + clock=lambda: ts, + ) + + +def _index_feed(monkeypatch, pro, cache=None): + feed = IndexConstituentsFeed("x.json", cache=cache) + monkeypatch.setattr(feed, "_client", lambda: pro) + return feed + + +def _flags_feed(monkeypatch, pro, cache=None): + feed = TushareFlagsFeed("x.json", cache=cache) + monkeypatch.setattr(feed, "_client", lambda: pro) + return feed + + +def _cov_feed(monkeypatch, pro, cache=None): + feed = TushareCovariatesFeed("x.json", cache=cache) + monkeypatch.setattr(feed, "_client", lambda: pro) + return feed + + +# --------------------------------------------------------------------------- # +# index_weight: cold/warm + equals direct + PIT preserved +# --------------------------------------------------------------------------- # +def test_index_weight_cold_warm_zero_calls(tmp_path, monkeypatch): + root = str(tmp_path / "cache") + pro = _FakeIndexPro() + feed = _index_feed(monkeypatch, pro, _make_cache(root)) + out1 = feed.get_constituents("000300.SH", "2024-01-01", "2024-03-31") + assert pro.calls > 0 # cold: paged window(s) fetched + assert out1["date"].nunique() == 2 + + # warm rerun on a FRESH cache over the same root -> zero index_weight calls. + pro2 = _FakeIndexPro() + warm_cache = _make_cache(root) + warm = _index_feed(monkeypatch, pro2, warm_cache) + out2 = warm.get_constituents("000300.SH", "2024-01-01", "2024-03-31") + assert pro2.calls == 0 + pd.testing.assert_frame_equal(out1, out2) + assert warm_cache.stats()[INDEX_WEIGHT] == 0 + + +def test_index_weight_cached_equals_direct(tmp_path, monkeypatch): + direct = _index_feed(monkeypatch, _FakeIndexPro()) + direct_out = direct.get_constituents("000300.SH", "2024-01-01", "2024-03-31") + + root = str(tmp_path / "cache") + cached = _index_feed(monkeypatch, _FakeIndexPro(), _make_cache(root)) + cached_out = cached.get_constituents("000300.SH", "2024-01-01", "2024-03-31") + + pd.testing.assert_frame_equal(direct_out, cached_out) + # PIT preserved: every stored snapshot date kept (latest-as-of stays the + # universe's job); the 2024-01-31 cross-section is both names, ascending. + jan = cached_out[cached_out["date"] == pd.Timestamp("2024-01-31")]["symbol"].tolist() + assert jan == ["000001.SZ", "000002.SZ"] + + +# --------------------------------------------------------------------------- # +# suspend_d: cold/warm + equals direct +# --------------------------------------------------------------------------- # +def test_suspend_d_cold_warm_zero_calls_equals_direct(tmp_path, monkeypatch): + symbols = ["000001.SZ", "000002.SZ"] + direct = _flags_feed(monkeypatch, _FakeFlagsPro()) + direct_out = direct.suspended(symbols, "2024-01-01", "2024-01-31") + + root = str(tmp_path / "cache") + pro = _FakeFlagsPro() + cached = _flags_feed(monkeypatch, pro, _make_cache(root)) + cold_out = cached.suspended(symbols, "2024-01-01", "2024-01-31") + assert pro.suspend_calls == 2 # one per symbol (full miss) + assert cold_out == direct_out # set equality + + pro2 = _FakeFlagsPro() + warm_cache = _make_cache(root) + warm = _flags_feed(monkeypatch, pro2, warm_cache) + warm_out = warm.suspended(symbols, "2024-01-01", "2024-01-31") + assert pro2.suspend_calls == 0 # fully covered + assert warm_out == direct_out + assert warm_cache.stats()[SUSPEND_D] == 0 + + +# --------------------------------------------------------------------------- # +# namechange / ST intervals: cold/warm + equals direct +# --------------------------------------------------------------------------- # +def test_namechange_cold_warm_zero_calls_equals_direct(tmp_path, monkeypatch): + symbols = ["000001.SZ", "000002.SZ"] + direct = _flags_feed(monkeypatch, _FakeFlagsPro()) + direct_out = direct.st_intervals(symbols) + + root = str(tmp_path / "cache") + pro = _FakeFlagsPro() + cached = _flags_feed(monkeypatch, pro, _make_cache(root)) + cold_out = cached.st_intervals(symbols) + assert pro.namechange_calls == 2 + + pro2 = _FakeFlagsPro() + warm_cache = _make_cache(root) + warm = _flags_feed(monkeypatch, pro2, warm_cache) + warm_out = warm.st_intervals(symbols) + assert pro2.namechange_calls == 0 # dimension snapshot fresh -> no refetch + assert warm_cache.stats()[NAMECHANGE] == 0 + + # interval SETS equal across direct / cold / warm (dedupe + ST flag preserved) + assert set(direct_out) == set(cold_out) == set(warm_out) + for sym in symbols: + assert set(direct_out[sym]) == set(cold_out[sym]) == set(warm_out[sym]) + assert any(is_st for _, _, is_st in cold_out[sym]) + assert any(not is_st for _, _, is_st in cold_out[sym]) + # the open (active) interval keeps a None end exactly like the direct path + assert any(end is None for _, end, _ in cold_out[sym]) + + +# --------------------------------------------------------------------------- # +# stk_limit: cold/warm + equals direct (raw price terms) +# --------------------------------------------------------------------------- # +def test_stk_limit_cold_warm_zero_calls_equals_direct(tmp_path, monkeypatch): + symbols = ["000001.SZ", "000002.SZ"] + direct = _flags_feed(monkeypatch, _FakeFlagsPro()) + direct_out = direct.limits(symbols, "2024-01-01", "2024-01-31") + + root = str(tmp_path / "cache") + pro = _FakeFlagsPro() + cached = _flags_feed(monkeypatch, pro, _make_cache(root)) + cold_out = cached.limits(symbols, "2024-01-01", "2024-01-31") + assert pro.stk_limit_calls == 2 + pd.testing.assert_frame_equal(direct_out, cold_out) + # raw price terms preserved (no qfq scaling in the cache) + assert cold_out.iloc[0]["up_limit"] == 11.0 + + pro2 = _FakeFlagsPro() + warm_cache = _make_cache(root) + warm = _flags_feed(monkeypatch, pro2, warm_cache) + warm_out = warm.limits(symbols, "2024-01-01", "2024-01-31") + assert pro2.stk_limit_calls == 0 + pd.testing.assert_frame_equal(direct_out, warm_out) + assert warm_cache.stats()[STK_LIMIT] == 0 + + +# --------------------------------------------------------------------------- # +# stock_basic / listing_dates: cold/warm + equals direct +# --------------------------------------------------------------------------- # +def test_stock_basic_cold_warm_zero_calls_equals_direct(tmp_path, monkeypatch): + symbols = ["000001.SZ", "000002.SZ"] + direct = _cov_feed(monkeypatch, _FakeBasicPro()) + direct_out = direct.listing_dates(symbols) + + root = str(tmp_path / "cache") + pro = _FakeBasicPro() + cached = _cov_feed(monkeypatch, pro, _make_cache(root)) + cold_out = cached.listing_dates(symbols) + assert pro.calls == 1 # ONE global snapshot, not per-symbol + assert cold_out == direct_out + assert cold_out["000001.SZ"] == pd.Timestamp("1991-04-03") + + pro2 = _FakeBasicPro() + warm_cache = _make_cache(root) + warm = _cov_feed(monkeypatch, pro2, warm_cache) + warm_out = warm.listing_dates(symbols) + assert pro2.calls == 0 # dimension snapshot fresh -> no refetch + assert warm_out == direct_out + assert warm_cache.stats()[STOCK_BASIC] == 0 + + +# --------------------------------------------------------------------------- # +# empty endpoint return still records coverage (no needless refetch) +# --------------------------------------------------------------------------- # +def test_empty_return_records_coverage_no_refetch(tmp_path): + root = str(tmp_path / "cache") + cache = _make_cache(root) + calls = {"n": 0} + + def fetch(symbol, s, e): + calls["n"] += 1 + return pd.DataFrame() # always empty + + out = cache.suspend_d(["000999.SZ"], "2024-01-01", "2024-01-31", fetch) + assert out.empty and calls["n"] == 1 + + ledger = CoverageLedger(root).read() + rows = ledger[ledger["endpoint"] == SUSPEND_D] + assert len(rows) == 1 and rows.iloc[0]["status"] == "empty" + + # a second run sees the empty range as covered -> no extra call. + cache2 = _make_cache(root) + cache2.suspend_d(["000999.SZ"], "2024-01-01", "2024-01-31", fetch) + assert calls["n"] == 1 + + +def test_stock_basic_empty_snapshot_records_coverage(tmp_path): + root = str(tmp_path / "cache") + cache = _make_cache(root) + calls = {"n": 0} + + def fetch(): + calls["n"] += 1 + return pd.DataFrame() + + assert cache.stock_basic(fetch).empty and calls["n"] == 1 + # fresh instance, same root: snapshot recorded (empty) -> no refetch. + _make_cache(root).stock_basic(fetch) + assert calls["n"] == 1 + + +# --------------------------------------------------------------------------- # +# failed fetch does NOT record coverage (a later run retries) +# --------------------------------------------------------------------------- # +def test_failed_fetch_records_no_coverage(tmp_path): + root = str(tmp_path / "cache") + cache = _make_cache(root) + + def boom(symbol, s, e): + raise RuntimeError("transient network error") + + with pytest.raises(RuntimeError): + cache.suspend_d(["000001.SZ"], "2024-01-01", "2024-01-31", boom) + + # nothing covered -> a later good fetch retries and populates. + assert CoverageLedger(root).read().empty + cache2 = _make_cache(root) + good_calls = {"n": 0} + + def good(symbol, s, e): + good_calls["n"] += 1 + return pd.DataFrame( + {"ts_code": [symbol], "trade_date": ["20240110"], "suspend_type": ["S"]} + ) + + out = cache2.suspend_d(["000001.SZ"], "2024-01-01", "2024-01-31", good) + assert good_calls["n"] == 1 and len(out) == 1 + + +def test_failed_snapshot_fetch_records_no_coverage(tmp_path): + root = str(tmp_path / "cache") + cache = _make_cache(root) + + def boom(): + raise RuntimeError("transient network error") + + with pytest.raises(RuntimeError): + cache.stock_basic(boom) + assert CoverageLedger(root).read().empty + + +# --------------------------------------------------------------------------- # +# duplicate upsert keeps one row per natural key (index_weight / suspend_d) +# --------------------------------------------------------------------------- # +def test_duplicate_upsert_one_row_per_natural_key(tmp_path): + store = CacheParquetStore(str(tmp_path / "cache")) + rows = pd.DataFrame( + { + "index_code": ["000300.SH", "000300.SH"], + "date": pd.to_datetime(["2024-01-31", "2024-01-31"]), + "symbol": ["000001.SZ", "000002.SZ"], + "weight": [2.5, 1.5], + } + ) + store.upsert_symbol(INDEX_WEIGHT, "000300.SH", rows, ["date", "symbol"]) + rows2 = rows.copy() + rows2["weight"] = [9.9, 8.8] # re-fetch same keys, new weights + n = store.upsert_symbol(INDEX_WEIGHT, "000300.SH", rows2, ["date", "symbol"]) + assert n == 2 + cached = store.read_symbol(INDEX_WEIGHT, "000300.SH") + w = cached.loc[cached["symbol"] == "000001.SZ", "weight"].iloc[0] + assert w == 9.9 # latest wins + + +def test_suspend_dedup_keeps_one_row_per_key(tmp_path): + store = CacheParquetStore(str(tmp_path / "cache")) + rows = pd.DataFrame( + { + "date": pd.to_datetime(["2024-01-10", "2024-01-10"]), + "symbol": ["000001.SZ", "000001.SZ"], + "suspend_type": ["S", "S"], + } + ) + store.upsert_symbol(SUSPEND_D, "000001.SZ", rows, ["date", "symbol", "suspend_type"]) + n = store.upsert_symbol( + SUSPEND_D, "000001.SZ", rows, ["date", "symbol", "suspend_type"] + ) + assert n == 1 # both upserts collapse to one natural-key row + + +# --------------------------------------------------------------------------- # +# cache-disabled path is the direct path (no cache files created) +# --------------------------------------------------------------------------- # +def test_cache_disabled_path_unchanged(tmp_path, monkeypatch): + root = tmp_path / "cache" + direct = _flags_feed(monkeypatch, _FakeFlagsPro(), cache=None) + out = direct.limits(["000001.SZ"], "2024-01-01", "2024-01-31") + assert list(out.columns) == ["date", "symbol", "up_limit", "down_limit"] + assert not root.exists() # cache=None never touches the cache tree + + +# --------------------------------------------------------------------------- # +# force_refresh re-pulls a dimension snapshot even when fresh +# --------------------------------------------------------------------------- # +def test_force_refresh_repulls_dimension(tmp_path, monkeypatch): + root = str(tmp_path / "cache") + pro = _FakeBasicPro() + cached = _cov_feed(monkeypatch, pro, _make_cache(root)) + cached.listing_dates(["000001.SZ"]) + assert pro.calls == 1 + + pro2 = _FakeBasicPro() + forced = _cov_feed( + monkeypatch, pro2, _make_cache(root, force_refresh=(STOCK_BASIC,)) + ) + forced.listing_dates(["000001.SZ"]) + assert pro2.calls == 1 # force_refresh ignores freshness + + +# --------------------------------------------------------------------------- # +# run-log stats line names the new endpoints +# --------------------------------------------------------------------------- # +def test_cache_stats_line_includes_new_endpoints(): + from qt.pipeline import _format_cache_stats + + line = _format_cache_stats( + {"market_daily": 1, "index_weight": 2, "suspend_d": 3, + "namechange": 4, "stk_limit": 5, "stock_basic": 6} + ) + assert line.startswith("data cache: market_daily_gap_fetches=1") + for token in ( + "index_weight_gap_fetches=2", + "suspend_d_gap_fetches=3", + "namechange_gap_fetches=4", + "stk_limit_gap_fetches=5", + "stock_basic_gap_fetches=6", + ): + assert token in line + + +def test_log_run_cache_stats_emits_full_line(caplog): + import logging + + from qt.pipeline import _log_run_cache_stats + + class _Cache: + def stats(self): + return {"market_daily": 0, "index_weight": 0, "suspend_d": 0, + "namechange": 0, "stk_limit": 0, "stock_basic": 0} + + logger = logging.getLogger("qt.test_p42_cache_line") + with caplog.at_level(logging.INFO, logger="qt.test_p42_cache_line"): + _log_run_cache_stats(_Cache(), logger) + _log_run_cache_stats(None, logger) # disabled -> no line + assert "stock_basic_gap_fetches=0" in caplog.text + assert caplog.text.count("data cache:") == 1 + + +# --------------------------------------------------------------------------- # +# no token / secret-file content in cache files or ledger +# --------------------------------------------------------------------------- # +def test_cache_files_and_ledger_contain_no_secret(tmp_path, monkeypatch): + root = str(tmp_path / "cache") + cache = _make_cache(root) + + # exercise every new endpoint so all parquet partitions exist. + idx = _index_feed(monkeypatch, _FakeIndexPro(), cache) + idx.get_constituents("000300.SH", "2024-01-01", "2024-03-31") + flags = _flags_feed(monkeypatch, _FakeFlagsPro(), cache) + flags.suspended(["000001.SZ"], "2024-01-01", "2024-01-31") + flags.st_intervals(["000001.SZ"]) + flags.limits(["000001.SZ"], "2024-01-01", "2024-01-31") + cov = _cov_feed(monkeypatch, _FakeBasicPro(), cache) + cov.listing_dates(["000001.SZ"]) + + from pathlib import Path + + blobs = b"" + for p in Path(root).rglob("*.parquet"): + blobs += p.read_bytes() + assert FAKE_TOKEN.encode() not in blobs + assert SECRET_PATH_MARKER.encode() not in blobs + + ledger = CoverageLedger(root).read() + assert "token" not in [c.lower() for c in ledger.columns] + for col in ledger.columns: + joined = "".join(str(v) for v in ledger[col].tolist()) + assert FAKE_TOKEN not in joined and SECRET_PATH_MARKER not in joined