feat(data): D3b default-off data-update quality-report hook

config/data_update.yaml

@@ -71,6 +71,20 @@ data_update:
     - grossprofit_margin
   rate_limit_per_min: 450   # conservative (official stk_mins ceiling is 500/min)
   force_refresh: []
+  # D3b report-only data-quality hook. OFF by default: with enabled=false the job
+  # behaves exactly as before. When enabled, the updater runs the D3 data/quality
+  # STRUCTURAL checks on the frames it ALREADY warmed (NO extra API calls) and
+  # writes a deterministic Markdown report under output.report_dir. It never
+  # filters/repairs data, never fails the job, never changes cache coverage or the
+  # per-endpoint request summary. endpoints accepts only the structural surfaces
+  # the updater loads as frames; report_name must be a bare filename.
+  quality:
+    enabled: false
+    endpoints:
+      - market_daily
+      - adj_factor
+      - stk_mins_1min
+    report_name: data_update_quality_report.md
 
 # --- unused by data-update (schema-only stubs) ------------------------------- #
 factors:

qt/cli.py

@@ -212,11 +212,20 @@ def _cmd_data_update(args: argparse.Namespace) -> int:
     except (ConfigError, ValueError, FileNotFoundError) as exc:
         print(f"ERROR: {exc}", file=sys.stderr)
         return 1
-    print(
+    out = (
         f"OK data-update: {len(result.endpoints)} endpoints, "
         f"{len(result.symbols)} symbols ({result.elapsed_seconds:.1f}s)\n"
         f"{format_summary(result)}"
     )
+    # D3b: only surfaced when the report-only quality hook is enabled; with the
+    # default (disabled) hook the output is materially unchanged.
+    if result.quality_report_path is not None:
+        out += (
+            f"\nquality: findings={result.quality_findings_count} "
+            f"hard={result.quality_hard_count} "
+            f"report={result.quality_report_path}"
+        )
+    print(out)
     return 0
 
 

qt/config.py

@@ -592,6 +592,66 @@ def _check_sections(self) -> "SubsetValidationCfg":
     "index_member_all", "stk_mins_1min",
 })
 
+# D3b quality can only check the STRUCTURAL endpoints the updater already loads as
+# in-memory frames (market bars + 1min minutes). Universe / financial / dimension
+# endpoints are not re-materialized as frames here, so they are out of scope.
+_DATA_UPDATE_QUALITY_ENDPOINTS = frozenset({
+    "market_daily", "adj_factor", "stk_mins_1min",
+})
+
+
+class DataUpdateQualityCfg(_Strict):
+    """D3b report-only data-quality hook for ``data-update`` (default OFF).
+
+    When ``enabled`` the updater runs the accepted D3 ``data/quality`` STRUCTURAL
+    checks on the frames it ALREADY warmed (no extra API call) and writes a
+    deterministic Markdown report under ``output.report_dir``. It is report-only:
+    it never filters / repairs / mutates data, never fails the job, never changes
+    cache coverage or the per-endpoint request summary. With ``enabled=false``
+    (the default) every existing config behaves exactly as before.
+    """
+
+    enabled: bool = False
+    # Which warmed surfaces to quality-check (only the structural ones above).
+    endpoints: list[str] = Field(
+        default_factory=lambda: ["market_daily", "adj_factor", "stk_mins_1min"]
+    )
+    # A BARE filename written under output.report_dir (never absolute, never a
+    # path with separators or '..').
+    report_name: str = "data_update_quality_report.md"
+
+    @field_validator("endpoints")
+    @classmethod
+    def _check_quality_endpoints(cls, v: list[str]) -> list[str]:
+        unknown = [e for e in v if e not in _DATA_UPDATE_QUALITY_ENDPOINTS]
+        if unknown:
+            raise ValueError(
+                f"data_update.quality.endpoints {unknown} unknown; must be in "
+                f"{sorted(_DATA_UPDATE_QUALITY_ENDPOINTS)}."
+            )
+        return v
+
+    @field_validator("report_name")
+    @classmethod
+    def _check_report_name(cls, v: str) -> str:
+        name = str(v)
+        if not name.strip():
+            raise ValueError(
+                "data_update.quality.report_name must be a non-empty filename."
+            )
+        if (
+            "/" in name
+            or "\\" in name
+            or ".." in name
+            or name in (".", "..")
+        ):
+            raise ValueError(
+                "data_update.quality.report_name must be a bare filename under "
+                "output.report_dir (no path separators, no '..', not absolute); "
+                f"got {name!r}."
+            )
+        return name
+
 
 class DataUpdateCfg(_Strict):
     """Standalone data-updater section (P4-3) — consumed ONLY by ``data-update``.
@@ -614,6 +674,8 @@ class DataUpdateCfg(_Strict):
     fina_fields: list[str] = Field(default_factory=lambda: ["roe", "netprofit_yoy"])
     rate_limit_per_min: int = 450
     force_refresh: list[str] = Field(default_factory=list)
+    # D3b report-only quality hook (default OFF; see DataUpdateQualityCfg).
+    quality: DataUpdateQualityCfg = Field(default_factory=DataUpdateQualityCfg)
 
     @field_validator("endpoints", "force_refresh")
     @classmethod

qt/data_update_quality.py

@@ -0,0 +1,136 @@
+"""D3b: report-only data-quality hook for the ``data-update`` job.
+
+Surfaces the accepted D3 ``data/quality`` checks in operations. When the hook is
+enabled it runs the D3 STRUCTURAL checks on the frames the updater ALREADY warmed
+(market bars + 1min minutes) and writes a deterministic Markdown report. This is
+report-only: it NEVER filters / repairs / mutates data, never fails the job, never
+changes cache coverage or the per-endpoint request summary, and makes NO extra API
+call for quality.
+
+No exchange calendar is synthesized here — the missing-date / missing-minute D3
+checks need an explicit calendar and stay off in D3b (structural checks only).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+import pandas as pd
+
+from data.quality import (
+    HARD,
+    QualityFinding,
+    render_report,
+    run_adj_factor_checks,
+    run_intraday_checks,
+    run_market_checks,
+)
+
+# Only the structural endpoints the updater loads as in-memory frames can be
+# quality-checked without new API calls (mirrors the config-level allow-list).
+QUALITY_ENDPOINTS: tuple[str, ...] = ("market_daily", "adj_factor", "stk_mins_1min")
+
+_REPORT_TITLE = "Data Update Quality Report"
+
+
+@dataclass(frozen=True)
+class QualityOutcome:
+    """Outcome of the D3b quality hook (immutable)."""
+
+    report_path: Path
+    findings_count: int
+    hard_count: int
+    checked_endpoints: tuple[str, ...]
+
+
+def collect_findings(
+    *,
+    selected: list[str],
+    warmed: set[str],
+    market_frame: pd.DataFrame | None,
+    intraday_frame: pd.DataFrame | None,
+) -> tuple[list[QualityFinding], list[str]]:
+    """Run the D3 checks on already-warmed frames; return (findings, checked).
+
+    A surface is checked only when it is BOTH selected for quality (``selected``)
+    AND was warmed by ``data_update.endpoints`` (``warmed``) — so the hook never
+    checks a frame the updater did not load. Structural checks only: no calendar
+    is passed, so missing-date / missing-minute checks stay off (D3b scope).
+    """
+    chosen = set(selected)
+    findings: list[QualityFinding] = []
+    checked: list[str] = []
+    if "market_daily" in chosen and "market_daily" in warmed and market_frame is not None:
+        findings.extend(run_market_checks(market_frame))
+        checked.append("market_daily")
+    if "adj_factor" in chosen and "adj_factor" in warmed and market_frame is not None:
+        findings.extend(run_adj_factor_checks(market_frame))
+        checked.append("adj_factor")
+    if (
+        "stk_mins_1min" in chosen
+        and "stk_mins_1min" in warmed
+        and intraday_frame is not None
+    ):
+        findings.extend(run_intraday_checks(intraday_frame))
+        checked.append("stk_mins_1min")
+    return findings, checked
+
+
+def _run_context(
+    *, window_start, window_end, n_symbols: int, checked_endpoints: list[str]
+) -> str:
+    """A small, non-secret run-context block (window / symbol COUNT / endpoints).
+
+    Carries only metadata: the date window, the NUMBER of symbols (never the
+    symbol list), and the endpoint names checked. No token, no secret-file path,
+    no raw config.
+    """
+    eps = ", ".join(checked_endpoints) if checked_endpoints else "(none)"
+    return (
+        "## Run context\n\n"
+        f"- window: {pd.Timestamp(window_start).date()} .. "
+        f"{pd.Timestamp(window_end).date()}\n"
+        f"- symbols checked: {int(n_symbols)}\n"
+        f"- endpoints checked: {eps}\n"
+    )
+
+
+def write_quality_report(
+    findings: list[QualityFinding],
+    *,
+    report_dir: str,
+    report_name: str,
+    window_start,
+    window_end,
+    n_symbols: int,
+    checked_endpoints: list[str],
+) -> QualityOutcome:
+    """Render + write the deterministic report (even when clean); return the outcome.
+
+    Uses the D3 ``render_report`` (bounded, redacted examples) plus a small
+    non-secret run-context block. The report is written even with zero findings so
+    operations get an explicit "no findings" artifact. ``report_name`` is assumed
+    pre-validated as a bare filename (see ``DataUpdateQualityCfg``).
+    """
+    body = render_report(findings, title=_REPORT_TITLE)
+    context = _run_context(
+        window_start=window_start,
+        window_end=window_end,
+        n_symbols=n_symbols,
+        checked_endpoints=checked_endpoints,
+    )
+    text = f"{body}\n\n{context}"
+
+    out_dir = Path(report_dir)
+    out_dir.mkdir(parents=True, exist_ok=True)
+    path = out_dir / report_name
+    path.write_text(text, encoding="utf-8")
+
+    hard_count = sum(1 for f in findings if f.severity == HARD)
+    return QualityOutcome(
+        report_path=path,
+        findings_count=len(findings),
+        hard_count=hard_count,
+        checked_endpoints=tuple(checked_endpoints),
+    )

qt/data_updater.py

@@ -20,10 +20,12 @@
 
 import time
 from dataclasses import dataclass, field
+from pathlib import Path
 
 import pandas as pd
 
 from qt.config import RootConfig, load_config
+from qt.data_update_quality import collect_findings, write_quality_report
 
 # the intraday window the 21:00 job warms (the bulk historical minute backfill is
 # a separate manual run; the daily job only tops up the recent tail).
@@ -53,6 +55,10 @@ class UpdateResult:
     summary: dict[str, dict[str, int]]
     elapsed_seconds: float = 0.0
     notes: list[str] = field(default_factory=list)
+    # D3b report-only quality hook (None / 0 when the hook is disabled).
+    quality_report_path: Path | None = None
+    quality_findings_count: int = 0
+    quality_hard_count: int = 0
 
 
 def update_endpoints(
@@ -68,16 +74,25 @@ def update_endpoints(
     sw_level: str = "L1",
     intraday_cache=None,
     intraday_window: tuple[str, str] | None = None,
+    capture: dict | None = None,
 ) -> dict[str, dict[str, int]]:
     """Warm each requested endpoint through its feed; return the per-endpoint summary.
 
     Pure orchestration: every feed shares the read-through ``cache``, so only
     uncovered ranges hit the API. Calls NOTHING but the cache-warming feed
     methods — no factor / alpha / portfolio / backtest / PanelStore.
+
+    ``capture`` is an optional D3b sink: when a dict is passed, the already-fetched
+    market / intraday frames are stored under ``"market"`` / ``"intraday"`` for the
+    report-only quality hook. When ``None`` (the default) nothing is captured and
+    the warm path is byte-identical to before — the feed calls, their arguments,
+    their order, and the returned summary are unchanged.
     """
     eps = set(endpoints)
     if ({"market_daily", "adj_factor"} & eps) and feeds.market is not None:
-        feeds.market.get_bars(symbols, start, end)
+        bars = feeds.market.get_bars(symbols, start, end)
+        if capture is not None:
+            capture["market"] = bars
     if "index_weight" in eps and feeds.index is not None:
         for code in index_codes:
             feeds.index.get_constituents(code, start, end)
@@ -99,7 +114,9 @@ def update_endpoints(
     summary = cache.update_summary()
     if "stk_mins_1min" in eps and feeds.intraday is not None and intraday_window:
         s, e = intraday_window
-        feeds.intraday.get_minutes(symbols, s, e)
+        minutes = feeds.intraday.get_minutes(symbols, s, e)
+        if capture is not None:
+            capture["intraday"] = minutes
         st = intraday_cache.stats() if intraday_cache is not None else {}
         summary["stk_mins_1min"] = {
             "requests": int(st.get("stk_mins_1min", 0)),
@@ -213,21 +230,55 @@ def run_data_update(config_path: str, *, today=None) -> UpdateResult:
         (today_ts - pd.Timedelta(days=_INTRADAY_TAIL_DAYS)).strftime("%Y-%m-%d 00:00:00"),
         today_ts.strftime("%Y-%m-%d 23:59:59"),
     )
+    # D3b: capture the warmed frames only when the report-only quality hook is on.
+    # Disabled (the default) keeps capture=None, so the warm path is unchanged.
+    capture: dict | None = {} if du.quality.enabled else None
     summary = update_endpoints(
         cache, feeds, symbols,
         start=start, end=end,
         endpoints=du.endpoints, index_codes=du.index_codes,
         fina_fields=du.fina_fields,
         sw_level=cfg.processing.neutralize.industry_level,
         intraday_cache=intraday_cache, intraday_window=intraday_window,
+        capture=capture,
     )
+    quality = _maybe_run_quality(cfg, du, capture, symbols, start, end)
     return UpdateResult(
         window_start=pd.Timestamp(start),
         window_end=pd.Timestamp(end),
         symbols=symbols,
         endpoints=list(du.endpoints),
         summary=summary,
         elapsed_seconds=time.perf_counter() - t0,
+        quality_report_path=quality.report_path if quality is not None else None,
+        quality_findings_count=quality.findings_count if quality is not None else 0,
+        quality_hard_count=quality.hard_count if quality is not None else 0,
+    )
+
+
+def _maybe_run_quality(cfg: RootConfig, du, capture, symbols, start, end):
+    """Run the D3b report-only quality hook when enabled; return its outcome or None.
+
+    Report-only: reads the frames the updater ALREADY warmed (``capture``), runs
+    the D3 structural checks for the selected-and-warmed endpoints, and writes a
+    deterministic report. It never touches the cache, the summary, or any feed.
+    """
+    if not du.quality.enabled or capture is None:
+        return None
+    findings, checked = collect_findings(
+        selected=du.quality.endpoints,
+        warmed=set(du.endpoints),
+        market_frame=capture.get("market"),
+        intraday_frame=capture.get("intraday"),
+    )
+    return write_quality_report(
+        findings,
+        report_dir=cfg.output.report_dir,
+        report_name=du.quality.report_name,
+        window_start=pd.Timestamp(start),
+        window_end=pd.Timestamp(end),
+        n_symbols=len(symbols),
+        checked_endpoints=checked,
     )