Data quality and validation

An agent wrote the ingest step for a supplier feed. It reads a CSV, loops over the rows, builds dicts, and writes them to your database. It works. Six weeks later the client asks why satisfaction dropped four points, and nobody can tell whether the market moved or the code did. This topic is about the checks that would have answered that in two minutes, and the habit of failing loudly rather than writing bad rows.

Validate at the boundary

The boundary is the exact point data enters your system: a FastAPI request body, the function that reads the supplier file, the parser for a scraped page, the JSON coming back from an LLM step. Validate there, coerce there, and let nothing untyped past it. Everything downstream then gets to assume the data is well formed.

Agent-written ingest code usually looks like this:

# Before: dicts all the way down. Nothing is checked; everything is optional.
def load_responses(path: str) -> list[dict]:
    rows = []
    with open(path) as f:                      # platform default encoding
        for row in csv.DictReader(f):
            rows.append({
                "id": row["response_id"],
                "score": float(row.get("csat") or 0),   # missing becomes 0
                "weight": float(row.get("resp_weight", 1)),
            })
    return rows

Two silent bugs are already in there. A missing satisfaction score becomes a real zero, dragging the mean down. A missing weight becomes 1.0, quietly un-weighting the data. The typed version makes both impossible:

from pydantic import BaseModel, ConfigDict, Field, field_validator

class Response(BaseModel):
    model_config = ConfigDict(extra="forbid")   # unknown columns raise, not ignored
    response_id: str
    csat: int | None = Field(ge=1, le=5)        # explicit: missing is None, never 0
    resp_weight: float = Field(gt=0, le=20)     # a weight of 0 or 900 is a bug

    @field_validator("response_id")
    @classmethod
    def not_blank(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("blank response_id")
        return v

extra="forbid" is the line that earns its keep. Without it, a supplier renaming resp_weight to respondent_weight produces a perfectly valid model with the default weight, and you find out weeks later. With it, the first row raises.

Use Response.model_validate(row) to build one, model_validate_json(raw) when the source is JSON bytes, and model_dump() to get a dict back out. For relationships between fields — “if csat is null then refused must be true” — use @model_validator(mode="after"), which runs once the individual fields are already parsed.

Checks over a whole table, not one row

Row-level validation cannot see that a column’s mean halved, or that a quota cell is empty. Dataframe validation can. Pandera expresses this as a typed schema class:

import pandera.pandas as pa
from pandera.typing.pandas import DataFrame, Series

class Wave(pa.DataFrameModel):
    response_id: Series[str] = pa.Field(unique=True)
    csat: Series[int] = pa.Field(ge=1, le=5, nullable=True)
    resp_weight: Series[float] = pa.Field(gt=0, le=20)
    country: Series[str] = pa.Field(isin=["GB", "US", "DE"])

@pa.check_types                                 # validates in and out on every call
def clean(df: DataFrame[Wave]) -> DataFrame[Wave]:
    return df.drop_duplicates("response_id")

Great Expectations does the same job with a different vocabulary — an expectation suite of named assertions, run as a checkpoint, with a generated report. Either is fine. The point is that a check like “country must be one of these three” or “csat must not be more than 5% null” lives in version control next to the code, rather than in someone’s head.

Schema drift

Schema drift is the shape or meaning of an incoming feed changing without warning. Ranked by how much trouble they cause:

The bottom half of that table is the dangerous half. Nothing crashes, the values are plausible, and the only symptom is a trend that looks like a real market change. Bounded checks (ge, le, isin) are what catch them, because they encode what you believe about the data rather than what Python can parse.

Two habits help more than any library. Never read a CSV by column position — read by name, and assert the header matches what you expect. And always pass encoding="utf-8" explicitly; relying on the platform default is why a curly apostrophe turns into mojibake on one machine and not another, which then breaks every exact string match downstream.

Nulls, duplicates, units

Nulls are not one thing. “Not asked” (routed past the question), “asked and refused”, “don’t know”, and “lost in processing” are four different facts with four different denominators. Collapsing them into a single null — or worse, into zero — changes every percentage you report. Keep them as explicit codes and decide the denominator table by table.

# Before: the shape of your data now depends on a default
df["csat"] = df["csat"].fillna(0)               # a real score of zero. it isn't.

# After: missingness stays visible and the denominator is a decision
answered = df[df["csat"].notna()]
pct_top_box = (answered["csat"] >= 4).mean()    # base = people who answered

Duplicates arrive from a retried job that half-wrote, an overlapping incremental window, a supplier resending a file, or a join that fanned out. The overlapping window is usually deliberate and correct; the missing unique key is the bug. Let the database refuse the duplicate:

CREATE TABLE responses (
  response_id TEXT PRIMARY KEY,                -- natural key, not an autoincrement id
  wave        TEXT NOT NULL,
  csat        INTEGER CHECK (csat BETWEEN 1 AND 5),
  resp_weight REAL    NOT NULL CHECK (resp_weight > 0)
);

-- idempotent: re-running yesterday's window changes nothing
INSERT INTO responses (response_id, wave, csat, resp_weight)
VALUES (?, ?, ?, ?)
ON CONFLICT(response_id) DO UPDATE SET csat = excluded.csat;

Units cause the errors that survive review because both numbers look reasonable. Percentages stored as 0.42 in one table and 42 in another. Seconds versus milliseconds. Currency without a code. Timestamps without a timezone. Put the unit in the column name — duration_ms, revenue_gbp_minor — and store the unit next to the number when it can vary.

Weighting and base sizes

A weight scales each respondent so the sample matches the population on known quotas. A base size is how many real people are behind a figure. The reporting failure is publishing a weighted percentage without its unweighted base, so “62% of IT directors would switch” rests on eleven respondents and reads like a finding.

# Before: agent-written, and wrong for weighted data
pct = (df["would_switch"] == 1).mean()

# After: weighted proportion, carrying the unweighted base with it
w = df["resp_weight"]
pct = (df["would_switch"].eq(1) * w).sum() / w.sum()
base_unweighted = int(df["would_switch"].notna().sum())
if base_unweighted < 30:
    log.warning("low_base", extra={"base": base_unweighted})   # flag, don't hide

Never sum weights and call the result a count of people.

”The numbers moved — was it the world or the code?”

This question should take two minutes, and it does if every run writes a manifest: the code commit, the input file hashes, the row counts at each stage, and the config. Then you diff manifests instead of guessing.

{
  "run_id": "2026-09-24T02:00Z",
  "git_sha": "9f2c1ab",
  "inputs": [{"path": "wave12.csv", "sha256": "4d9e…", "rows": 2841}],
  "stages": {"parsed": 2841, "rejected": 0, "deduped": 2803, "written": 2803},
  "config": {"weight_scheme": "quota_v3", "scale_max": 5}
}

Without one, the fallback still works: re-run the previous commit against the previous input. If the old number reproduces, the code changed. If it does not, the data changed.

Reconciliation, and failing loudly

A reconciliation count asserts that rows are conserved across a stage boundary. It is the only check that catches silent loss — a filter with an accidental and, an inner join that dropped unmatched rows, a truncated read.

assert parsed + rejected == fetched, f"lost {fetched - parsed - rejected} rows at parse"
before = len(df)
df = df.merge(quotas, on="segment", how="left")   # left, not inner: don't drop silently
assert len(df) == before, f"join fanned out: {before} -> {len(df)}"

And the single worst pattern an agent will hand you:

# Before: unknown gaps, no error, table looks complete
for row in rows:
    try:
        save(Response.model_validate(row))
    except Exception:
        continue

A crash costs you a delayed report and a stack trace. A swallowed exception costs you a table with unknown holes that gets joined, aggregated and sent to a client. If partial processing is genuinely acceptable, quarantine the rejects with their error attached and alert on the rate:

rejects = []
for row in rows:
    try:
        good.append(Response.model_validate(row))
    except ValidationError as exc:
        rejects.append({"raw": row, "stage": "parse", "error": exc.json()})

save_quarantine(rejects)                     # replayable once you fix the cause
if len(rejects) / max(len(rows), 1) > 0.01:  # a baseline, not a guess
    raise RuntimeError(f"reject rate {len(rejects)}/{len(rows)} above threshold")

A quarantine table nobody reads is the same as discarding the rows, so the alert is not optional.

What to check in your own code

• Grep for fillna(0), or 0, .get( with a default, and errors="coerce". Each one is a place a missing value became a real number.
• Grep for except Exception followed by pass or continue anywhere in an ingest or transform path.
• Does any Pydantic model reading external data set extra="forbid"? If not, renamed columns are invisible to you.
• Look for a unique constraint or primary key on the natural id of every table a scheduled job writes to. Missing one plus an overlapping window equals doubled counts.
• Search for open( and read_csv( without an explicit encoding=.
• Check every merge and join: is it how="inner" where it should be how="left", and is there a row-count assertion around it?
• Is any percentage computed with .mean() on data that has weights? That number is wrong.
• Does every reported figure carry an unweighted base somewhere a human can see it?
• Does a run write down its commit, input hashes and per-stage row counts? If not, build the manifest before you need it.