validation

name	validation
description	Use for tasks that check data, artifacts, citations, schemas, or cross-format records against explicit rules and produce clean outputs, violation lists, anomaly reports, reconciliation diffs, or validation summaries.

Validation Toolkit

Validation tasks decide what passes, what fails, and why. The deliverable is usually a paired output: clean records (or values that satisfied the rules) plus a failure ledger (violations, diffs, or anomaly flags) that is fully traceable to the source rows.

Three questions before any code:

1. What is the validation unit — row, cell, claim, record, file, schema field, or a cross-format entity?
2. Are violations counted per row or per check? One row can contain several violations.
3. What identity carries from input to output? Preserve original indices and source IDs so every line in the failure ledger points back.

Pick the tool by task shape

Task shape	Tool	Why this one
Tabular data against typed rules	pandera	Schema-as-code, lazy validation, structured failure cases
Two sources for the same records	pandas + normalization rules	Diff is the deliverable, no off-the-shelf library wins
Outliers, drift, distribution checks	numpy/scipy robust statistics	Mean/std are not robust; use median, MAD, quantiles
Citation / claim adjudication	Custom matching + evidence packets	Generic libraries do not understand "claim with evidence"

Schema validation (Pandera)

Pandera's lazy validation collects every failure case in one pass. For reports that need exact row indices, validate row-by-row — whole-frame failures conflate row positions with failure_cases.index. Uniqueness checks across the frame need a separate pass because a single row cannot tell.

import json
import pandas as pd
import pandera as pa

schema = pa.DataFrameSchema({
    "<id_column>":     pa.Column(int, checks=pa.Check.gt(0), unique=True),
    "<group_column>":  pa.Column(int, checks=pa.Check.gt(0)),
    "<numeric_column>":pa.Column(float, checks=pa.Check.in_range(<low>, <high>)),
    "<enum_column>":   pa.Column(str, checks=pa.Check.isin([<allowed_values>])),
    "<timestamp>":     pa.Column(str, nullable=False),
})

def to_native(value):
    if pd.isna(value):
        return None
    if hasattr(value, "item"):
        return value.item()
    return value

violations, bad_rows = [], set()
for idx in range(len(df)):
    row = df.iloc[[idx]].reset_index(drop=True)
    try:
        schema.validate(row, lazy=True)
    except pa.errors.SchemaErrors as exc:
        for _, fc in exc.failure_cases.iterrows():
            violations.append({
                "row_index": idx,
                "column": str(fc.get("column", )),
                : (fc.get(, )),
                : to_native(fc.get()),
            })
            bad_rows.add(idx)


 idx  df[df[].duplicated(keep=)].index:
    violations.append({
        : (idx),
        : ,
        : ,
        : to_native(df.at[idx, ]),
    })
    bad_rows.add((idx))

 (, )  f:
    json.dump(violations, f, ensure_ascii=, indent=)

The to_native helper is required: numpy scalar types (np.int64, np.float64) raise TypeError from json.dump without explicit casting. default=str masks the problem instead of fixing it.

Cross-format reconciliation

Two sources, same records, find the differences. Normalization is most of the work — comparing raw strings nearly always reports false positives.

Normalize before comparing:

• IDs — trim whitespace, casefold, preserve leading zeros where IDs are strings.
• Dates — parse to ISO date with explicit timezone.
• Money — strip currency symbols, thousands separators, parentheses-as-negative; cast to Decimal for exact compare.
• Text — collapse runs of whitespace; normalize Unicode quotes (" " → "); NFKC normalize.

Emit a diff record with: record_id, field, left_value, right_value, difference (string description), and source locations (pdf_page, workbook_sheet:cell, row_index). The location field turns a violation list into something a human can audit.

Anomaly and profiling

Output two things: row-level flags and summary statistics. Use robust statistics — outliers contaminate mean/std, so a 3-sigma rule chases its own tail.

• Median + MAD (median absolute deviation) for the centre + scale.
• Quantile cuts (1st/99th, or 5th/95th) for tail flagging.
• Per-group baselines when records belong to entities — global thresholds dilute group signal.
• Time-window baselines for series — recent trend, not lifetime mean.

Write both: detail flags traceable to row indices, and a summary that reconciles (sum of group counts equals total flagged).

Citation and claim checking

Adjudicate one claim at a time against a closed evidence pack:

• true_positive — claim matches evidence directly.
• false_positive — claim contradicts evidence (or no permitted source supports it).
• duplicate — same claim raised more than once.
• insufficient_evidence — pack does not let you decide either way.
• out_of_scope — claim is about something outside the evidence horizon.

Do not borrow task labels, hidden oracle values, or fixture-specific expected counts as evidence. The evidence pack you adjudicate against is what you build from permitted sources, not what the test expects.

Self-checks before finishing

• The clean-output file contains only records with zero violations.
• Violation count equals the number of failed checks (not failed rows) when the schema asks for check counts.
• Re-read every JSON/CSV from disk and confirm types are native Python (no numpy escapes).
• Summary numbers reconcile with detail rows.
• For cross-format, manually sample a handful of mismatches against both sources.

Pitfalls

• Resetting indices too early — original row positions are lost.
• One row with three failed fields counted as one violation.
• numpy scalar types break json.dump without explicit casting.
• Comparing formatted strings (currency-symbol, thousands-separator) instead of normalized values.
• A summary that cannot be traced back to its detail rows.
• 3-sigma anomaly flagging on data that contains the very outliers you are chasing.