diff --git a/.github/workflows/gauntlet.yml b/.github/workflows/gauntlet.yml new file mode 100644 index 0000000..ad0b2fb --- /dev/null +++ b/.github/workflows/gauntlet.yml @@ -0,0 +1,59 @@ +# Validation Gauntlet: gold-labelled disposition benchmark for the validation, +# domain and text-cleaning surfaces. Runs the lightweight fixtures on every PR +# and gates on the absolute thresholds plus no-regression vs the stored +# baseline (benchmarks/gauntlet/baseline.json). Heavier sizes stay manual. +name: Validation Gauntlet + +on: + pull_request: + paths-ignore: + - "docs/**" + - "*.md" + workflow_dispatch: + inputs: + rows: + description: "rows per fixture" + default: "300" + update_baseline: + description: "re-pin baseline.json from this run (commit it manually)" + type: boolean + default: false + +permissions: + contents: read + +jobs: + gauntlet: + runs-on: ubuntu-latest + timeout-minutes: 20 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: pip + - name: Install + run: | + python -m pip install --upgrade pip + pip install -e ".[dev]" + - name: Run gauntlet with gates + run: | + ROWS="${{ github.event.inputs.rows || '300' }}" + EXTRA="" + if [ "${{ github.event.inputs.update_baseline }}" = "true" ]; then + EXTRA="--update-baseline" + fi + python -m benchmarks.gauntlet run --rows "$ROWS" --check $EXTRA + - name: Job summary + if: always() + run: | + if [ -f benchmarks/gauntlet/results/gauntlet.md ]; then + cat benchmarks/gauntlet/results/gauntlet.md >> "$GITHUB_STEP_SUMMARY" + fi + - name: Upload results + if: always() + uses: actions/upload-artifact@v4 + with: + name: gauntlet-results + path: benchmarks/gauntlet/results/ + if-no-files-found: warn diff --git a/.gitignore b/.gitignore index 06efe6b..1efa0b6 100644 --- a/.gitignore +++ b/.gitignore @@ -34,3 +34,4 @@ crates/freshcore/target/ # dev artifact (regenerated by running teacher tasks), not committed content. training/cache/* !training/cache/.gitkeep +benchmarks/gauntlet/results/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 698314b..f5f2886 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,63 @@ adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Added +- **Validation Gauntlet** (`benchmarks/gauntlet/`, `docs/validation-gauntlet.md`): + a gold-labelled disposition benchmark for the validation, domain and + text-cleaning surfaces. Five deterministic fixtures (finance, healthcare, + CRM, e-commerce, adversarial text) label every injected defect with the + disposition FreshData should choose (preserve / repair / flag / review) and + the harness scores detection P/R/F1, repair accuracy, review routing, + preservation, corruption, escapes, false positives, audit completeness, + determinism, trust monotonicity and runtime/memory. Runs on every PR + (`gauntlet.yml`) with absolute gates plus no-regression checks against the + stored `baseline.json`. +- `CleanReport.coerced_cells`: per-cell record (`{column: {row: original}}`) + of values that `fix_dtypes` nulled because they did not parse as the + column's inferred type — the recovery source for quarantined cells, also + included in `report.to_dict()`. +- Date-field range validation in `fd.validate_fields`: `FieldSpec.min_value` + / `max_value` now accept a date string or timestamp for `date`/`datetime` + fields, so a future date of birth or an 1875 admission date is flagged as a + `domain_mismatch` (gauntlet finding). +- Case-variant vocabulary suggestions in `fd.validate_fields`: a value that + matches an `allowed_values` entry except for case (`ACTIVE` vs `active`) is + no longer silently accepted — it gets a warning-severity issue with the + canonical form as `suggestion` and action `accept_with_warning` (gauntlet + finding). + ### Fixed +- **Unparseable values are quarantined, never fabricated** (gauntlet finding, + the `'apple'`-in-a-price-column case): when `fix_dtypes` converts a + mostly-numeric (or datetime) text column, cells that fail to parse used to + become `NaN` and then be silently imputed by the auto engine — turning + junk into a fabricated median. They now stay missing, are excluded from + auto-imputation, keep their originals in `report.coerced_cells`, and the + decision is a `human_review` action in the audit trail. Genuine missing + values (true `NaN`, sentinels like `"N/A"`) keep the documented + auto-impute behaviour, and an explicit `impute=` request still fills + everything. +- Formatted-number stragglers (`"$1,234.56"`, `"1,200,500.00"`) in a + mostly-plain numeric column are now parsed by the existing locale-aware + rescue instead of being coerced to missing — the rescue previously only + engaged when the plain parse failed the threshold entirely (gauntlet + finding). +- `fd.validate_fields` consensus inference now honours the same + contamination boundary as the `fix_dtypes` warning that points users at it + (dominant share ≥ 60% with at most a handful of stragglers). Previously + the warning fired from a 60% parse share but the consensus gate required + 80%, so the exact frame the warning named sailed through + `validate_fields` silently (gauntlet finding). +- Explicitly allowed values are no longer swallowed by null-marker + heuristics in `fd.validate_fields`: with + `FieldSpec(allowed_values={"US", "DE", "NA"})`, `"NA"` is Namibia, not a + missing value (gauntlet finding). +- `clean_text` / `validate_fields` text normalization no longer rewrites + typography in content-bearing fields: for `free_text`, `text` and entity + name types, the punctuation→ASCII mapping (curly quotes, em-dashes, prime + marks — `12″` became `12"`) is withheld, matching the field-aware safety + contract. Untyped columns keep the existing behaviour (gauntlet finding). + - `anonymize()` called with no `rules` and no `detection_config` now emits a `UserWarning` instead of silently returning the data unchanged — a privacy call that does nothing must say so. Behavior is otherwise diff --git a/benchmarks/README.md b/benchmarks/README.md index fadd5a6..c8ae4dd 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -9,6 +9,12 @@ pyjanitor baselines. > The harness calls FreshData exactly as a user would. It never modifies library > internals. +> Sibling harness: `benchmarks/gauntlet/` (the **Validation Gauntlet**) +> scores per-cell dispositions — preserve / repair / flag / review — for +> the validation, domain and text-cleaning surfaces against gold labels, +> and gates every PR via `.github/workflows/gauntlet.yml`. See +> `docs/validation-gauntlet.md`. + ## Layout ``` diff --git a/benchmarks/gauntlet/__init__.py b/benchmarks/gauntlet/__init__.py new file mode 100644 index 0000000..a845793 --- /dev/null +++ b/benchmarks/gauntlet/__init__.py @@ -0,0 +1,28 @@ +"""FreshData Validation Gauntlet. + +Gold-labelled adversarial fixtures plus a harness that measures how +FreshData's validation surfaces (``fd.clean``, ``fd.validate_fields``, +``fd.clean_text``, domain packs, the semantic layer, PII detection) treat +each labelled cell: preserve, repair, flag, or route to review. + +Unlike CleanBench (which scores whole-frame repair fidelity against a clean +oracle), the gauntlet scores *dispositions*: every injected defect carries the +disposition FreshData should choose, and every adversarial trap is a valid +value that must survive cleaning untouched. + +Run ``python -m benchmarks.gauntlet run`` from the repo root. +""" + +from .fixtures import FIXTURES, GauntletFixture, GoldCell, build_fixture +from .metrics import compute_metrics +from .runner import run_fixture, run_gauntlet + +__all__ = [ + "FIXTURES", + "GauntletFixture", + "GoldCell", + "build_fixture", + "compute_metrics", + "run_fixture", + "run_gauntlet", +] diff --git a/benchmarks/gauntlet/__main__.py b/benchmarks/gauntlet/__main__.py new file mode 100644 index 0000000..0ee2dd1 --- /dev/null +++ b/benchmarks/gauntlet/__main__.py @@ -0,0 +1,67 @@ +"""Validation Gauntlet CLI. + +Run from the repo root:: + + python -m benchmarks.gauntlet run # run + write results/ + python -m benchmarks.gauntlet run --check # also gate (CI mode) + python -m benchmarks.gauntlet run --update-baseline +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +from .fixtures import DEFAULT_ROWS, DEFAULT_SEED, FIXTURES +from .metrics import compute_metrics +from .report import check_gates, render_markdown, results_payload, write_json +from .runner import run_gauntlet + +RESULTS_DIR = Path(__file__).parent / "results" +BASELINE_PATH = Path(__file__).parent / "baseline.json" + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="python -m benchmarks.gauntlet") + sub = parser.add_subparsers(dest="command", required=True) + run_p = sub.add_parser("run", help="run the gauntlet and write JSON + Markdown") + run_p.add_argument("--rows", type=int, default=DEFAULT_ROWS) + run_p.add_argument("--seed", type=int, default=DEFAULT_SEED) + run_p.add_argument("--fixtures", nargs="*", choices=sorted(FIXTURES)) + run_p.add_argument("--check", action="store_true", + help="exit 1 when a gate fails or the baseline regresses") + run_p.add_argument("--update-baseline", action="store_true", + help="write this run as the stored baseline") + args = parser.parse_args(argv) + + runs = run_gauntlet(n_rows=args.rows, seed=args.seed, fixtures=args.fixtures) + metrics = {name: compute_metrics(r) for name, r in runs.items()} + payload = results_payload(metrics, n_rows=args.rows, seed=args.seed) + + write_json(payload, RESULTS_DIR / "gauntlet.json") + markdown = render_markdown(payload) + (RESULTS_DIR / "gauntlet.md").write_text(markdown) + print(markdown) + print(f"results: {RESULTS_DIR / 'gauntlet.json'}") + + if args.update_baseline: + write_json(payload, BASELINE_PATH) + print(f"baseline updated: {BASELINE_PATH}") + + if args.check: + baseline = (json.loads(BASELINE_PATH.read_text()) + if BASELINE_PATH.exists() else None) + problems = check_gates(payload, baseline) + if problems: + print("\nGATE FAILURES:", file=sys.stderr) + for p in problems: + print(f" - {p}", file=sys.stderr) + return 1 + print("all gates passed") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/gauntlet/baseline.json b/benchmarks/gauntlet/baseline.json new file mode 100644 index 0000000..f22eed5 --- /dev/null +++ b/benchmarks/gauntlet/baseline.json @@ -0,0 +1,228 @@ +{ + "schema_version": 1, + "generated_at": "2026-07-13T04:14:15+00:00", + "n_rows": 300, + "seed": 42, + "fixtures": { + "crm": { + "fixture": "crm", + "n_rows": 304, + "labelled_cells": 19, + "detection": { + "precision": 1.0, + "recall": 0.9167, + "f1": 0.9565, + "true_positives": 11, + "false_negatives": 1, + "false_positives": 0 + }, + "repair_accuracy": 1.0, + "repair_sources": { + "clean": 3, + "clean_text_opt_in": 1 + }, + "review_routing": 1.0, + "preservation_rate": 1.0, + "corruption_count": 0, + "escape_rate": 0.0833, + "false_positive_rate": 0.0, + "duplicates": { + "expected": 4, + "removed": 4, + "ok": true + }, + "audit_completeness": 1.0, + "deterministic": true, + "trust": { + "pristine_frame": 100.0, + "dirty_frame": 99.688, + "monotonic": true + }, + "performance": { + "clean_seconds": 0.2098, + "validate_seconds": 0.0133, + "peak_memory_mb": 0.269 + }, + "failures": [ + { + "row": 24, + "column": "full_name", + "kind": "injection_text", + "expect": "flag", + "detected": false, + "dirty": "\"ROBERT'); DROP TABLE users;--\"", + "verdict": "escaped" + } + ], + "detected_only": [] + }, + "ecommerce": { + "fixture": "ecommerce", + "n_rows": 303, + "labelled_cells": 15, + "detection": { + "precision": 1.0, + "recall": 1.0, + "f1": 1.0, + "true_positives": 11, + "false_negatives": 0, + "false_positives": 0 + }, + "repair_accuracy": 1.0, + "repair_sources": { + "clean": 2, + "clean_text_opt_in": 1 + }, + "review_routing": 1.0, + "preservation_rate": 1.0, + "corruption_count": 0, + "escape_rate": 0.0, + "false_positive_rate": 0.0, + "duplicates": { + "expected": 3, + "removed": 3, + "ok": true + }, + "audit_completeness": 1.0, + "deterministic": true, + "trust": { + "pristine_frame": 100.0, + "dirty_frame": 94.74, + "monotonic": true + }, + "performance": { + "clean_seconds": 0.1883, + "validate_seconds": 0.0189, + "peak_memory_mb": 0.157 + }, + "failures": [], + "detected_only": [] + }, + "finance": { + "fixture": "finance", + "n_rows": 303, + "labelled_cells": 25, + "detection": { + "precision": 1.0, + "recall": 1.0, + "f1": 1.0, + "true_positives": 21, + "false_negatives": 0, + "false_positives": 0 + }, + "repair_accuracy": 1.0, + "repair_sources": { + "clean": 10 + }, + "review_routing": 1.0, + "preservation_rate": 1.0, + "corruption_count": 0, + "escape_rate": 0.0, + "false_positive_rate": 0.0, + "duplicates": { + "expected": 3, + "removed": 3, + "ok": true + }, + "audit_completeness": 1.0, + "deterministic": true, + "trust": { + "pristine_frame": 100.0, + "dirty_frame": 92.178, + "monotonic": true + }, + "performance": { + "clean_seconds": 0.223, + "validate_seconds": 0.021, + "peak_memory_mb": 0.181 + }, + "failures": [], + "detected_only": [] + }, + "healthcare": { + "fixture": "healthcare", + "n_rows": 303, + "labelled_cells": 16, + "detection": { + "precision": 1.0, + "recall": 1.0, + "f1": 1.0, + "true_positives": 15, + "false_negatives": 0, + "false_positives": 0 + }, + "repair_accuracy": 1.0, + "repair_sources": { + "clean": 1, + "validate_fields": 1 + }, + "review_routing": 1.0, + "preservation_rate": 1.0, + "corruption_count": 0, + "escape_rate": 0.0, + "false_positive_rate": 0.0, + "duplicates": { + "expected": 3, + "removed": 3, + "ok": true + }, + "audit_completeness": 1.0, + "deterministic": true, + "trust": { + "pristine_frame": 100.0, + "dirty_frame": 94.74, + "monotonic": true + }, + "performance": { + "clean_seconds": 0.2001, + "validate_seconds": 0.0217, + "peak_memory_mb": 0.16 + }, + "failures": [], + "detected_only": [] + }, + "text": { + "fixture": "text", + "n_rows": 302, + "labelled_cells": 16, + "detection": { + "precision": 1.0, + "recall": 1.0, + "f1": 1.0, + "true_positives": 10, + "false_negatives": 0, + "false_positives": 0 + }, + "repair_accuracy": 1.0, + "repair_sources": { + "clean": 1, + "clean_text_opt_in": 2, + "validate_fields": 4 + }, + "review_routing": null, + "preservation_rate": 1.0, + "corruption_count": 0, + "escape_rate": 0.0, + "false_positive_rate": 0.0, + "duplicates": { + "expected": 2, + "removed": 2, + "ok": true + }, + "audit_completeness": 1.0, + "deterministic": true, + "trust": { + "pristine_frame": 100.0, + "dirty_frame": 99.801, + "monotonic": true + }, + "performance": { + "clean_seconds": 0.0689, + "validate_seconds": 0.0054, + "peak_memory_mb": 0.103 + }, + "failures": [], + "detected_only": [] + } + } +} diff --git a/benchmarks/gauntlet/fixtures.py b/benchmarks/gauntlet/fixtures.py new file mode 100644 index 0000000..b1c8702 --- /dev/null +++ b/benchmarks/gauntlet/fixtures.py @@ -0,0 +1,520 @@ +"""Gold-labelled gauntlet fixtures. + +Every fixture is deterministic for a given seed: a base frame of valid +records, plus injected problem cells whose *expected disposition* is known. + +Dispositions (``GoldCell.expect``): + +- ``preserve`` — valid (often unusual) data; must survive every cleaning + surface byte-identical and must not draw an error-severity issue. +- ``repair`` — a safe deterministic repair exists; ``repaired`` is the gold + value. Silently leaving it is an escape; changing it to anything else is a + corruption. +- ``flag`` — must be *detected* (issue / warning / finding) but never + auto-changed by the default pipeline. +- ``review`` — ambiguous; must be routed to a quarantine / manual-review + action, never auto-deleted or auto-accepted. + +Whole-row duplicate injections are tracked separately in ``dup_row_count`` +because row removal is a row-level (not cell-level) disposition. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Callable + +import numpy as np +import pandas as pd + +from freshdata import FieldSpec + +DEFAULT_ROWS = 300 +DEFAULT_SEED = 42 + + +@dataclass(frozen=True) +class GoldCell: + row: int + column: str + kind: str #: defect / trap family, e.g. "text_in_numeric" + dirty: Any #: the value placed in the frame + expect: str #: preserve | repair | flag | review + repaired: Any = None + #: the cell may legitimately end up imputed by the audited missing-value + #: engine after its repair-to-missing (sentinels, empty strings): both + #: "left missing" and "audited fill" count as the gold repair. + accept_impute: bool = False + pii: str | None = None #: expected PII entity_type, when the cell is PII + note: str = "" + replaced: Any = None #: the valid value the injection overwrote + + +@dataclass +class GauntletFixture: + name: str + df: pd.DataFrame + cells: list[GoldCell] + schema: dict[str, FieldSpec | str] + field_types: dict[str, str] = field(default_factory=dict) + domain: str | None = None + dup_row_count: int = 0 + n_rows: int = 0 + + def labelled(self, *expects: str) -> list[GoldCell]: + return [c for c in self.cells if not expects or c.expect in expects] + + def pristine(self) -> pd.DataFrame: + """The base frame before injection: labelled cells restored, dups gone.""" + base = self.df.iloc[: self.n_rows].copy() + for c in self.cells: + base.iloc[c.row, base.columns.get_loc(c.column)] = c.replaced + return base + + +class _Injector: + """Places labelled values on distinct (row, column) slots, deterministically.""" + + def __init__(self, df: pd.DataFrame, seed: int) -> None: + self.df = df + self.cells: list[GoldCell] = [] + self._order = { + col: list(np.random.default_rng(seed + i).permutation(len(df))) + for i, col in enumerate(df.columns) + } + + def place(self, column: str, dirty: Any, kind: str, expect: str, *, + repaired: Any = None, accept_impute: bool = False, + pii: str | None = None, note: str = "") -> int: + row = self._order[column].pop() + loc = self.df.columns.get_loc(column) + replaced = self.df.iloc[row, loc] + self.df.iloc[row, loc] = dirty + self.cells.append(GoldCell(row=row, column=column, kind=kind, dirty=dirty, + expect=expect, repaired=repaired, + accept_impute=accept_impute, pii=pii, note=note, + replaced=replaced)) + return row + + +def _rng(seed: int) -> np.random.Generator: + return np.random.default_rng(seed) + + +# --------------------------------------------------------------------------- +# finance +# --------------------------------------------------------------------------- + +_TICKERS = ("AAPL", "MSFT", "TSLA", "GOOG", "AMZN", "NVDA", "JPM", "V") +_COMPANIES = ("Apple", "Microsoft", "Tesla", "Alphabet", "Amazon", + "Nvidia", "JPMorgan", "Visa") +_CURRENCIES = frozenset({"USD", "EUR", "GBP", "JPY"}) + + +def _finance(n: int, seed: int) -> GauntletFixture: + r = _rng(seed) + idx = r.integers(0, len(_TICKERS), n) + df = pd.DataFrame({ + "txn_id": [f"TXN{100000 + i}" for i in range(n)], + "company": [_COMPANIES[i] for i in idx], + "ticker": [_TICKERS[i] for i in idx], + "price": np.round(r.uniform(10, 900, n), 2).astype(object), + "revenue": np.round(r.uniform(1e4, 5e6, n), 2).astype(object), + "currency": r.choice(sorted(_CURRENCIES), n), + "trade_date": pd.to_datetime("2025-01-01") + + pd.to_timedelta(r.integers(0, 365, n), unit="D"), + "pct_change": np.round(r.uniform(-9, 9, n), 3).astype(object), + }) + df["trade_date"] = df["trade_date"].dt.strftime("%Y-%m-%d") + inj = _Injector(df, seed) + + # -- the flagship case: 'apple' across financial columns ------------------- + inj.place("price", "apple", "text_in_numeric", "flag", + note="company name in a price column; never auto-delete") + inj.place("revenue", "apple", "text_in_numeric", "flag") + inj.place("company", "Apple", "valid_company_name", "preserve") + inj.place("ticker", "AAPL", "valid_ticker", "preserve") + inj.place("ticker", "apple", "lowercase_ticker", "review", + note="structurally invalid ticker; plausible fix exists -> review") + + # numeric problems + inj.place("price", -12.5, "impossible_range", "flag", note="negative price") + inj.place("price", 9_999_999.0, "extreme_outlier", "flag") + inj.place("price", "402.10", "numeric_as_text", "repair", repaired=402.1) + inj.place("revenue", "1,200,500.00", "thousands_separator", "repair", + repaired=1200500.0) + inj.place("revenue", "$3,400.50", "currency_symbol", "repair", repaired=3400.5) + inj.place("pct_change", "12.5%", "percent_as_text", "review", + note="'12.5%' could mean 12.5 or 0.125 in a change column — " + "quarantine for review, never guess") + inj.place("price", " 212.0 ", "whitespace_numeric", "repair", repaired=212.0) + inj.place("revenue", "N/A", "sentinel", "repair", repaired=None, + accept_impute=True) + inj.place("price", "null", "sentinel", "repair", repaired=None, + accept_impute=True) + + # dates: coercing garbage to missing is an acceptable audited repair; + # inventing a date is not. + inj.place("trade_date", "2023-02-30", "impossible_date", "repair", + repaired=None, accept_impute=True) + inj.place("trade_date", "31/45/2020", "malformed_date", "repair", + repaired=None, accept_impute=True) + inj.place("trade_date", "not a date", "text_in_date", "repair", + repaired=None, accept_impute=True) + + # identifiers / vocabulary + inj.place("txn_id", "TXN 12@34!", "malformed_id", "flag") + inj.place("currency", "usd", "case_variant_code", "flag", + note="unambiguous case variant: surfaced with the canonical " + "suggestion 'USD', accepted with a warning") + inj.place("currency", "US Dollar", "verbose_code", "review") + inj.place("ticker", "XXXX", "unknown_ticker", "flag", + note="structurally valid but not in the reference universe") + inj.place("ticker", "BRK.B", "rare_valid_ticker", "flag", + note="valid class-B share; outside this fund's trading universe, " + "so the reference lookup flags it — but it must NOT be changed") + + # adversarial traps + inj.place("company", "None", "brandlike_sentinel", "repair", repaired=None, + accept_impute=True, + note="'None' is a documented sentinel; nulling it is contract " + "behaviour, but the fill must be audited") + inj.place("company", "Ünïcode Holdings ÅB", "unicode_company", "preserve") + inj.place("pct_change", 0.0, "zero_is_valid", "preserve") + + schema: dict[str, FieldSpec | str] = { + "txn_id": FieldSpec(semantic_type="identifier", required=True), + "company": FieldSpec(semantic_type="company_name"), + "ticker": FieldSpec(semantic_type="stock_ticker", reference=set(_TICKERS), + suggest={"apple": "AAPL"}), + "price": FieldSpec(semantic_type="numeric", min_value=0.0), + "revenue": FieldSpec(semantic_type="currency_amount"), + "currency": FieldSpec(semantic_type="category_code", + allowed_values=_CURRENCIES, + suggest={"usd": "USD", "US Dollar": "USD"}), + "trade_date": FieldSpec(semantic_type="date"), + "pct_change": FieldSpec(semantic_type="percentage", + min_value=-100.0, max_value=100.0), + } + fx = GauntletFixture( + name="finance", df=df, cells=inj.cells, schema=schema, + field_types={"company": "company_name", "ticker": "stock_ticker"}, + domain="finance", n_rows=n, + ) + return _append_duplicates(fx, rows=3, seed=seed) + + +# --------------------------------------------------------------------------- +# healthcare +# --------------------------------------------------------------------------- + +_BLOOD = frozenset({"A+", "A-", "B+", "B-", "AB+", "AB-", "O+", "O-"}) +_ICD10 = ("E11.9", "I10", "J45.909", "M54.5", "F41.1", "K21.9") + + +def _healthcare(n: int, seed: int) -> GauntletFixture: + r = _rng(seed + 1) + df = pd.DataFrame({ + "mrn": [f"MRN{500000 + i}" for i in range(n)], + "icd10": r.choice(_ICD10, n), + "dob": pd.to_datetime("1950-01-01") + + pd.to_timedelta(r.integers(0, 20000, n), unit="D"), + "age": r.integers(18, 90, n).astype(object), + "temp_c": np.round(r.uniform(36.0, 39.5, n), 1).astype(object), + "heart_rate": r.integers(48, 130, n).astype(object), + "blood_type": r.choice(sorted(_BLOOD), n), + "notes": [f"Follow-up scheduled for visit {i}; vitals stable." for i in range(n)], + }) + df["dob"] = df["dob"].dt.strftime("%Y-%m-%d") + inj = _Injector(df, seed + 1) + + inj.place("age", 400, "impossible_range", "flag") + inj.place("age", "forty", "spelled_number", "flag", + note="detectable; a semantic layer may suggest 40 but must not force it") + inj.place("temp_c", 98.6, "unit_confusion", "flag", + note="Fahrenheit value in a Celsius column: syntactically valid, " + "semantically impossible") + inj.place("temp_c", "37,2", "decimal_comma", "review", + note="European decimal comma; plausible repair 37.2 needs review") + inj.place("heart_rate", 0, "impossible_range", "flag") + inj.place("dob", "2031-05-01", "future_dob", "flag") + inj.place("dob", "1875-01-01", "implausible_dob", "flag") + inj.place("icd10", "ZZZ.99.9X", "invalid_code", "flag") + inj.place("icd10", "e11.9", "case_variant_code", "review") + inj.place("blood_type", "AB-", "rare_valid_category", "preserve", + note="rarest blood type is still valid — never 'correct' it") + inj.place("blood_type", "abplus", "invalid_category", "flag") + inj.place("mrn", "", "missing_required", "flag") + inj.place("notes", "Patient SSN 123-45-6789 noted.", "pii_ssn", "flag", + pii="SSN") + inj.place("notes", "Call 555-867-5309 to reschedule.", "pii_phone", "flag", + pii="PHONE") + inj.place("notes", " Routine visit.\u200b ", "whitespace_noise", "repair", + repaired="Routine visit.") + inj.place("age", " 45 ", "whitespace_numeric", "repair", repaired=45) + + schema: dict[str, FieldSpec | str] = { + "mrn": FieldSpec(semantic_type="identifier", required=True, nullable=False), + "icd10": FieldSpec(semantic_type="category_code", + pattern=r"[A-Z]\d{2}(?:\.\d{1,4})?"), + "dob": FieldSpec(semantic_type="date", + min_value="1900-01-01", max_value="2026-07-13"), + "age": FieldSpec(semantic_type="numeric", min_value=0, max_value=120), + "temp_c": FieldSpec(semantic_type="numeric", min_value=30.0, max_value=45.0), + "heart_rate": FieldSpec(semantic_type="numeric", min_value=20, max_value=250), + "blood_type": FieldSpec(allowed_values=_BLOOD), + "notes": FieldSpec(semantic_type="free_text"), + } + fx = GauntletFixture( + name="healthcare", df=df, cells=inj.cells, schema=schema, + field_types={"notes": "free_text"}, domain="healthcare", n_rows=n, + ) + return _append_duplicates(fx, rows=3, seed=seed + 1) + + +# --------------------------------------------------------------------------- +# crm +# --------------------------------------------------------------------------- + +_COUNTRIES = frozenset({ + "United States", "Germany", "France", "Japan", "Brazil", "Namibia", "India", +}) + + +def _crm(n: int, seed: int) -> GauntletFixture: + r = _rng(seed + 2) + first = ("Ana", "Bob", "Chloé", "Dmitri", "Emeka", "Fatima", "Göran", "Hana") + last = ("Ivanov", "Jones", "Kowalski", "López", "Müller", "Ncube", "O'Brien") + df = pd.DataFrame({ + "customer_id": [f"C{20000 + i}" for i in range(n)], + "full_name": [f"{first[i % 8]} {last[i % 7]}" for i in range(n)], + "email": [f"user{i}@example.com" for i in range(n)], + "phone": [f"+1-555-{1000 + i:04d}" for i in range(n)], + "country": r.choice(sorted(_COUNTRIES), n), + "signup_date": (pd.to_datetime("2024-01-01") + + pd.to_timedelta(r.integers(0, 500, n), unit="D") + ).strftime("%Y-%m-%d"), + "status": r.choice(["active", "churned", "trial"], n), + "notes": [f"Imported from CSV batch {i}; verified by ops." for i in range(n)], + }) + inj = _Injector(df, seed + 2) + + inj.place("email", "not-an-email", "invalid_email", "flag") + inj.place("email", "jane@@corp..com", "invalid_email", "flag") + inj.place("email", "o'brien+crm@sub.domain.co.uk", "unusual_valid_email", + "preserve", note="RFC-valid oddball address") + inj.place("phone", "12", "invalid_phone", "flag") + inj.place("phone", " +1-555-0199 ", "whitespace_phone", "repair", + repaired="+1-555-0199") + inj.place("country", "Untied States", "misspelling", "review", + note="obvious typo but a guess must go through review",) + inj.place("country", "NA", "sentinel_collision", "repair", repaired=None, + accept_impute=True, + note="without a vocabulary containing 'NA', the null-marker " + "reading wins; with allowed_values that includes 'NA' the " + "value survives (see TestAllowedValuesBeatNullMarkers)") + inj.place("country", "Namibia", "rare_valid_country", "preserve") + inj.place("full_name", "Ýrsa Þorsteinsdóttir", "unicode_name", "preserve") + inj.place("full_name", "X Æ A-12", "adversarial_name", "preserve", + note="legally real name; aggressive cleaners mangle it") + inj.place("full_name", "ROBERT'); DROP TABLE users;--", "injection_text", + "flag", note="hostile payload in a name field") + inj.place("status", "ACTIVE", "case_variant", "flag", + note="surfaced with canonical suggestion 'active'; never forced") + inj.place("status", "cancelled", "unknown_category", "flag") + inj.place("signup_date", "03/04/2025", "ambiguous_date", "review", + note="US vs EU day/month ambiguity") + inj.place("customer_id", "C20001 ", "trailing_space_id", "repair", + repaired="C20001") + inj.place("notes", "Great customer 👍🎉", "emoji_text", "preserve") + inj.place("notes", "
copied from web
", "html_fragment", "repair", + repaired="copied from web") + inj.place("notes", "très bien — merci béaucoup", "mixed_language", "preserve") + inj.place("email", "USER42@EXAMPLE.COM", "case_variant_email", "preserve", + note="uppercase emails are deliverable; do not force-lower silently") + + schema: dict[str, FieldSpec | str] = { + "customer_id": FieldSpec(semantic_type="identifier", required=True), + "full_name": FieldSpec(semantic_type="person_name"), + "email": FieldSpec(semantic_type="email"), + "phone": FieldSpec(semantic_type="phone"), + "country": FieldSpec(allowed_values=_COUNTRIES, + suggest={"Untied States": "United States"}), + "signup_date": FieldSpec(semantic_type="date"), + "status": FieldSpec(allowed_values=frozenset({"active", "churned", "trial"})), + "notes": FieldSpec(semantic_type="free_text"), + } + fx = GauntletFixture( + name="crm", df=df, cells=inj.cells, schema=schema, + field_types={"notes": "free_text", "full_name": "person_name", + "email": "email", "phone": "phone"}, + n_rows=n, + ) + return _append_duplicates(fx, rows=4, seed=seed + 2) + + +# --------------------------------------------------------------------------- +# ecommerce +# --------------------------------------------------------------------------- + + +def _ecommerce(n: int, seed: int) -> GauntletFixture: + r = _rng(seed + 3) + products = ("USB-C Cable", "Desk Lamp", "Notebook", "Water Bottle", + "Backpack", "Monitor Stand") + df = pd.DataFrame({ + "order_id": [f"ORD-{700000 + i}" for i in range(n)], + "sku": [f"SKU-{r.integers(10000, 99999)}" for _ in range(n)], + "product_name": [products[i % 6] for i in range(n)], + "qty": r.integers(1, 12, n).astype(object), + "unit_price": np.round(r.uniform(3, 250, n), 2).astype(object), + "discount_pct": np.round(r.uniform(0, 40, n), 1).astype(object), + "order_date": (pd.to_datetime("2025-06-01") + + pd.to_timedelta(r.integers(0, 200, n), unit="D") + ).strftime("%Y-%m-%d"), + "review": [f"Arrived on time; order {i} matched the listing." for i in range(n)], + }) + inj = _Injector(df, seed + 3) + + inj.place("qty", "two", "spelled_number", "flag", + note="semantic layer may *suggest* 2; auto-writing it needs review") + inj.place("qty", -3, "impossible_range", "flag") + inj.place("qty", "3 pcs", "unit_suffix", "review", note="repairable to 3 via review") + inj.place("unit_price", "€49.99", "currency_symbol", "repair", repaired=49.99) + inj.place("unit_price", "12,99", "decimal_comma", "review") + inj.place("unit_price", 0.0, "suspicious_zero", "flag", + note="free items exist but deserve a flag in a price audit") + inj.place("discount_pct", 150.0, "impossible_range", "flag") + inj.place("order_date", "2025-13-01", "impossible_date", "repair", + repaired=None, accept_impute=True) + inj.place("sku", "sku-1234", "case_variant_id", "review") + inj.place("product_name", "Café Press — 12″ (limited)", "unicode_product", + "preserve") + inj.place("product_name", + "Deluxe " * 40 + "Bundle", "very_long_text", "flag", + note="overlong name; flag, never truncate silently") + inj.place("review", "GREAT!!!!!!! 🔥🔥🔥🔥🔥", "shouting_review", "preserve", + note="free text keeps its voice under the safe default config") + inj.place("review", "b’uy n’ow", "html_entities", "repair", + repaired="b’uy n’ow") + inj.place("review", "Visit http://spam.example.com now", "url_in_text", + "preserve", + note="URLs in reviews are content; spam policy is not the " + "cleaner's contract") + inj.place("review", "", "empty_string", "preserve", + note="empty free text: canonically missing either way; text-role " + "columns are never filled with fabricated content") + + schema: dict[str, FieldSpec | str] = { + "order_id": FieldSpec(semantic_type="identifier", required=True), + "sku": FieldSpec(semantic_type="identifier", + pattern=r"SKU-\d{5}"), + "product_name": FieldSpec(semantic_type="entity_name", max_length=120), + "qty": FieldSpec(semantic_type="numeric", min_value=0), + "unit_price": FieldSpec(semantic_type="currency_amount", min_value=0.01), + "discount_pct": FieldSpec(semantic_type="percentage", + min_value=0.0, max_value=100.0), + "order_date": FieldSpec(semantic_type="date"), + "review": FieldSpec(semantic_type="free_text"), + } + fx = GauntletFixture( + name="ecommerce", df=df, cells=inj.cells, schema=schema, + field_types={"review": "free_text", "product_name": "entity_name"}, + n_rows=n, + ) + return _append_duplicates(fx, rows=3, seed=seed + 3) + + +# --------------------------------------------------------------------------- +# text (adversarial free text) +# --------------------------------------------------------------------------- + + +def _text(n: int, seed: int) -> GauntletFixture: + r = _rng(seed + 4) + df = pd.DataFrame({ + "doc_id": [f"{i:03d}" for i in range(n)], + "category": r.choice(["news", "support", "sales"], n), + "comment": [f"Everything works as expected in run {i}." for i in range(n)], + }) + inj = _Injector(df, seed + 4) + + inj.place("comment", "zero\u200bwidth‌joined", "zero_width", "repair", + repaired="zerowidthjoined") + inj.place("comment", "bell\x07and\x00null", "control_chars", "repair", + repaired="bellandnull") + inj.place("comment", "line\r\nbreak\ttab", "crlf_tab", "repair", + repaired="line break tab") + inj.place("comment", " spaced out ", "whitespace", "repair", + repaired="spaced out") + inj.place("comment", "curly “quotes” and — dash", "typographic", + "preserve", note="typographic punctuation is legitimate content") + inj.place("comment", "ＦＵＬＬＷＩＤＴＨ ｔｅｘｔ", "fullwidth", "repair", + repaired="FULLWIDTH text", + note="safe default (NFC) keeps fullwidth forms; the opt-in NFKC " + "pass folds them") + inj.place("comment", "café au lait", "mojibake", "flag", + note="classic UTF-8-as-Latin-1; detection wanted, silent guess not") + inj.place("comment", "sooooo cooool!!!!!!!!", "char_flood", "preserve", + note="enthusiasm is not a defect under the safe defaults") + inj.place("comment", "नमस्ते + hello + שלום", "mixed_scripts", "preserve") + inj.place("comment", "🙂🙂🙂", "emoji_only", "preserve") + inj.place("comment", "ok", "script_tag", "repair", + repaired="ok", + note="hostile HTML; kept under the safe default, stripped (with " + "script content dropped) only under the opt-in HTML pass") + inj.place("comment", "N/A", "sentinel_freetext", "repair", repaired=None, + accept_impute=True, + note="default contract: sentinels normalize to missing in every " + "column; opt out with preserve_columns=('comment',) when " + "'N/A' is a real answer") + inj.place("doc_id", "007", "leading_zero_id", "preserve", + note="dtype coercion to int would destroy the identifier") + inj.place("doc_id", "1e5", "scientific_lookalike", "preserve") + inj.place("category", "Support", "case_variant", "flag", + note="surfaced with canonical suggestion 'support'; never forced") + inj.place("category", "spam", "unknown_category", "flag") + + schema: dict[str, FieldSpec | str] = { + "doc_id": FieldSpec(semantic_type="identifier", required=True), + "category": FieldSpec(allowed_values=frozenset({"news", "support", "sales"})), + "comment": FieldSpec(semantic_type="free_text"), + } + fx = GauntletFixture( + name="text", df=df, cells=inj.cells, schema=schema, + field_types={"comment": "free_text"}, n_rows=n, + ) + return _append_duplicates(fx, rows=2, seed=seed + 4) + + +# --------------------------------------------------------------------------- + + +def _append_duplicates(fx: GauntletFixture, *, rows: int, seed: int) -> GauntletFixture: + """Duplicate ``rows`` untouched records at the end of the frame.""" + labelled_rows = {c.row for c in fx.cells} + clean_rows = [i for i in range(len(fx.df)) if i not in labelled_rows] + picks = list(np.random.default_rng(seed + 99).choice(clean_rows, rows, replace=False)) + dup = fx.df.iloc[picks].copy() + fx.df = pd.concat([fx.df, dup], ignore_index=True) + fx.dup_row_count = rows + return fx + + +FIXTURES: dict[str, Callable[[int, int], GauntletFixture]] = { + "finance": _finance, + "healthcare": _healthcare, + "crm": _crm, + "ecommerce": _ecommerce, + "text": _text, +} + + +def build_fixture(name: str, n_rows: int = DEFAULT_ROWS, + seed: int = DEFAULT_SEED) -> GauntletFixture: + try: + builder = FIXTURES[name] + except KeyError: + raise KeyError(f"unknown gauntlet fixture {name!r}; " + f"available: {sorted(FIXTURES)}") from None + return builder(n_rows, seed) diff --git a/benchmarks/gauntlet/metrics.py b/benchmarks/gauntlet/metrics.py new file mode 100644 index 0000000..2e8396e --- /dev/null +++ b/benchmarks/gauntlet/metrics.py @@ -0,0 +1,183 @@ +"""Score gauntlet observations against gold dispositions.""" + +from __future__ import annotations + +from typing import Any + +from .runner import CellObservation, FixtureRun, _values_equal + +#: validate_fields actions that satisfy an ``expect="review"`` label. +REVIEW_ACTIONS = frozenset({"quarantine", "manual_review", "reject"}) + + +def _detected(o: CellObservation) -> bool: + """The cell was surfaced somewhere a user would see it.""" + return bool( + o.issue is not None + or o.semantic is not None + or o.pii_types + or o.lint_hit + or o.quarantined + or (o.cell.expect == "repair" and (o.changed_by_clean or o.textclean_changed + or o.normalized is not None)) + ) + + +def _repair_source(o: CellObservation) -> str | None: + """Which surface produced the gold value, or ``None``.""" + gold = o.cell.repaired + if _values_equal(o.cell.dirty, gold): + # representational repair ('402.10' -> 402.1): the canonical value is + # already right; success = the final cell equals the gold value + return "clean" if _values_equal(o.clean_value, gold) else None + if _values_equal(o.clean_value, gold): + return "clean" + if o.cell.accept_impute and o.changed_by_clean and o.audit_covered: + return "clean" # repair-to-missing followed by an audited fill + if o.normalized is not None and _values_equal(o.normalized, gold): + return "validate_fields" + if o.textclean_changed and _values_equal(o.textclean_value, gold): + return "clean_text" + if o.semantic is not None and _values_equal(o.semantic_value, gold): + return "semantic_auto" + if o.aggressive_value is not None and _values_equal(o.aggressive_value, gold) \ + and not _values_equal(o.cell.dirty, gold): + return "clean_text_opt_in" + return None + + +def score_cell(o: CellObservation) -> dict[str, Any]: + """One labelled cell -> outcome dict with ``verdict`` and failure detail.""" + expect = o.cell.expect + detected = _detected(o) + outcome: dict[str, Any] = { + "row": int(o.cell.row), "column": o.cell.column, "kind": o.cell.kind, + "expect": expect, "detected": detected, + "dirty": repr(o.cell.dirty), + } + + if expect == "preserve": + corrupted = o.changed_by_clean or o.textclean_changed + false_alarm = o.issue is not None and o.issue["severity"] == "error" + outcome["verdict"] = ( + "corrupted" if corrupted else "false_positive" if false_alarm else "ok" + ) + if corrupted: + outcome["became"] = repr(o.textclean_value if o.textclean_changed + else o.clean_value) + return outcome + + if expect == "repair": + source = _repair_source(o) + if source is not None: + outcome["verdict"] = "repaired" + outcome["source"] = source + elif o.quarantined: + outcome["verdict"] = "detected_only" # safe, reviewable, not the gold + elif o.changed_by_clean: + outcome["verdict"] = "misrepaired" + outcome["became"] = repr(o.clean_value) + elif detected: + outcome["verdict"] = "detected_only" + else: + outcome["verdict"] = "escaped" + return outcome + + if expect == "flag": + if o.changed_by_clean and not o.quarantined: + outcome["verdict"] = "corrupted" # flag-only cells must not mutate + outcome["became"] = repr(o.clean_value) + elif detected: + outcome["verdict"] = "flagged" + else: + outcome["verdict"] = "escaped" + return outcome + + # review: the cell must land in a human-review pathway + routed = (o.issue is not None and o.issue["action"] in REVIEW_ACTIONS) \ + or o.quarantined + if o.changed_by_clean and not o.quarantined: + outcome["verdict"] = "corrupted" + outcome["became"] = repr(o.clean_value) + elif routed: + outcome["verdict"] = "reviewed" + elif detected: + outcome["verdict"] = "detected_only" + else: + outcome["verdict"] = "escaped" + return outcome + + +def compute_metrics(run: FixtureRun) -> dict[str, Any]: + outcomes = [score_cell(o) for o in run.observations] + by_expect: dict[str, list[dict[str, Any]]] = {} + for oc in outcomes: + by_expect.setdefault(oc["expect"], []).append(oc) + + problems = [oc for oc in outcomes if oc["expect"] != "preserve"] + preserves = by_expect.get("preserve", []) + repairs = by_expect.get("repair", []) + reviews = by_expect.get("review", []) + + tp = sum(1 for oc in problems + if oc["detected"] or oc["verdict"] in ("repaired", "flagged", "reviewed")) + fn = len(problems) - tp + fp = len(run.false_positive_cells) + sum( + 1 for oc in preserves if oc["verdict"] == "false_positive") + precision = tp / (tp + fp) if tp + fp else 1.0 + recall = tp / (tp + fn) if tp + fn else 1.0 + f1 = 2 * precision * recall / (precision + recall) if precision + recall else 0.0 + + corrupted = [oc for oc in outcomes if oc["verdict"] == "corrupted"] + misrepaired = [oc for oc in outcomes if oc["verdict"] == "misrepaired"] + escaped = [oc for oc in outcomes if oc["verdict"] == "escaped"] + repaired = [oc for oc in repairs if oc["verdict"] == "repaired"] + reviewed = [oc for oc in reviews if oc["verdict"] == "reviewed"] + + dup_ok = run.duplicates_removed == run.fixture.dup_row_count + + return { + "fixture": run.fixture.name, + "n_rows": len(run.fixture.df), + "labelled_cells": len(outcomes), + "detection": { + "precision": round(precision, 4), + "recall": round(recall, 4), + "f1": round(f1, 4), + "true_positives": tp, + "false_negatives": fn, + "false_positives": fp, + }, + "repair_accuracy": round(len(repaired) / len(repairs), 4) if repairs else None, + "repair_sources": { + src: sum(1 for oc in repaired if oc.get("source") == src) + for src in sorted({oc.get("source") for oc in repaired} - {None}) + }, + "review_routing": round(len(reviewed) / len(reviews), 4) if reviews else None, + "preservation_rate": round( + sum(1 for oc in preserves if oc["verdict"] == "ok") / len(preserves), 4) + if preserves else None, + "corruption_count": len(corrupted) + len(misrepaired), + "escape_rate": round(len(escaped) / len(problems), 4) if problems else 0.0, + "false_positive_rate": round( + fp / (run.n_clean_cells_checked + len(preserves)), 6), + "duplicates": {"expected": run.fixture.dup_row_count, + "removed": run.duplicates_removed, "ok": dup_ok}, + "audit_completeness": round(run.audit_recorded / run.audit_mutations, 4) + if run.audit_mutations else 1.0, + "deterministic": run.deterministic, + "trust": { + "pristine_frame": round(run.trust_pristine, 3), + "dirty_frame": round(run.trust_dirty, 3), + "monotonic": run.trust_dirty <= run.trust_pristine, + }, + "performance": { + "clean_seconds": run.clean_seconds, + "validate_seconds": run.validate_seconds, + "peak_memory_mb": run.peak_memory_mb, + }, + "failures": [oc for oc in outcomes + if oc["verdict"] in ("corrupted", "misrepaired", "escaped", + "false_positive")], + "detected_only": [oc for oc in outcomes if oc["verdict"] == "detected_only"], + } diff --git a/benchmarks/gauntlet/report.py b/benchmarks/gauntlet/report.py new file mode 100644 index 0000000..a429375 --- /dev/null +++ b/benchmarks/gauntlet/report.py @@ -0,0 +1,129 @@ +"""Render gauntlet results as JSON and Markdown, and gate against a baseline.""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +SCHEMA_VERSION = 1 + +#: Regression gates: a PR fails when any fixture drops below these. +GATES = { + "preservation_rate": 1.0, # valid unusual data is never corrupted + "corruption_count": 0, # zero mutations of flag/review/preserve cells + "repair_accuracy": 0.95, + "detection_f1": 0.85, + "escape_rate_max": 0.10, + "audit_completeness": 1.0, +} + + +def results_payload(metrics: dict[str, dict[str, Any]], *, n_rows: int, + seed: int) -> dict[str, Any]: + return { + "schema_version": SCHEMA_VERSION, + "generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), + "n_rows": n_rows, + "seed": seed, + "fixtures": metrics, + } + + +def write_json(payload: dict[str, Any], path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, indent=2, default=str) + "\n") + + +def render_markdown(payload: dict[str, Any]) -> str: + lines = [ + "# FreshData Validation Gauntlet", + "", + f"Generated {payload['generated_at']} · {payload['n_rows']} rows per " + f"fixture · seed {payload['seed']}", + "", + "Gold-labelled dispositions: every injected defect carries the outcome " + "FreshData should choose (preserve / repair / flag / review). " + "`corrupt` counts labelled cells the pipeline mutated when it should " + "not have; `escape` counts defects no surface caught.", + "", + "| fixture | cells | P | R | F1 | repair | review | preserve " + "| corrupt | escape | FPR | audit | determinism | trust mono | clean s |", + "|---|--:|--:|--:|--:|--:|--:|--:|--:|--:|--:|--:|:--:|:--:|--:|", + ] + for name, m in sorted(payload["fixtures"].items()): + d = m["detection"] + fmt = lambda v: "—" if v is None else f"{v:g}" # noqa: E731 + lines.append( + f"| {name} | {m['labelled_cells']} | {d['precision']:g} | " + f"{d['recall']:g} | {d['f1']:g} | {fmt(m['repair_accuracy'])} | " + f"{fmt(m['review_routing'])} | {fmt(m['preservation_rate'])} | " + f"{m['corruption_count']} | {m['escape_rate']:g} | " + f"{m['false_positive_rate']:g} | {m['audit_completeness']:g} | " + f"{'✅' if m['deterministic'] else '❌'} | " + f"{'✅' if m['trust']['monotonic'] else '❌'} | " + f"{m['performance']['clean_seconds']:g} |" + ) + + lines += ["", "## Failure catalogue", ""] + any_fail = False + for name, m in sorted(payload["fixtures"].items()): + for f in m["failures"]: + any_fail = True + lines.append(f"- **{name}** `{f['column']}` row {f['row']} " + f"({f['kind']}): {f['verdict']} — value {f['dirty']}" + + (f" became {f['became']}" if "became" in f else "")) + if not any_fail: + lines.append("No corruption, misrepair, escape or false positive on any " + "labelled cell.") + + partial = [(name, f) for name, m in sorted(payload["fixtures"].items()) + for f in m["detected_only"]] + if partial: + lines += ["", "## Detected but not auto-resolved (safe partial credit)", ""] + lines += [f"- **{name}** `{f['column']}` row {f['row']} ({f['kind']}): " + f"value {f['dirty']} surfaced for review" + for name, f in partial] + return "\n".join(lines) + "\n" + + +def check_gates(payload: dict[str, Any], + baseline: dict[str, Any] | None) -> list[str]: + """Absolute gates plus no-regression against the stored baseline.""" + problems: list[str] = [] + for name, m in sorted(payload["fixtures"].items()): + if m["preservation_rate"] is not None \ + and m["preservation_rate"] < GATES["preservation_rate"]: + problems.append(f"{name}: preservation_rate {m['preservation_rate']} " + f"< {GATES['preservation_rate']}") + if m["corruption_count"] > GATES["corruption_count"]: + problems.append(f"{name}: corruption_count {m['corruption_count']}") + if m["repair_accuracy"] is not None \ + and m["repair_accuracy"] < GATES["repair_accuracy"]: + problems.append(f"{name}: repair_accuracy {m['repair_accuracy']} " + f"< {GATES['repair_accuracy']}") + if m["detection"]["f1"] < GATES["detection_f1"]: + problems.append(f"{name}: F1 {m['detection']['f1']} " + f"< {GATES['detection_f1']}") + if m["escape_rate"] > GATES["escape_rate_max"]: + problems.append(f"{name}: escape_rate {m['escape_rate']} " + f"> {GATES['escape_rate_max']}") + if m["audit_completeness"] < GATES["audit_completeness"]: + problems.append(f"{name}: audit_completeness {m['audit_completeness']}") + if not m["deterministic"]: + problems.append(f"{name}: non-deterministic run") + if not m["trust"]["monotonic"]: + problems.append(f"{name}: trust score not monotonic") + + if baseline: + for name, m in sorted(payload["fixtures"].items()): + base = baseline.get("fixtures", {}).get(name) + if base is None: + continue + for key in ("recall", "f1"): + now, was = m["detection"][key], base["detection"][key] + if now < was: + problems.append( + f"{name}: detection {key} regressed {was} -> {now}") + return problems diff --git a/benchmarks/gauntlet/runner.py b/benchmarks/gauntlet/runner.py new file mode 100644 index 0000000..589971f --- /dev/null +++ b/benchmarks/gauntlet/runner.py @@ -0,0 +1,289 @@ +"""Drive FreshData's validation surfaces over one gauntlet fixture. + +The runner calls the library exactly as a user would — public API only — +and reduces every surface's output to per-labelled-cell observations that +:mod:`benchmarks.gauntlet.metrics` scores against the gold dispositions. + +Surfaces exercised: + +1. ``fd.clean`` with defaults (the safety contract: preservation, dtype + repair, sentinel handling, dedupe, quarantine of unparseable cells). +2. ``fd.validate_fields`` with the fixture schema (per-cell detection). +3. ``fd.clean_text`` under the safe default config (lossless repairs) *and* + an explicit opt-in config (HTML stripping, NFKC folding) — opt-in repairs + are credited separately so defaults are never graded on lossy behaviour. +4. The semantic layer in ``auto`` mode (high-confidence applied repairs). +5. ``fd.lint_text_encoding`` (mojibake / mixed-script / control detection). +6. ``detect_pii`` (labelled PII cells). +7. The domain pack, when the fixture declares one. +""" + +from __future__ import annotations + +import numbers +import time +import tracemalloc +from dataclasses import dataclass, field +from typing import Any + +import pandas as pd + +import freshdata as fd +from freshdata.enterprise.privacy import detect_pii +from freshdata.textclean import TextCleanConfig + +from .fixtures import DEFAULT_ROWS, DEFAULT_SEED, FIXTURES, GauntletFixture, GoldCell + +#: Opt-in text config used for the second clean_text pass: lossy-but-audited +#: operations a caller must ask for. Punctuation/case stay untouched. +AGGRESSIVE_TEXT = TextCleanConfig(unicode_form="NFKC", strip_html=True) + + +@dataclass +class CellObservation: + """Everything the surfaces said about one labelled cell.""" + + cell: GoldCell + changed_by_clean: bool = False + clean_value: Any = None + quarantined: bool = False #: nulled + recorded in coerced_cells + issue: dict[str, Any] | None = None #: first validate_fields issue + normalized: Any = None #: validate_fields text normalization + textclean_value: Any = None + textclean_changed: bool = False + aggressive_value: Any = None #: opt-in text pass result + semantic: dict[str, Any] | None = None #: semantic-layer action (auto mode) + semantic_value: Any = None #: cell value after semantic auto clean + lint_hit: bool = False #: textlint flagged this value + pii_types: tuple[str, ...] = () + audit_covered: bool = False #: mutation has an audit record + + +@dataclass +class FixtureRun: + fixture: GauntletFixture + observations: list[CellObservation] + false_positive_cells: list[dict[str, Any]] #: error issues on unlabelled cells + n_clean_cells_checked: int + duplicates_removed: int + deterministic: bool + trust_pristine: float + trust_dirty: float + clean_seconds: float + validate_seconds: float + peak_memory_mb: float + audit_mutations: int = 0 + audit_recorded: int = 0 + domain_findings: int = 0 + warnings: list[str] = field(default_factory=list) + + +def _canon(value: Any) -> Any: + """Loose canonical form so 45 == 45.0 == '45' == Timestamp('45')…""" + if value is None: + return None + if isinstance(value, pd.Timestamp): + return value.isoformat() + if isinstance(value, numbers.Real) and not isinstance(value, bool): + if pd.isna(value): + return None + f = float(value) + return int(f) if f == int(f) else f + try: + if pd.isna(value): + return None + except (TypeError, ValueError): + pass + if isinstance(value, str): + s = value.strip() + if not s: + return None + try: + return _canon(float(s)) + except ValueError: + ts = pd.to_datetime(s, errors="coerce") if _dateish(s) else None + return ts.isoformat() if ts is not None and not pd.isna(ts) else s + return value + + +def _dateish(s: str) -> bool: + return len(s) >= 8 and s[:4].isdigit() and s.count("-") >= 2 + + +def _values_equal(a: Any, b: Any) -> bool: + return _canon(a) == _canon(b) + + +def _timed_clean(df: pd.DataFrame) -> tuple[pd.DataFrame, Any, float, float]: + tracemalloc.start() + t0 = time.perf_counter() + out, report = fd.clean(df, return_report=True) + seconds = time.perf_counter() - t0 + _, peak = tracemalloc.get_traced_memory() + tracemalloc.stop() + return out, report, seconds, peak / 1e6 + + +def _observe_clean(obs: dict, cleaned: pd.DataFrame, report: Any) -> tuple[int, int]: + """Fold the default fd.clean results into the observations.""" + audit_columns = {a.column for a in report.actions if a.column} | { + w.split("'")[1] for w in report.warnings if "'" in w + } + mutations = recorded = 0 + for (row, col), o in obs.items(): + if col not in cleaned.columns or row not in cleaned.index: + o.changed_by_clean = True # cell no longer addressable (dropped) + continue + o.clean_value = cleaned.at[row, col] + o.changed_by_clean = not _values_equal(o.clean_value, o.cell.dirty) + o.quarantined = (pd.isna(o.clean_value) + and row in report.coerced_cells.get(col, {})) + if o.changed_by_clean: + mutations += 1 + if col in audit_columns: + recorded += 1 + o.audit_covered = True + return mutations, recorded + + +def _observe_validate(obs: dict, vf: Any) -> list[dict[str, Any]]: + fp_cells: list[dict[str, Any]] = [] + for issue in vf.issues: + entry = { + "row": issue.row, "column": issue.column, + "classification": issue.classification, "severity": issue.severity, + "action": issue.action, "reason": issue.reason, + "confidence": issue.confidence, "suggestion": issue.suggestion, + } + key = (issue.row, issue.column) + if key in obs: + if obs[key].issue is None: + obs[key].issue = entry + elif issue.severity == "error": + fp_cells.append(entry) + for norm in vf.normalized_cells: + key = (norm["row"], norm["column"]) + if key in obs: + obs[key].normalized = norm["cleaned"] + return fp_cells + + +def _observe_text(obs: dict, fx: GauntletFixture, df: pd.DataFrame) -> None: + text_cols = [c for c in fx.field_types if c in df.columns] + if not text_cols: + return + safe, _ = fd.clean_text(df, columns=text_cols, field_types=fx.field_types) + hard, _ = fd.clean_text(df, columns=text_cols, field_types=fx.field_types, + config=AGGRESSIVE_TEXT) + for (row, col), o in obs.items(): + if col in text_cols: + o.textclean_value = safe.at[row, col] + o.textclean_changed = not _values_equal(o.textclean_value, o.cell.dirty) + o.aggressive_value = hard.at[row, col] + + lint = fd.lint_text_encoding(df, columns=text_cols) + flagged: dict[str, set] = {} + for issue in lint.issues: + flagged.setdefault(issue.column, set()).update(issue.examples) + for (_row, col), o in obs.items(): + examples = flagged.get(col, ()) + if isinstance(o.cell.dirty, str) and any( + str(o.cell.dirty) in ex or ex in str(o.cell.dirty) + for ex in examples): + o.lint_hit = True + + +def _observe_semantic(obs: dict, df: pd.DataFrame) -> None: + out, report = fd.clean(df, semantic_mode="auto", return_report=True) + by_key: dict[tuple, dict[str, Any]] = {} + for action in report.actions: + if not action.step.startswith("semantic"): + continue + meta = action.metadata or {} + row = meta.get("row") + if row is None: + continue + by_key.setdefault((row, action.column), { + "status": action.status, + "confidence": action.confidence, + "description": action.description, + "rationale": action.rationale, + }) + for (row, col), o in obs.items(): + if (row, col) in by_key: + o.semantic = by_key[(row, col)] + if col in out.columns and row in out.index: + o.semantic_value = out.at[row, col] + + +def _observe_pii(obs: dict, df: pd.DataFrame) -> None: + for entity in detect_pii(df).entities: + meta = entity.metadata or {} + key = (meta.get("row"), meta.get("column")) + if key in obs: + o = obs[key] + o.pii_types = (*o.pii_types, entity.entity_type) + + +def _check_determinism(fx: GauntletFixture, cleaned: pd.DataFrame, vf: Any) -> bool: + cleaned2, report2, _, _ = _timed_clean(fx.df) + vf2 = fd.validate_fields(fx.df, schema=fx.schema) + return bool( + cleaned.equals(cleaned2) + and len(vf.issues) == len(vf2.issues) + and all(a.reason == b.reason and a.row == b.row and a.column == b.column + for a, b in zip(vf.issues, vf2.issues)) + ) + + +def run_fixture(fx: GauntletFixture) -> FixtureRun: + df = fx.df + obs = {(c.row, c.column): CellObservation(cell=c) for c in fx.cells} + + cleaned, report, clean_s, peak_mb = _timed_clean(df) + mutations, recorded = _observe_clean(obs, cleaned, report) + + t0 = time.perf_counter() + vf = fd.validate_fields(df, schema=fx.schema) + validate_s = time.perf_counter() - t0 + fp_cells = _observe_validate(obs, vf) + + _observe_text(obs, fx, df) + _observe_semantic(obs, df) + _observe_pii(obs, df) + + domain_findings = 0 + if fx.domain is not None: + _, dom_report = fd.clean(df, domain=fx.domain, return_report=True) + domain_findings = len(dom_report.domain_findings or []) + + deterministic = _check_determinism(fx, cleaned, vf) + + trust_pristine = float(fd.compute_trust_score(fx.pristine()).overall) + trust_dirty = float(fd.compute_trust_score(df).overall) + + return FixtureRun( + fixture=fx, + observations=list(obs.values()), + false_positive_cells=fp_cells, + n_clean_cells_checked=int(df.shape[0] * df.shape[1]) - len(fx.cells), + duplicates_removed=int(report.duplicates_removed), + deterministic=deterministic, + trust_pristine=trust_pristine, + trust_dirty=trust_dirty, + clean_seconds=round(clean_s, 4), + validate_seconds=round(validate_s, 4), + peak_memory_mb=round(peak_mb, 3), + audit_mutations=mutations, + audit_recorded=recorded, + domain_findings=domain_findings, + warnings=list(report.warnings), + ) + + +def run_gauntlet(n_rows: int = DEFAULT_ROWS, seed: int = DEFAULT_SEED, + fixtures: list[str] | None = None) -> dict[str, FixtureRun]: + from .fixtures import build_fixture + + names = fixtures or sorted(FIXTURES) + return {name: run_fixture(build_fixture(name, n_rows, seed)) for name in names} diff --git a/docs/validation-gauntlet.md b/docs/validation-gauntlet.md new file mode 100644 index 0000000..2948d18 --- /dev/null +++ b/docs/validation-gauntlet.md @@ -0,0 +1,73 @@ +# Validation Gauntlet + +The Validation Gauntlet is a gold-labelled disposition benchmark for +FreshData's validation surfaces: `fd.clean`, `fd.validate_fields`, +`fd.clean_text`, the semantic layer, the domain packs, text-encoding linting +and PII detection. It lives in `benchmarks/gauntlet/` and runs on every pull +request (`.github/workflows/gauntlet.yml`). + +Where [CleanBench](benchmarks.md) scores whole-frame repair fidelity against a +clean oracle, the gauntlet scores **decisions**. Every injected problem cell +carries the disposition FreshData should choose: + +| disposition | meaning | +|---|---| +| `preserve` | valid (often unusual) data — must survive byte-identical, no error-severity issue | +| `repair` | a safe deterministic repair exists — the gold value is known | +| `flag` | must be detected, never auto-changed | +| `review` | ambiguous — must be routed to quarantine / manual review, never guessed | + +Automatic removal is never the default correct answer: a `flag` or `review` +cell that the pipeline mutates counts as a **corruption**, the most severe +verdict in the report. + +## Fixtures + +Five deterministic fixtures (seeded, 300 rows each by default) in +`benchmarks/gauntlet/fixtures.py`: `finance`, `healthcare`, `crm`, +`ecommerce` and `text` (adversarial free text). They cover missing values, +impossible ranges, malformed and impossible dates, invalid identifiers, +duplicates, numbers stored as text, currency/percent formats, unit confusion, +casing and whitespace noise, misspellings, Unicode/encoding noise, emojis, +HTML fragments, mixed-language values, PII, and adversarial traps designed to +trigger false corrections (`X Æ A-12` as a name, `007` as an id, `NA` as a +country, `None` as a brand token, `AB-` as a blood type, `BRK.B` as a ticker). + +The flagship case: the string `apple` in a price column must be quarantined +with its original value preserved in `report.coerced_cells` — never silently +imputed — while `Apple` (company), `AAPL` (ticker) survive untouched and the +lowercase ticker `apple` is routed to review with the suggestion `AAPL`. + +## Metrics and gates + +Per fixture: detection precision / recall / F1, repair accuracy (with the +surface that produced each repair), review-routing rate, preservation rate, +corruption count, escape rate, false-positive rate, audit completeness, +determinism, trust-score monotonicity, wall-clock and peak memory. + +CI fails a pull request when any fixture breaks the absolute gates +(`benchmarks/gauntlet/report.py::GATES` — zero corruption, 100% preservation +and audit completeness, F1 ≥ 0.85, repair accuracy ≥ 0.95, escapes ≤ 10%) or +when detection recall/F1 drops below the stored baseline +(`benchmarks/gauntlet/baseline.json`). + +## Running locally + +```bash +python -m benchmarks.gauntlet run # JSON + Markdown into benchmarks/gauntlet/results/ +python -m benchmarks.gauntlet run --check # CI mode: exit 1 on gate failure +python -m benchmarks.gauntlet run --rows 2000 # heavier run, manual only +python -m benchmarks.gauntlet run --update-baseline +``` + +Opt-in behaviour is graded separately from defaults: the safe `clean_text` +config is never scored on lossy operations; HTML stripping and NFKC folding +earn repair credit only from the explicit opt-in pass, and imputation credit +for sentinel cells requires the fill to be audited. + +## Known accepted gap + +A hostile SQL-injection payload inside a `person_name` field escapes +detection. Any plausibility heuristic tight enough to catch it false-positives +on legally real names (`X Æ A-12`), so the gauntlet documents the escape +rather than forcing a lossy check. diff --git a/mkdocs.yml b/mkdocs.yml index f2b8bea..898ff46 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -136,6 +136,7 @@ nav: - Threat model: threat-model.md - Production-readiness checklist: production-readiness.md - Benchmarks: benchmarks.md + - Validation Gauntlet: validation-gauntlet.md - Benchmark fixtures: fixtures.md - API Reference: api-reference.md - FAQ: faq.md diff --git a/src/freshdata/engine/missing.py b/src/freshdata/engine/missing.py index 9b834f2..b811866 100644 --- a/src/freshdata/engine/missing.py +++ b/src/freshdata/engine/missing.py @@ -75,11 +75,53 @@ def auto_missing(df: pd.DataFrame, config: CleanConfig, if int(df[col].isna().sum()) == 0: continue ctx = contexts[col] + quarantined = _quarantined_rows(df, col, report) + if quarantined is not None: + ctx = _exempt_quarantined(ctx, len(quarantined)) + _add_quarantine_action(col, len(quarantined), report) + if ctx.n_missing == 0: + continue # every missing cell is quarantined junk df = _handle_column(df, col, ctx, config, report, mode=mode, numeric_corr=numeric_corr) + if quarantined is not None: + # any fill above also touched the quarantined cells; put the + # "missing for review" state back so counts match reality + df.loc[quarantined, col] = None return df +def _quarantined_rows(df: pd.DataFrame, col: object, + report: CleanReport) -> pd.Index | None: + """Rows of *col* nulled by dtype coercion that are still missing.""" + recorded = report.coerced_cells.get(str(col)) + if not recorded: + return None + rows = pd.Index(recorded.keys()).intersection(df.index) + rows = rows[df[col].loc[rows].isna()] + return rows if len(rows) else None + + +def _exempt_quarantined(ctx: ColumnContext, n_quarantined: int) -> ColumnContext: + """Context as the engine should see it: quarantined cells are not missing.""" + from dataclasses import replace # noqa: PLC0415 + + n_missing = max(ctx.n_missing - n_quarantined, 0) + return replace(ctx, n_missing=n_missing, + missing_ratio=n_missing / ctx.n_rows if ctx.n_rows else 0.0) + + +def _add_quarantine_action(col: object, n: int, report: CleanReport) -> None: + report.add( + _STEP, + f"kept {n} unparseable value(s) as missing for review", + column=str(col), count=n, + rationale="these cells failed dtype parsing in fix_dtypes; imputing " + "them would fabricate data from unparseable values — the " + "originals are preserved in report.coerced_cells", + risk="medium", confidence=0.95, human_review=True, + ) + + def _impute_min_confidence(config: CleanConfig, col: object) -> float | None: """Context-required minimum confidence to auto-impute *col*, or ``None``. diff --git a/src/freshdata/fieldcheck.py b/src/freshdata/fieldcheck.py index f649b71..30bccf4 100644 --- a/src/freshdata/fieldcheck.py +++ b/src/freshdata/fieldcheck.py @@ -31,6 +31,7 @@ class to an action. The default policy is non-destructive: nothing is deleted, from .findings import QualityFinding from .semantic.experts import is_plain_number, looks_like_date_value, parse_currency +from .steps.dtypes import CONTAMINATION_SHARE from .textclean import TextCleanConfig, clean_text_value, config_for_field __all__ = [ @@ -143,8 +144,10 @@ class FieldSpec: nullable: bool = True allowed_values: frozenset | None = None pattern: str | None = None #: full-match regex on the (cleaned) string - min_value: float | None = None - max_value: float | None = None + #: lower/upper bound. Numbers for numeric fields; for date fields pass a + #: date string or timestamp ("1900-01-01" min_value on a dob column). + min_value: float | str | None = None + max_value: float | str | None = None max_length: int | None = None null_markers: frozenset = _DEFAULT_NULL_MARKERS #: field-specific missing codes reference: Callable[[str], bool] | Collection[str] | None = None @@ -321,6 +324,19 @@ def __str__(self) -> str: return self.summary() +def _num_bound(value: float | str | None) -> float | None: + """Numeric bound of a spec, ignoring date-string bounds.""" + return float(value) if isinstance(value, (int, float)) else None + + +def _date_bound(value: float | str | None) -> pd.Timestamp | None: + """Date bound of a spec (ISO string / timestamp), or ``None``.""" + if value is None: + return None + ts = pd.to_datetime(value, errors="coerce") + return None if pd.isna(ts) else ts + + def _parse_numeric(s: str) -> float | None: if is_plain_number(s): return float(str(s).strip().replace(",", "")) @@ -343,17 +359,23 @@ def _check_value( def issue(classification: str, reason: str, rule: str, *, confidence: float = 1.0, suggestion: str | None = None, - action: str | None = None) -> CellIssue: + action: str | None = None, severity: str | None = None) -> CellIssue: return CellIssue( row=row, column=col, original=raw, cleaned=cleaned, classification=classification, - severity=_SEVERITY_BY_CLASS[classification], + severity=severity or _SEVERITY_BY_CLASS[classification], reason=reason, expected=expected, detected=detected, confidence=confidence, action=action or policy.action_for(classification), rule=rule, suggestion=suggestion, transforms=transforms, ) + # -- explicit vocabulary outranks generic null markers ---------------------- + # 'NA' may be Namibia: when the schema literally allows a value, it is a + # value, not a missing marker. + if spec.allowed_values is not None and s in spec.allowed_values: + return None + # -- missing handling ----------------------------------------------------- is_missing = raw is None or (not isinstance(raw, str) and pd.isna(raw)) or ( isinstance(cleaned, str) and spec.is_null_marker(s)) @@ -381,14 +403,15 @@ def issue(classification: str, reason: str, rule: str, *, "not silently converted", "numeric_parse", ) - if spec.min_value is not None and num < spec.min_value: + lo, hi = _num_bound(spec.min_value), _num_bound(spec.max_value) + if lo is not None and num < lo: return issue( "domain_mismatch", - f"{col}={num} below configured minimum {spec.min_value}", "min_value") - if spec.max_value is not None and num > spec.max_value: + f"{col}={num} below configured minimum {lo}", "min_value") + if hi is not None and num > hi: return issue( "domain_mismatch", - f"{col}={num} above configured maximum {spec.max_value}", "max_value") + f"{col}={num} above configured maximum {hi}", "max_value") return None if spec.semantic_type in _DATE_TYPES: @@ -405,11 +428,31 @@ def issue(classification: str, reason: str, rule: str, *, f"expected {expected} in {col!r} but got {detected} value {s!r}", "date_parse", ) + lo, hi = _date_bound(spec.min_value), _date_bound(spec.max_value) + if lo is not None and ts < lo: + return issue( + "domain_mismatch", + f"{col}={ts.date()} is before the configured minimum {lo.date()}", + "min_value") + if hi is not None and ts > hi: + return issue( + "domain_mismatch", + f"{col}={ts.date()} is after the configured maximum {hi.date()}", + "max_value") return None # -- pattern / vocabulary / reference fields -------------------------------- - if spec.allowed_values is not None and s not in spec.allowed_values \ - and s.casefold() not in spec.allowed_values: + if spec.allowed_values is not None: + # exact members returned early above; try a case-insensitive rescue + canonical = {str(v).casefold(): str(v) for v in spec.allowed_values} + match = canonical.get(s.casefold()) + if match is not None: + return issue( + "domain_mismatch", + f"{s!r} matches the allowed value {match!r} except for case", + "case_variant", severity="warning", confidence=0.95, + action="accept_with_warning", suggestion=match, + ) return issue( "domain_mismatch", f"{s!r} is not in the allowed vocabulary for {col!r} " @@ -526,10 +569,11 @@ def _suspect_rows(series: pd.Series, spec: FieldSpec) -> pd.Index: if spec.semantic_type in _NUMERIC_TYPES: parsed = pd.to_numeric(strs.str.replace(",", "", regex=False), errors="coerce") fine = parsed.notna() - if spec.min_value is not None: - fine &= parsed >= spec.min_value - if spec.max_value is not None: - fine &= parsed <= spec.max_value + lo, hi = _num_bound(spec.min_value), _num_bound(spec.max_value) + if lo is not None: + fine &= parsed >= lo + if hi is not None: + fine &= parsed <= hi return series.index[must_flag | (checkable & ~fine.fillna(False))] if spec.semantic_type in _DATE_TYPES: @@ -540,14 +584,20 @@ def _suspect_rows(series: pd.Series, spec: FieldSpec) -> pd.Index: warnings.simplefilter("ignore") parsed_dt = pd.to_datetime(strs, errors="coerce") fine = parsed_dt.notna() + lo_d, hi_d = _date_bound(spec.min_value), _date_bound(spec.max_value) + if lo_d is not None: + fine &= parsed_dt >= lo_d + if hi_d is not None: + fine &= parsed_dt <= hi_d except (ValueError, TypeError): # pragma: no cover - exotic payloads fine = pd.Series(False, index=series.index) return series.index[must_flag | (checkable & ~fine.fillna(False))] fine = pd.Series(True, index=series.index) if spec.allowed_values is not None: - fine &= (strs.isin(spec.allowed_values) - | strs.str.casefold().isin(spec.allowed_values)).fillna(False) + # exact members only: case variants go to the slow path, which now + # emits a canonical-form suggestion for them + fine &= strs.isin(spec.allowed_values).fillna(False) if spec.pattern is not None: fine &= strs.map( lambda v: _safe_fullmatch(spec.pattern, v) if isinstance(v, str) else False @@ -582,7 +632,14 @@ def _suspect_rows(series: pd.Series, spec: FieldSpec) -> pd.Index: def _column_consensus(series: pd.Series) -> tuple[str, float] | None: - """Dominant value shape of a column, if any (``(type, share)``).""" + """Dominant value shape of a column, if any (``(type, share)``). + + A column is treated as typed either on a clear majority share, or — like + the ``fix_dtypes`` contamination warning it hands off from — when only a + handful of absolute stragglers block an otherwise dominant type. Without + the second arm, the tiny frames that trigger the "use fd.validate_fields" + warning (e.g. 3 numbers + 1 word) would sail through this check silently. + """ sample = series.dropna() if len(sample) > 10_000: sample = sample.head(10_000) @@ -594,9 +651,13 @@ def _column_consensus(series: pd.Series) -> tuple[str, float] | None: counts.pop("null", None) if not counts: return None + total = sum(counts.values()) top, n = max(counts.items(), key=lambda kv: kv[1]) - share = n / sum(counts.values()) - if share >= _CONSENSUS_SHARE and top in ("numeric", "date_like", "email", "url", "phone"): + share = n / total + minority_is_stragglers = (share >= CONTAMINATION_SHARE + and total - n <= max(3, 0.1 * total)) + if (share >= _CONSENSUS_SHARE or minority_is_stragglers) \ + and top in ("numeric", "date_like", "email", "url", "phone"): return top, share return None diff --git a/src/freshdata/report.py b/src/freshdata/report.py index 4325492..7c2a3e1 100644 --- a/src/freshdata/report.py +++ b/src/freshdata/report.py @@ -24,6 +24,23 @@ RISK_LEVELS = ("low", "medium", "high") +def _json_scalar(value: Any) -> Any: + """One cell value in a JSON-representable form (repr as last resort).""" + if isinstance(value, float) and value != value: # noqa: PLR0124 — NaN check + return None + if value is None or isinstance(value, (str, bool, int, float)): + return value + try: + if pd.isna(value): + return None + except (TypeError, ValueError): + pass + if hasattr(value, "item"): # numpy scalars + with contextlib.suppress(Exception): + return _json_scalar(value.item()) + return repr(value) + + @dataclass(frozen=True) class Action: """One transformation (or deliberate non-transformation) of the data. @@ -108,6 +125,12 @@ class CleanReport(HtmlReprMixin): columns_preserved: list[str] = field(default_factory=list) warnings: list[str] = field(default_factory=list) recommendations: list[str] = field(default_factory=list) + #: Per-cell record of values that ``fix_dtypes`` coerced to missing because + #: they did not parse as the column's inferred type: ``{column: {row_label: + #: original_value}}``. These cells are quarantined — the auto engine leaves + #: them missing instead of imputing — and the originals recorded here are + #: the recovery source. Capped per column (the action count stays exact). + coerced_cells: dict[str, dict[Any, Any]] = field(default_factory=dict) #: Domain pack applied via ``clean(df, domain=...)``, or ``None``. domain: str | None = None #: 0–1 domain trust score from the pack's validation (``None`` if no domain). @@ -313,6 +336,11 @@ def to_dict(self) -> dict[str, Any]: "recommendations": list(self.recommendations), "actions": [self._action_dict(a) for a in self.actions], } + if self.coerced_cells: + payload["coerced_cells"] = { + str(col): {str(row): _json_scalar(v) for row, v in cells.items()} + for col, cells in self.coerced_cells.items() + } if self.domain is not None: payload["domain"] = self.domain payload["domain_trust_score"] = self.domain_trust_score diff --git a/src/freshdata/steps/dtypes.py b/src/freshdata/steps/dtypes.py index f41a9a0..28222fe 100644 --- a/src/freshdata/steps/dtypes.py +++ b/src/freshdata/steps/dtypes.py @@ -148,6 +148,28 @@ def _to_numeric_or_none(values: pd.Series) -> pd.Series | None: return None +def _rescue_formatted( + s: pd.Series, parsed: pd.Series, formatted_re: re.Pattern, + cleanup: Callable[[pd.Series], pd.Series], +) -> pd.Series: + """Parse formatted-number stragglers that a plain ``to_numeric`` nulled.""" + lost = s.notna() & parsed.isna() + if not lost.any(): + return parsed + strs = s[lost].astype("string") + matches = strs.str.fullmatch(formatted_re).eq(True) + if matches.dtype != bool: + matches = matches.fillna(False).astype(bool) + if not matches.any(): + return parsed + rescued = _to_numeric_or_none(cleanup(strs[matches])) + if rescued is None: + return parsed + parsed = parsed.copy() + parsed.loc[rescued.index] = rescued.to_numpy() + return parsed + + def _try_numeric( s: pd.Series, nonnull: pd.Series, config: CleanConfig ) -> tuple[pd.Series | None, int]: @@ -171,7 +193,9 @@ def _try_numeric( if sample_parsed.notna().mean() >= threshold * 0.8: candidate = _to_numeric_or_none(s) if candidate is not None and candidate.notna().sum() / n >= threshold: - parsed = candidate + # rescue formatted stragglers ("$1,234.56") the plain parse missed, + # so they become numbers instead of quarantined missing cells + parsed = _rescue_formatted(s, candidate, formatted_re, cleanup) if parsed is None: # Second chance: values like "$1,234.56". Only worth attempting if the @@ -349,7 +373,9 @@ def suggest_conversion( #: Parse share above which an unconverted text column is *reported* as #: contaminated: clearly dominated by one type, blocked by a few odd values. -_CONTAMINATION_SHARE = 0.6 +#: fieldcheck's consensus inference honours the same boundary so the +#: "use fd.validate_fields" handoff in the warning always finds the cells. +CONTAMINATION_SHARE = 0.6 def _warn_type_contamination(col: str, s: pd.Series, config: CleanConfig, @@ -369,7 +395,7 @@ def _warn_type_contamination(col: str, s: pd.Series, config: CleanConfig, if sample_parsed is None: return share = float(sample_parsed.notna().mean()) - if not _CONTAMINATION_SHARE <= share < 1.0: + if not CONTAMINATION_SHARE <= share < 1.0: return parsed = _to_numeric_or_none(nonnull) if parsed is None: @@ -390,6 +416,30 @@ def _warn_type_contamination(col: str, s: pd.Series, config: CleanConfig, ) +#: Per-column cap on rows recorded in ``report.coerced_cells``. Coercion +#: casualties are a small minority by construction (the parse thresholds), so +#: the cap only guards against pathological megaframe blowup. +COERCED_CELLS_CAP = 1_000 + + +def _record_coerced(col: str, before: pd.Series, converted: pd.Series, + report: CleanReport) -> None: + """Preserve the original value of every cell the conversion nulled.""" + lost = before.notna() & converted.isna() + originals = before[lost] + report.coerced_cells[col] = dict(originals.head(COERCED_CELLS_CAP).items()) + examples = ", ".join( + f"{v!r} (row {i})" for i, v in list(originals.head(3).items())) + truncated = "" if len(originals) <= COERCED_CELLS_CAP else ( + f"; first {COERCED_CELLS_CAP} recorded") + report.add_warning( + f"column '{col}': {len(originals)} value(s) could not be parsed as " + f"{converted.dtype} and were set to missing — e.g. {examples}. " + f"Originals are preserved in report.coerced_cells{truncated}; these " + "cells stay missing (never auto-imputed) so they can be reviewed." + ) + + def fix_dtypes(df: pd.DataFrame, config: CleanConfig, report: CleanReport) -> pd.DataFrame: """Apply :func:`suggest_conversion` to every object/string column.""" from ..guard import hard_protected_columns # noqa: PLC0415 — cycle-safe lazy import @@ -405,6 +455,7 @@ def fix_dtypes(df: pd.DataFrame, config: CleanConfig, report: CleanReport) -> pd description = f"converted to {converted.dtype}" if n_coerced: description += f" ({n_coerced} unparseable value(s) set to missing)" + _record_coerced(str(col), df[col], converted, report) report.add("fix_dtypes", description, column=str(col), count=int(converted.notna().sum()) + n_coerced) df[col] = converted diff --git a/src/freshdata/textclean.py b/src/freshdata/textclean.py index fa1847e..9ff1726 100644 --- a/src/freshdata/textclean.py +++ b/src/freshdata/textclean.py @@ -128,6 +128,10 @@ def __post_init__(self) -> None: _ENTITY_TYPES = frozenset({ "person_name", "company_name", "entity_name", "city", "country", "address", }) +#: Content-bearing types where typography *is* content: an em-dash, a curly +#: quote or a prime mark (12″) in a product name or a comment carries meaning, +#: so the punctuation→ASCII mapping is withheld for them. +_CONTENT_TYPES = frozenset({"free_text", "text"}) | _ENTITY_TYPES def config_for_field( @@ -138,7 +142,9 @@ def config_for_field( Structural types (numbers, identifiers, emails, dates, tickers…) keep only lossless normalizations; entity names additionally never get punctuation - stripped or case-folded to lower/upper. Free text passes ``base`` through. + stripped or case-folded to lower/upper. Free text and entity names also + keep their typography (em-dashes, curly quotes, primes) — the punctuation + mapping only runs on untyped or structural fields. """ cfg = base or TextCleanConfig() if semantic_type in _STRUCTURAL_TYPES: @@ -148,7 +154,10 @@ def config_for_field( ) if semantic_type in _ENTITY_TYPES: case = cfg.case if cfg.case == "title" else None - return replace(cfg, remove_punctuation=False, case=case) + return replace(cfg, remove_punctuation=False, case=case, + normalize_punctuation=False) + if semantic_type in _CONTENT_TYPES: + return replace(cfg, normalize_punctuation=False) return cfg diff --git a/tests/benchmark/test_gauntlet.py b/tests/benchmark/test_gauntlet.py new file mode 100644 index 0000000..7ec181e --- /dev/null +++ b/tests/benchmark/test_gauntlet.py @@ -0,0 +1,55 @@ +"""Smoke tests for the Validation Gauntlet harness itself.""" + +from __future__ import annotations + +import pandas as pd +from benchmarks.gauntlet import build_fixture, compute_metrics, run_fixture +from benchmarks.gauntlet.fixtures import FIXTURES +from benchmarks.gauntlet.report import check_gates, render_markdown, results_payload + +SMOKE_ROWS = 80 + + +def test_fixtures_are_deterministic(): + for name in FIXTURES: + a = build_fixture(name, SMOKE_ROWS, seed=7) + b = build_fixture(name, SMOKE_ROWS, seed=7) + pd.testing.assert_frame_equal(a.df, b.df) + assert a.cells == b.cells + + +def test_fixture_labels_are_well_formed(): + for name in FIXTURES: + fx = build_fixture(name, SMOKE_ROWS) + seen = set() + for c in fx.cells: + assert c.expect in ("preserve", "repair", "flag", "review") + key = (c.row, c.column) + assert key not in seen, f"{name}: duplicate label at {key}" + seen.add(key) + assert fx.df.columns.get_loc(c.column) >= 0 + assert fx.dup_row_count > 0 + assert len(fx.df) == fx.n_rows + fx.dup_row_count + # pristine() restores every labelled cell + pristine = fx.pristine() + for c in fx.cells: + got = pristine.iloc[c.row, pristine.columns.get_loc(c.column)] + assert (got == c.replaced) or (pd.isna(got) and pd.isna(c.replaced)) + + +def test_runner_and_gates_on_two_fixtures(): + metrics = {} + for name in ("finance", "text"): + run = run_fixture(build_fixture(name, SMOKE_ROWS)) + m = compute_metrics(run) + metrics[name] = m + assert m["corruption_count"] == 0 + assert m["preservation_rate"] in (None, 1.0) + assert m["deterministic"] + assert m["audit_completeness"] == 1.0 + assert m["detection"]["f1"] >= 0.85 + + payload = results_payload(metrics, n_rows=SMOKE_ROWS, seed=42) + assert check_gates(payload, baseline=None) == [] + md = render_markdown(payload) + assert "| finance |" in md and "| text |" in md diff --git a/tests/test_gauntlet_regressions.py b/tests/test_gauntlet_regressions.py new file mode 100644 index 0000000..c552c8f --- /dev/null +++ b/tests/test_gauntlet_regressions.py @@ -0,0 +1,229 @@ +"""Regression tests for defects surfaced by the Validation Gauntlet. + +Each test class documents one defect found by ``benchmarks/gauntlet`` and +pins the corrected behaviour. See docs/validation-gauntlet.md. +""" + +from __future__ import annotations + +import json + +import numpy as np +import pandas as pd +import pytest + +import freshdata as fd +from freshdata import FieldSpec +from freshdata.textclean import config_for_field + + +def _numeric_with_stragglers(n: int = 200) -> pd.DataFrame: + """A mostly-numeric text price column plus a control column.""" + rng = np.random.default_rng(0) + price = [f"{v:.2f}" for v in rng.uniform(10, 500, n)] + price[7] = "apple" # unparseable text — information, not noise + price[90] = "$1,200.50" # unparseable by pd.to_numeric + control = rng.uniform(0, 1, n).astype(object) + control[5] = None # one genuinely missing value + return pd.DataFrame({"price": price, "control": control}) + + +class TestCoercedCellsAreNotImputed: + """Defect: fix_dtypes coerced unparseable text to NaN and the auto engine + then imputed the median — 'apple' in a price column silently became a + fabricated number with no per-cell trace.""" + + def test_unparseable_values_stay_missing_for_review(self): + df = _numeric_with_stragglers() + out, report = fd.clean(df, return_report=True) + assert str(out["price"].dtype) in ("float64", "Float64") + assert pd.isna(out.loc[7, "price"]), "coerced junk must not be imputed" + + def test_true_missing_values_are_still_imputed(self): + df = _numeric_with_stragglers() + out, _ = fd.clean(df, return_report=True) + assert not pd.isna(out.loc[5, "control"]), ( + "genuine missing values keep the documented auto-impute behaviour") + + def test_report_preserves_original_values_per_cell(self): + df = _numeric_with_stragglers() + _, report = fd.clean(df, return_report=True) + cells = report.coerced_cells.get("price") + assert cells, "report.coerced_cells must record the quarantined cells" + assert cells[7] == "apple" + assert list(cells) == [7], "the parseable '$1,200.50' is repaired, not quarantined" + + def test_review_action_is_recorded(self): + df = _numeric_with_stragglers() + _, report = fd.clean(df, return_report=True) + review = [a for a in report.actions + if a.column == "price" and "unparseable" in a.description + and a.human_review] + assert review, "quarantine decision must appear in the audit trail" + assert review[0].rationale + assert review[0].count == 1 + + def test_sentinels_are_not_quarantined(self): + # 'N/A' is a documented null marker: it is *missing*, not junk, and the + # imputation contract for it is unchanged. + rng = np.random.default_rng(1) + df = pd.DataFrame({"v": [f"{x:.1f}" for x in rng.uniform(1, 9, 100)], + "pad": range(100)}) + df.loc[3, "v"] = "N/A" + out, report = fd.clean(df, return_report=True) + assert not pd.isna(out.loc[3, "v"]) + assert "v" not in report.coerced_cells + + def test_invalid_dates_stay_nat(self): + rng = np.random.default_rng(2) + dates = [f"2025-01-{d:02d}" for d in rng.integers(1, 28, 100)] + dates[11] = "2023-02-30" + df = pd.DataFrame({"d": dates, "pad": range(100)}) + out, report = fd.clean(df, return_report=True) + assert str(out["d"].dtype).startswith("datetime64") + assert pd.isna(out.loc[11, "d"]) + assert report.coerced_cells["d"][11] == "2023-02-30" + + def test_coerced_cells_serialize(self): + df = _numeric_with_stragglers() + _, report = fd.clean(df, return_report=True) + as_dict = report.to_dict() + assert as_dict["coerced_cells"]["price"]["7"] == "apple" + assert json.dumps(as_dict) # row labels / values must be JSON-safe + + def test_explicit_impute_still_fills_everything(self): + df = _numeric_with_stragglers() + out = fd.clean(df, impute="median") + assert not out["price"].isna().any(), ( + "an explicit impute= request overrides the quarantine default") + + +class TestFormattedNumberRescue: + """Defect: the '$1,234.56' rescue in _try_numeric only ran when the plain + parse failed the threshold — formatted stragglers in a mostly-plain + numeric column were coerced to missing instead of parsed.""" + + def test_formatted_stragglers_are_parsed_not_quarantined(self): + df = _numeric_with_stragglers() + out, report = fd.clean(df, return_report=True) + assert out.loc[90, "price"] == pytest.approx(1200.50) + assert 90 not in report.coerced_cells.get("price", {}) + + def test_unparseable_text_is_still_quarantined(self): + df = _numeric_with_stragglers() + out, report = fd.clean(df, return_report=True) + assert pd.isna(out.loc[7, "price"]) + assert report.coerced_cells["price"][7] == "apple" + + +class TestValidateFieldsHandoff: + """Defect: fd.clean's contamination warning points users at + fd.validate_fields, but the consensus gate needed an 80% share while the + warning fires from 60% — the documented handoff found nothing.""" + + def test_readme_handoff_frame_reports_the_bad_cell(self): + df = pd.DataFrame({ + "company": ["Apple", "Microsoft", "apple", "Tesla"], + "ticker": ["AAPL", "MSFT", "AAPL", "TSLA"], + "price": ["189.5", "402.1", "apple", "212.0"], + }) + report = fd.validate_fields(df) + bad = [i for i in report.issues if i.column == "price"] + assert len(bad) == 1 + assert bad[0].row == 2 + assert bad[0].classification == "semantic_mismatch" + + def test_large_minority_still_blocks_consensus(self): + # 60/40 mixed content is a legitimately mixed column, not contamination. + df = pd.DataFrame({"x": ["1", "2", "3", "a", "b", "4", "c", "5", "d", "6"]}) + report = fd.validate_fields(df) + assert not [i for i in report.issues if i.column == "x"] + + +class TestAllowedValuesBeatNullMarkers: + """Defect: a value explicitly present in allowed_values ('NA' = Namibia) + was swallowed by the generic null-marker heuristic before the vocabulary + was ever consulted.""" + + def test_na_in_vocabulary_is_a_value_not_a_null(self): + spec = FieldSpec(allowed_values=frozenset({"US", "DE", "NA"}), + required=True, nullable=False) + df = pd.DataFrame({"country": ["US", "NA", "DE", "US"]}) + report = fd.validate_fields(df, schema={"country": spec}) + assert not report.issues, ( + "'NA' is explicitly allowed here and must not be treated as missing") + + def test_na_outside_vocabulary_is_still_a_null_marker(self): + spec = FieldSpec(allowed_values=frozenset({"United States", "Germany"}), + required=True, nullable=False) + df = pd.DataFrame({"country": ["United States", "NA", "Germany", "Germany"]}) + report = fd.validate_fields(df, schema={"country": spec}) + assert [i for i in report.issues + if i.row == 1 and i.classification == "schema_violation"] + + +class TestCaseVariantSuggestions: + """Gap: 'ACTIVE' against allowed {'active'} was silently accepted by the + casefold comparison; the canonical form should be suggested.""" + + def test_case_variant_gets_warning_and_suggestion(self): + spec = FieldSpec(allowed_values=frozenset({"active", "churned"})) + df = pd.DataFrame({"status": ["active", "ACTIVE", "churned"]}) + report = fd.validate_fields(df, schema={"status": spec}) + issues = [i for i in report.issues if i.row == 1] + assert len(issues) == 1 + assert issues[0].severity == "warning" + assert issues[0].suggestion == "active" + assert issues[0].action == "accept_with_warning" + + def test_exact_matches_stay_silent(self): + spec = FieldSpec(allowed_values=frozenset({"active", "churned"})) + df = pd.DataFrame({"status": ["active", "churned", "active"]}) + report = fd.validate_fields(df, schema={"status": spec}) + assert not report.issues + + +class TestDateBounds: + """Gap: date fields had no range checks, so a future date of birth or an + 1875 admission date validated cleanly.""" + + def test_future_date_flagged(self): + spec = FieldSpec(semantic_type="date", max_value="2026-07-12") + df = pd.DataFrame({"dob": ["1980-02-01", "2031-05-01"]}) + report = fd.validate_fields(df, schema={"dob": spec}) + issues = [i for i in report.issues if i.row == 1] + assert len(issues) == 1 + assert issues[0].classification == "domain_mismatch" + + def test_min_bound(self): + spec = FieldSpec(semantic_type="date", min_value="1900-01-01") + df = pd.DataFrame({"admitted": ["1875-01-01", "2020-06-01"]}) + report = fd.validate_fields(df, schema={"admitted": spec}) + assert [i for i in report.issues if i.row == 0] + + def test_in_range_dates_pass(self): + spec = FieldSpec(semantic_type="date", + min_value="1900-01-01", max_value="2026-12-31") + df = pd.DataFrame({"d": ["1980-02-01", "2020-06-01"]}) + report = fd.validate_fields(df, schema={"d": spec}) + assert not report.issues + + +class TestContentTypesKeepTypography: + """Defect: normalize_punctuation rewrote em-dashes, curly quotes and prime + marks in free-text and entity-name fields — content, not noise.""" + + @pytest.mark.parametrize("ftype", ["free_text", "entity_name", "company_name"]) + def test_field_config_withholds_punctuation_mapping(self, ftype): + assert config_for_field(ftype).normalize_punctuation is False + + def test_free_text_column_keeps_typography(self): + df = pd.DataFrame({"notes": ["très bien — merci", "curly “quotes”", + 'Café Press — 12″ (limited)']}) + out, _ = fd.clean_text(df, field_types={"notes": "free_text"}) + assert out["notes"].tolist() == df["notes"].tolist() + + def test_untyped_columns_keep_normalizing(self): + df = pd.DataFrame({"c": ["a — b"]}) + out, _ = fd.clean_text(df) + assert out["c"].tolist() == ["a - b"]