Text-cleaning pipeline + context-aware per-cell field validation

benchmarks/bench_fieldcheck.py

@@ -0,0 +1,161 @@
+"""Benchmark fd.validate_fields against regex and pandas-coercion baselines.
+
+Reproducible (seeded) invalid-cell detection benchmark on a synthetic
+financial feed with labeled corruptions::
+
+    python benchmarks/bench_fieldcheck.py            # default 20k rows
+    python benchmarks/bench_fieldcheck.py 100000     # custom row count
+
+For each validator we report precision / recall / F1 over the labeled
+invalid cells, plus wall time and throughput. Baselines:
+
+* ``regex``   — per-column regular expressions only (no context, no ranges);
+* ``pandas``  — ``pd.to_numeric`` / ``pd.to_datetime`` coercion NaN-diffing
+  (types only: no vocabulary, format or semantic checks);
+* ``fieldcheck`` — ``fd.validate_fields`` with a declared schema.
+"""
+
+from __future__ import annotations
+
+import json
+import random
+import re
+import sys
+import time
+
+import pandas as pd
+
+from freshdata.fieldcheck import FieldSpec, validate_fields
+
+SEED = 20260711
+CORRUPTION_RATE = 0.02
+
+SCHEMA = {
+    "transaction_id": FieldSpec(semantic_type="identifier", required=True, nullable=False),
+    "transaction_amount": FieldSpec(semantic_type="currency_amount"),
+    "currency": FieldSpec(allowed_values=frozenset({"USD", "EUR", "GBP", "INR"})),
+    "stock_ticker": FieldSpec(semantic_type="ticker"),
+    "interest_rate": FieldSpec(semantic_type="rate", min_value=0, max_value=1),
+    "transaction_date": FieldSpec(semantic_type="date"),
+}
+
+#: (column, corrupt value) pool — all are invalid for their column.
+CORRUPTIONS = [
+    ("transaction_amount", "apple"),
+    ("transaction_amount", "approved"),
+    ("transaction_amount", "12O.50"),
+    ("currency", "BTC"),
+    ("currency", "dollars"),
+    ("stock_ticker", "apple"),
+    ("stock_ticker", "AА PL"),
+    ("interest_rate", "high"),
+    ("interest_rate", "5"),          # out of [0, 1]
+    ("transaction_date", "2026-02-30"),
+    ("transaction_date", "not a date"),
+]
+
+
+def make_dataset(n: int) -> tuple[pd.DataFrame, set]:
+    rng = random.Random(SEED)
+    tickers = ["AAPL", "MSFT", "GOOG", "TSLA", "BRK.B", "AMZN"]
+    rows = []
+    for i in range(n):
+        rows.append({
+            "transaction_id": f"T{i:07d}",
+            "transaction_amount": f"{rng.uniform(1, 5000):.2f}",
+            "currency": rng.choice(["USD", "EUR", "GBP", "INR"]),
+            "stock_ticker": rng.choice(tickers),
+            "interest_rate": f"{rng.uniform(0.001, 0.2):.4f}",
+            "transaction_date": f"2026-{rng.randint(1, 12):02d}-{rng.randint(1, 28):02d}",
+        })
+    df = pd.DataFrame(rows)
+    truth: set = set()
+    n_bad = max(1, int(n * CORRUPTION_RATE))
+    for row in rng.sample(range(n), n_bad):
+        col, bad = rng.choice(CORRUPTIONS)
+        df.loc[row, col] = bad
+        truth.add((row, col))
+    return df, truth
+
+
+# --- baselines ---------------------------------------------------------------
+
+REGEXES = {
+    "transaction_id": re.compile(r"^T\d{7}$"),
+    "transaction_amount": re.compile(r"^-?\d+(\.\d+)?$"),
+    "currency": re.compile(r"^[A-Z]{3}$"),
+    "stock_ticker": re.compile(r"^[A-Z]{1,6}([.\-][A-Z0-9]{1,4})?$"),
+    "interest_rate": re.compile(r"^-?\d+(\.\d+)?$"),
+    "transaction_date": re.compile(r"^\d{4}-\d{2}-\d{2}$"),
+}
+
+
+def regex_validator(df: pd.DataFrame) -> set:
+    flagged = set()
+    for col, rx in REGEXES.items():
+        bad = ~df[col].astype(str).str.fullmatch(rx)
+        flagged.update((row, col) for row in df.index[bad])
+    return flagged
+
+
+def pandas_validator(df: pd.DataFrame) -> set:
+    flagged = set()
+    for col in ("transaction_amount", "interest_rate"):
+        bad = pd.to_numeric(df[col], errors="coerce").isna() & df[col].notna()
+        flagged.update((row, col) for row in df.index[bad])
+    dates = pd.to_datetime(df["transaction_date"], errors="coerce", format="%Y-%m-%d")
+    bad = dates.isna() & df["transaction_date"].notna()
+    flagged.update((row, "transaction_date") for row in df.index[bad])
+    return flagged
+
+
+def fieldcheck_validator(df: pd.DataFrame) -> set:
+    report = validate_fields(df, SCHEMA)
+    return {(i.row, i.column) for i in report.issues if i.severity == "error"}
+
+
+# --- scoring -----------------------------------------------------------------
+
+
+def score(name: str, fn, df: pd.DataFrame, truth: set) -> dict:
+    start = time.perf_counter()
+    flagged = fn(df)
+    elapsed = time.perf_counter() - start
+    tp = len(flagged & truth)
+    fp = len(flagged - truth)
+    fn_ = len(truth - flagged)
+    precision = tp / (tp + fp) if tp + fp else 0.0
+    recall = tp / (tp + fn_) if tp + fn_ else 0.0
+    f1 = 2 * precision * recall / (precision + recall) if precision + recall else 0.0
+    return {
+        "validator": name,
+        "precision": round(precision, 4),
+        "recall": round(recall, 4),
+        "f1": round(f1, 4),
+        "false_accepts": fn_,
+        "false_rejects": fp,
+        "seconds": round(elapsed, 3),
+        "rows_per_s": int(len(df) / elapsed) if elapsed else None,
+    }
+
+
+def main() -> None:
+    n = int(sys.argv[1]) if len(sys.argv) > 1 else 20_000
+    df, truth = make_dataset(n)
+    print(f"rows={n} corrupted_cells={len(truth)} seed={SEED}\n")
+    results = [
+        score("regex", regex_validator, df, truth),
+        score("pandas_coercion", pandas_validator, df, truth),
+        score("fd.validate_fields", fieldcheck_validator, df, truth),
+    ]
+    print(json.dumps(results, indent=2))
+    print()
+    header = f"{'validator':<20}{'precision':>10}{'recall':>8}{'f1':>8}{'sec':>8}{'rows/s':>10}"
+    print(header)
+    for r in results:
+        print(f"{r['validator']:<20}{r['precision']:>10}{r['recall']:>8}"
+              f"{r['f1']:>8}{r['seconds']:>8}{r['rows_per_s']:>10}")
+
+
+if __name__ == "__main__":
+    main()

docs/field-validation.md

@@ -0,0 +1,179 @@
+# Text cleaning & context-aware field validation
+
+Two standalone, dependency-light layers for handling scraped or third-party
+feeds where the same string can be valid in one column and wrong in another:
+
+- **`fd.clean_text`** — a configurable, field-aware text-cleaning pipeline
+  that never destroys information;
+- **`fd.validate_fields`** — per-cell validation that judges each value *in
+  the context of its field*, with a configurable, non-destructive
+  remediation policy.
+
+Both work on plain pandas frames and are independent of the main
+`fd.clean` engine (which now also *warns* when a mostly-numeric column is
+kept as text because of a few unparseable values, naming the exact rows).
+
+## Text cleaning
+
+```python
+import pandas as pd
+import freshdata as fd
+
+df = pd.DataFrame({
+    "note": ["  <p>Great&nbsp;product…</p>  ", "fine"],
+    "amount": [" 1,200.50 ", "99.99"],
+})
+
+cleaned, report = fd.clean_text(
+    df,
+    config=fd.TextCleanConfig(strip_html=True),
+    field_types={"amount": "currency_amount"},   # guards aggressive ops
+)
+
+print(cleaned.loc[0, "note"])     # Great product...
+print(cleaned.loc[0, "amount"])   # 1,200.50  (only whitespace trimmed)
+print(report.summary())
+```
+
+Key properties:
+
+- **Non-destructive** — the input frame is never modified; every changed
+  cell is logged with `original`, `cleaned` and the ordered transform list.
+- **Field-aware** — lossy operations (case folding, punctuation or HTML
+  stripping) are automatically withheld from structural types (amounts,
+  identifiers, emails, URLs, dates, tickers) and from entity names.
+- **Deterministic and lossless by default** — HTML/URL removal, case
+  folding and punctuation removal are opt-in via `TextCleanConfig`.
+
+Scalar use: `fd.clean_text_value("  x  ")` returns a `CleanedText` with
+`.original`, `.cleaned`, `.transforms`.
+
+## Context-aware validation: the "apple" example
+
+A scraped feed inserts the string `"apple"` into different columns. The
+correct verdict depends entirely on the field:
+
+```python
+import pandas as pd
+import freshdata as fd
+
+df = pd.DataFrame([{
+    "transaction_amount": "apple",        # invalid: monetary field
+    "company_name": "Apple",              # valid: real-word names are fine
+    "stock_ticker": "apple",              # invalid format; AAPL would pass
+    "transaction_description": "Payment to Apple",  # valid free text
+}])
+
+schema = {
+    "transaction_amount": fd.FieldSpec(semantic_type="currency_amount"),
+    "company_name": fd.FieldSpec(semantic_type="company_name"),
+    "stock_ticker": fd.FieldSpec(semantic_type="ticker",
+                                 suggest={"apple": "AAPL"}),
+    "transaction_description": fd.FieldSpec(semantic_type="free_text"),
+}
+
+report = fd.validate_fields(df, schema)
+print(report.summary())
+for issue in report.issues:
+    print(issue.column, issue.classification, "->", issue.action,
+          "| suggestion:", issue.suggestion)
+```
+
+Output (abridged):
+
+```
+transaction_amount  semantic_mismatch -> quarantine  | suggestion: None
+stock_ticker        domain_mismatch   -> manual_review | suggestion: AAPL
+```
+
+Rules of the road:
+
+- values are **never silently converted** — `"apple"` does not become `0`
+  or `NaN`;
+- suggestions come **only from a trusted mapping or callable you supply**
+  (`FieldSpec.suggest`); nothing is invented;
+- reference verification (e.g. a ticker universe) is injected via
+  `FieldSpec.reference` — never hard-coded;
+- a column *without* a spec is still protected: when ≥80% of its values
+  share one shape, nonconforming cells are reported as `semantic_mismatch`
+  with the exact row, reason and confidence.
+
+## Failure classes and remediation policy
+
+Distinct problems stay distinct — there is no generic "outlier" bucket:
+
+| classification             | default action        | severity |
+|----------------------------|-----------------------|----------|
+| `parse_failure`            | `quarantine`          | error    |
+| `semantic_mismatch`        | `quarantine`          | error    |
+| `domain_mismatch`          | `manual_review`       | error    |
+| `schema_violation`         | `reject`              | error    |
+| `statistical_outlier`      | `accept_with_warning` | warning  |
+| `categorical_rare`         | `accept_with_warning` | warning  |
+| `cross_field_inconsistency`| `manual_review`       | warning  |
+
+Rare or extreme values are **warned about, never auto-rejected**: an
+unusually large transaction is still a transaction. Every action is
+configurable per class:
+
+```python
+policy = fd.RemediationPolicy(semantic_mismatch="replace_with_null")
+report = fd.validate_fields(df, schema, policy=policy)
+result = fd.apply_field_policy(df, report)
+result.accepted      # cleaned copy (replacements applied, audited)
+result.quarantined   # rows held back for review
+result.rejected      # rows dropped from the accepted set (still returned!)
+result.needs_review  # rows awaiting a human decision
+result.audit         # one record per issue: original, action, reason, rule
+```
+
+`apply_field_policy` never mutates the input frame, and every
+`replace_with_null` keeps the original value in the audit trail.
+
+Context-dependent missing codes are per-field: `"N/A"` can be a null marker
+in one column and a legitimate code in another
+(`FieldSpec(null_markers=frozenset({""}), allowed_values={"N/A", ...})`).
+
+Cross-column rules are plain callables:
+
+```python
+def settle_after_trade(row):
+    if str(row["settlement_date"]) < str(row["trade_date"]):
+        return "settlement_date precedes trade_date"
+    return None
+
+fd.validate_fields(df, schema, cross_rules=[settle_after_trade])
+```
+
+## Reporting
+
+`report.to_findings()` exports issues as standard
+[`QualityFinding`](api-reference.md) records (step `"fieldcheck"`), so they
+flow into the existing exporters (dbt tests, Great Expectations suites,
+exception tables, lineage). `report.to_frame()` gives a flat DataFrame;
+`report.normalized_cells` is the text-normalization audit.
+
+## Performance
+
+Validation runs a vectorized pre-screen per column and only pays the
+per-cell explanation path for suspect cells, so mostly-clean feeds validate
+at bulk speed. Reproduce the comparison against regex and pandas-coercion
+baselines with:
+
+```bash
+python benchmarks/bench_fieldcheck.py 100000
+```
+
+## Known limitations
+
+- Semantic-type coverage is deliberately narrow and deterministic (numbers,
+  dates, emails, URLs, phones, tickers, identifiers, entity names, free
+  text); embedding-based typing lives in the optional
+  [semantic layer](semantic-models.md).
+- Column-consensus checks need a dominant shape (≥80%) and at least 3
+  values; genuinely mixed columns are left alone by design.
+- `FieldSpec.reference` given as a callable disables the vectorized
+  pre-screen for that column (every value is checked individually).
+- Date validation accepts any format `pandas.to_datetime` can parse;
+  day-first/locale ambiguity is the cleaning engine's concern
+  (`fd.clean(dayfirst=...)`).

mkdocs.yml

@@ -121,6 +121,7 @@ nav:
       - Learning profiles: learning-profiles.md
       - Developer training pipeline: developer-training-pipeline.md
       - Format parsers & tick mode: parsers.md
+      - Text cleaning & field validation: field-validation.md
       - Compliance reports: compliance.md
       - Orchestration integrations: integrations.md
   - Examples: examples.md