diff --git a/data/cache/coverage.py b/data/cache/coverage.py index dd226be..467fc7c 100644 --- a/data/cache/coverage.py +++ b/data/cache/coverage.py @@ -158,6 +158,25 @@ def snapshot_fetched_at(self, endpoint: str, key: str) -> pd.Timestamp | None: self._snapshot_memo[memo_key] = result return result + def last_fields_hash(self, endpoint: str) -> str | None: + """``fields_hash`` of the most-recent (by ``fetched_at``) row for ``endpoint``. + + READ-ONLY, additive (D-series schema guard): the stored canonical-column + hash from the last fetch of any key of ``endpoint``, or ``None`` when the + endpoint was never recorded. Reuses the in-process frame cache; does NOT + change ``LEDGER_COLUMNS``, the parquet path, or coverage semantics. + """ + ledger = self._load() + if ledger.empty: + return None + sub = ledger[ledger["endpoint"] == endpoint] + if sub.empty: + return None + value = sub.loc[sub["fetched_at"].idxmax(), "fields_hash"] + if pd.isna(value): # pd.isna(None) is True — covers None / NaN / NaT + return None + return str(value) + # -- write -------------------------------------------------------------- # def _normalize_row(self, row: dict) -> dict: """Normalize one input row to the ledger dtypes/order (no secret fields).""" diff --git a/data/cache/schema_registry.py b/data/cache/schema_registry.py new file mode 100644 index 0000000..2f5422d --- /dev/null +++ b/data/cache/schema_registry.py @@ -0,0 +1,447 @@ +"""Declarative tushare endpoint schema registry + a default-off drift guard. + +THE CORE INSIGHT. Each ``_parse_*`` in :mod:`data.cache.tushare_parsers` returns +a FIXED canonical column set and defensively NaN-fills any canonical column whose +source column is absent (``if col not in df.columns: df[col] = nan``). So the +``fields_hash`` of the STORED canonical columns is IMMUNE to tushare changing its +source columns — the real corruption risk is: tushare removes/renames a SOURCE +column a parser depends on -> the parser silently fills NaN -> the cache stores +NaN with ZERO signal. The highest-value guard therefore checks the RAW INPUT +boundary, BEFORE parsing. + +This module: + * derives canonical info (``canonical_columns`` / ``natural_key`` / + ``expected_canonical_hash``) from :mod:`data.cache.tushare_specs` (SINGLE + SOURCE OF TRUTH — no duplicated canonical column lists), and + * declares, per endpoint, the RAW SOURCE columns each parser depends on, split + into ``required_source_columns`` (drift here is HARD), ``optional_source_ + columns`` (tushare may legitimately omit; absent => WARNING-free, parser + defaults), and ``known_extra_columns`` (documented-but-ignored tushare + response columns, so check #2 does not warn on them). The declarations are + LOCKED both ways (see ``tests/test_schema_registry``): a forward lock (feed + exactly the required set through the REAL parser => no NaN in the required- + derived canonical columns) and an inverse lock (drop any optional column => + the parser does NOT raise and still yields the canonical schema). + +Per-endpoint REQUIRED vs OPTIONAL rule (the HIGH-1 fix). A source column is +REQUIRED only when tushare GUARANTEES it back, i.e. one of: + (a) the parser reads it directly so its absence RAISES (the axis columns that + become ``date`` / ``symbol`` / a natural-key part, plus ``adj_factor`` / + ``start_date`` / ``name`` / ``in_date``); or + (b) it is explicitly requested via a ``fields=`` selector (daily_basic / + fina_indicator request their fields, so tushare returns those columns); or + (c) it is a core column of the endpoint's documented STANDARD response for a + no-``fields=`` endpoint (daily OHLCV+vol+amount, stk_limit up/down_limit, + index_weight weight) — always present, so a missing column is true drift. +A source column is OPTIONAL when the parser DEFENSIVELY fills it AND there is +evidence tushare legitimately omits it / it is not axis-critical: e.g. +index_member_all's ``l1/l2/l3_name`` (the DIRECT feed path guards their absence +at ``data/feed/tushare_covariates.py:155`` — proof tushare omits them), +``out_date`` / ``end_date`` (open-interval ends), ``ts_code`` (parser falls back +to the symbol arg), ``suspend_type`` (query-filter default 'S'), ``list_date`` +(the feed keeps + discloses a stock with no list_date). Optional drift is NOT a +HARD finding — it is benign, so it produces neither a HARD nor (being known) a +check-#2 WARNING. + +The :class:`SchemaGuard` is REPORT-ONLY by default and DEFAULT-OFF in the cache +(``TushareCache(schema_guard=None)`` is byte-identical to before). It never sees +a token or a data value — only endpoint names + column names — so a finding, +log, or summary can never carry a secret. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +import pandas as pd + +from data.cache.tushare_planning import _fields_hash +from data.cache.tushare_specs import ( + ADJ_FACTOR, + ALL_ENDPOINTS, + DAILY_BASIC, + FINA_FIELDS, + FINA_INDICATOR, + INDEX_MEMBER_ALL, + INDEX_WEIGHT, + MARKET_DAILY, + NAMECHANGE, + STK_LIMIT, + STOCK_BASIC, + SUSPEND_D, + _ADJ_COLUMNS, + _DAILY_BASIC_COLUMNS, + _DAILY_BASIC_KEY, + _DAILY_COLUMNS, + _FINA_COLUMNS, + _FINA_KEY, + _INDEX_MEMBER_COLUMNS, + _INDEX_MEMBER_KEY, + _INDEX_WEIGHT_COLUMNS, + _INDEX_WEIGHT_KEY, + _KEY_COLS, + _NAMECHANGE_COLUMNS, + _NAMECHANGE_KEY, + _STK_LIMIT_COLUMNS, + _STK_LIMIT_KEY, + _STOCK_BASIC_COLUMNS, + _STOCK_BASIC_KEY, + _SUSPEND_COLUMNS, + _SUSPEND_KEY, +) + +# guard modes. +REPORT_ONLY = "report_only" +STRICT = "strict" + +# finding severities. +HARD = "hard" +WARNING = "warning" + +# check ids (stable, secret-free identifiers used in findings + tests). +CHECK_MISSING_REQUIRED = "missing_required_source_columns" +CHECK_UNKNOWN_EXTRA = "unknown_extra_source_columns" +CHECK_CANONICAL_MISMATCH = "canonical_columns_mismatch" +CHECK_HASH_CHANGED = "stored_schema_hash_changed" + + +@dataclass(frozen=True) +class EndpointSchema: + """The raw-input + canonical-output contract for one cached tushare endpoint. + + ``required_source_columns`` are the RAW columns tushare GUARANTEES and the + parser depends on (their absence is true drift / the silent-NaN risk). + ``optional_source_columns`` are columns the parser reads but tolerates as + absent (defaults / arg fallbacks) and tushare may legitimately omit — so + their absence is neither HARD nor a check-#2 warning. ``known_extra_columns`` + is a best-effort catalogue of documented-but-IGNORED tushare response columns + (the no-``fields=`` endpoints return more than the parser keeps); check #2 + warns only on a column OUTSIDE required ∪ optional ∪ known_extra, i.e. a + genuinely new/uncatalogued column. ``canonical_columns`` / ``natural_key`` + come straight from the spec constants (no duplication). ``expected_canonical_ + hash`` uses the SAME hashing as the coverage ledger's ``fields_hash`` so + check #4 can compare them. + """ + + endpoint: str + required_source_columns: frozenset[str] + canonical_columns: tuple[str, ...] + natural_key: tuple[str, ...] + optional_source_columns: frozenset[str] = field(default_factory=frozenset) + known_extra_columns: frozenset[str] = field(default_factory=frozenset) + + @property + def known_source_columns(self) -> frozenset[str]: + """Required + optional + catalogued extras — columns NOT to flag as #2.""" + return ( + self.required_source_columns + | self.optional_source_columns + | self.known_extra_columns + ) + + @property + def expected_canonical_hash(self) -> str: + """Stable hash of the canonical columns (matches the ledger fields_hash).""" + return _fields_hash(list(self.canonical_columns)) + + +@dataclass(frozen=True) +class SchemaDriftFinding: + """One schema-drift finding. Carries ONLY endpoint + column names (no secret). + + ``columns`` is the sorted tuple of offending column names (missing required / + unknown extra / canonical diff); empty for the hash-change check. ``detail`` is + a human-readable, secret-free message (endpoint + column names + the short + column-set hashes only — never a token, path, or data value). + """ + + endpoint: str + check: str + severity: str + detail: str + columns: tuple[str, ...] = () + + +# --------------------------------------------------------------------------- # +# Registry. canonical_columns / natural_key are the spec constants verbatim. +# required / optional / known_extra are declared per the REQUIRED-vs-OPTIONAL rule +# in the module docstring (LOCKED both ways by tests). known_extra is a best-effort +# catalogue of documented tushare response columns the parser ignores — a brand-new +# tushare column simply yields one WARNING until it is catalogued here. +# --------------------------------------------------------------------------- # +REGISTRY: dict[str, EndpointSchema] = { + # _parse_daily (no fields=): ts_code->symbol, trade_date->date (axis, raise); + # open/high/low/close/vol/amount are tushare daily's guaranteed core (NaN-filled + # if a column vanished = the silent-NaN drift we want HARD). + MARKET_DAILY: EndpointSchema( + endpoint=MARKET_DAILY, + required_source_columns=frozenset( + {"ts_code", "trade_date", "open", "high", "low", "close", "vol", "amount"} + ), + canonical_columns=tuple(_DAILY_COLUMNS), + natural_key=tuple(_KEY_COLS), + known_extra_columns=frozenset({"pre_close", "change", "pct_chg"}), + ), + # _parse_adj (no fields=): adj_factor selected directly with NO defensive fill — + # absence raises; ts_code/trade_date are the axis. + ADJ_FACTOR: EndpointSchema( + endpoint=ADJ_FACTOR, + required_source_columns=frozenset({"ts_code", "trade_date", "adj_factor"}), + canonical_columns=tuple(_ADJ_COLUMNS), + natural_key=tuple(_KEY_COLS), + ), + # _parse_index_weight (no fields=): con_code->symbol, trade_date->date (axis); + # weight is the endpoint's core; index_code is INJECTED from the arg (raw's is + # overwritten / ignored => catalogued extra, never read). + INDEX_WEIGHT: EndpointSchema( + endpoint=INDEX_WEIGHT, + required_source_columns=frozenset({"con_code", "trade_date", "weight"}), + canonical_columns=tuple(_INDEX_WEIGHT_COLUMNS), + natural_key=tuple(_INDEX_WEIGHT_KEY), + known_extra_columns=frozenset({"index_code"}), + ), + # _parse_suspend (filter suspend_type='S', no fields=): ts_code->symbol, + # trade_date->date (axis). suspend_type is defensively defaulted to 'S' (matches + # the query filter) => optional; suspend_timing is ignored. + SUSPEND_D: EndpointSchema( + endpoint=SUSPEND_D, + required_source_columns=frozenset({"ts_code", "trade_date"}), + canonical_columns=tuple(_SUSPEND_COLUMNS), + natural_key=tuple(_SUSPEND_KEY), + optional_source_columns=frozenset({"suspend_type"}), + known_extra_columns=frozenset({"suspend_timing"}), + ), + # _parse_namechange (no fields=): start_date + name read directly (absence + # raises). ts_code falls back to the symbol arg, end_date defaults to NaT (open + # interval) => both optional; ann_date / change_reason are ignored. + NAMECHANGE: EndpointSchema( + endpoint=NAMECHANGE, + required_source_columns=frozenset({"start_date", "name"}), + canonical_columns=tuple(_NAMECHANGE_COLUMNS), + natural_key=tuple(_NAMECHANGE_KEY), + optional_source_columns=frozenset({"ts_code", "end_date"}), + known_extra_columns=frozenset({"ann_date", "change_reason"}), + ), + # _parse_stk_limit (no fields=): ts_code->symbol, trade_date->date (axis); + # up_limit/down_limit are the endpoint's core; pre_close is ignored. + STK_LIMIT: EndpointSchema( + endpoint=STK_LIMIT, + required_source_columns=frozenset( + {"ts_code", "trade_date", "up_limit", "down_limit"} + ), + canonical_columns=tuple(_STK_LIMIT_COLUMNS), + natural_key=tuple(_STK_LIMIT_KEY), + known_extra_columns=frozenset({"pre_close"}), + ), + # _parse_stock_basic (fields="ts_code,list_date"): ts_code->symbol read directly + # (raise). list_date is defensively defaulted to None AND the feed keeps + + # discloses a stock with no list_date (proof it is legitimately omitted) => + # optional, not HARD. + STOCK_BASIC: EndpointSchema( + endpoint=STOCK_BASIC, + required_source_columns=frozenset({"ts_code"}), + canonical_columns=tuple(_STOCK_BASIC_COLUMNS), + natural_key=tuple(_STOCK_BASIC_KEY), + optional_source_columns=frozenset({"list_date"}), + ), + # _parse_daily_basic (fields="ts_code,trade_date,pe,pb,total_mv"): every kept + # column is explicitly requested, so tushare guarantees the columns back => + # required (a missing COLUMN is drift; a row-level NaN pe is downstream's job). + DAILY_BASIC: EndpointSchema( + endpoint=DAILY_BASIC, + required_source_columns=frozenset( + {"ts_code", "trade_date", "pe", "pb", "total_mv"} + ), + canonical_columns=tuple(_DAILY_BASIC_COLUMNS), + natural_key=tuple(_DAILY_BASIC_KEY), + ), + # _parse_fina (fields="ts_code,ann_date,end_date,*FINA_FIELDS" superset): + # ts_code->symbol; the coverage axis ``date`` is DERIVED from end_date, and + # ann_date drives the downstream as-of — both are requested so guaranteed back + # => required. All FINA_FIELDS are requested too => required. + FINA_INDICATOR: EndpointSchema( + endpoint=FINA_INDICATOR, + required_source_columns=frozenset( + {"ts_code", "ann_date", "end_date", *FINA_FIELDS} + ), + canonical_columns=tuple(_FINA_COLUMNS), + natural_key=tuple(_FINA_KEY), + ), + # _parse_index_member (no fields=): in_date read directly (absence raises) — the + # ONLY required column. l1/l2/l3_name are defensively None-filled AND the direct + # feed path (tushare_covariates.py:155) guards their absence (proof tushare omits + # level columns) => optional, NEVER HARD. ts_code falls back to the symbol arg, + # out_date defaults to NaT (open membership) => optional. l*_code / name / is_new + # are ignored. + INDEX_MEMBER_ALL: EndpointSchema( + endpoint=INDEX_MEMBER_ALL, + required_source_columns=frozenset({"in_date"}), + canonical_columns=tuple(_INDEX_MEMBER_COLUMNS), + natural_key=tuple(_INDEX_MEMBER_KEY), + optional_source_columns=frozenset( + {"ts_code", "l1_name", "l2_name", "l3_name", "out_date"} + ), + known_extra_columns=frozenset( + {"l1_code", "l2_code", "l3_code", "name", "is_new"} + ), + ), +} + +# every endpoint the cache knows must have a registry entry. Use an explicit raise +# (NOT assert, which `python -O` strips) so a future endpoint can never slip in +# uncovered. +if set(REGISTRY) != set(ALL_ENDPOINTS): + raise RuntimeError( + "schema REGISTRY is out of sync with tushare_specs.ALL_ENDPOINTS: " + f"missing={sorted(set(ALL_ENDPOINTS) - set(REGISTRY))}, " + f"extra={sorted(set(REGISTRY) - set(ALL_ENDPOINTS))}." + ) + + +class SchemaGuard: + """Stateful, secret-free schema-drift guard for the tushare read-through cache. + + DEFAULT-OFF: a cache built with ``schema_guard=None`` never constructs one. In + ``report_only`` mode every check appends a :class:`SchemaDriftFinding`; in + ``strict`` mode a HARD finding RAISES a secret-free ``RuntimeError`` (warnings + still only append). Findings carry only endpoint + column names. + """ + + def __init__(self, mode: str = REPORT_ONLY) -> None: + if mode not in (REPORT_ONLY, STRICT): + raise ValueError( + f"SchemaGuard mode must be {REPORT_ONLY!r} or {STRICT!r}; got {mode!r}." + ) + self._mode = mode + self._findings: list[SchemaDriftFinding] = [] + # endpoints whose stored-schema hash has already been checked this run + # (check #4 fires at most once per endpoint per run). + self._hash_checked: set[str] = set() + + @property + def mode(self) -> str: + return self._mode + + # -- raw-input checks (#1 missing required, #2 unknown extra) ------------ # + def inspect_raw(self, endpoint: str, raw: "pd.DataFrame | None") -> None: + """Check a NON-EMPTY raw frame's source columns; skip None/empty returns. + + An empty / None return is legitimate coverage (a stock not listed / + suspended / out of index on a range), NOT drift — so it is skipped. + """ + if raw is None or len(raw) == 0: + return + schema = REGISTRY.get(endpoint) + if schema is None: + return + cols = {str(c) for c in raw.columns} + missing = schema.required_source_columns - cols + if missing: + names = tuple(sorted(missing)) + self._emit( + endpoint, + CHECK_MISSING_REQUIRED, + HARD, + f"endpoint {endpoint} raw frame is missing required source " + f"column(s) {list(names)}; the parser will silently NaN-fill the " + f"canonical data", + names, + ) + extra = cols - schema.known_source_columns + if extra: + names = tuple(sorted(extra)) + self._emit( + endpoint, + CHECK_UNKNOWN_EXTRA, + WARNING, + f"endpoint {endpoint} raw frame has unknown extra source " + f"column(s) {list(names)} the parser ignores", + names, + ) + + # -- canonical-output check (#3 parsed columns vs registry) -------------- # + def inspect_canonical(self, endpoint: str, parsed_columns) -> None: + """Check the parsed canonical columns equal the registry (order-independent). + + A mismatch means the parser / spec / registry disagree — a code bug, not + upstream drift — so it is HARD. + """ + schema = REGISTRY.get(endpoint) + if schema is None: + return + expected = set(schema.canonical_columns) + got = {str(c) for c in parsed_columns} + if expected != got: + diff = tuple(sorted((expected - got) | (got - expected))) + self._emit( + endpoint, + CHECK_CANONICAL_MISMATCH, + HARD, + f"endpoint {endpoint} parsed canonical columns " + f"{sorted(got)} != registry {sorted(expected)}", + diff, + ) + + # -- stored-schema hash check (#4 vs cached ledger history) -------------- # + def check_fields_hash( + self, endpoint: str, current_hash: str, prior_hash: "str | None" + ) -> None: + """Compare the registry canonical hash to the ledger's last fields_hash. + + Fires at most once per endpoint per run. A mismatch means the STORED + schema changed vs the cached ledger history (a canonical-column migration) + — HARD. This is a MIGRATION DETECTOR, not an init validator: + ``prior_hash is None`` (a fresh cache / an endpoint never recorded) is + ALWAYS clean, and an equal hash is clean. + """ + if endpoint in self._hash_checked: + return + self._hash_checked.add(endpoint) + if prior_hash is None or prior_hash == current_hash: + return + self._emit( + endpoint, + CHECK_HASH_CHANGED, + HARD, + f"endpoint {endpoint} stored schema changed vs cached ledger history " + f"(ledger fields_hash={prior_hash}, current={current_hash})", + (), + ) + + # -- accumulation / reporting ------------------------------------------- # + def _emit( + self, + endpoint: str, + check: str, + severity: str, + detail: str, + columns: tuple[str, ...], + ) -> None: + finding = SchemaDriftFinding( + endpoint=endpoint, + check=check, + severity=severity, + detail=detail, + columns=columns, + ) + if self._mode == STRICT and severity == HARD: + raise RuntimeError(detail) + self._findings.append(finding) + + def findings(self) -> tuple[SchemaDriftFinding, ...]: + """Accumulated findings (insertion order — stable + deterministic).""" + return tuple(self._findings) + + def summary(self) -> dict: + """One-line-friendly counts: total / hard / warning / by_endpoint.""" + hard = sum(1 for f in self._findings if f.severity == HARD) + warning = sum(1 for f in self._findings if f.severity == WARNING) + by_endpoint: dict[str, int] = {} + for f in self._findings: + by_endpoint[f.endpoint] = by_endpoint.get(f.endpoint, 0) + 1 + return { + "total": len(self._findings), + "hard": hard, + "warning": warning, + "by_endpoint": by_endpoint, + } diff --git a/data/cache/tushare_cache.py b/data/cache/tushare_cache.py index 7bccdbd..6d265f4 100644 --- a/data/cache/tushare_cache.py +++ b/data/cache/tushare_cache.py @@ -56,6 +56,7 @@ from data.cache.coverage import CoverageLedger from data.cache.intervals import merge_intervals, subtract_intervals from data.cache.parquet_store import CacheParquetStore +from data.cache.schema_registry import REGISTRY, SchemaGuard from data.cache.tushare_parsers import ( _parse_adj, _parse_daily, @@ -153,9 +154,15 @@ def __init__( not_ready_days: int = 0, recent_tail_overrides: dict[str, int] | None = None, max_workers: int = 1, + schema_guard: SchemaGuard | None = None, ) -> None: self._store = store self._ledger = ledger + # D-series schema drift guard. None (default) keeps every parse site a + # byte-identical passthrough (zero behaviour change); a guard runs the raw/ + # canonical/hash drift checks around the SAME parse call (report-only or + # strict). The guard never sees a token or a data value — only column names. + self._schema_guard = schema_guard self._refresh_recent_days = int(refresh_recent_days) self._refresh_dimension_days = int(refresh_dimension_days) self._force_refresh = set(force_refresh or ()) @@ -205,6 +212,41 @@ def update_summary(self) -> dict[str, dict[str, int]]: for ep in ALL_ENDPOINTS } + # -- D-series schema drift guard --------------------------------------- # + def _guarded_parse(self, endpoint, raw, parse): + """Parse ``raw`` for ``endpoint``, optionally running the drift guard. + + With no guard this is an EXACT passthrough (``parse(raw)``) — zero + behaviour change. With a guard: check #4 (stored-schema hash, deduped to + once per endpoint per run) runs first, then #1/#2 on the raw frame, then + the SAME parse, then #3 on the parsed canonical columns. The guard only + ever inspects column names, never a value or token. + """ + if self._schema_guard is None: + return parse(raw) + guard = self._schema_guard + guard.check_fields_hash( + endpoint, + REGISTRY[endpoint].expected_canonical_hash, + self._ledger.last_fields_hash(endpoint), + ) + guard.inspect_raw(endpoint, raw) + parsed = parse(raw) + guard.inspect_canonical(endpoint, list(parsed.columns)) + return parsed + + def schema_findings(self) -> tuple: + """Accumulated schema-drift findings (empty when no guard is attached).""" + if self._schema_guard is None: + return () + return self._schema_guard.findings() + + def schema_summary(self) -> "dict | None": + """Schema-guard count summary, or None when no guard is attached.""" + if self._schema_guard is None: + return None + return self._schema_guard.summary() + # -- dense per-symbol date-range endpoints ----------------------------- # def daily_bars( self, symbols: list[str], start: str, end: str, fetch: FetchOne @@ -484,7 +526,7 @@ def _read_through_concurrent( first_exc = res.exc continue self.fetch_counts[endpoint] = self.fetch_counts.get(endpoint, 0) + 1 - parsed = parse(res.raw) + parsed = self._guarded_parse(endpoint, res.raw, parse) if len(parsed): self._store.upsert_symbol(endpoint, symbol, parsed, key_cols) self.written_counts[endpoint] = ( @@ -590,7 +632,7 @@ def _fetch_gap( """Fetch one gap, upsert raw rows, record coverage (incl. empty/not-ready).""" raw = fetch(symbol, _compact(gap_start), _compact(gap_end)) self.fetch_counts[endpoint] = self.fetch_counts.get(endpoint, 0) + 1 - parsed = parse(raw) + parsed = self._guarded_parse(endpoint, raw, parse) if len(parsed): self._store.upsert_symbol(endpoint, symbol, parsed, key_cols) self.written_counts[endpoint] = ( @@ -677,7 +719,9 @@ def _fetch_index_gap(self, index_code, gap_start, gap_end, fetch, fields_hash): ) raw = fetch(index_code, _compact(win_start), _compact(win_end)) self.fetch_counts[INDEX_WEIGHT] += 1 - parsed = _parse_index_weight(raw, index_code) + parsed = self._guarded_parse( + INDEX_WEIGHT, raw, lambda r: _parse_index_weight(r, index_code) + ) if not parsed.empty: frames.append(parsed) total_rows += len(parsed) @@ -717,7 +761,7 @@ def _fetch_snapshot( """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) + parsed = self._guarded_parse(endpoint, raw, parse) row_count = len(parsed) if row_count: self._store.upsert_symbol(endpoint, key, parsed, key_cols) diff --git a/qt/config.py b/qt/config.py index f1b15fb..9e69cd7 100644 --- a/qt/config.py +++ b/qt/config.py @@ -35,6 +35,21 @@ class ProjectCfg(_Strict): timezone: str = "Asia/Shanghai" +class SchemaGuardCfg(_Strict): + """Default-off endpoint schema drift guard for the tushare cache (D-series). + + Disabled by DEFAULT — ``enabled=False`` keeps every existing config byte/ + behaviour identical (no guard is constructed, every cache parse site stays a + passthrough). When ``enabled``, the cache runs report-only (or ``strict``, + which raises on a HARD drift) checks on each endpoint's RAW source columns, + parsed canonical columns, and stored-schema hash vs the ledger history. The + guard sees only column + endpoint names — never a token or a data value. + """ + + enabled: bool = False + mode: Literal["report_only", "strict"] = "report_only" + + class CacheCfg(_Strict): """Persistent endpoint-level raw cache (P4-1 market bars + P4-2 universe/tradability). @@ -68,6 +83,8 @@ class CacheCfg(_Strict): # 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) + # D-series schema drift guard (default-off; see SchemaGuardCfg). + schema_guard: SchemaGuardCfg = Field(default_factory=SchemaGuardCfg) @field_validator("refresh_recent_days") @classmethod diff --git a/qt/pipeline.py b/qt/pipeline.py index 11b4541..27a8cc4 100644 --- a/qt/pipeline.py +++ b/qt/pipeline.py @@ -563,6 +563,14 @@ def _build_cache(cfg: RootConfig): return None from data.cache import CacheParquetStore, CoverageLedger, TushareCache + # D-series schema drift guard: built ONLY when opted in (default-off => None => + # every cache parse site stays a byte-identical passthrough). + schema_guard = None + if cache_cfg.schema_guard.enabled: + from data.cache.schema_registry import SchemaGuard + + schema_guard = SchemaGuard(mode=cache_cfg.schema_guard.mode) + root = cache_cfg.root_dir return TushareCache( CacheParquetStore(root), @@ -570,6 +578,7 @@ def _build_cache(cfg: RootConfig): refresh_recent_days=cache_cfg.refresh_recent_days, refresh_dimension_days=cache_cfg.refresh_dimension_days, force_refresh=tuple(cache_cfg.force_refresh), + schema_guard=schema_guard, ) @@ -681,6 +690,15 @@ def _log_run_cache_stats(cache, logger: logging.Logger) -> None: if not stats: return logger.info("%s", _format_cache_stats(stats)) + # D-series schema drift guard: one secret-free count line, only when a guard + # is attached (default-off => schema_summary() is None => nothing logged). + summary_getter = getattr(cache, "schema_summary", None) + summary = summary_getter() if summary_getter is not None else None + if summary is not None: + logger.info( + "schema guard: hard=%d warning=%d total=%d", + int(summary["hard"]), int(summary["warning"]), int(summary["total"]), + ) def _load_panel( diff --git a/tests/test_schema_registry.py b/tests/test_schema_registry.py new file mode 100644 index 0000000..8a564fe --- /dev/null +++ b/tests/test_schema_registry.py @@ -0,0 +1,438 @@ +"""D-series schema registry + default-off drift guard (network-free, fakes only). + +Locks: + * the registry declarations are consistent with the spec column lists AND the + REAL parsers (feeding exactly ``required_source_columns`` introduces no NaN); + * ``expected_canonical_hash`` matches the ledger's ``fields_hash`` hashing; + * check #1 (missing required) -> HARD / strict raises (endpoint+column, no secret); + * check #2 (unknown extra) -> WARNING, never hard; + * empty / None raw -> no findings (legitimate coverage, not drift); + * check #3 (parsed canonical mismatch) -> HARD; + * check #4 (stored-schema hash changed) -> HARD once per endpoint, deduped; + * default-off ``_guarded_parse`` is a byte-identical passthrough; + * report_only records coverage + a finding; strict raises BEFORE upsert/coverage; + * ``CoverageLedger.last_fields_hash`` returns latest-by-fetched_at / None; + * findings + summary carry no token / secret-file content. + +No test hits the network or reads the real token: fake raw frames + a fake fetch +closure drive everything. +""" + +from __future__ import annotations + +import pandas as pd +import pytest + +from data.cache import tushare_parsers as P +from data.cache.coverage import CoverageLedger +from data.cache.parquet_store import CacheParquetStore +from data.cache.schema_registry import ( + CHECK_CANONICAL_MISMATCH, + CHECK_HASH_CHANGED, + CHECK_MISSING_REQUIRED, + CHECK_UNKNOWN_EXTRA, + REGISTRY, + EndpointSchema, + SchemaDriftFinding, + SchemaGuard, +) +from data.cache.tushare_cache import TushareCache +from data.cache.tushare_planning import _fields_hash +from data.cache.tushare_specs import ( + ALL_ENDPOINTS, + INDEX_MEMBER_ALL, + MARKET_DAILY, + _DAILY_COLUMNS, +) + +# bench of one valid value per known source column (drives the lock test). +_SAMPLE: dict[str, object] = { + "ts_code": "000001.SZ", "con_code": "000001.SZ", "trade_date": "20240102", + "start_date": "20240102", "in_date": "20240102", "out_date": "20240630", + "ann_date": "20240102", + "end_date": "20240331", "open": 1.0, "high": 2.0, "low": 0.5, "close": 1.5, + "vol": 1000.0, "amount": 1.0e6, "adj_factor": 1.0, "up_limit": 2.2, + "down_limit": 1.8, "weight": 5.0, "pe": 10.0, "pb": 1.2, "total_mv": 1.0e9, + "name": "PINGAN", "l1_name": "Bank", "l2_name": "Bank2", "l3_name": "Bank3", + "roe": 12.0, "netprofit_yoy": 8.0, "grossprofit_margin": 30.0, +} + +# the real parser per endpoint (index_weight/namechange/index_member take an arg). +_PARSERS = { + "market_daily": P._parse_daily, + "adj_factor": P._parse_adj, + "suspend_d": P._parse_suspend, + "stk_limit": P._parse_stk_limit, + "daily_basic": P._parse_daily_basic, + "fina_indicator": P._parse_fina, + "index_weight": lambda raw: P._parse_index_weight(raw, "000300.SH"), + "namechange": lambda raw: P._parse_namechange(raw, "000001.SZ"), + "stock_basic": P._parse_stock_basic, + "index_member_all": lambda raw: P._parse_index_member(raw, "000001.SZ"), +} + +# canonical columns derived from OPTIONAL source columns: feeding only the required +# source columns legitimately leaves these None/NaT (open-interval ends, omittable +# SW level names), so they are excluded from the forward no-NaN lock. +_NULLABLE = { + "namechange": {"end_date"}, + "index_member_all": {"l1_name", "l2_name", "l3_name", "out_date"}, +} + + +def _raw_with(cols) -> pd.DataFrame: + return pd.DataFrame({c: [_SAMPLE[c]] for c in cols}) + + +# --------------------------------------------------------------------------- # +# 1. registry <-> spec/parser consistency (THE LOCK) +# --------------------------------------------------------------------------- # +def test_registry_covers_every_endpoint(): + assert set(REGISTRY) == set(ALL_ENDPOINTS) + + +@pytest.mark.parametrize("endpoint", list(ALL_ENDPOINTS)) +def test_required_source_columns_are_sufficient(endpoint): + """Feeding EXACTLY required_source_columns through the REAL parser yields the + registry canonical columns with no NaN in the data columns.""" + schema = REGISTRY[endpoint] + raw = _raw_with(sorted(schema.required_source_columns)) + out = _PARSERS[endpoint](raw) + assert list(out.columns) == list(schema.canonical_columns) + data_cols = [c for c in out.columns if c not in _NULLABLE.get(endpoint, set())] + assert not out[data_cols].isna().any().any(), ( + endpoint, out[data_cols].isna().any().to_dict() + ) + + +@pytest.mark.parametrize("endpoint", list(ALL_ENDPOINTS)) +def test_canonical_columns_match_real_parser_shape(endpoint): + """The registry canonical columns equal the parser's full (empty) shape.""" + empty = _PARSERS[endpoint](None) + assert list(REGISTRY[endpoint].canonical_columns) == list(empty.columns) + + +# --------------------------------------------------------------------------- # +# 2. expected_canonical_hash == ledger fields_hash hashing +# --------------------------------------------------------------------------- # +@pytest.mark.parametrize("endpoint", list(ALL_ENDPOINTS)) +def test_expected_canonical_hash_matches_fields_hash(endpoint): + schema = REGISTRY[endpoint] + assert schema.expected_canonical_hash == _fields_hash(list(schema.canonical_columns)) + + +# --------------------------------------------------------------------------- # +# 3. check #1 (missing required) -> HARD / strict raises, no secret +# --------------------------------------------------------------------------- # +def test_missing_required_is_hard_in_report_only(): + guard = SchemaGuard(mode="report_only") + schema = REGISTRY[MARKET_DAILY] + raw = _raw_with(sorted(schema.required_source_columns - {"amount"})) + guard.inspect_raw(MARKET_DAILY, raw) + findings = guard.findings() + hard = [f for f in findings if f.severity == "hard"] + assert len(hard) == 1 + f = hard[0] + assert f.check == CHECK_MISSING_REQUIRED + assert f.endpoint == MARKET_DAILY + assert "amount" in f.columns + + +def test_missing_required_raises_in_strict_with_endpoint_and_column(): + guard = SchemaGuard(mode="strict") + schema = REGISTRY[MARKET_DAILY] + raw = _raw_with(sorted(schema.required_source_columns - {"amount"})) + with pytest.raises(RuntimeError) as exc: + guard.inspect_raw(MARKET_DAILY, raw) + msg = str(exc.value) + assert MARKET_DAILY in msg + assert "amount" in msg + # strict raise carries no secret. + assert ".config.json" not in msg and "token" not in msg.lower() + assert guard.findings() == () # nothing accumulated on a raise + + +# --------------------------------------------------------------------------- # +# 4. check #2 (unknown extra) -> WARNING, never hard +# --------------------------------------------------------------------------- # +def test_unknown_extra_source_column_is_warning(): + guard = SchemaGuard(mode="report_only") + schema = REGISTRY[MARKET_DAILY] + raw = _raw_with(sorted(schema.required_source_columns)) + raw["surprise_col"] = [123.0] + guard.inspect_raw(MARKET_DAILY, raw) + findings = guard.findings() + assert all(f.severity != "hard" for f in findings) + warn = [f for f in findings if f.check == CHECK_UNKNOWN_EXTRA] + assert len(warn) == 1 + assert warn[0].severity == "warning" + assert "surprise_col" in warn[0].columns + + +def test_unknown_extra_never_raises_in_strict(): + guard = SchemaGuard(mode="strict") + schema = REGISTRY[MARKET_DAILY] + raw = _raw_with(sorted(schema.required_source_columns)) + raw["surprise_col"] = [123.0] + guard.inspect_raw(MARKET_DAILY, raw) # must NOT raise (warning only) + assert any(f.check == CHECK_UNKNOWN_EXTRA for f in guard.findings()) + + +# HIGH-2: a catalogued known-extra column must NOT warn; only a truly-new column does. +def test_known_extra_column_is_not_flagged(): + guard = SchemaGuard(mode="report_only") + schema = REGISTRY[MARKET_DAILY] + raw = _raw_with(sorted(schema.required_source_columns)) + for extra in sorted(schema.known_extra_columns): # pre_close/change/pct_chg + raw[extra] = [0.0] + guard.inspect_raw(MARKET_DAILY, raw) + assert guard.findings() == () # all extras catalogued => silent + + +def test_only_uncatalogued_extra_warns_once(): + guard = SchemaGuard(mode="report_only") + schema = REGISTRY[MARKET_DAILY] + raw = _raw_with(sorted(schema.required_source_columns)) + for extra in sorted(schema.known_extra_columns): + raw[extra] = [0.0] + raw["brand_new_tushare_col"] = [1.0] + guard.inspect_raw(MARKET_DAILY, raw) + warn = [f for f in guard.findings() if f.check == CHECK_UNKNOWN_EXTRA] + assert len(warn) == 1 + assert warn[0].columns == ("brand_new_tushare_col",) # only the uncatalogued one + + +def test_no_fields_endpoints_do_not_warn_on_standard_extras(): + """The documented tushare response columns the parser ignores are catalogued, + so a normal fetch raises no check-#2 noise (the HIGH-2 fix).""" + guard = SchemaGuard(mode="report_only") + standard_response = { + "market_daily": ["ts_code", "trade_date", "open", "high", "low", "close", + "pre_close", "change", "pct_chg", "vol", "amount"], + "stk_limit": ["ts_code", "trade_date", "pre_close", "up_limit", "down_limit"], + "suspend_d": ["ts_code", "trade_date", "suspend_timing", "suspend_type"], + "namechange": ["ts_code", "name", "start_date", "end_date", + "ann_date", "change_reason"], + "index_member_all": ["l1_code", "l1_name", "l2_code", "l2_name", "l3_code", + "l3_name", "ts_code", "name", "in_date", "out_date", + "is_new"], + } + for ep, cols in standard_response.items(): + raw = pd.DataFrame({c: [_SAMPLE.get(c, "x")] for c in cols}) + guard.inspect_raw(ep, raw) + extra = [f for f in guard.findings() if f.check == CHECK_UNKNOWN_EXTRA] + assert extra == [], [f.endpoint + ":" + str(f.columns) for f in extra] + + +# --------------------------------------------------------------------------- # +# MED-3: inverse lock — every optional column is legitimately droppable, and the +# index_member_all required/optional split is pinned (prevents HIGH-1 regression). +# --------------------------------------------------------------------------- # +@pytest.mark.parametrize("endpoint", list(ALL_ENDPOINTS)) +def test_optional_columns_are_truly_droppable(endpoint): + """Dropping ANY single optional source column from a full raw frame must NOT + raise and must still yield the registry canonical schema.""" + schema = REGISTRY[endpoint] + full_cols = sorted(schema.required_source_columns | schema.optional_source_columns) + for drop in sorted(schema.optional_source_columns): + raw = _raw_with([c for c in full_cols if c != drop]) + out = _PARSERS[endpoint](raw) # must not raise + assert list(out.columns) == list(schema.canonical_columns), (endpoint, drop) + + +def test_index_member_required_is_only_in_date(): + """HIGH-1 regression lock: the defensively-filled level names are OPTIONAL, + never REQUIRED (else strict mode loops on a stock missing a level column).""" + schema = REGISTRY[INDEX_MEMBER_ALL] + assert schema.required_source_columns == frozenset({"in_date"}) + assert {"l1_name", "l2_name", "l3_name", "out_date"} <= schema.optional_source_columns + # and a real raw frame missing every level name must NOT raise in strict mode. + guard = SchemaGuard(mode="strict") + raw = pd.DataFrame({"in_date": ["20240102"], "ts_code": ["000001.SZ"]}) + guard.inspect_raw(INDEX_MEMBER_ALL, raw) # no level cols -> must not raise + assert all(f.severity != "hard" for f in guard.findings()) + + +# --------------------------------------------------------------------------- # +# 5. empty / None raw -> no findings (legitimate coverage, not drift) +# --------------------------------------------------------------------------- # +def test_empty_or_none_raw_is_not_drift(): + guard = SchemaGuard(mode="strict") # strict: would raise if it inspected + guard.inspect_raw(MARKET_DAILY, None) + guard.inspect_raw(MARKET_DAILY, pd.DataFrame(columns=["ts_code"])) + assert guard.findings() == () + + +# --------------------------------------------------------------------------- # +# 6. check #3 (parsed canonical mismatch) -> HARD +# --------------------------------------------------------------------------- # +def test_canonical_mismatch_is_hard(): + guard = SchemaGuard(mode="report_only") + wrong = list(_DAILY_COLUMNS) + ["unexpected_col"] + guard.inspect_canonical(MARKET_DAILY, wrong) + hard = [f for f in guard.findings() if f.severity == "hard"] + assert len(hard) == 1 + assert hard[0].check == CHECK_CANONICAL_MISMATCH + assert "unexpected_col" in hard[0].columns + + +def test_canonical_match_has_no_finding(): + guard = SchemaGuard(mode="report_only") + guard.inspect_canonical(MARKET_DAILY, list(_DAILY_COLUMNS)) + assert guard.findings() == () + + +# --------------------------------------------------------------------------- # +# 7. check #4 (stored-schema hash changed) -> HARD once per endpoint, deduped +# --------------------------------------------------------------------------- # +def test_fields_hash_change_is_hard_once_then_deduped(): + guard = SchemaGuard(mode="report_only") + cur = REGISTRY[MARKET_DAILY].expected_canonical_hash + guard.check_fields_hash(MARKET_DAILY, cur, prior_hash="STALEHASH") + guard.check_fields_hash(MARKET_DAILY, cur, prior_hash="STALEHASH") # deduped + hard = [f for f in guard.findings() if f.check == CHECK_HASH_CHANGED] + assert len(hard) == 1 + assert hard[0].severity == "hard" + assert hard[0].endpoint == MARKET_DAILY + + +def test_fields_hash_none_prior_no_finding(): + guard = SchemaGuard(mode="report_only") + cur = REGISTRY[MARKET_DAILY].expected_canonical_hash + guard.check_fields_hash(MARKET_DAILY, cur, prior_hash=None) + assert guard.findings() == () + + +def test_fields_hash_equal_no_finding(): + guard = SchemaGuard(mode="report_only") + cur = REGISTRY[MARKET_DAILY].expected_canonical_hash + guard.check_fields_hash(MARKET_DAILY, cur, prior_hash=cur) + assert guard.findings() == () + + +def test_fields_hash_change_strict_raises(): + guard = SchemaGuard(mode="strict") + cur = REGISTRY[MARKET_DAILY].expected_canonical_hash + with pytest.raises(RuntimeError) as exc: + guard.check_fields_hash(MARKET_DAILY, cur, prior_hash="STALEHASH") + msg = str(exc.value) + assert MARKET_DAILY in msg + assert ".config.json" not in msg and "token" not in msg.lower() + + +# --------------------------------------------------------------------------- # +# 8. default-off passthrough is byte-identical +# --------------------------------------------------------------------------- # +def _daily_raw(ts_code, dates, drop=()): + rows = [ + {"ts_code": ts_code, "trade_date": d, "open": 1.0, "high": 2.0, + "low": 0.5, "close": 1.5, "vol": 1000.0, "amount": 1.0e6} + for d in dates + ] + return pd.DataFrame(rows).drop(columns=list(drop)) + + +def test_guarded_parse_passthrough_when_no_guard(tmp_path): + cache = TushareCache( + CacheParquetStore(str(tmp_path)), CoverageLedger(str(tmp_path)), + ) + raw = _daily_raw("000001.SZ", ["20240102", "20240103"]) + out = cache._guarded_parse(MARKET_DAILY, raw, P._parse_daily) + pd.testing.assert_frame_equal(out, P._parse_daily(raw)) + + +# --------------------------------------------------------------------------- # +# 9. report_only records coverage + finding; strict raises before upsert/coverage +# --------------------------------------------------------------------------- # +def _drift_fetch(symbol, s_compact, e_compact): + # a non-empty frame MISSING the required 'amount' column (parser NaN-fills it). + return _daily_raw(symbol, ["20240102", "20240103"], drop=("amount",)) + + +def test_report_only_still_upserts_and_records_coverage(tmp_path): + cache = TushareCache( + CacheParquetStore(str(tmp_path)), CoverageLedger(str(tmp_path)), + schema_guard=SchemaGuard(mode="report_only"), + ) + out = cache.daily_bars(["000001.SZ"], "2024-01-02", "2024-01-03", _drift_fetch) + # row is upserted (NaN amount) + coverage recorded — report-only changes nothing. + assert not out.empty + assert not cache._store.read_symbol(MARKET_DAILY, "000001.SZ").empty + assert cache._ledger.covered_intervals(MARKET_DAILY, "000001.SZ") + # and a HARD missing-required finding is recorded. + hard = [f for f in cache.schema_findings() if f.severity == "hard"] + assert any(f.check == CHECK_MISSING_REQUIRED and "amount" in f.columns for f in hard) + + +def test_strict_raises_before_upsert_and_coverage(tmp_path): + cache = TushareCache( + CacheParquetStore(str(tmp_path)), CoverageLedger(str(tmp_path)), + schema_guard=SchemaGuard(mode="strict"), + ) + with pytest.raises(RuntimeError): + cache.daily_bars(["000001.SZ"], "2024-01-02", "2024-01-03", _drift_fetch) + # the failing gap recorded NO coverage and upserted NO rows (retryable). + assert cache._store.read_symbol(MARKET_DAILY, "000001.SZ").empty + assert cache._ledger.covered_intervals(MARKET_DAILY, "000001.SZ") == [] + + +# --------------------------------------------------------------------------- # +# 10. CoverageLedger.last_fields_hash returns latest-by-fetched_at / None +# --------------------------------------------------------------------------- # +def test_last_fields_hash_latest_and_absent(tmp_path): + led = CoverageLedger(str(tmp_path)) + assert led.last_fields_hash(MARKET_DAILY) is None # never recorded + base = { + "endpoint": MARKET_DAILY, "key_type": "symbol", "key": "000001.SZ", + "start_date": pd.Timestamp("2024-01-02"), "end_date": pd.Timestamp("2024-01-03"), + "row_count": 1, "status": "ok", "source_version": None, + } + led.record_many([ + {**base, "fields_hash": "OLDHASH", "fetched_at": pd.Timestamp("2026-06-01")}, + {**base, "fields_hash": "NEWHASH", "fetched_at": pd.Timestamp("2026-06-10")}, + ]) + assert led.last_fields_hash(MARKET_DAILY) == "NEWHASH" + assert led.last_fields_hash("adj_factor") is None # other endpoint absent + + +# --------------------------------------------------------------------------- # +# 11. findings / summary carry no secret +# --------------------------------------------------------------------------- # +def test_findings_and_summary_have_no_secret(): + guard = SchemaGuard(mode="report_only") + schema = REGISTRY[MARKET_DAILY] + raw = _raw_with(sorted(schema.required_source_columns - {"amount"})) + raw["surprise_col"] = [1.0] + guard.inspect_raw(MARKET_DAILY, raw) + guard.check_fields_hash(MARKET_DAILY, schema.expected_canonical_hash, "STALE") + # also exercise check #3's detail string (canonical mismatch). + guard.inspect_canonical(MARKET_DAILY, list(_DAILY_COLUMNS) + ["unexpected_col"]) + rendered = " ".join( + f"{f.endpoint} {f.check} {f.severity} {f.detail} {f.columns}" + for f in guard.findings() + ) + rendered += " " + str(guard.summary()) + for needle in (".config.json", "tushare.token", "token="): + assert needle not in rendered + summary = guard.summary() + assert summary["total"] == len(guard.findings()) + assert summary["hard"] >= 1 and summary["warning"] >= 1 + + +def test_endpoint_schema_and_finding_are_frozen(): + schema = REGISTRY[MARKET_DAILY] + with pytest.raises(Exception): + schema.endpoint = "x" # type: ignore[misc] + finding = SchemaDriftFinding( + endpoint=MARKET_DAILY, check=CHECK_MISSING_REQUIRED, severity="hard", + detail="d", columns=("amount",), + ) + with pytest.raises(Exception): + finding.severity = "warning" # type: ignore[misc] + # known_source_columns is required + optional. + custom = EndpointSchema( + endpoint="x", required_source_columns=frozenset({"a"}), + canonical_columns=("a",), natural_key=("a",), + optional_source_columns=frozenset({"b"}), + ) + assert custom.known_source_columns == frozenset({"a", "b"})