From b781d5f0ad5f506839215af737fcdf768aab4780 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Sat, 13 Jun 2026 12:23:43 +0800 Subject: [PATCH 1/4] feat: P4-1 persistent Tushare market-data cache (daily + adj_factor) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Endpoint-level raw cache below the feeds: real runs no longer refetch full daily bars + adj_factor every time. Disabled by default (backward compatible); an opted-in config reads market bars read-through so only uncovered date ranges hit the API. - config: DataCfg.cache (CacheCfg) — enabled=False, root_dir (default artifacts/cache/tushare/v1), refresh_recent_days=14, force_refresh=[]. Validated; every existing config still loads. - data/cache/ package: - intervals.py: closed-day interval algebra (merge/subtract) for gap planning — coverage is by calendar range, not row presence. - parquet_store.py: per-(endpoint, symbol) parquet under a symbol_prefix shard; atomic upsert-by-(date,symbol) dedup (latest wins); best-effort per-symbol lock. - coverage.py: append-only ledger (endpoint, key_type, key, start/end_date, fields_hash, fetched_at, row_count, status ok/empty/failed, schema_version, source_version). Only ok/empty count as coverage; a failed fetch stays uncovered. No token or secret-file content is ever written. - tushare_cache.py: read-through daily_bars/adj_factor — fetch only gaps, upsert raw rows, record coverage incl. empty returns, then return the requested range from cache. refresh_recent_days refetches a recent tail; force_refresh re-pulls an endpoint in full. Per-run fetch-count stats logged. - TushareFeed.get_bars: cache=None -> unchanged direct path; cache set -> read-through, then the SAME join + select as the direct path so the panel is byte-identical before front_adjust(). The per-symbol retry/throttle stay in the feed via _call closures. - pipeline._build_market_cache wires the cache only when data.cache.enabled. front_adjust, factor/alpha/portfolio/execution/ OOS/report math all unchanged. PanelStore stays a per-run artifact. - config/phase2_real_baseline_cached.yaml: cache-enabled smoke variant. - tests: +24 (cache config validation + all-configs parametrized; market cache: full miss/hit, partial gap, empty coverage, duplicate upsert one-key, qfq equivalence vs direct fetch, no-secret in cache files) -> 406 passed; ruff clean; demo run-phase0 unchanged (ic 0.96 / annual 0.8408). --- config/phase2_real_baseline_cached.yaml | 85 ++++++++ data/cache/__init__.py | 25 +++ data/cache/coverage.py | 136 ++++++++++++ data/cache/intervals.py | 76 +++++++ data/cache/parquet_store.py | 125 +++++++++++ data/cache/tushare_cache.py | 212 ++++++++++++++++++ data/feed/tushare_feed.py | 45 ++++ qt/config.py | 38 ++++ qt/pipeline.py | 22 ++ tests/test_cache_config.py | 79 +++++++ tests/test_phase0_pipeline.py | 6 +- tests/test_tushare_cache_market.py | 272 ++++++++++++++++++++++++ 12 files changed, 1120 insertions(+), 1 deletion(-) create mode 100644 config/phase2_real_baseline_cached.yaml create mode 100644 data/cache/__init__.py create mode 100644 data/cache/coverage.py create mode 100644 data/cache/intervals.py create mode 100644 data/cache/parquet_store.py create mode 100644 data/cache/tushare_cache.py create mode 100644 tests/test_cache_config.py create mode 100644 tests/test_tushare_cache_market.py diff --git a/config/phase2_real_baseline_cached.yaml b/config/phase2_real_baseline_cached.yaml new file mode 100644 index 0000000..b9b559f --- /dev/null +++ b/config/phase2_real_baseline_cached.yaml @@ -0,0 +1,85 @@ +# Phase 2-1 real baseline WITH the P4-1 persistent market cache enabled. +# Identical to phase2_real_baseline.yaml except data.cache.enabled=true + +# a distinct report name. Used for the P4-1 real smoke: run twice; the +# second run makes ZERO market_daily / adj_factor API calls (historical +# window, fully covered). Report metrics must equal the non-cached run. + +project: + name: quantitative_trading_phase2_baseline_cached + timezone: Asia/Shanghai +data: + source: tushare + freq: D + start: '2023-07-01' + end: '2024-06-30' + external_secret_file: /home/shaofl/Projects/financial_projects/.config.json + tushare_token_key: tushare.token + output_name: phase2_daily + cache: + enabled: true + root_dir: artifacts/cache/tushare/v1 + refresh_recent_days: 14 + force_refresh: [] +universe: + type: index + index_code: 000016.SH + symbols: [] + min_listing_days: 60 + filters: + missing_close: true + suspended: true + st: true + limit_up_down: true +factors: +- name: momentum_20 + enabled: true + params: + window: 20 + price_col: close +processing: + drop_missing: true + standardize: + enabled: true + method: zscore + winsorize: + enabled: false + method: mad + n: 3.0 + neutralize: + enabled: true + industry_col: industry + size_col: market_cap + industry_level: L1 +alpha: + model: equal_weight + params: {} +portfolio: + constructor: topn_equal_weight + top_n: 20 + long_only: true + max_weight: null + turnover_cap: null +backtest: + initial_nav: 1.0 + rebalance: monthly + event_order: close_to_next_period + cash_return: 0.0 +cost: + fee_rate: 0.001 + slippage_rate: 0.0 + turnover_formula: l1 +analytics: + forward_return_periods: + - 1 + - 5 + - 20 + quantiles: 5 + benchmark: null +output: + root_dir: artifacts + data_dir: artifacts/data + factor_dir: artifacts/factors + report_dir: artifacts/reports + log_dir: artifacts/logs + overwrite: true + baseline_report_name: phase2_real_baseline_cached.md diff --git a/data/cache/__init__.py b/data/cache/__init__.py new file mode 100644 index 0000000..3c8666f --- /dev/null +++ b/data/cache/__init__.py @@ -0,0 +1,25 @@ +"""Persistent endpoint-level raw cache for tushare (P4-1: market bars). + +This package sits BELOW the feeds: it stores RAW endpoint facts (unadjusted +OHLCV/amount and raw adj_factor) keyed by their natural key, plus a coverage +ledger that records which (endpoint, key, date-range) tuples have been fetched — +so the read-through planner can fetch ONLY uncovered gaps and a repeated run +makes zero API calls for already-covered ranges. + +Design invariants (see tmp/context/tushare_cache_architecture.md): + * the durable cache stays RAW — never qfq prices, never research semantics; + * ``front_adjust`` still runs in memory downstream, unchanged; + * PanelStore remains a per-run artifact layer, NOT the cache source of truth; + * the cache and its ledger never store a token or any secret-file content. + +P4-1 implements ``market_daily`` and ``adj_factor`` only. Universe/tradability +(P4-2) and factor-support endpoints (P4-3) are deliberately out of scope here. +""" + +from __future__ import annotations + +from data.cache.coverage import CoverageLedger +from data.cache.parquet_store import CacheParquetStore +from data.cache.tushare_cache import TushareCache + +__all__ = ["CacheParquetStore", "CoverageLedger", "TushareCache"] diff --git a/data/cache/coverage.py b/data/cache/coverage.py new file mode 100644 index 0000000..86f86d6 --- /dev/null +++ b/data/cache/coverage.py @@ -0,0 +1,136 @@ +"""Coverage ledger for the persistent cache (P4-1). + +Row presence alone cannot tell "missing because not fetched" from "missing +because the source has no row" (a stock not listed / suspended / out of index on +a date legitimately has no row). The ledger records, per fetch, which +``(endpoint, key, [start_date, end_date])`` range was retrieved and with what +outcome, so the planner can compute genuine gaps. + +Schema (one append-only row per fetch attempt): + + endpoint market_daily | adj_factor | ... + key_type symbol | index_code | global + key e.g. 000001.SZ + start_date closed interval start (NaT for snapshot endpoints) + end_date closed interval end (NaT for snapshot endpoints) + fields_hash stable hash of the requested field set + fetched_at local timestamp of the fetch + row_count rows the endpoint returned + status ok | empty | failed + schema_version cache schema version + source_version optional tushare/source version string + +Only ``ok`` / ``empty`` rows count as COVERAGE (a failed fetch leaves the range +uncovered so a later run retries). The ledger NEVER stores a token or any +secret-file content — it holds endpoint metadata only. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +import pandas as pd + +from data.cache.intervals import Interval + +LEDGER_COLUMNS: list[str] = [ + "endpoint", + "key_type", + "key", + "start_date", + "end_date", + "fields_hash", + "fetched_at", + "row_count", + "status", + "schema_version", + "source_version", +] + +_COVERING_STATUSES = ("ok", "empty") + + +class CoverageLedger: + """Append-only parquet ledger of fetched (endpoint, key, range) coverage.""" + + def __init__(self, root: str, schema_version: str = "v1") -> None: + self._root = Path(root) + self._schema_version = schema_version + + @property + def path(self) -> Path: + return self._root / "manifest" / "coverage.parquet" + + # -- read --------------------------------------------------------------- # + def read(self) -> pd.DataFrame: + """Return the full ledger (empty, correctly-typed frame if absent).""" + if not self.path.exists(): + return pd.DataFrame(columns=LEDGER_COLUMNS) + return pd.read_parquet(self.path) + + def covered_intervals(self, endpoint: str, key: str) -> list[Interval]: + """Closed date intervals already covered for ``(endpoint, key)``. + + Only ``ok`` / ``empty`` rows count (a ``failed`` fetch is not coverage). + Snapshot rows (NaT range) are ignored here — this is the dense-endpoint + planner's view (P4-1 only uses dense endpoints). + """ + ledger = self.read() + if ledger.empty: + return [] + mask = ( + (ledger["endpoint"] == endpoint) + & (ledger["key"] == key) + & (ledger["status"].isin(_COVERING_STATUSES)) + & ledger["start_date"].notna() + & ledger["end_date"].notna() + ) + sub = ledger[mask] + return [ + (pd.Timestamp(s), pd.Timestamp(e)) + for s, e in zip(sub["start_date"], sub["end_date"]) + ] + + # -- write -------------------------------------------------------------- # + def record( + self, + *, + endpoint: str, + key_type: str, + key: str, + start_date, + end_date, + fields_hash: str, + row_count: int, + status: str, + fetched_at: pd.Timestamp, + source_version: str | None = None, + ) -> None: + """Append one coverage row (atomic). No secret is ever written here.""" + row = { + "endpoint": endpoint, + "key_type": key_type, + "key": str(key), + "start_date": pd.Timestamp(start_date) if start_date is not None else pd.NaT, + "end_date": pd.Timestamp(end_date) if end_date is not None else pd.NaT, + "fields_hash": str(fields_hash), + "fetched_at": pd.Timestamp(fetched_at), + "row_count": int(row_count), + "status": str(status), + "schema_version": self._schema_version, + "source_version": source_version, + } + existing = self.read() + new_row = pd.DataFrame([row]) + # Avoid concat-with-empty (it warns and can shift dtypes): the first + # record IS the new row; later records append to a non-empty ledger. + combined = new_row if existing.empty else pd.concat( + [existing, new_row], ignore_index=True + ) + # enforce column order / presence + combined = combined.reindex(columns=LEDGER_COLUMNS) + self.path.parent.mkdir(parents=True, exist_ok=True) + tmp = self.path.with_suffix(".parquet.tmp") + combined.to_parquet(tmp, engine="pyarrow", index=False) + os.replace(tmp, self.path) diff --git a/data/cache/intervals.py b/data/cache/intervals.py new file mode 100644 index 0000000..74f5ca3 --- /dev/null +++ b/data/cache/intervals.py @@ -0,0 +1,76 @@ +"""Closed-interval day algebra for the cache's missing-range planning (P4-1). + +Coverage is tracked as a set of CLOSED calendar-date intervals ``[start, end]`` +(day resolution). The read-through planner needs two operations: + + * ``merge_intervals`` — collapse overlapping / day-adjacent intervals so the + covered set is canonical; + * ``subtract_intervals`` — given a requested ``[start, end]`` and the covered + set, return the uncovered sub-intervals (the gaps to fetch). + +Calendar (not trading-day) granularity is deliberate: a fetched ``[s, e]`` range +COVERS that whole calendar span even on days the endpoint legitimately returned +no row (not listed / suspended / holiday). Tracking coverage by calendar range — +not by row presence — is exactly what lets the cache distinguish "missing because +not fetched" from "missing because the source has no row" (the coverage-ledger +rationale). + +Pure functions; one day == ``pd.Timedelta(days=1)``. +""" + +from __future__ import annotations + +import pandas as pd + +_ONE_DAY = pd.Timedelta(days=1) + +Interval = tuple[pd.Timestamp, pd.Timestamp] + + +def _norm(ts) -> pd.Timestamp: + return pd.Timestamp(ts).normalize() + + +def merge_intervals(intervals: list[Interval]) -> list[Interval]: + """Sort and merge overlapping or day-adjacent closed intervals.""" + cleaned = [(_norm(s), _norm(e)) for s, e in intervals if _norm(s) <= _norm(e)] + if not cleaned: + return [] + cleaned.sort() + merged: list[Interval] = [cleaned[0]] + for start, end in cleaned[1:]: + last_start, last_end = merged[-1] + # adjacent (gap of exactly one day) intervals merge into one. + if start <= last_end + _ONE_DAY: + merged[-1] = (last_start, max(last_end, end)) + else: + merged.append((start, end)) + return merged + + +def subtract_intervals( + start, end, covered: list[Interval] +) -> list[Interval]: + """Return the closed sub-intervals of ``[start, end]`` NOT in ``covered``. + + Both endpoints are inclusive. An empty result means the request is fully + covered (a repeated run fetches nothing). + """ + req_start, req_end = _norm(start), _norm(end) + if req_start > req_end: + return [] + gaps: list[Interval] = [] + cursor = req_start + for cov_start, cov_end in merge_intervals(covered): + if cov_end < cursor: + continue + if cov_start > req_end: + break + if cov_start > cursor: + gaps.append((cursor, min(cov_start - _ONE_DAY, req_end))) + cursor = max(cursor, cov_end + _ONE_DAY) + if cursor > req_end: + break + if cursor <= req_end: + gaps.append((cursor, req_end)) + return gaps diff --git a/data/cache/parquet_store.py b/data/cache/parquet_store.py new file mode 100644 index 0000000..2d26aa4 --- /dev/null +++ b/data/cache/parquet_store.py @@ -0,0 +1,125 @@ +"""Per-symbol parquet store for raw endpoint rows (P4-1). + +One parquet file per ``(endpoint, symbol)`` under a ``symbol_prefix`` shard: + + //symbol_prefix=000/000001.SZ.parquet + +Why per-symbol files: the feed fetches per symbol, the natural key is +``(symbol, date)``, and a per-symbol file makes upsert-by-key a trivial local +read-merge-write with no cross-symbol contention. ``symbol_prefix`` (the first +three characters) shards the directory so it never holds thousands of files. + +Writes are ATOMIC (write ``*.tmp`` then ``os.replace``) and idempotent: an +upsert drops duplicates by the endpoint's natural key, keeping the latest +fetched row, so re-fetching a recent tail never doubles a ``(symbol, date)``. +A simple per-(endpoint, symbol) lock file guards the read-modify-write. + +The store holds RAW rows only and never sees a token — nothing secret is ever +written here. +""" + +from __future__ import annotations + +import os +import time +from contextlib import contextmanager +from pathlib import Path + +import pandas as pd + + +def _symbol_prefix(symbol: str) -> str: + """Directory shard for a symbol (its first 3 chars, or the whole symbol).""" + s = str(symbol) + return s[:3] if len(s) >= 3 else s + + +class CacheParquetStore: + """Persist raw endpoint rows as per-(endpoint, symbol) parquet files.""" + + def __init__(self, root: str) -> None: + self._root = Path(root) + + # -- paths -------------------------------------------------------------- # + def symbol_path(self, endpoint: str, symbol: str) -> Path: + return ( + self._root + / endpoint + / f"symbol_prefix={_symbol_prefix(symbol)}" + / f"{symbol}.parquet" + ) + + def _lock_path(self, endpoint: str, symbol: str) -> Path: + return self._root / ".locks" / f"{endpoint}__{symbol}.lock" + + # -- locking (best-effort, per endpoint+symbol) ------------------------- # + @contextmanager + def _locked(self, endpoint: str, symbol: str, timeout: float = 10.0): + lock = self._lock_path(endpoint, symbol) + lock.parent.mkdir(parents=True, exist_ok=True) + deadline = time.monotonic() + timeout + fd = None + while True: + try: + fd = os.open(str(lock), os.O_CREAT | os.O_EXCL | os.O_WRONLY) + break + except FileExistsError: + if time.monotonic() > deadline: + # Best-effort: a stale lock must never wedge a single-process + # run; proceed (atomic replace still keeps the file coherent). + break + time.sleep(0.05) + try: + yield + finally: + if fd is not None: + os.close(fd) + try: + lock.unlink() + except FileNotFoundError: + pass + + # -- read / upsert ------------------------------------------------------ # + def read_symbol(self, endpoint: str, symbol: str) -> pd.DataFrame: + """Return all cached raw rows for ``(endpoint, symbol)`` (empty if none).""" + path = self.symbol_path(endpoint, symbol) + if not path.exists(): + return pd.DataFrame() + return pd.read_parquet(path) + + def upsert_symbol( + self, + endpoint: str, + symbol: str, + rows: pd.DataFrame, + key_cols: list[str], + ) -> int: + """Merge ``rows`` into the symbol's parquet, dedup by ``key_cols``. + + Returns the resulting row count. New rows win on a key collision (a + re-fetched recent tail replaces the stale row). Atomic write; never + mutates the caller's frame. + """ + if rows is None or rows.empty: + return self._row_count(endpoint, symbol) + path = self.symbol_path(endpoint, symbol) + with self._locked(endpoint, symbol): + existing = self.read_symbol(endpoint, symbol) + if existing.empty: + combined = rows.copy() + else: + # new rows last so keep="last" lets a refetch overwrite. + combined = pd.concat([existing, rows], ignore_index=True) + combined = combined.drop_duplicates(subset=key_cols, keep="last") + combined = combined.sort_values(key_cols).reset_index(drop=True) + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(".parquet.tmp") + combined.to_parquet(tmp, engine="pyarrow", index=False) + os.replace(tmp, path) + return len(combined) + + def _row_count(self, endpoint: str, symbol: str) -> int: + path = self.symbol_path(endpoint, symbol) + if not path.exists(): + return 0 + return len(pd.read_parquet(path, columns=[])) diff --git a/data/cache/tushare_cache.py b/data/cache/tushare_cache.py new file mode 100644 index 0000000..818643a --- /dev/null +++ b/data/cache/tushare_cache.py @@ -0,0 +1,212 @@ +"""Read-through cache for tushare market endpoints (P4-1: daily + adj_factor). + +``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. + +Behaviour the P4-1 acceptance pins down: + * full cache miss -> fetch every symbol's full 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)``. + +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. +""" + +from __future__ import annotations + +import hashlib +import logging +from collections.abc import Callable + +import pandas as pd + +from data.cache.coverage import CoverageLedger +from data.cache.intervals import merge_intervals, subtract_intervals +from data.cache.parquet_store import CacheParquetStore + +_LOGGER = logging.getLogger("data.cache.tushare") + +# endpoint identifiers (also the names accepted in data.cache.force_refresh). +MARKET_DAILY = "market_daily" +ADJ_FACTOR = "adj_factor" + +# canonical-raw column sets stored per endpoint. +_DAILY_COLUMNS = ["date", "symbol", "open", "high", "low", "close", "volume", "amount"] +_ADJ_COLUMNS = ["date", "symbol", "adj_factor"] +_KEY_COLS = ["date", "symbol"] + +# 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"] + + +def _fields_hash(columns: list[str]) -> str: + """Stable short hash of a field set (order-independent).""" + return hashlib.sha1(",".join(sorted(columns)).encode("utf-8")).hexdigest()[:16] + + +def _compact(ts: pd.Timestamp) -> str: + return pd.Timestamp(ts).strftime("%Y%m%d") + + +def _parse_daily(raw: pd.DataFrame | None) -> pd.DataFrame: + """tushare ``daily`` frame -> canonical-raw rows (or empty).""" + if raw is None or len(raw) == 0: + return pd.DataFrame(columns=_DAILY_COLUMNS) + df = raw.rename(columns=_DAILY_RENAME).copy() + df["date"] = pd.to_datetime(df["date"].astype(str), format="%Y%m%d") + df["symbol"] = df["symbol"].astype(str) + for col in _DAILY_COLUMNS: + if col not in df.columns: + df[col] = float("nan") + return df[_DAILY_COLUMNS] + + +def _parse_adj(raw: pd.DataFrame | None) -> pd.DataFrame: + """tushare ``adj_factor`` frame -> canonical-raw rows (or empty).""" + if raw is None or len(raw) == 0: + return pd.DataFrame(columns=_ADJ_COLUMNS) + df = raw.rename(columns=_ADJ_RENAME).copy() + df["date"] = pd.to_datetime(df["date"].astype(str), format="%Y%m%d") + df["symbol"] = df["symbol"].astype(str) + return df[_ADJ_COLUMNS] + + +class TushareCache: + """Endpoint-level read-through cache for ``market_daily`` and ``adj_factor``.""" + + def __init__( + self, + store: CacheParquetStore, + ledger: CoverageLedger, + *, + refresh_recent_days: int = 14, + force_refresh: tuple[str, ...] | list[str] = (), + today: pd.Timestamp | None = None, + clock: Callable[[], pd.Timestamp] | None = None, + source_version: str | None = None, + ) -> None: + self._store = store + self._ledger = ledger + self._refresh_recent_days = int(refresh_recent_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} + + 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 ------------------------------------------- # + 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 + ) + + 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 + ) + + # -- read-through engine ----------------------------------------------- # + def _read_through( + self, + endpoint: str, + symbols: list[str], + start: str, + end: str, + fetch: FetchOne, + parse: Callable[["pd.DataFrame | None"], pd.DataFrame], + columns: list[str], + ) -> pd.DataFrame: + req_start = pd.Timestamp(start).normalize() + req_end = pd.Timestamp(end).normalize() + fields_hash = _fields_hash(columns) + forced = endpoint in self._force_refresh + + out: list[pd.DataFrame] = [] + n_fetches = 0 + n_covered = 0 + for symbol in symbols: + gaps = self._gaps_for(endpoint, symbol, req_start, req_end, forced) + if gaps: + n_fetches += len(gaps) + else: + n_covered += 1 + for gap_start, gap_end in gaps: + self._fetch_gap( + endpoint, symbol, gap_start, gap_end, fetch, parse, fields_hash + ) + cached = self._store.read_symbol(endpoint, symbol) + if not cached.empty: + mask = (cached["date"] >= req_start) & (cached["date"] <= req_end) + hit = cached.loc[mask, columns] + if not hit.empty: + out.append(hit) + _LOGGER.info( + "cache %s: %d symbols, %d gap-fetches, %d fully-covered (api calls=%d)", + endpoint, len(symbols), n_fetches, n_covered, self.fetch_counts[endpoint], + ) + if not out: + return pd.DataFrame(columns=columns) + return pd.concat(out, ignore_index=True) + + def _gaps_for(self, endpoint, symbol, req_start, req_end, forced): + """The uncovered sub-intervals to fetch (+ a forced recent tail).""" + if forced: + return [(req_start, req_end)] + covered = self._ledger.covered_intervals(endpoint, symbol) + gaps = subtract_intervals(req_start, req_end, covered) + recent = self._recent_tail(req_start, req_end) + if recent is not None: + gaps = merge_intervals(gaps + [recent]) + return gaps + + def _recent_tail(self, req_start, req_end): + """Force-refetch the recent tail within ``refresh_recent_days`` of today.""" + if self._refresh_recent_days <= 0: + return None + today = self._today if self._today is not None else self._clock().normalize() + threshold = today - pd.Timedelta(days=self._refresh_recent_days) + if req_end < threshold: + 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): + """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._ledger.record( + endpoint=endpoint, + key_type="symbol", + key=symbol, + start_date=gap_start, + end_date=gap_end, + 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/tushare_feed.py b/data/feed/tushare_feed.py index 25fc323..7a9a33b 100644 --- a/data/feed/tushare_feed.py +++ b/data/feed/tushare_feed.py @@ -69,6 +69,7 @@ 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 @@ -76,6 +77,11 @@ def __init__( self._rate_limit = rate_limit self._max_retries = max(1, int(max_retries)) self._pro = None # lazily built tushare pro client + # P4-1: optional persistent market cache (read-through). None keeps the + # historical direct-fetch behaviour EXACTLY; only an opted-in config + # passes one in. The cache stores RAW rows only; front_adjust still runs + # downstream, unchanged. + self._cache = cache # -- secret handling ---------------------------------------------------- # def _read_token(self) -> str: @@ -132,6 +138,9 @@ def get_bars( raise ValueError("TushareFeed.get_bars requires a non-empty symbol list.") pro = self._client() + if self._cache is not None: + return self._get_bars_cached(pro, symbols, start, end) + start_compact = pd.Timestamp(start).strftime("%Y%m%d") end_compact = pd.Timestamp(end).strftime("%Y%m%d") @@ -155,6 +164,42 @@ def get_bars( combined = pd.concat(frames, ignore_index=True) return normalize_panel(combined) + # -- read-through cache path (P4-1) ------------------------------------- # + def _get_bars_cached(self, pro, symbols, start, end) -> pd.DataFrame: + """Build the canonical panel from the read-through market cache. + + The cache fetches only uncovered date ranges (the second identical run + makes zero ``daily`` / ``adj_factor`` calls); the per-symbol throttle + + retry stay HERE via ``self._call`` closures. The cache returns RAW + canonical rows; the join + select below mirror ``_to_canonical`` so the + result is byte-identical to the direct path before ``front_adjust``. + """ + adj_getter = getattr(pro, "adj_factor", None) + + def fetch_daily(symbol, s_compact, e_compact): + return self._call( + pro.daily, ts_code=symbol, start_date=s_compact, end_date=e_compact + ) + + def fetch_adj(symbol, s_compact, e_compact): + if adj_getter is None: + return None + return self._call( + adj_getter, ts_code=symbol, start_date=s_compact, end_date=e_compact + ) + + daily = self._cache.daily_bars(symbols, start, end, fetch_daily) + if daily.empty: + return self._empty_panel() + adjf = self._cache.adj_factor(symbols, start, end, fetch_adj) + merged = daily.merge(adjf, on=["symbol", "date"], how="left") + if "adj_factor" not in merged.columns: + merged["adj_factor"] = 1.0 + merged["adj_factor"] = merged["adj_factor"].fillna(1.0) + keep = ["date", "symbol", *CORE_COLUMNS] + present = [c for c in keep if c in merged.columns] + return normalize_panel(merged[present]) + # -- mapping helpers ---------------------------------------------------- # def _fetch_adj_factor(self, pro, symbol: str, start_compact: str, end_compact: str): """Fetch the adj_factor series for ``symbol``; tolerate absence.""" diff --git a/qt/config.py b/qt/config.py index 837bb4a..9261ff4 100644 --- a/qt/config.py +++ b/qt/config.py @@ -35,6 +35,43 @@ class ProjectCfg(_Strict): timezone: str = "Asia/Shanghai" +class CacheCfg(_Strict): + """P4-1 persistent endpoint-level raw cache (market_daily + adj_factor). + + 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. + """ + + 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). + refresh_recent_days: int = 14 + # Endpoint names (e.g. "market_daily", "adj_factor") to always refetch in + # full, ignoring coverage — for forcing a clean re-pull of one endpoint. + force_refresh: list[str] = Field(default_factory=list) + + @field_validator("refresh_recent_days") + @classmethod + def _check_recent_days(cls, v: int) -> int: + if v < 0: + raise ValueError( + f"data.cache.refresh_recent_days must be >= 0; got {v}." + ) + return v + + @field_validator("root_dir") + @classmethod + def _check_root_dir(cls, v: str) -> str: + if not v or not str(v).strip(): + raise ValueError("data.cache.root_dir must be a non-empty path.") + return v + + class DataCfg(_Strict): source: Literal["demo", "tushare"] = "demo" freq: str = "D" @@ -43,6 +80,7 @@ class DataCfg(_Strict): external_secret_file: str | None = None tushare_token_key: str = "tushare.token" output_name: str = "daily" + cache: CacheCfg = Field(default_factory=CacheCfg) @field_validator("start", "end", mode="before") @classmethod diff --git a/qt/pipeline.py b/qt/pipeline.py index 8ac0c6c..8986ff6 100644 --- a/qt/pipeline.py +++ b/qt/pipeline.py @@ -535,12 +535,34 @@ def _build_feed(cfg: RootConfig) -> DataFeed: return TushareFeed( secret_file=secret_file, token_key=cfg.data.tushare_token_key, + cache=_build_market_cache(cfg), ) 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). + + 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. + """ + cache_cfg = cfg.data.cache + if not cache_cfg.enabled: + return None + from data.cache import CacheParquetStore, CoverageLedger, TushareCache + + root = cache_cfg.root_dir + return TushareCache( + CacheParquetStore(root), + CoverageLedger(root), + refresh_recent_days=cache_cfg.refresh_recent_days, + force_refresh=tuple(cache_cfg.force_refresh), + ) + + def _build_universe( cfg: RootConfig, logger: logging.Logger ) -> tuple[Universe, list[str]]: diff --git a/tests/test_cache_config.py b/tests/test_cache_config.py new file mode 100644 index 0000000..723cb25 --- /dev/null +++ b/tests/test_cache_config.py @@ -0,0 +1,79 @@ +"""P4-1: cache config validation (network-free). + +Locks the cache config contract: + * ``data.cache`` defaults to disabled (backward compatible — an existing + config runs exactly as before); + * the four configured knobs exist with the documented defaults; + * bad values (negative refresh window, empty root) fail readably; + * every existing config still validates with the new (defaulted) section. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import yaml + +from qt.config import ConfigError, DataCfg, load_config + +_CONFIG_DIR = Path(__file__).resolve().parents[1] / "config" + + +def test_cache_defaults_disabled_and_backward_compatible(): + # A DataCfg built without a cache section gets the disabled default. + cfg = DataCfg(start="2024-01-01", end="2024-06-30") + assert cfg.cache.enabled is False + assert cfg.cache.root_dir == "artifacts/cache/tushare/v1" + assert cfg.cache.refresh_recent_days == 14 + assert cfg.cache.force_refresh == [] + + +def test_cache_enabled_with_overrides(tmp_path): + raw = yaml.safe_load((_CONFIG_DIR / "example.yaml").read_text(encoding="utf-8")) + raw["data"]["cache"] = { + "enabled": True, + "root_dir": "artifacts/cache/tushare/v1", + "refresh_recent_days": 7, + "force_refresh": ["market_daily", "adj_factor"], + } + p = tmp_path / "cache.yaml" + p.write_text(yaml.safe_dump(raw), encoding="utf-8") + cfg = load_config(str(p)) + assert cfg.data.cache.enabled is True + assert cfg.data.cache.refresh_recent_days == 7 + assert cfg.data.cache.force_refresh == ["market_daily", "adj_factor"] + + +def test_cache_rejects_negative_refresh_window(tmp_path): + raw = yaml.safe_load((_CONFIG_DIR / "example.yaml").read_text(encoding="utf-8")) + raw["data"]["cache"] = {"enabled": True, "refresh_recent_days": -1} + p = tmp_path / "bad.yaml" + p.write_text(yaml.safe_dump(raw), encoding="utf-8") + with pytest.raises(ConfigError, match="refresh_recent_days"): + load_config(str(p)) + + +def test_cache_rejects_empty_root_dir(tmp_path): + raw = yaml.safe_load((_CONFIG_DIR / "example.yaml").read_text(encoding="utf-8")) + raw["data"]["cache"] = {"enabled": True, "root_dir": " "} + p = tmp_path / "bad2.yaml" + p.write_text(yaml.safe_dump(raw), encoding="utf-8") + with pytest.raises(ConfigError, match="root_dir"): + load_config(str(p)) + + +def test_cache_rejects_unknown_key(tmp_path): + raw = yaml.safe_load((_CONFIG_DIR / "example.yaml").read_text(encoding="utf-8")) + raw["data"]["cache"] = {"enabled": True, "ttl_days": 3} # not a field + p = tmp_path / "bad3.yaml" + p.write_text(yaml.safe_dump(raw), encoding="utf-8") + with pytest.raises(ConfigError): + load_config(str(p)) + + +@pytest.mark.parametrize("cfg_path", sorted(_CONFIG_DIR.glob("*.yaml"))) +def test_all_existing_configs_still_validate(cfg_path): + cfg = load_config(str(cfg_path)) + # the cache section is present (defaulted) and disabled unless opted in. + assert cfg.data.cache.enabled in (True, False) diff --git a/tests/test_phase0_pipeline.py b/tests/test_phase0_pipeline.py index 7622441..185f604 100644 --- a/tests/test_phase0_pipeline.py +++ b/tests/test_phase0_pipeline.py @@ -196,14 +196,18 @@ def test_phase0_tushare_source_routes_to_tushare_feed( captured: dict[str, object] = {} class _FakeTushareFeed: - def __init__(self, secret_file: str, token_key: str = "tushare.token"): + def __init__(self, secret_file: str, token_key: str = "tushare.token", + cache=None): captured["secret_file"] = secret_file captured["token_key"] = token_key + captured["cache"] = cache # P4-1: None unless data.cache.enabled monkeypatch.setattr(pipeline, "TushareFeed", _FakeTushareFeed) feed = _build_feed(cfg) assert isinstance(feed, _FakeTushareFeed) + # cache disabled by default -> feed built without a cache (unchanged path). + assert captured["cache"] is None assert captured["secret_file"] == str(tmp_path / "fake.config.json") # The demo source must still route to the offline DemoFeed. demo_cfg = load_config( diff --git a/tests/test_tushare_cache_market.py b/tests/test_tushare_cache_market.py new file mode 100644 index 0000000..bbcafc2 --- /dev/null +++ b/tests/test_tushare_cache_market.py @@ -0,0 +1,272 @@ +"""P4-1: market-bars read-through cache (network-free, fake tushare client). + +Pins the P4-1 acceptance: + * full cache miss populates daily + adj_factor cache; + * full cache hit makes ZERO endpoint calls; + * a partial symbol/date gap fetches ONLY the missing range; + * an empty endpoint return still records coverage (no needless refetch); + * a duplicate upsert keeps one row per (symbol, date); + * cached raw + adj_factor output equals the direct full-fetch output after + front_adjust(); + * the cache + ledger contain no token / secret-file content. + +No test hits the network or reads the real token: a fake ``pro`` client with +call counters drives both the cache path and the direct path. +""" + +from __future__ import annotations + +import json + +import pandas as pd + +from data.cache import CacheParquetStore, CoverageLedger, TushareCache +from data.cache.intervals import subtract_intervals +from data.clean.adjust import front_adjust +from data.clean.schema import validate_panel +from data.feed.tushare_feed import TushareFeed + +FAKE_TOKEN = "FAKE_TUSHARE_TOKEN_do_not_leak_0123456789abcdef" +SECRET_PATH_MARKER = "/abs/path/to/.config.json" + + +# --------------------------------------------------------------------------- # +# Fake tushare client: deterministic daily/adj rows + per-endpoint call counts. +# --------------------------------------------------------------------------- # +class FakePro: + """A fake ``pro`` whose ``daily``/``adj_factor`` count calls and return + deterministic rows for the requested compact [start, end] interval.""" + + def __init__(self): + self.daily_calls = 0 + self.adj_calls = 0 + self.daily_ranges: list[tuple[str, str, str]] = [] + + def _trading_days(self, start_compact, end_compact): + s = pd.Timestamp(start_compact) + e = pd.Timestamp(end_compact) + return pd.bdate_range(s, e) + + def daily(self, ts_code, start_date, end_date, **_): + self.daily_calls += 1 + self.daily_ranges.append((ts_code, start_date, end_date)) + days = self._trading_days(start_date, end_date) + if len(days) == 0: + return pd.DataFrame( + columns=["ts_code", "trade_date", "open", "high", "low", + "close", "vol", "amount"] + ) + base = 10.0 + 0.1 * (hash(ts_code) % 7) + rows = [] + for i, d in enumerate(days): + px = base + i # strictly rising raw close + rows.append({ + "ts_code": ts_code, + "trade_date": d.strftime("%Y%m%d"), + "open": px - 0.2, "high": px + 0.3, "low": px - 0.4, + "close": px, "vol": 1000.0 + i, "amount": 1.0e6 + i, + }) + return pd.DataFrame(rows) + + def adj_factor(self, ts_code, start_date, end_date, **_): + self.adj_calls += 1 + days = self._trading_days(start_date, end_date) + if len(days) == 0: + return pd.DataFrame(columns=["ts_code", "trade_date", "adj_factor"]) + # a mid-window split so front_adjust is non-trivial. + return pd.DataFrame({ + "ts_code": [ts_code] * len(days), + "trade_date": [d.strftime("%Y%m%d") for d in days], + "adj_factor": [1.0 if i < len(days) // 2 else 2.0 + for i in range(len(days))], + }) + + +class FakeProEmpty: + """A fake ``pro`` that always returns empty frames (records coverage).""" + + def __init__(self): + self.daily_calls = 0 + self.adj_calls = 0 + + def daily(self, ts_code, start_date, end_date, **_): + self.daily_calls += 1 + return pd.DataFrame( + columns=["ts_code", "trade_date", "open", "high", "low", + "close", "vol", "amount"] + ) + + def adj_factor(self, ts_code, start_date, end_date, **_): + self.adj_calls += 1 + return pd.DataFrame(columns=["ts_code", "trade_date", "adj_factor"]) + + +def _write_fake_config(tmp_path): + cfg = {"tushare": {"token": FAKE_TOKEN}, "secret_path": SECRET_PATH_MARKER} + path = tmp_path / "fake_config.json" + path.write_text(json.dumps(cfg), encoding="utf-8") + return path + + +def _feed_with_cache(tmp_path, monkeypatch, pro, *, force_refresh=(), today=None): + """A TushareFeed wired to a cache rooted in tmp, with ``pro`` injected. + + ``today`` is fixed far in the future-relative-past so refresh_recent_days + never triggers (the windows here are historical).""" + import tushare as ts + + monkeypatch.setattr(ts, "pro_api", lambda token=None: pro) + root = str(tmp_path / "cache") + cache = TushareCache( + CacheParquetStore(root), + CoverageLedger(root), + refresh_recent_days=14, + force_refresh=force_refresh, + today=pd.Timestamp(today) if today else pd.Timestamp("2024-12-31"), + ) + feed = TushareFeed(secret_file=str(_write_fake_config(tmp_path)), cache=cache) + return feed, root + + +# --------------------------------------------------------------------------- # +# interval algebra (the gap planner's core) +# --------------------------------------------------------------------------- # +def test_subtract_intervals_full_partial_none(): + s, e = pd.Timestamp("2024-01-01"), pd.Timestamp("2024-01-31") + # nothing covered -> one gap == the whole request + assert subtract_intervals(s, e, []) == [(s, e)] + # fully covered -> no gap + assert subtract_intervals(s, e, [(s, e)]) == [] + # left covered -> right gap only + gaps = subtract_intervals(s, e, [(pd.Timestamp("2024-01-01"), + pd.Timestamp("2024-01-15"))]) + assert gaps == [(pd.Timestamp("2024-01-16"), pd.Timestamp("2024-01-31"))] + + +# --------------------------------------------------------------------------- # +# full miss / full hit +# --------------------------------------------------------------------------- # +def test_full_miss_then_full_hit(tmp_path, monkeypatch): + pro = FakePro() + feed, root = _feed_with_cache(tmp_path, monkeypatch, pro) + symbols = ["000001.SZ", "000002.SZ"] + + panel1 = feed.get_bars(symbols, "2024-01-01", "2024-01-31") + validate_panel(panel1) + assert pro.daily_calls == 2 and pro.adj_calls == 2 # one per symbol, full miss + # cache files exist + assert (CacheParquetStore(root).symbol_path("market_daily", "000001.SZ")).exists() + assert (CacheParquetStore(root).symbol_path("adj_factor", "000001.SZ")).exists() + + # second identical run: ZERO endpoint calls, identical panel. + panel2 = feed.get_bars(symbols, "2024-01-01", "2024-01-31") + assert pro.daily_calls == 2 and pro.adj_calls == 2 # unchanged + pd.testing.assert_frame_equal(panel1, panel2) + + +# --------------------------------------------------------------------------- # +# partial gap +# --------------------------------------------------------------------------- # +def test_partial_gap_fetches_only_missing_range(tmp_path, monkeypatch): + pro = FakePro() + feed, _ = _feed_with_cache(tmp_path, monkeypatch, pro) + feed.get_bars(["000001.SZ"], "2024-01-01", "2024-01-15") + assert pro.daily_calls == 1 + pro.daily_ranges.clear() + + # extend the window forward: only the new tail [01-16, 01-31] is fetched. + feed.get_bars(["000001.SZ"], "2024-01-01", "2024-01-31") + assert pro.daily_calls == 2 # exactly one more fetch + sym, s, e = pro.daily_ranges[-1] + assert s == "20240116" and e == "20240131" + + +# --------------------------------------------------------------------------- # +# empty endpoint return still records coverage +# --------------------------------------------------------------------------- # +def test_empty_return_records_coverage_no_refetch(tmp_path, monkeypatch): + pro = FakeProEmpty() + feed, root = _feed_with_cache(tmp_path, monkeypatch, pro) + panel = feed.get_bars(["000001.SZ"], "2024-01-01", "2024-01-31") + assert panel.empty # no rows, but no error + assert pro.daily_calls == 1 + + ledger = CoverageLedger(root).read() + daily_rows = ledger[ledger["endpoint"] == "market_daily"] + assert len(daily_rows) == 1 + assert daily_rows.iloc[0]["status"] == "empty" + assert daily_rows.iloc[0]["row_count"] == 0 + + # a second run sees the empty range as covered -> no extra daily call. + feed.get_bars(["000001.SZ"], "2024-01-01", "2024-01-31") + assert pro.daily_calls == 1 + + +# --------------------------------------------------------------------------- # +# duplicate upsert keeps one row per (symbol, date) +# --------------------------------------------------------------------------- # +def test_duplicate_upsert_keeps_one_row_per_key(tmp_path): + store = CacheParquetStore(str(tmp_path / "cache")) + rows = pd.DataFrame({ + "date": pd.to_datetime(["2024-01-01", "2024-01-02"]), + "symbol": ["000001.SZ", "000001.SZ"], + "close": [10.0, 11.0], + }) + store.upsert_symbol("market_daily", "000001.SZ", rows, ["date", "symbol"]) + # re-upsert the SAME keys with new values -> latest wins, still 2 rows. + rows2 = rows.copy() + rows2["close"] = [99.0, 98.0] + n = store.upsert_symbol("market_daily", "000001.SZ", rows2, ["date", "symbol"]) + assert n == 2 + cached = store.read_symbol("market_daily", "000001.SZ") + assert len(cached) == 2 + jan1 = cached.loc[cached["date"] == pd.Timestamp("2024-01-01"), "close"].iloc[0] + assert jan1 == 99.0 # the re-fetched value replaced the old one + + +# --------------------------------------------------------------------------- # +# qfq equivalence: cache path == direct path, after front_adjust +# --------------------------------------------------------------------------- # +def test_cached_equals_direct_after_front_adjust(tmp_path, monkeypatch): + symbols = ["000001.SZ", "000002.SZ"] + start, end = "2024-01-01", "2024-01-31" + + # direct path (no cache) + pro_direct = FakePro() + import tushare as ts + monkeypatch.setattr(ts, "pro_api", lambda token=None: pro_direct) + direct_feed = TushareFeed(secret_file=str(_write_fake_config(tmp_path))) + direct_panel = direct_feed.get_bars(symbols, start, end) + direct_qfq = front_adjust(direct_panel) + + # cache path (fresh fake client, fresh cache) + pro_cached = FakePro() + cache_feed, _ = _feed_with_cache(tmp_path, monkeypatch, pro_cached) + cached_panel = cache_feed.get_bars(symbols, start, end) + cached_qfq = front_adjust(cached_panel) + + pd.testing.assert_frame_equal(direct_panel, cached_panel) + pd.testing.assert_frame_equal(direct_qfq, cached_qfq) + + +# --------------------------------------------------------------------------- # +# no secret leaks into cache files +# --------------------------------------------------------------------------- # +def test_cache_files_contain_no_secret(tmp_path, monkeypatch): + pro = FakePro() + feed, root = _feed_with_cache(tmp_path, monkeypatch, pro) + feed.get_bars(["000001.SZ"], "2024-01-01", "2024-01-31") + + 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 + # the ledger columns carry no token/secret-path fields + 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 From baf234cbdbc8a5bafa5e7b47c528e0f1e2bd4c03 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Sat, 13 Jun 2026 13:15:15 +0800 Subject: [PATCH 2/4] docs: record P4-1 persistent market cache + real smoke results MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - CLAUDE.md / AGENTS.md (verbatim copy): P4-1 progress entry — design (raw-only endpoint cache below feeds, coverage ledger as SoT, gap planning by calendar range, front_adjust unchanged downstream), the cache-disabled backward-compat invariant, and the real phase2 smoke: non-cached reference vs cached cold vs cached warm metrics ALL IDENTICAL (IC 0.0083 / annual -10.19% / maxDD -16.52% / vol 16.59% / sharpe -0.5703 / turnover 1.0818 / cost 1.19%); warm run made ZERO market_daily/adj_factor calls (coverage ledger 68+68 unchanged between cold and warm); wall ref 998s / cold 1025s / warm 734s; cache files + ledger carry no token/secret. Gates updated to 407 passed; roadmap now points at P4-2 (universe/tradability) then P4-3 (factor-support endpoints). - TEST_REPORT.md: P4-1 per-file breakdown (cache config 17 + market cache 7 = 24) + real-smoke entry; total 407. - RUNBOOK.md: Phase 4-1 section (opt-in config, read-through behaviour, cache layout, real smoke commands, market-bars-only scope note). --- AGENTS.md | 15 ++++++++++++-- CLAUDE.md | 15 ++++++++++++-- RUNBOOK.md | 53 ++++++++++++++++++++++++++++++++++++++++++++++++++ TEST_REPORT.md | 27 +++++++++++++++++++------ 4 files changed, 100 insertions(+), 10 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index d8478d4..de85330 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -154,6 +154,17 @@ data → universe → factors(特征) → alpha(合成/预测) → portfolio(+ri - **CSI500 verdict：SUPPORTED**（21 settled vs min 8）：value_ep **+0.0083/+0.0145**（train/test）、value_bp **+0.0230/+0.0127**、volatility_20 **−0.0350/−0.0272**——三假设双子期全保持。**且衰减更小**：CSI500 test 子期量级（0.0127~0.0272）明显高于 SSE50/CSI300 holdout 的后段（0.003~0.016）——value/低波信号在中盘股上更强，**P3-7 结论泛化成立（GENERALIZES，未减弱）**。 - combo_ic_weighted CSI500 test IC 全 4 组正（0.0243/0.0294/0.0286/0.0285，为三个 holdout cells 最高）；**组合净值 CSI500 base 全组正且 4× 高成本仍全正**（trio +17.80%→+10.57% / full_pack +7.83%→+2.95% / value_lowvol +4.19%→+1.14% / liq +4.02%→+0.84%）——首个全成本阶梯为正的 cell。⚠️ 仍然诚实：trio +17.80% 又是组间排名跨 cell 翻转的例证（中盘动量 regime）；单窗口 ~21 调仓小样本；**非收益声明**。 - **报告标题配置化**（`output.subset_report_title`，沿 `subset_report_name` 先例）：P3-8 报告 H1 自报 study 名「Phase 3-8 — CSI500 Independent Generalization Check」而非机器默认 P3-7 标签（review 同类 stale-wording 问题修复，测试断言首行非 P3-7；P3-6/P3-7 配置不设此项、保持 sample-aware 默认，回归锁定）；文案/分区检查全过（无 stale P3-6 措辞、verdict 节只含独立 cells）；secret scan 报告+日志 0 处；demo 0.96/0.84 不变。 -- ✅ 质量门：`pytest` **383 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）；`ruff` clean；`validate-config`（全部 11 配置）+ `run-phase0`（demo）均 OK。 +- ✅ **Phase 4-1 持久化 Tushare 行情缓存**（分支 `data-persistent-market-cache`，daily + adj_factor）：feed 之下的 **endpoint 级 raw 缓存**——真实 run 不再每次重抓全量日线+复权因子。**默认 disabled（向后兼容，旧配置行为一字不变）**；opt-in 后行情走 read-through，只有未覆盖的日期区间打 API。 + - **缓存只存 raw**（未复权 OHLCV/amount + 原始 adj_factor），绝不存 qfq、绝不存任何 secret；`front_adjust` 仍在内存跑，公式/时机零改动；`PanelStore` 仍是 per-run artifact，**不是**缓存 SoT。 + - `data/cache/`：`intervals`（闭区间日历算法做 gap 规划——按"日历区间是否抓过"而非"行是否存在"判定，正确区分"未抓"vs"源本无行"）/ `parquet_store`（按 `(endpoint,symbol)` 分文件存 `symbol_prefix` 分片，原子 upsert 按 `(date,symbol)` 去重，latest 胜）/ `coverage`（append-only ledger，11 列含 status ok/empty/failed，**只 ok/empty 算覆盖**，failed 留待重试）/ `tushare_cache`（read-through：只抓 gap、upsert、记 coverage 含空返回、再从缓存读全区间；`refresh_recent_days` 重抓近端 tail；`force_refresh` 整端点重拉；per-run 抓取计数 stats 日志）。 + - `TushareFeed.get_bars`：`cache=None` → 旧直抓路径**逐字不变**；`cache` 注入 → read-through 后用**与直抓完全相同的 join+select**，故 `front_adjust` 前面板逐字节一致（qfq 等价单测锁定）；per-symbol 限流/重试仍留在 feed 的 `_call` 闭包里。`_build_market_cache` 仅在 `data.cache.enabled` 时接线。 + - **配置**：`data.cache`（enabled=False / root_dir=`artifacts/cache/tushare/v1`（gitignored）/ refresh_recent_days=14 / force_refresh=[]）；校验拒绝负刷新窗/空 root/未知键；全部旧配置仍 validate。 + - **真实 smoke（phase2 baseline；非缓存参照 ref / 缓存冷 cold / 缓存暖 warm 三轮）**： + - **三轮 report 指标完全一致** ✓（IC 0.0083 / annual −10.19% / maxDD −16.52% / vol 16.59% / sharpe −0.5703 / turnover 1.0818 / cost 1.19%，REF==COLD==WARM）——qfq 等价单测已证 cached==direct 数据层逐字节一致，真实 run 三轮一致是端到端实证。 + - **warm 轮零市场端点调用**：cold 后 coverage ledger = 68 market_daily + 68 adj_factor 行（68 成分全 ok）；warm 后**仍 68+68 不变**（零新 coverage 行 = 零 gap-fetch = 零 daily/adj 调用）。 + - wall：ref 998s / cold 1025s / warm **734s**（暖跑省 ~290s = 行情抓取部分；index_weight/财务/covariates 仍 live——P4-1 只缓存行情，诚实标注）。 + - secret scan：缓存 parquet + ledger 0 处 token / `.config.json`；ledger 列只有端点元数据。 + - **不变量守住**：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` **407 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=24）；`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 起不再互覆盖）。 -- 路线图下一步：更长 holdout 积累（2026-06 之后滚动复检）/ 中证1000 或全市场扩展 / 成本模型细化（印花税卖侧不对称、冲击成本）,或分钟级（architecture.html §11）。 +- 路线图下一步：**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）。 diff --git a/CLAUDE.md b/CLAUDE.md index d8478d4..de85330 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -154,6 +154,17 @@ data → universe → factors(特征) → alpha(合成/预测) → portfolio(+ri - **CSI500 verdict：SUPPORTED**（21 settled vs min 8）：value_ep **+0.0083/+0.0145**（train/test）、value_bp **+0.0230/+0.0127**、volatility_20 **−0.0350/−0.0272**——三假设双子期全保持。**且衰减更小**：CSI500 test 子期量级（0.0127~0.0272）明显高于 SSE50/CSI300 holdout 的后段（0.003~0.016）——value/低波信号在中盘股上更强，**P3-7 结论泛化成立（GENERALIZES，未减弱）**。 - combo_ic_weighted CSI500 test IC 全 4 组正（0.0243/0.0294/0.0286/0.0285，为三个 holdout cells 最高）；**组合净值 CSI500 base 全组正且 4× 高成本仍全正**（trio +17.80%→+10.57% / full_pack +7.83%→+2.95% / value_lowvol +4.19%→+1.14% / liq +4.02%→+0.84%）——首个全成本阶梯为正的 cell。⚠️ 仍然诚实：trio +17.80% 又是组间排名跨 cell 翻转的例证（中盘动量 regime）；单窗口 ~21 调仓小样本；**非收益声明**。 - **报告标题配置化**（`output.subset_report_title`，沿 `subset_report_name` 先例）：P3-8 报告 H1 自报 study 名「Phase 3-8 — CSI500 Independent Generalization Check」而非机器默认 P3-7 标签（review 同类 stale-wording 问题修复，测试断言首行非 P3-7；P3-6/P3-7 配置不设此项、保持 sample-aware 默认，回归锁定）；文案/分区检查全过（无 stale P3-6 措辞、verdict 节只含独立 cells）；secret scan 报告+日志 0 处；demo 0.96/0.84 不变。 -- ✅ 质量门：`pytest` **383 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）；`ruff` clean；`validate-config`（全部 11 配置）+ `run-phase0`（demo）均 OK。 +- ✅ **Phase 4-1 持久化 Tushare 行情缓存**（分支 `data-persistent-market-cache`，daily + adj_factor）：feed 之下的 **endpoint 级 raw 缓存**——真实 run 不再每次重抓全量日线+复权因子。**默认 disabled（向后兼容，旧配置行为一字不变）**；opt-in 后行情走 read-through，只有未覆盖的日期区间打 API。 + - **缓存只存 raw**（未复权 OHLCV/amount + 原始 adj_factor），绝不存 qfq、绝不存任何 secret；`front_adjust` 仍在内存跑，公式/时机零改动；`PanelStore` 仍是 per-run artifact，**不是**缓存 SoT。 + - `data/cache/`：`intervals`（闭区间日历算法做 gap 规划——按"日历区间是否抓过"而非"行是否存在"判定，正确区分"未抓"vs"源本无行"）/ `parquet_store`（按 `(endpoint,symbol)` 分文件存 `symbol_prefix` 分片，原子 upsert 按 `(date,symbol)` 去重，latest 胜）/ `coverage`（append-only ledger，11 列含 status ok/empty/failed，**只 ok/empty 算覆盖**，failed 留待重试）/ `tushare_cache`（read-through：只抓 gap、upsert、记 coverage 含空返回、再从缓存读全区间；`refresh_recent_days` 重抓近端 tail；`force_refresh` 整端点重拉；per-run 抓取计数 stats 日志）。 + - `TushareFeed.get_bars`：`cache=None` → 旧直抓路径**逐字不变**；`cache` 注入 → read-through 后用**与直抓完全相同的 join+select**，故 `front_adjust` 前面板逐字节一致（qfq 等价单测锁定）；per-symbol 限流/重试仍留在 feed 的 `_call` 闭包里。`_build_market_cache` 仅在 `data.cache.enabled` 时接线。 + - **配置**：`data.cache`（enabled=False / root_dir=`artifacts/cache/tushare/v1`（gitignored）/ refresh_recent_days=14 / force_refresh=[]）；校验拒绝负刷新窗/空 root/未知键；全部旧配置仍 validate。 + - **真实 smoke（phase2 baseline；非缓存参照 ref / 缓存冷 cold / 缓存暖 warm 三轮）**： + - **三轮 report 指标完全一致** ✓（IC 0.0083 / annual −10.19% / maxDD −16.52% / vol 16.59% / sharpe −0.5703 / turnover 1.0818 / cost 1.19%，REF==COLD==WARM）——qfq 等价单测已证 cached==direct 数据层逐字节一致，真实 run 三轮一致是端到端实证。 + - **warm 轮零市场端点调用**：cold 后 coverage ledger = 68 market_daily + 68 adj_factor 行（68 成分全 ok）；warm 后**仍 68+68 不变**（零新 coverage 行 = 零 gap-fetch = 零 daily/adj 调用）。 + - wall：ref 998s / cold 1025s / warm **734s**（暖跑省 ~290s = 行情抓取部分；index_weight/财务/covariates 仍 live——P4-1 只缓存行情，诚实标注）。 + - secret scan：缓存 parquet + ledger 0 处 token / `.config.json`；ledger 列只有端点元数据。 + - **不变量守住**：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` **407 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=24）；`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 起不再互覆盖）。 -- 路线图下一步：更长 holdout 积累（2026-06 之后滚动复检）/ 中证1000 或全市场扩展 / 成本模型细化（印花税卖侧不对称、冲击成本）,或分钟级（architecture.html §11）。 +- 路线图下一步：**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）。 diff --git a/RUNBOOK.md b/RUNBOOK.md index 809ccf4..6a3373f 100644 --- a/RUNBOOK.md +++ b/RUNBOOK.md @@ -497,6 +497,59 @@ Documented by `config/phase3_real_csi500_generalization.yaml`. sample-aware default (P3-7 independent / P3-6 post-hoc), locked by tests. CSI500|2022-2024 is skip_cells-listed (runtime budget) and disclosed. +## Phase 4-1 — persistent Tushare market-data cache (daily + adj_factor) + +A persistent endpoint-level RAW cache below the feeds so real runs stop +refetching full daily bars + adj_factor every time. **Disabled by default** +(backward compatible — an existing config runs exactly as before). The cache +stores RAW rows only (unadjusted OHLCV/amount, raw adj_factor); `front_adjust` +still runs in memory downstream, unchanged. `PanelStore` stays a per-run +artifact, NOT the cache source of truth. + +Opt in via `data.cache`: + +```yaml +data: + cache: + enabled: true + root_dir: artifacts/cache/tushare/v1 # gitignored + refresh_recent_days: 14 # refetch a recent tail near today + force_refresh: [] # e.g. ["market_daily","adj_factor"] +``` + +Read-through behaviour (`TushareFeed.get_bars`, cache enabled): + +- the coverage ledger (`/manifest/coverage.parquet`) records which + `(endpoint, symbol, [start,end])` ranges were fetched (status ok/empty/failed) + — so the planner fetches ONLY uncovered date gaps; +- a second identical run over a historical window makes **zero** `daily` / + `adj_factor` API calls; +- a partial window extension fetches only the new tail; +- raw rows are upserted per `(symbol, date)` (latest wins; no duplicates); +- the cache + ledger never store a token or secret-file content. + +Cache layout (per-symbol parquet under a symbol_prefix shard): + +```text +artifacts/cache/tushare/v1/ + manifest/coverage.parquet + market_daily/symbol_prefix=600/600519.SH.parquet + adj_factor/symbol_prefix=600/600519.SH.parquet +``` + +Real smoke (small real config; metrics must equal the non-cached run): + +```bash +# populate, then re-run: the second run hits cache for market bars +... -m qt.cli run-phase2-baseline --config config/phase2_real_baseline_cached.yaml +... -m qt.cli run-phase2-baseline --config config/phase2_real_baseline_cached.yaml +``` + +Scope note: P4-1 caches market bars ONLY. `index_weight`, `daily_basic`, +`fina_indicator`, `stk_limit`, `suspend_d`, `namechange`, `stock_basic`, +`index_member_all` are still fetched live (cached in P4-2/P4-3); the warm run +therefore saves the market-bar calls, not those. + ## Quality gate ```bash diff --git a/TEST_REPORT.md b/TEST_REPORT.md index 87088c0..013099c 100644 --- a/TEST_REPORT.md +++ b/TEST_REPORT.md @@ -1,4 +1,4 @@ -# TEST_REPORT — Phase 0 + Phase 1 + Phase 2 + Phase 3 (bias-boundary → execution realism → PIT industry → standard analytics → multi-factor → walk-forward IC alpha → OOS stability → robustness matrix → factor candidates → subset + cost sensitivity → independent validation → CSI500 generalization) +# TEST_REPORT — Phase 0 + Phase 1 + Phase 2 + Phase 3 (bias-boundary → execution realism → PIT industry → standard analytics → multi-factor → walk-forward IC alpha → OOS stability → robustness matrix → factor candidates → subset + cost sensitivity → independent validation → CSI500 generalization → persistent market cache) ## Commands @@ -18,6 +18,7 @@ Run from the repo root with the project python (env `quant_mf`): /home/shaofl/Development/env_tools/envs/quant_mf/bin/python -m qt.cli validate-config --config config/phase3_real_subset_costs.yaml /home/shaofl/Development/env_tools/envs/quant_mf/bin/python -m qt.cli validate-config --config config/phase3_real_independent_validation.yaml /home/shaofl/Development/env_tools/envs/quant_mf/bin/python -m qt.cli validate-config --config config/phase3_real_csi500_generalization.yaml +/home/shaofl/Development/env_tools/envs/quant_mf/bin/python -m qt.cli validate-config --config config/phase2_real_baseline_cached.yaml /home/shaofl/Development/env_tools/envs/quant_mf/bin/python -m qt.cli run-phase0 --config config/example.yaml ``` @@ -25,12 +26,12 @@ Run from the repo root with the project python (env `quant_mf`): | Gate | Command | Result | |---|---|---| -| Unit + integration | `pytest -q` | **383 passed, 0 failed** | +| Unit + integration | `pytest -q` | **407 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`) | exit `0`, prints `OK` | +| 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 = 383). +Counts below are the actual per-file `pytest` numbers (sum = 407). ## Per-file breakdown — Phase 0 core (97) @@ -155,7 +156,15 @@ Counts below are the actual per-file `pytest` numbers (sum = 383). | Test file | Tests | Red-line / feature | |---|---|---| | `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 (P0 + P1 + P2-1..P2-4 + P3-1..P3-8)** | **383** | | +| **Total through P3-8** | **383** | | + +## Per-file breakdown — Phase 4-1 persistent market cache (24) + +| 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` | 7 | 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 | +| **Total (P0 + P1 + P2-1..P2-4 + P3-1..P3-8 + P4-1)** | **407** | | ## Real-data validation (manual, not in CI — TEST-002 keeps the suite network-free) @@ -178,7 +187,13 @@ Counts below are the actual per-file `pytest` numbers (sum = 383). in 2/2 holdout cells** (21 settled rebalances each vs minimum 8; value_ep/value_bp positive and volatility_20 negative in BOTH subperiods of BOTH cells) with visible magnitude attenuation in the later subperiod; - secret scan 0 occurrences in report and log. Two earlier run attempts died + secret scan 0 occurrences in report and log. +- **P4-1** persistent market cache (real smoke = phase2 baseline, non-cached + reference vs cached cold vs cached warm): report metrics IDENTICAL across all + three (the qfq-equivalence unit test proves cached==direct at the data layer); + the cached WARM run makes zero market_daily / adj_factor API calls (coverage + 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. ## Notes From 17867f5be9f91db6ba1026d94a302944d4bdd4b8 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Sat, 13 Jun 2026 13:44:25 +0800 Subject: [PATCH 3/4] feat: surface P4-1 market-cache gap-fetch counts in the run log MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-up (non-blocking): the run log did not directly show the cache hit rate — warm-hit evidence relied on coverage-ledger row counts. Now the run-scoped logger emits the per-endpoint gap-fetch tally so a warm historical rerun is self-evident. - TushareFeed.cache_stats() -> dict | None: the read-through cache's per-endpoint API-call tally (None when caching is disabled). Endpoint counts only — no token, secret path, or per-symbol detail. - pipeline._log_cache_stats(feed, logger): called in _load_panel right after get_bars; logs one line through the existing run-scoped logger when the feed exposes cache_stats() (no-op for DemoFeed / cache off): data cache: market_daily_gap_fetches= adj_factor_gap_fetches= A cold run shows nonzero counts; a warm historical rerun shows 0/0. - No change to cache behaviour, schema, or any factor/alpha/portfolio/ execution/report math; no new endpoints cached. - tests: +4 (feed exposes cold {2,2} then warm {0,0} stats over a shared cache root; cache_stats None without a cache; pipeline logs the exact line; no line when cache absent/disabled) -> 411 passed; ruff clean. --- data/feed/tushare_feed.py | 13 +++++ qt/pipeline.py | 25 ++++++++++ tests/test_tushare_cache_market.py | 76 ++++++++++++++++++++++++++++++ 3 files changed, 114 insertions(+) diff --git a/data/feed/tushare_feed.py b/data/feed/tushare_feed.py index 7a9a33b..7ed1ffe 100644 --- a/data/feed/tushare_feed.py +++ b/data/feed/tushare_feed.py @@ -120,6 +120,19 @@ def _call(self, fn, **kwargs): **kwargs, ) + # -- cache stats -------------------------------------------------------- # + def cache_stats(self) -> dict[str, int] | None: + """Per-endpoint gap-fetch counts since this feed's cache was built. + + ``None`` when caching is disabled (no cache wired in). The counts are + the read-through cache's API-call tally — a warm rerun over a covered + historical window reports zeros. Returns endpoint metadata only; no + token, secret path, or per-symbol detail. + """ + if self._cache is None: + return None + return self._cache.stats() + # -- DataFeed API ------------------------------------------------------- # def get_bars( self, diff --git a/qt/pipeline.py b/qt/pipeline.py index 8986ff6..0963c5d 100644 --- a/qt/pipeline.py +++ b/qt/pipeline.py @@ -622,6 +622,26 @@ def _build_universe( raise ValueError(f"Unsupported universe.type {cfg.universe.type!r}.") +def _log_cache_stats(feed, logger: logging.Logger) -> None: + """Log the market cache's per-endpoint gap-fetch counts (P4-1), 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. + """ + getter = getattr(feed, "cache_stats", None) + if getter is None: + return + 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)), + ) + + def _load_panel( cfg: RootConfig, symbols: list[str], logger: logging.Logger ) -> pd.DataFrame: @@ -630,6 +650,11 @@ def _load_panel( raise ValueError("No symbols to load; the universe produced an empty set.") feed = _build_feed(cfg) 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 " diff --git a/tests/test_tushare_cache_market.py b/tests/test_tushare_cache_market.py index bbcafc2..05acabe 100644 --- a/tests/test_tushare_cache_market.py +++ b/tests/test_tushare_cache_market.py @@ -164,6 +164,82 @@ def test_full_miss_then_full_hit(tmp_path, monkeypatch): pd.testing.assert_frame_equal(panel1, panel2) +# --------------------------------------------------------------------------- # +# feed exposes cache stats (review follow-up: visible warm-hit evidence) +# --------------------------------------------------------------------------- # +def test_feed_exposes_cache_stats_cold_then_warm(tmp_path, monkeypatch): + pro = FakePro() + feed, root = _feed_with_cache(tmp_path, monkeypatch, pro) + symbols = ["000001.SZ", "000002.SZ"] + + # cold run: gap-fetches == one per symbol per endpoint. + feed.get_bars(symbols, "2024-01-01", "2024-01-31") + assert feed.cache_stats() == {"market_daily": 2, "adj_factor": 2} + + # warm rerun on a fresh feed + fresh cache over the SAME root -> 0 gap-fetches. + pro2 = FakePro() + import tushare as ts + monkeypatch.setattr(ts, "pro_api", lambda token=None: pro2) + warm_cache = TushareCache( + CacheParquetStore(root), CoverageLedger(root), + refresh_recent_days=14, today=pd.Timestamp("2024-12-31"), + ) + 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} + assert pro2.daily_calls == 0 and pro2.adj_calls == 0 + + +def test_feed_cache_stats_none_without_cache(tmp_path, monkeypatch): + pro = FakePro() + import tushare as ts + monkeypatch.setattr(ts, "pro_api", lambda token=None: pro) + feed = TushareFeed(secret_file=str(_write_fake_config(tmp_path))) # no cache + feed.get_bars(["000001.SZ"], "2024-01-01", "2024-01-05") + assert feed.cache_stats() is None + + +# --------------------------------------------------------------------------- # +# pipeline logs the cache stats line through the run-scoped logger +# --------------------------------------------------------------------------- # +def test_load_panel_logs_cache_stats_line(caplog): + import logging + + from qt.pipeline import _log_cache_stats + + class _FeedWithStats: + def cache_stats(self): + return {"market_daily": 3, "adj_factor": 0} + + logger = logging.getLogger("qt.test_cache_stats") + with caplog.at_level(logging.INFO, logger="qt.test_cache_stats"): + _log_cache_stats(_FeedWithStats(), logger) + assert ( + "data cache: market_daily_gap_fetches=3 adj_factor_gap_fetches=0" + in caplog.text + ) + + +def test_load_panel_no_cache_line_when_cache_absent(caplog): + import logging + + from qt.pipeline import _log_cache_stats + + class _DemoFeedNoStats: # DemoFeed has no cache_stats attribute + pass + + class _FeedCacheDisabled: + def cache_stats(self): + return None # cache wired off + + logger = logging.getLogger("qt.test_cache_stats_off") + with caplog.at_level(logging.INFO, logger="qt.test_cache_stats_off"): + _log_cache_stats(_DemoFeedNoStats(), logger) + _log_cache_stats(_FeedCacheDisabled(), logger) + assert "data cache:" not in caplog.text + + # --------------------------------------------------------------------------- # # partial gap # --------------------------------------------------------------------------- # From 8463c4ddf16857cb5e9c514cb5c030d4c015dc1d Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Sat, 13 Jun 2026 13:56:52 +0800 Subject: [PATCH 4/4] docs: note the run-log cache-stats line + bump gate count to 411 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Records the P4-1 follow-up: TushareFeed.cache_stats() + the run-log line `data cache: market_daily_gap_fetches=N adj_factor_gap_fetches=M` (cold nonzero, warm historical rerun 0/0 — confirmed in run_phase2_baseline.log, secret scan 0). Gate count 407 -> 411 across CLAUDE.md / AGENTS.md / TEST_REPORT.md; RUNBOOK P4-1 read-through bullets mention the visible-hit-rate line. --- AGENTS.md | 5 +++-- CLAUDE.md | 5 +++-- RUNBOOK.md | 5 ++++- TEST_REPORT.md | 8 ++++---- 4 files changed, 14 insertions(+), 9 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index de85330..f46f030 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -164,7 +164,8 @@ data → universe → factors(特征) → alpha(合成/预测) → portfolio(+ri - **warm 轮零市场端点调用**：cold 后 coverage ledger = 68 market_daily + 68 adj_factor 行（68 成分全 ok）；warm 后**仍 68+68 不变**（零新 coverage 行 = 零 gap-fetch = 零 daily/adj 调用）。 - wall：ref 998s / cold 1025s / warm **734s**（暖跑省 ~290s = 行情抓取部分；index_weight/财务/covariates 仍 live——P4-1 只缓存行情，诚实标注）。 - secret scan：缓存 parquet + ledger 0 处 token / `.config.json`；ledger 列只有端点元数据。 - - **不变量守住**：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` **407 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=24）；`ruff` clean；`validate-config`（全部 12 配置）+ `run-phase0`（demo）均 OK。 + - **缓存命中直接可见（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。 - ⚠️ 剩余（已显式披露）：日线 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）。 diff --git a/CLAUDE.md b/CLAUDE.md index de85330..f46f030 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -164,7 +164,8 @@ data → universe → factors(特征) → alpha(合成/预测) → portfolio(+ri - **warm 轮零市场端点调用**：cold 后 coverage ledger = 68 market_daily + 68 adj_factor 行（68 成分全 ok）；warm 后**仍 68+68 不变**（零新 coverage 行 = 零 gap-fetch = 零 daily/adj 调用）。 - wall：ref 998s / cold 1025s / warm **734s**（暖跑省 ~290s = 行情抓取部分；index_weight/财务/covariates 仍 live——P4-1 只缓存行情，诚实标注）。 - secret scan：缓存 parquet + ledger 0 处 token / `.config.json`；ledger 列只有端点元数据。 - - **不变量守住**：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` **407 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=24）；`ruff` clean；`validate-config`（全部 12 配置）+ `run-phase0`（demo）均 OK。 + - **缓存命中直接可见（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。 - ⚠️ 剩余（已显式披露）：日线 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）。 diff --git a/RUNBOOK.md b/RUNBOOK.md index 6a3373f..7675cf9 100644 --- a/RUNBOOK.md +++ b/RUNBOOK.md @@ -526,7 +526,10 @@ Read-through behaviour (`TushareFeed.get_bars`, cache enabled): `adj_factor` API calls; - a partial window extension fetches only the new tail; - raw rows are upserted per `(symbol, date)` (latest wins; no duplicates); -- the cache + ledger never store a token or secret-file content. +- the cache + ledger never store a token or secret-file content; +- the run log shows the hit rate directly — `_load_panel` emits + `data cache: market_daily_gap_fetches= adj_factor_gap_fetches=` + (cold run nonzero; a warm historical rerun shows 0/0). Cache layout (per-symbol parquet under a symbol_prefix shard): diff --git a/TEST_REPORT.md b/TEST_REPORT.md index 013099c..d8d78d0 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` | **407 passed, 0 failed** | +| Unit + integration | `pytest -q` | **411 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 = 407). +Counts below are the actual per-file `pytest` numbers (sum = 411). ## Per-file breakdown — Phase 0 core (97) @@ -163,8 +163,8 @@ Counts below are the actual per-file `pytest` numbers (sum = 407). | 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` | 7 | 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 | -| **Total (P0 + P1 + P2-1..P2-4 + P3-1..P3-8 + P4-1)** | **407** | | +| `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** | | ## Real-data validation (manual, not in CI — TEST-002 keeps the suite network-free)