From 22e0aad8f62577bab9e9507c243b2abb34cba2b2 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Fri, 19 Jun 2026 15:58:58 +0800 Subject: [PATCH 1/2] feat(data): add report-only data quality checks (D3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A small, pure, report-only data-quality layer that flags suspicious upstream daily-market / adj_factor / 1min-intraday data near ingestion. It NEVER mutates data, filters rows, repairs values, alters cache coverage, or touches qfq / factor / alpha / portfolio / runtime logic — it only surfaces findings. New package data/quality/: - report.py: QualityFinding (immutable) + make_finding (bounds examples to 5, cleans values), findings_to_frame (stable, deterministically ordered), render_report (deterministic Markdown; "no findings" on clean), has_hard, clean_value. Findings carry only dataset/check metadata + bounded {date/time, symbol, value} examples — no token, no secret path, no unbounded dump. - market.py: daily checks — duplicate (date,symbol), non-positive OHLC, highthreshold) and missing dates vs an explicit calendar (warning). run_market_checks / run_adj_factor_checks orchestrators. - intraday.py: 1min checks — duplicate bars, non-monotonic timestamps, non-positive OHLC, high findings; inputs are never mutated. Clean input yields zero hard findings. Tests (tests/test_data_quality_{market,intraday,report}.py): each daily/intraday issue is caught; clean panels produce zero hard findings; examples are bounded (<=5) and deterministic; rendering is deterministic and asserts no secret-looking paths in output. Integration deferred to D3b: this is the library + tests only; no data-update hook, no config change, no extra Tushare calls. Non-goals (unchanged): cache semantics, CoverageLedger storage, concurrency, schema registry, PanelStore, factor/alpha/portfolio/runtime/fills/OOS/analytics math. --- data/quality/__init__.py | 37 +++++ data/quality/_frames.py | 38 +++++ data/quality/intraday.py | 213 ++++++++++++++++++++++++ data/quality/market.py | 241 ++++++++++++++++++++++++++++ data/quality/report.py | 159 ++++++++++++++++++ tests/test_data_quality_intraday.py | 110 +++++++++++++ tests/test_data_quality_market.py | 145 +++++++++++++++++ tests/test_data_quality_report.py | 104 ++++++++++++ 8 files changed, 1047 insertions(+) create mode 100644 data/quality/__init__.py create mode 100644 data/quality/_frames.py create mode 100644 data/quality/intraday.py create mode 100644 data/quality/market.py create mode 100644 data/quality/report.py create mode 100644 tests/test_data_quality_intraday.py create mode 100644 tests/test_data_quality_market.py create mode 100644 tests/test_data_quality_report.py diff --git a/data/quality/__init__.py b/data/quality/__init__.py new file mode 100644 index 0000000..3b3a399 --- /dev/null +++ b/data/quality/__init__.py @@ -0,0 +1,37 @@ +"""Report-only data-quality layer (D3). + +Pure, library-only checks that flag suspicious upstream daily-market / adj_factor +/ 1min-intraday data near ingestion. This layer NEVER mutates data, filters rows, +repairs values, or touches cache / feed / factor / alpha / portfolio / runtime +logic — it only surfaces findings (with bounded, secret-free examples) for logs or +a future data-update artifact. +""" + +from __future__ import annotations + +from data.quality.intraday import run_intraday_checks +from data.quality.market import run_adj_factor_checks, run_market_checks +from data.quality.report import ( + HARD, + INFO, + WARNING, + QualityFinding, + findings_to_frame, + has_hard, + make_finding, + render_report, +) + +__all__ = [ + "HARD", + "INFO", + "WARNING", + "QualityFinding", + "findings_to_frame", + "has_hard", + "make_finding", + "render_report", + "run_adj_factor_checks", + "run_intraday_checks", + "run_market_checks", +] diff --git a/data/quality/_frames.py b/data/quality/_frames.py new file mode 100644 index 0000000..1afe16c --- /dev/null +++ b/data/quality/_frames.py @@ -0,0 +1,38 @@ +"""Internal pure frame helpers shared by the D3 quality checks. + +No mutation of inputs, no I/O, no secrets — just index normalization and bounded +deterministic example extraction. +""" + +from __future__ import annotations + +import pandas as pd + + +def reset_keys(df: pd.DataFrame) -> pd.DataFrame: + """Promote a MultiIndex (e.g. (date, symbol) / (time, symbol)) to columns. + + Returns a NEW frame (``reset_index`` copies); a frame whose keys are already + plain columns is returned unchanged. The input is never mutated. + """ + if isinstance(df.index, pd.MultiIndex): + return df.reset_index() + return df + + +def row_examples( + d: pd.DataFrame, + mask, + key_cols: list[str], + *, + extra: str | None = None, + limit: int = 5, +) -> list[dict]: + """Bounded, deterministic example rows for a boolean ``mask`` over ``d``. + + Sorted by ``key_cols`` so the sample is stable; values are returned raw + (``make_finding`` cleans them). At most ``limit`` rows. + """ + cols = list(key_cols) + ([extra] if extra else []) + sub = d.loc[mask, cols].sort_values(list(key_cols)).head(limit) + return [{c: row[c] for c in cols} for _, row in sub.iterrows()] diff --git a/data/quality/intraday.py b/data/quality/intraday.py new file mode 100644 index 0000000..e481f51 --- /dev/null +++ b/data/quality/intraday.py @@ -0,0 +1,213 @@ +"""Report-only data-quality checks for normalized 1min intraday frames (D3). + +Pure functions over a normalized intraday frame (raw cache-shaped columns or the +MultiIndex ``(time, symbol)`` panel; the bar timestamp is ``bar_end`` when +present, else ``time``). Report-only: nothing here enforces raw frequency, drops +rows, or replaces the existing intraday schema / cache guards — it surfaces +suspicious patterns alongside them. +""" + +from __future__ import annotations + +import pandas as pd + +from data.quality._frames import reset_keys, row_examples +from data.quality.report import HARD, WARNING, QualityFinding, make_finding + +_OHLC = ["open", "high", "low", "close"] +_SYMBOL = "symbol" +_DEFAULT_DATASET = "stk_mins_1min" + + +def _time_col(d: pd.DataFrame) -> str | None: + """The bar-timestamp column: ``bar_end`` (preferred) else ``time``.""" + if "bar_end" in d.columns: + return "bar_end" + if "time" in d.columns: + return "time" + return None + + +def check_duplicate_bars( + df: pd.DataFrame, *, dataset: str = _DEFAULT_DATASET +) -> QualityFinding | None: + """Duplicate ``(bar_end/time, symbol)`` rows.""" + d = reset_keys(df) + tcol = _time_col(d) + if tcol is None or _SYMBOL not in d.columns: + return None + mask = d.duplicated(subset=[tcol, _SYMBOL], keep=False) + if not mask.any(): + return None + keys = ( + d.loc[mask, [tcol, _SYMBOL]] + .drop_duplicates() + .sort_values([_SYMBOL, tcol]) + .head(5) + ) + examples = [{_SYMBOL: r[_SYMBOL], tcol: r[tcol]} for _, r in keys.iterrows()] + return make_finding( + dataset, "duplicate_bars", HARD, count=int(mask.sum()), + examples=examples, note=f"duplicate ({tcol}, symbol) rows", + ) + + +def check_non_monotonic_time( + df: pd.DataFrame, *, dataset: str = _DEFAULT_DATASET +) -> QualityFinding | None: + """Bar timestamps that step backwards within a symbol (in row order).""" + d = reset_keys(df) + tcol = _time_col(d) + if tcol is None or _SYMBOL not in d.columns: + return None + d = d.assign(_t=pd.to_datetime(d[tcol], errors="coerce")) + backward = d.groupby(_SYMBOL, sort=False)["_t"].diff() < pd.Timedelta(0) + backward = backward.fillna(False) + if not backward.any(): + return None + examples = row_examples(d, backward, [_SYMBOL, tcol]) + return make_finding( + dataset, "non_monotonic_time", HARD, count=int(backward.sum()), + examples=examples, note=f"{tcol} steps backwards within a symbol", + ) + + +def check_non_positive_ohlc( + df: pd.DataFrame, *, dataset: str = _DEFAULT_DATASET +) -> QualityFinding | None: + """Any of open/high/low/close <= 0 (or NaN).""" + d = reset_keys(df) + tcol = _time_col(d) + present = [c for c in _OHLC if c in d.columns] + if not present or tcol is None or _SYMBOL not in d.columns: + return None + vals = d[present].apply(pd.to_numeric, errors="coerce") + mask = ~(vals > 0).all(axis=1) + if not mask.any(): + return None + examples = row_examples(d, mask, [_SYMBOL, tcol]) + return make_finding( + dataset, "non_positive_ohlc", HARD, count=int(mask.sum()), + examples=examples, note="open/high/low/close must be > 0", + ) + + +def check_high_low_inversion( + df: pd.DataFrame, *, dataset: str = _DEFAULT_DATASET +) -> QualityFinding | None: + """``high < low``.""" + d = reset_keys(df) + tcol = _time_col(d) + if tcol is None or "high" not in d.columns or "low" not in d.columns: + return None + hi = pd.to_numeric(d["high"], errors="coerce") + lo = pd.to_numeric(d["low"], errors="coerce") + mask = (hi < lo).fillna(False) + if not mask.any(): + return None + examples = row_examples(d, mask, [_SYMBOL, tcol]) + return make_finding( + dataset, "high_lt_low", HARD, count=int(mask.sum()), + examples=examples, note="high must be >= low", + ) + + +def check_close_outside_range( + df: pd.DataFrame, *, dataset: str = _DEFAULT_DATASET +) -> QualityFinding | None: + """``close`` outside ``[low, high]``.""" + d = reset_keys(df) + tcol = _time_col(d) + if tcol is None or not {"low", "high", "close"}.issubset(d.columns): + return None + hi = pd.to_numeric(d["high"], errors="coerce") + lo = pd.to_numeric(d["low"], errors="coerce") + cl = pd.to_numeric(d["close"], errors="coerce") + mask = ((cl < lo) | (cl > hi)).fillna(False) + if not mask.any(): + return None + examples = row_examples(d, mask, [_SYMBOL, tcol], extra="close") + return make_finding( + dataset, "close_outside_low_high", HARD, count=int(mask.sum()), + examples=examples, note="close must be within [low, high]", + ) + + +def check_negative_volume_amount( + df: pd.DataFrame, *, dataset: str = _DEFAULT_DATASET +) -> QualityFinding | None: + """Negative ``volume`` or ``amount``.""" + d = reset_keys(df) + tcol = _time_col(d) + cols = [c for c in ("volume", "amount") if c in d.columns] + if tcol is None or not cols: + return None + vals = d[cols].apply(pd.to_numeric, errors="coerce") + mask = (vals < 0).any(axis=1) + if not mask.any(): + return None + examples = row_examples(d, mask, [_SYMBOL, tcol]) + return make_finding( + dataset, "negative_volume_amount", HARD, count=int(mask.sum()), + examples=examples, note="volume / amount must be >= 0", + ) + + +def check_missing_minutes( + df: pd.DataFrame, expected_times, *, dataset: str = _DEFAULT_DATASET +) -> QualityFinding | None: + """Bars missing per symbol vs an explicitly-provided minute calendar/window. + + WARNING severity (a gap may be a real halt / illiquid minute); reported only + when a calendar is passed. Examples are bounded ``{symbol, time}``. + """ + if expected_times is None: + return None + d = reset_keys(df) + tcol = _time_col(d) + if tcol is None or _SYMBOL not in d.columns: + return None + expected = pd.DatetimeIndex(pd.to_datetime(list(expected_times))).unique() + if len(expected) == 0: + return None + total = 0 + examples: list[dict] = [] + for sym in sorted(d[_SYMBOL].astype(str).unique()): + present = set(pd.to_datetime(d.loc[d[_SYMBOL].astype(str) == sym, tcol])) + missing = [t for t in expected if t not in present] + total += len(missing) + for t in missing: + if len(examples) < 5: + examples.append({_SYMBOL: sym, tcol: t}) + if total == 0: + return None + return make_finding( + dataset, "missing_minutes", WARNING, count=total, + examples=examples, note="bars missing vs expected minute calendar", + ) + + +def run_intraday_checks( + df: pd.DataFrame, + *, + dataset: str = _DEFAULT_DATASET, + expected_times=None, +) -> list[QualityFinding]: + """Run the 1min intraday checks; return all non-clean findings (deterministic).""" + findings = [] + for fn in ( + check_duplicate_bars, + check_non_monotonic_time, + check_non_positive_ohlc, + check_high_low_inversion, + check_close_outside_range, + check_negative_volume_amount, + ): + f = fn(df, dataset=dataset) + if f is not None: + findings.append(f) + if expected_times is not None: + mf = check_missing_minutes(df, expected_times, dataset=dataset) + if mf is not None: + findings.append(mf) + return findings diff --git a/data/quality/market.py b/data/quality/market.py new file mode 100644 index 0000000..30198d0 --- /dev/null +++ b/data/quality/market.py @@ -0,0 +1,241 @@ +"""Report-only data-quality checks for daily market / adj_factor frames (D3). + +Each check is a pure function: it takes a canonical daily frame (raw cache-shaped +``[date, symbol, open, high, low, close, volume, amount]`` or the MultiIndex +``(date, symbol)`` panel) and returns one :class:`~data.quality.report. +QualityFinding` (or ``None`` when clean). Nothing here filters the panel, repairs +values, or touches ``front_adjust`` / cache coverage — it only surfaces findings. +""" + +from __future__ import annotations + +import pandas as pd + +from data.quality._frames import reset_keys, row_examples +from data.quality.report import HARD, WARNING, QualityFinding, make_finding + +_OHLC = ["open", "high", "low", "close"] +_DATE = "date" +_SYMBOL = "symbol" + + +def check_duplicate_keys( + df: pd.DataFrame, *, dataset: str = "market_daily" +) -> QualityFinding | None: + """Duplicate ``(date, symbol)`` rows (corrupts joins / coverage assumptions).""" + d = reset_keys(df) + if _DATE not in d.columns or _SYMBOL not in d.columns: + return None + mask = d.duplicated(subset=[_DATE, _SYMBOL], keep=False) + if not mask.any(): + return None + keys = ( + d.loc[mask, [_DATE, _SYMBOL]] + .drop_duplicates() + .sort_values([_SYMBOL, _DATE]) + .head(5) + ) + examples = [{_DATE: r[_DATE], _SYMBOL: r[_SYMBOL]} for _, r in keys.iterrows()] + return make_finding( + dataset, "duplicate_keys", HARD, count=int(mask.sum()), + examples=examples, note="duplicate (date, symbol) rows", + ) + + +def check_non_positive_ohlc( + df: pd.DataFrame, *, dataset: str = "market_daily" +) -> QualityFinding | None: + """Any of open/high/low/close <= 0 (or NaN).""" + d = reset_keys(df) + present = [c for c in _OHLC if c in d.columns] + if not present or _DATE not in d.columns or _SYMBOL not in d.columns: + return None + vals = d[present].apply(pd.to_numeric, errors="coerce") + mask = ~(vals > 0).all(axis=1) + if not mask.any(): + return None + examples = row_examples(d, mask, [_SYMBOL, _DATE]) + return make_finding( + dataset, "non_positive_ohlc", HARD, count=int(mask.sum()), + examples=examples, note="open/high/low/close must be > 0", + ) + + +def check_high_low_inversion( + df: pd.DataFrame, *, dataset: str = "market_daily" +) -> QualityFinding | None: + """``high < low``.""" + d = reset_keys(df) + if "high" not in d.columns or "low" not in d.columns: + return None + hi = pd.to_numeric(d["high"], errors="coerce") + lo = pd.to_numeric(d["low"], errors="coerce") + mask = (hi < lo).fillna(False) + if not mask.any(): + return None + examples = row_examples(d, mask, [_SYMBOL, _DATE]) + return make_finding( + dataset, "high_lt_low", HARD, count=int(mask.sum()), + examples=examples, note="high must be >= low", + ) + + +def check_close_outside_range( + df: pd.DataFrame, *, dataset: str = "market_daily" +) -> QualityFinding | None: + """``close`` outside ``[low, high]``.""" + d = reset_keys(df) + needed = {"low", "high", "close"} + if not needed.issubset(d.columns): + return None + hi = pd.to_numeric(d["high"], errors="coerce") + lo = pd.to_numeric(d["low"], errors="coerce") + cl = pd.to_numeric(d["close"], errors="coerce") + mask = ((cl < lo) | (cl > hi)).fillna(False) + if not mask.any(): + return None + examples = row_examples(d, mask, [_SYMBOL, _DATE], extra="close") + return make_finding( + dataset, "close_outside_low_high", HARD, count=int(mask.sum()), + examples=examples, note="close must be within [low, high]", + ) + + +def check_negative_volume_amount( + df: pd.DataFrame, *, dataset: str = "market_daily" +) -> QualityFinding | None: + """Negative ``volume`` or ``amount``.""" + d = reset_keys(df) + cols = [c for c in ("volume", "amount") if c in d.columns] + if not cols: + return None + vals = d[cols].apply(pd.to_numeric, errors="coerce") + mask = (vals < 0).any(axis=1) + if not mask.any(): + return None + examples = row_examples(d, mask, [_SYMBOL, _DATE]) + return make_finding( + dataset, "negative_volume_amount", HARD, count=int(mask.sum()), + examples=examples, note="volume / amount must be >= 0", + ) + + +def check_adj_factor( + df: pd.DataFrame, *, dataset: str = "adj_factor" +) -> QualityFinding | None: + """Non-positive / invalid ``adj_factor``.""" + d = reset_keys(df) + if "adj_factor" not in d.columns: + return None + vals = pd.to_numeric(d["adj_factor"], errors="coerce") + mask = ~(vals > 0) + if not mask.any(): + return None + key_cols = [c for c in (_SYMBOL, _DATE) if c in d.columns] + examples = row_examples(d, mask, key_cols, extra="adj_factor") if key_cols else [] + return make_finding( + dataset, "invalid_adj_factor", HARD, count=int(mask.sum()), + examples=examples, note="adj_factor must be > 0", + ) + + +def check_extreme_returns( + df: pd.DataFrame, *, dataset: str = "market_daily", threshold: float = 0.5 +) -> QualityFinding | None: + """Suspicious raw ``close`` day-over-day moves: ``abs(pct_change) > threshold``. + + Per symbol, sorted by date. WARNING severity (suspicious, not proven wrong — + e.g. an un-adjusted split or a real streak); never used to filter the panel. + """ + d = reset_keys(df) + if not {"close", _SYMBOL, _DATE}.issubset(d.columns): + return None + d = d.sort_values([_SYMBOL, _DATE]).copy() + close_num = pd.to_numeric(d["close"], errors="coerce") + d["pct_change"] = close_num.groupby(d[_SYMBOL]).pct_change().round(4) + mask = (d["pct_change"].abs() > threshold).fillna(False) + if not mask.any(): + return None + examples = row_examples(d, mask, [_SYMBOL, _DATE], extra="pct_change") + return make_finding( + dataset, "extreme_close_move", WARNING, count=int(mask.sum()), + examples=examples, note=f"|close pct_change| > {threshold}", + ) + + +def check_missing_dates( + df: pd.DataFrame, expected_dates, *, dataset: str = "market_daily" +) -> QualityFinding | None: + """Rows missing per symbol vs an explicitly-provided ``expected_dates`` calendar. + + WARNING severity (a gap may be a legitimate halt); reported only when a + calendar is passed. Examples are bounded ``{symbol, date}`` of missing rows. + """ + if expected_dates is None: + return None + d = reset_keys(df) + if _DATE not in d.columns or _SYMBOL not in d.columns: + return None + expected = pd.DatetimeIndex(pd.to_datetime(list(expected_dates))).normalize().unique() + if len(expected) == 0: + return None + total = 0 + examples: list[dict] = [] + for sym in sorted(d[_SYMBOL].astype(str).unique()): + present = set( + pd.to_datetime(d.loc[d[_SYMBOL].astype(str) == sym, _DATE]).dt.normalize() + ) + missing = [dt for dt in expected if dt not in present] + total += len(missing) + for dt in missing: + if len(examples) < 5: + examples.append({_SYMBOL: sym, _DATE: dt}) + if total == 0: + return None + return make_finding( + dataset, "missing_dates", WARNING, count=total, + examples=examples, note="rows missing vs expected_dates calendar", + ) + + +def run_market_checks( + df: pd.DataFrame, + *, + dataset: str = "market_daily", + expected_dates=None, + return_threshold: float = 0.5, +) -> list[QualityFinding]: + """Run the OHLCV daily checks; return all non-clean findings (deterministic).""" + findings = [] + for fn in ( + check_duplicate_keys, + check_non_positive_ohlc, + check_high_low_inversion, + check_close_outside_range, + check_negative_volume_amount, + ): + f = fn(df, dataset=dataset) + if f is not None: + findings.append(f) + rf = check_extreme_returns(df, dataset=dataset, threshold=return_threshold) + if rf is not None: + findings.append(rf) + if expected_dates is not None: + mf = check_missing_dates(df, expected_dates, dataset=dataset) + if mf is not None: + findings.append(mf) + return findings + + +def run_adj_factor_checks( + df: pd.DataFrame, *, dataset: str = "adj_factor" +) -> list[QualityFinding]: + """Run the adj_factor frame checks (duplicate keys + positivity).""" + findings = [] + dup = check_duplicate_keys(df, dataset=dataset) + if dup is not None: + findings.append(dup) + adj = check_adj_factor(df, dataset=dataset) + if adj is not None: + findings.append(adj) + return findings diff --git a/data/quality/report.py b/data/quality/report.py new file mode 100644 index 0000000..8308c0c --- /dev/null +++ b/data/quality/report.py @@ -0,0 +1,159 @@ +"""Report-only data-quality findings model + deterministic renderer (D3). + +A :class:`QualityFinding` is a small immutable record describing one suspicious +pattern found in an upstream data frame. The helpers here aggregate findings into +a stable frame and render a concise, bounded, deterministic text/Markdown summary +suitable for logs or a future data-update artifact. + +This layer is REPORT-ONLY: it never mutates data, never drops/repairs rows, and +never touches cache, feed, factor, alpha, portfolio, or runtime logic. Findings +and rendered text carry only dataset/check metadata + bounded {date/time, symbol, +value} examples — never a token, a secret-file path, or an unbounded symbol dump. +""" + +from __future__ import annotations + +import numbers +from dataclasses import dataclass, field + +import pandas as pd + +# severity levels. "hard" = a severe structural finding in the report; it is NOT +# a default exception (the layer stays report-only unless a caller opts in). +INFO = "info" +WARNING = "warning" +HARD = "hard" + +_SEVERITY_ORDER = {HARD: 0, WARNING: 1, INFO: 2} + +# bound on how many example rows a finding may carry (keeps reports small + safe). +MAX_EXAMPLES = 5 + +_FRAME_COLUMNS = ["dataset", "check", "severity", "count", "examples", "note"] + + +@dataclass(frozen=True) +class QualityFinding: + """One report-only data-quality finding (immutable).""" + + dataset: str + check: str + severity: str + count: int + examples: tuple[dict, ...] = field(default_factory=tuple) + note: str | None = None + + +def clean_value(value: object) -> object: + """Coerce a cell to a small, deterministic, JSON-friendly example value. + + Timestamps -> ISO date (or date-time if intraday); numbers -> rounded + python scalars; everything else -> ``str``. No paths, no secrets. + """ + if isinstance(value, pd.Timestamp): + if value == value.normalize(): + return value.strftime("%Y-%m-%d") + return value.strftime("%Y-%m-%d %H:%M:%S") + if isinstance(value, bool): + return bool(value) + if isinstance(value, numbers.Integral): + return int(value) + if isinstance(value, numbers.Real): + return round(float(value), 6) + return str(value) + + +def make_finding( + dataset: str, + check: str, + severity: str, + count: int, + examples: list[dict] | tuple[dict, ...] = (), + note: str | None = None, +) -> QualityFinding: + """Build a finding with bounded, cleaned examples (at most ``MAX_EXAMPLES``).""" + if severity not in _SEVERITY_ORDER: + raise ValueError(f"unknown severity {severity!r}; expected info/warning/hard") + bounded = tuple( + {k: clean_value(v) for k, v in dict(ex).items()} + for ex in list(examples)[:MAX_EXAMPLES] + ) + return QualityFinding( + dataset=dataset, + check=check, + severity=severity, + count=int(count), + examples=bounded, + note=note, + ) + + +def has_hard(findings: list[QualityFinding]) -> bool: + """Whether any finding is ``hard`` (a report flag, not a default failure).""" + return any(f.severity == HARD for f in findings) + + +def sort_findings(findings: list[QualityFinding]) -> list[QualityFinding]: + """Deterministic order: severity (hard first), then dataset, then check.""" + return sorted( + findings, + key=lambda f: (_SEVERITY_ORDER.get(f.severity, 9), f.dataset, f.check), + ) + + +def findings_to_frame(findings: list[QualityFinding]) -> pd.DataFrame: + """Collect findings into a stable, deterministically-ordered DataFrame.""" + if not findings: + return pd.DataFrame(columns=_FRAME_COLUMNS) + rows = [ + { + "dataset": f.dataset, + "check": f.check, + "severity": f.severity, + "count": f.count, + "examples": f.examples, + "note": f.note, + } + for f in sort_findings(findings) + ] + return pd.DataFrame(rows, columns=_FRAME_COLUMNS) + + +def _format_examples(examples: tuple[dict, ...]) -> str: + """Compact deterministic ``k=v`` join of bounded example rows.""" + parts = [] + for ex in examples: + kv = ", ".join(f"{k}={ex[k]}" for k in ex) + parts.append(f"{{{kv}}}") + return "; ".join(parts) + + +def render_report( + findings: list[QualityFinding], *, title: str = "Data Quality Report" +) -> str: + """Render a concise, bounded, deterministic Markdown summary. + + Clean input -> an explicit "no findings" line. Otherwise one bullet per + finding (severity / dataset / check / count / note) plus its bounded examples. + """ + lines = [f"# {title}", ""] + if not findings: + lines.append("No data-quality findings. ✓") + return "\n".join(lines) + + ordered = sort_findings(findings) + n_hard = sum(1 for f in ordered if f.severity == HARD) + n_warn = sum(1 for f in ordered if f.severity == WARNING) + n_info = sum(1 for f in ordered if f.severity == INFO) + lines.append( + f"{len(ordered)} finding(s): {n_hard} hard / {n_warn} warning / {n_info} info." + ) + lines.append("") + for f in ordered: + head = f"- [{f.severity}] `{f.dataset}` / {f.check}: count={f.count}" + if f.note: + head += f" — {f.note}" + lines.append(head) + if f.examples: + lines.append(f" examples: {_format_examples(f.examples)}") + return "\n".join(lines) diff --git a/tests/test_data_quality_intraday.py b/tests/test_data_quality_intraday.py new file mode 100644 index 0000000..b7d0881 --- /dev/null +++ b/tests/test_data_quality_intraday.py @@ -0,0 +1,110 @@ +"""D3 report-only 1min intraday quality checks — synthetic, no network/cache.""" + +from __future__ import annotations + +import pandas as pd + +from data.quality.intraday import ( + check_close_outside_range, + check_duplicate_bars, + check_high_low_inversion, + check_missing_minutes, + check_negative_volume_amount, + check_non_monotonic_time, + check_non_positive_ohlc, + run_intraday_checks, +) +from data.quality.report import HARD, WARNING, has_hard + + +def _clean() -> pd.DataFrame: + """A small clean 1min frame (1 symbol x 3 consecutive minutes).""" + times = pd.to_datetime( + ["2024-01-03 09:31:00", "2024-01-03 09:32:00", "2024-01-03 09:33:00"] + ) + return pd.DataFrame( + { + "bar_end": times, + "symbol": ["000001.SZ"] * 3, + "open": [10.0, 10.1, 10.2], "high": [10.3, 10.4, 10.5], + "low": [9.9, 10.0, 10.1], "close": [10.1, 10.2, 10.3], + "volume": [100.0, 110.0, 120.0], "amount": [1000.0, 1100.0, 1200.0], + } + ) + + +def test_clean_intraday_has_zero_hard_findings(): + findings = run_intraday_checks(_clean()) + assert findings == [] + assert not has_hard(findings) + + +def test_clean_intraday_as_multiindex_also_clean(): + panel = _clean().rename(columns={"bar_end": "time"}).set_index(["time", "symbol"]) + # MultiIndex (time, symbol); reset_keys promotes -> 'time' is the bar column + assert run_intraday_checks(panel) == [] + + +def test_duplicate_bars_caught(): + df = _clean() + df2 = pd.concat([df, df.iloc[[0]]], ignore_index=True) + f = check_duplicate_bars(df2) + assert f is not None and f.severity == HARD and f.check == "duplicate_bars" + assert f.count == 2 + + +def test_non_monotonic_time_caught(): + df = _clean() + # swap rows 1 and 2 so the timestamp steps backwards within the symbol + df2 = df.iloc[[0, 2, 1]].reset_index(drop=True) + f = check_non_monotonic_time(df2) + assert f is not None and f.severity == HARD and f.count == 1 + + +def test_non_monotonic_clean_in_order(): + assert check_non_monotonic_time(_clean()) is None + + +def test_non_positive_ohlc_caught(): + df = _clean() + df.loc[0, "low"] = 0.0 + f = check_non_positive_ohlc(df) + assert f is not None and f.severity == HARD and f.count == 1 + + +def test_high_low_inversion_caught(): + df = _clean() + df.loc[1, "high"] = 1.0 + df.loc[1, "low"] = 5.0 + f = check_high_low_inversion(df) + assert f is not None and f.severity == HARD and f.count == 1 + + +def test_close_outside_range_caught(): + df = _clean() + df.loc[2, "close"] = 999.0 + f = check_close_outside_range(df) + assert f is not None and f.severity == HARD and f.count == 1 + + +def test_negative_volume_amount_caught(): + df = _clean() + df.loc[0, "amount"] = -1.0 + f = check_negative_volume_amount(df) + assert f is not None and f.severity == HARD and f.count == 1 + + +def test_missing_minutes_caught_with_calendar(): + df = _clean() + # drop the 09:32 bar + df = df[df["bar_end"] != pd.Timestamp("2024-01-03 09:32:00")] + cal = [ + "2024-01-03 09:31:00", "2024-01-03 09:32:00", "2024-01-03 09:33:00", + ] + f = check_missing_minutes(df, cal) + assert f is not None and f.severity == WARNING and f.count == 1 + assert f.examples[0]["symbol"] == "000001.SZ" + + +def test_missing_minutes_none_without_calendar(): + assert check_missing_minutes(_clean(), None) is None diff --git a/tests/test_data_quality_market.py b/tests/test_data_quality_market.py new file mode 100644 index 0000000..1cf3830 --- /dev/null +++ b/tests/test_data_quality_market.py @@ -0,0 +1,145 @@ +"""D3 report-only daily-market quality checks — synthetic, no network/cache.""" + +from __future__ import annotations + +import pandas as pd + +from data.quality.market import ( + check_adj_factor, + check_close_outside_range, + check_duplicate_keys, + check_extreme_returns, + check_high_low_inversion, + check_missing_dates, + check_negative_volume_amount, + check_non_positive_ohlc, + run_adj_factor_checks, + run_market_checks, +) +from data.quality.report import HARD, WARNING, has_hard + + +def _clean(rows=None) -> pd.DataFrame: + """A small clean daily panel (2 symbols x 3 days), no quality issues.""" + data = [] + for sym, base in (("000001.SZ", 10.0), ("000002.SZ", 20.0)): + for i, d in enumerate(("2024-01-03", "2024-01-04", "2024-01-05")): + px = base + i * 0.1 + data.append( + { + "date": pd.Timestamp(d), "symbol": sym, + "open": px, "high": px + 0.5, "low": px - 0.5, "close": px, + "volume": 1000.0 + i, "amount": 1_000_000.0 + i, + } + ) + return pd.DataFrame(data if rows is None else rows) + + +def test_clean_panel_has_zero_hard_findings(): + findings = run_market_checks(_clean()) + assert findings == [] + assert not has_hard(findings) + + +def test_clean_panel_as_multiindex_also_clean(): + panel = _clean().set_index(["date", "symbol"]) + assert run_market_checks(panel) == [] + + +def test_duplicate_keys_caught(): + df = _clean() + dup = df.iloc[[0]].copy() + df2 = pd.concat([df, dup], ignore_index=True) + f = check_duplicate_keys(df2) + assert f is not None and f.severity == HARD and f.check == "duplicate_keys" + assert f.count == 2 # the original + the duplicate row + + +def test_non_positive_ohlc_caught(): + df = _clean() + df.loc[0, "close"] = 0.0 + f = check_non_positive_ohlc(df) + assert f is not None and f.severity == HARD + assert f.count == 1 + assert f.examples[0]["symbol"] == "000001.SZ" + + +def test_high_low_inversion_caught(): + df = _clean() + df.loc[2, "high"] = 1.0 + df.loc[2, "low"] = 5.0 + f = check_high_low_inversion(df) + assert f is not None and f.severity == HARD and f.count == 1 + + +def test_close_outside_range_caught(): + df = _clean() + df.loc[1, "close"] = 999.0 # above high + f = check_close_outside_range(df) + assert f is not None and f.severity == HARD and f.count == 1 + assert f.examples[0]["close"] == 999.0 + + +def test_negative_volume_amount_caught(): + df = _clean() + df.loc[3, "volume"] = -5.0 + f = check_negative_volume_amount(df) + assert f is not None and f.severity == HARD and f.count == 1 + + +def test_extreme_returns_caught_as_warning(): + df = _clean() + # 000001.SZ close jumps 10.x -> 50 on day 2 (pct_change > 0.5) + df.loc[1, "close"] = 50.0 + df.loc[1, "high"] = 50.5 + f = check_extreme_returns(df, threshold=0.5) + assert f is not None and f.severity == WARNING + assert f.count >= 1 + assert "pct_change" in f.examples[0] + + +def test_extreme_returns_clean_when_below_threshold(): + assert check_extreme_returns(_clean(), threshold=0.5) is None + + +def test_adj_factor_non_positive_caught(): + adj = pd.DataFrame( + { + "date": [pd.Timestamp("2024-01-03"), pd.Timestamp("2024-01-04")], + "symbol": ["000001.SZ", "000001.SZ"], + "adj_factor": [1.0, 0.0], # second is invalid + } + ) + f = check_adj_factor(adj) + assert f is not None and f.severity == HARD and f.count == 1 + findings = run_adj_factor_checks(adj) + assert any(x.check == "invalid_adj_factor" for x in findings) + + +def test_missing_dates_caught_with_calendar(): + df = _clean() + # drop 000001.SZ on 2024-01-04 + df = df[~((df["symbol"] == "000001.SZ") & (df["date"] == pd.Timestamp("2024-01-04")))] + cal = ["2024-01-03", "2024-01-04", "2024-01-05"] + f = check_missing_dates(df, cal) + assert f is not None and f.severity == WARNING and f.count == 1 + assert f.examples[0] == {"symbol": "000001.SZ", "date": "2024-01-04"} + + +def test_missing_dates_none_without_calendar(): + assert check_missing_dates(_clean(), None) is None + + +def test_examples_bounded_to_five(): + # 8 rows all with close == 0 -> finding capped at 5 examples + rows = [ + { + "date": pd.Timestamp("2024-01-03"), "symbol": f"{i:06d}.SZ", + "open": 1.0, "high": 1.0, "low": 1.0, "close": 0.0, + "volume": 1.0, "amount": 1.0, + } + for i in range(8) + ] + f = check_non_positive_ohlc(pd.DataFrame(rows)) + assert f.count == 8 + assert len(f.examples) == 5 # bounded diff --git a/tests/test_data_quality_report.py b/tests/test_data_quality_report.py new file mode 100644 index 0000000..9c62505 --- /dev/null +++ b/tests/test_data_quality_report.py @@ -0,0 +1,104 @@ +"""D3 report-only findings model + renderer — deterministic, bounded, secret-free.""" + +from __future__ import annotations + +import pandas as pd + +from data.quality.report import ( + HARD, + INFO, + WARNING, + clean_value, + findings_to_frame, + has_hard, + make_finding, + render_report, + sort_findings, +) + + +def test_make_finding_bounds_examples_to_five(): + examples = [{"symbol": f"{i:06d}.SZ", "date": pd.Timestamp("2024-01-03")} for i in range(9)] + f = make_finding("market_daily", "non_positive_ohlc", HARD, count=9, examples=examples) + assert len(f.examples) == 5 + assert f.count == 9 + + +def test_make_finding_cleans_values(): + f = make_finding( + "market_daily", "x", WARNING, count=1, + examples=[{"date": pd.Timestamp("2024-01-03"), "v": 1.23456789}], + ) + assert f.examples[0]["date"] == "2024-01-03" + assert f.examples[0]["v"] == round(1.23456789, 6) + + +def test_clean_value_formats(): + assert clean_value(pd.Timestamp("2024-01-03")) == "2024-01-03" + assert clean_value(pd.Timestamp("2024-01-03 09:31:00")) == "2024-01-03 09:31:00" + assert clean_value(5) == 5 + assert clean_value(1.0 / 3.0) == round(1.0 / 3.0, 6) + + +def test_has_hard(): + assert has_hard([make_finding("d", "c", HARD, 1)]) + assert not has_hard([make_finding("d", "c", WARNING, 1)]) + assert not has_hard([]) + + +def test_findings_to_frame_stable_columns_and_order(): + findings = [ + make_finding("market_daily", "extreme_close_move", WARNING, 2), + make_finding("adj_factor", "invalid_adj_factor", HARD, 1), + make_finding("market_daily", "non_positive_ohlc", HARD, 3), + ] + frame = findings_to_frame(findings) + assert list(frame.columns) == ["dataset", "check", "severity", "count", "examples", "note"] + # hard first, then by dataset, then check + assert frame["severity"].tolist() == ["hard", "hard", "warning"] + assert frame.iloc[0]["dataset"] == "adj_factor" + assert frame.iloc[1]["check"] == "non_positive_ohlc" + + +def test_findings_to_frame_empty(): + frame = findings_to_frame([]) + assert list(frame.columns) == ["dataset", "check", "severity", "count", "examples", "note"] + assert len(frame) == 0 + + +def test_render_clean_is_explicit(): + out = render_report([]) + assert "No data-quality findings" in out + assert out.startswith("# Data Quality Report") + + +def test_render_is_deterministic_and_bounded(): + findings = [ + make_finding( + "market_daily", "non_positive_ohlc", HARD, count=2, + examples=[{"symbol": "000001.SZ", "date": pd.Timestamp("2024-01-03")}], + note="open/high/low/close must be > 0", + ), + make_finding("market_daily", "extreme_close_move", WARNING, count=1), + ] + out1 = render_report(findings) + out2 = render_report(list(reversed(findings))) + assert out1 == out2 # deterministic regardless of input order + assert "[hard] `market_daily` / non_positive_ohlc: count=2" in out1 + assert "1 hard / 1 warning / 0 info" in out1 + + +def test_render_has_no_secret_looking_paths(): + findings = [make_finding("market_daily", "non_positive_ohlc", HARD, 1, + examples=[{"symbol": "000001.SZ", "date": pd.Timestamp("2024-01-03")}])] + out = render_report(findings) + assert ".config.json" not in out + assert "tushare.token" not in out + assert "secret" not in out.lower() + + +def test_sort_findings_severity_first(): + findings = [make_finding("z", "c", INFO, 1), make_finding("a", "c", HARD, 1)] + ordered = sort_findings(findings) + assert ordered[0].severity == HARD + assert ordered[1].severity == INFO From 08919e9c160b5dcf758d4faa9b9d534db193a98f Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Fri, 19 Jun 2026 16:45:49 +0800 Subject: [PATCH 2/2] fix(data): redact secret-looking strings from quality report (D3 review fix) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex review found a contract gap: render_report() (and stored findings) could emit a secret-looking config path or token key if a caller passed one in a finding note or example value — the checks never do, but the public report API did not defend against it. Add a small, deterministic report-safety guard in data/quality/report.py: - sanitize_text() redacts -> "[REDACTED]": any path ending in .config.json, the literal tushare.token, and obvious token=.../token: ... values. Idempotent; benign values (symbols, ISO dates, numbers) are untouched. - applied at the three string entry points: clean_value (string example values), make_finding (note), and _format_examples (rendered output, so a secret-looking example KEY is covered too). Update tests/test_data_quality_report.py to pass secret-looking inputs (a full .config.json path + tushare.token in example values AND in note) and assert they are redacted from both the stored finding and the rendered report, while benign symbols/ISO dates still render and examples stay bounded to 5. Report-only and scope-tight: no data-update, config, cache/ledger, feed, factor/ alpha/portfolio/runtime, or PanelStore change; D3 stays library + tests only. --- data/quality/report.py | 37 ++++++++++++++-- tests/test_data_quality_report.py | 72 +++++++++++++++++++++++++++++-- 2 files changed, 101 insertions(+), 8 deletions(-) diff --git a/data/quality/report.py b/data/quality/report.py index 8308c0c..7d29eb4 100644 --- a/data/quality/report.py +++ b/data/quality/report.py @@ -14,6 +14,7 @@ from __future__ import annotations import numbers +import re from dataclasses import dataclass, field import pandas as pd @@ -31,6 +32,30 @@ _FRAME_COLUMNS = ["dataset", "check", "severity", "count", "examples", "note"] +# Defensive report-safety guard (NOT a secrets-detection product): redact +# secret-looking strings that a caller might pass into a finding note or example +# value, so a rendered/stored report never carries a token-file path or key. +# Deterministic + order-independent: each pattern -> a neutral marker. +_REDACTED = "[REDACTED]" +_SECRET_PATTERNS = ( + re.compile(r"\S*\.config\.json", re.IGNORECASE), # any path ending in .config.json + re.compile(r"tushare\.token", re.IGNORECASE), # the secret config key + re.compile(r"token\s*[=:]\s*\S+", re.IGNORECASE), # token=... / token: ... values +) + + +def sanitize_text(value: str) -> str: + """Redact secret-looking substrings (config paths / token keys/values). + + A bounded report-safety guard applied to every string that may reach a + rendered or stored finding. Benign values (symbols, ISO dates, numbers) are + untouched. Idempotent and deterministic. + """ + out = value + for pattern in _SECRET_PATTERNS: + out = pattern.sub(_REDACTED, out) + return out + @dataclass(frozen=True) class QualityFinding: @@ -60,7 +85,7 @@ def clean_value(value: object) -> object: return int(value) if isinstance(value, numbers.Real): return round(float(value), 6) - return str(value) + return sanitize_text(str(value)) def make_finding( @@ -84,7 +109,7 @@ def make_finding( severity=severity, count=int(count), examples=bounded, - note=note, + note=sanitize_text(note) if note is not None else None, ) @@ -120,12 +145,16 @@ def findings_to_frame(findings: list[QualityFinding]) -> pd.DataFrame: def _format_examples(examples: tuple[dict, ...]) -> str: - """Compact deterministic ``k=v`` join of bounded example rows.""" + """Compact deterministic ``k=v`` join of bounded example rows. + + Sanitized once more at render time so a secret-looking example KEY (not just + a value) can never reach the output. + """ parts = [] for ex in examples: kv = ", ".join(f"{k}={ex[k]}" for k in ex) parts.append(f"{{{kv}}}") - return "; ".join(parts) + return sanitize_text("; ".join(parts)) def render_report( diff --git a/tests/test_data_quality_report.py b/tests/test_data_quality_report.py index 9c62505..f41ee22 100644 --- a/tests/test_data_quality_report.py +++ b/tests/test_data_quality_report.py @@ -13,9 +13,15 @@ has_hard, make_finding, render_report, + sanitize_text, sort_findings, ) +# Secret-looking inputs a caller might (wrongly) pass into a finding; the report +# layer must redact them. These are SCAN TARGETS, not real secrets. +_SECRET_PATH = "/home/shaofl/Projects/financial_projects/.config.json" +_SECRET_KEY = "tushare.token" + def test_make_finding_bounds_examples_to_five(): examples = [{"symbol": f"{i:06d}.SZ", "date": pd.Timestamp("2024-01-03")} for i in range(9)] @@ -89,12 +95,70 @@ def test_render_is_deterministic_and_bounded(): def test_render_has_no_secret_looking_paths(): - findings = [make_finding("market_daily", "non_positive_ohlc", HARD, 1, - examples=[{"symbol": "000001.SZ", "date": pd.Timestamp("2024-01-03")}])] + """Secret-looking path/key passed into a finding must be redacted from output.""" + findings = [ + make_finding( + "market_daily", "non_positive_ohlc", HARD, 1, + examples=[{"path": _SECRET_PATH, "key": _SECRET_KEY, "symbol": "000001.SZ"}], + note=f"leaked at {_SECRET_PATH}", + ) + ] out = render_report(findings) + assert _SECRET_PATH not in out assert ".config.json" not in out - assert "tushare.token" not in out - assert "secret" not in out.lower() + assert _SECRET_KEY not in out + assert "[REDACTED]" in out + # benign content in the same finding still renders + assert "000001.SZ" in out + + +def test_secret_path_in_example_value_redacted(): + f = make_finding("x", "y", HARD, 1, examples=[{"path": _SECRET_PATH}]) + assert f.examples[0]["path"] == "[REDACTED]" + assert _SECRET_PATH not in render_report([f]) + + +def test_tushare_token_in_example_value_redacted(): + f = make_finding("x", "y", HARD, 1, examples=[{"key": _SECRET_KEY}]) + assert f.examples[0]["key"] == "[REDACTED]" + assert _SECRET_KEY not in render_report([f]) + + +def test_secret_path_in_note_redacted(): + f = make_finding("x", "y", HARD, 1, note=f"see {_SECRET_PATH} for the token") + assert _SECRET_PATH not in (f.note or "") + assert "[REDACTED]" in (f.note or "") + assert _SECRET_PATH not in render_report([f]) + + +def test_token_kv_value_redacted(): + out = render_report([make_finding("x", "y", HARD, 1, examples=[{"v": "token=abc123secret"}])]) + assert "abc123secret" not in out + + +def test_benign_values_still_render(): + f = make_finding( + "market_daily", "high_lt_low", HARD, 1, + examples=[{"symbol": "000001.SZ", "date": pd.Timestamp("2024-01-03")}], + ) + out = render_report([f]) + assert "000001.SZ" in out + assert "2024-01-03" in out + assert "[REDACTED]" not in out + + +def test_sanitize_text_is_idempotent_and_deterministic(): + once = sanitize_text(f"path {_SECRET_PATH} key {_SECRET_KEY}") + twice = sanitize_text(once) + assert once == twice + assert _SECRET_PATH not in once and _SECRET_KEY not in once + + +def test_secret_inputs_examples_still_bounded_to_five(): + examples = [{"path": _SECRET_PATH, "i": i} for i in range(9)] + f = make_finding("x", "y", HARD, 9, examples=examples) + assert len(f.examples) == 5 + assert all(ex["path"] == "[REDACTED]" for ex in f.examples) def test_sort_findings_severity_first():