Make CI Catch Bad Data

My green pipeline from the last lesson stayed green while the model quietly lost several points of accuracy over a week. Every code test I had written passed. A schema change in an upstream feed had started sending annual_inc, the borrower’s stated annual income, as a string instead of a number, pandas silently coerced the whole column to its catch-all object dtype, and the model trained on garbage that every code test happily ignored. The pipeline was green because it tested the code, and the code was fine. The data was the bug, and nothing in CI could see it. I owned the whole pipeline, so there was no second reviewer and no data-platform team to catch it; the green check was the only thing watching, and it was looking at the wrong half of the system.

The pipeline from the last lesson is green and it tests the code. This lesson hardens the gap that makes a green ML pipeline a lie: it never looks at the data. The fix is a new pipeline stage, carrying schema, range, and leakage assertions, that slots in after the code tests and before the model fit, and that fails the build on a bad-DATA commit, not only a bad-code commit. By the end you will have a data stage that turns the corrupted-income incident above into a red check on the commit that introduced it.

A green code suite says nothing about the data

Here is the mental model that breaks first: a green pipeline is a statement about the whole running system. A passing test suite reads as “the system is healthy,” so a green check on a commit is taken to mean the model that commit produces is fine. That reading holds for ordinary software, where the program is the code and the tests exercise the code. It does not hold for a machine-learning system, because the served model is a function of two arguments, not one: model = f(code, data). Continuous integration, the checks that run automatically on a clean machine on every change from the last lesson, re-runs the code’s assertions on every push, but it only ever exercises the code half against a fixture frozen when the test was written. The data half changes in production between pushes, on its own schedule, observed by nothing.

A unit test pins its input and checks its output, so it can never observe a change in production data: that data is not in the test, it is a fixture written into the test file. Hold the code fixed and corrupt the second argument, and the model’s behaviour moves with zero code diff and zero failing test. Here is the failure the opening incident describes, reduced to the mechanism. Watch the model train without raising anything on a column that is no longer numeric.

import pandas as pd
from sklearn.linear_model import LogisticRegression

# upstream "fixed" a parser and started sending annual_inc as a string
raw = {"annual_inc": ["55000", "?", "90000"], "default": [0, 0, 1]}
df = pd.DataFrame(raw)

print(df["annual_inc"].dtype)   # str on pandas 3.0 (object on pandas 2.x) — not float64
model = LogisticRegression().fit(df[["annual_inc"]], df["default"])
# no exception. the model fit on a column of Python string pointers.

The "?" is the killer. A pandas column holds a numeric dtype only while every value parses as a number; the instant one non-numeric value enters during parsing, pandas cannot keep the column numeric and falls back to a non-numeric dtype (object on pandas 2.x, the dedicated str dtype on pandas 3.0, which is what the example below prints), the widest type that holds anything. No exception is raised, because object is a perfectly valid dtype: it is a column of Python pointers rather than a typed NumPy float64 array. Arithmetic still runs, but it now runs at the Python level on those pointers with much more overhead than the vectorised native-type math the model relied on, so the fit completes on a column that is no longer the feature it was. Every code test passes because every code path is exercised on the test’s own clean fixture; the corruption lives only in production data the test never loads.

Treating the code and the data as two separate arguments is the correct model. The data stage asserts on the shape and dtype of the data at ingest, the only place CI can see the function’s second argument at all. Here is that assertion as the smallest thing that would have turned the incident red.

import pandas as pd


def assert_numeric_dtype(df: pd.DataFrame, column: str) -> None:
    """Fail loudly if a column the model treats as numeric is not numeric."""
    dtype = df[column].dtype
    if not pd.api.types.is_numeric_dtype(dtype):
        raise AssertionError(
            f"{column!r} must be numeric, got dtype={dtype}. "
            f"An upstream change likely sent a non-numeric value."
        )


def main() -> None:
    good = pd.DataFrame({"annual_inc": [55000.0, 90000.0, 42000.0]})
    bad = pd.DataFrame({"annual_inc": ["55000", "?", "90000"]})

    assert_numeric_dtype(good, "annual_inc")
    print("good frame: passed")

    try:
        assert_numeric_dtype(bad, "annual_inc")
    except AssertionError as err:
        print(f"bad frame: caught -> {err}")


if __name__ == "__main__":
    main()

One frame passes silently; the bad frame raises with the column named, so a developer reading the CI log sees exactly which feed regressed. The check reads one attribute, df[column].dtype, and touches no model fit, which is why it belongs at the front of the data stage and why it costs almost nothing to run on every push. Why does pandas return object for the bad frame instead of raising at construction time, the way a typed language would reject a string assigned to a float field? Because pandas has no declared schema to violate; object is its escape hatch for “I cannot find a common numeric type,” and silence is the default.

This is a new pipeline stage, not a new test framework. It slots into the ordered pipeline from the last lesson, after the code tests and before the model fit, so that a bad-data commit fails the build at the data stage before any compute is spent fitting a model over garbage. The named failure mode is the opening incident exactly: accuracy decayed with no code change and a fully green suite, because an upstream type change was silently coerced to object and the model trained on a column it could no longer use numerically. In an ML system the data is effectively code, because a change to the data changes the production system’s behaviour the same way a code change would, so a code-only suite cannot establish system health. The data stage is the half of the system the code suite was never built to watch.

import pandas as pd


def check_numeric(df: pd.DataFrame, column: str) -> bool:
    # this returns True for ANY column that exists -- that is the bug
    return column in df.columns


def main() -> None:
    bad = pd.DataFrame({"annual_inc": ["55000", "?", "90000"]})
    print("passes:", check_numeric(bad, "annual_inc"))


if __name__ == "__main__":
    main()

import pandas as pd


def check_numeric(df: pd.DataFrame, column: str) -> None:
    dtype = df[column].dtype
    if not pd.api.types.is_numeric_dtype(dtype):
        raise AssertionError(f"{column!r} is {dtype}, expected numeric")


def main() -> None:
    bad = pd.DataFrame({"annual_inc": ["55000", "?", "90000"]})

    try:
        check_numeric(bad, "annual_inc")
    except AssertionError as err:
        print(f"caught: {err}")


if __name__ == "__main__":
    main()

Schema and range: the cheapest data gate

The natural next move, once data clearly needs checking, is to reach for the most powerful check available: a statistical drift detector that compares this batch’s distribution to last week’s. That instinct is backwards. Most bad-data incidents are not distributional at all: they are a renamed column, a flipped encoding, a null where the contract said non-null, a label that is not one of the known classes. A drift check is blind to every one of those, because the distribution of a renamed column is undefined and the distribution of a flipped boolean looks fine. Skipping the boring structural check and going straight to distribution misses the common case while paying for the expensive one.

Assert the data contract at the boundary, structure before statistics. There is no compiler across a data boundary: upstream systems rename columns, change encodings, and introduce nulls without telling downstream consumers, and the change arrives as data, not as a broken build, so the consumer keeps running on input that no longer means what it did. A schema assertion reintroduces the missing compiler. It compares the incoming frame’s structure (the set of column names, each column’s dtype, each column’s nullability) against a declared contract, and fails on the first divergence, naming the offending column. This is the typed data contract from Module 1 enforced one layer out: in Module 1 the contract lived in-process at a function boundary; here the same contract runs in CI on every push, against data that arrives before the process even starts.

Why does the order matter? Because it is the design, not tidiness: structure, then statistics, then model, cheapest and most-common failure first. A structure check reads only metadata (df.columns is a list, df.dtypes is one entry per column, df.isna().any() is a single pass), none of which touches a model fit, so a renamed column fails the build in the time it takes to read the frame’s header instead of after a multi-minute fit on garbage. This mirrors the production pattern of running cheap freshness and volume checks before deep data-quality checks: the cheap one-pass check gates the expensive statistical one, so noisy or costly checks never run on data a single pass would already have rejected. The scrolly below walks one commit through that ordered stage.

A range check is the second layer because structure alone cannot catch a value that is structurally valid but semantically impossible. annual_inc = -3 passes the dtype check, because -3 is a perfectly good float, but no applicant earns negative income; a label outside {fully_paid, charged_off} is a valid string but not a valid class. The contract has to assert bounds, allowed sets, and nullability on top of dtype, or the cheap gate catches the wrong-type failure and waves the wrong-value failure straight through. Watch the schema check reject a renamed column and the range check reject a structurally-valid-but-impossible income.

import pandas as pd

SCHEMA = {
    "annual_inc": "float64",
    "loan_amnt": "float64",
    "default": "int64",
}


def check_schema(df: pd.DataFrame, schema: dict[str, str]) -> None:
    for col, want in schema.items():
        if col not in df.columns:
            raise AssertionError(f"missing column {col!r}")
        got = str(df[col].dtype)
        if got != want:
            raise AssertionError(f"{col!r}: expected {want}, got {got}")


def check_ranges(df: pd.DataFrame) -> None:
    if (df["annual_inc"] < 0).any():
        raise AssertionError("annual_inc has negative values")
    allowed = {0, 1}
    bad = set(df["default"].unique()) - allowed
    if bad:
        raise AssertionError(f"default has values outside {allowed}: {bad}")


def main() -> None:
    good = pd.DataFrame(
        {"annual_inc": [55000.0], "loan_amnt": [10000.0], "default": [0]}
    )
    check_schema(good, SCHEMA)
    check_ranges(good)
    print("good frame: passed structure and range")

    renamed = good.rename(columns={"annual_inc": "annualIncome"})
    try:
        check_schema(renamed, SCHEMA)
    except AssertionError as err:
        print(f"structure: caught -> {err}")

    negative = pd.DataFrame(
        {"annual_inc": [-3.0], "loan_amnt": [10000.0], "default": [0]}
    )
    check_schema(negative, SCHEMA)  # passes -- dtype is fine
    try:
        check_ranges(negative)
    except AssertionError as err:
        print(f"range: caught -> {err}")


if __name__ == "__main__":
    main()

Run it and the renamed frame fails structure, while the negative-income frame clears structure and is caught only by range, because its dtype is a valid float. Notice that the negative frame passed check_schema: a value being the right type tells you nothing about whether it is a possible value, which is precisely why the range layer cannot be folded into the dtype check. What is the cost of pinning a range bound? A range learned from one data snapshot becomes a false alarm the day the legitimate range shifts: a new high-income loan product raises the real maximum income, and a too-tight annual_inc <= 500000 bound now fails good data. The bound is a judgment call about the domain, not a fact about the current sample, so pinning it to the observed min and max bakes today’s sample into tomorrow’s gate.

Name the failure mode here and it is the boring, common one a drift check never surfaces: a renamed annual_inc, an out-of-range label, a null in a non-null column. None of these are distributional, so the statistical layer is blind to all of them, and they are the cheapest to catch. A data contract formalizes the producer-to-consumer relationship, where the producer publishes and adheres to it and the consumer builds on its guarantees, and the data stage is where that contract is enforced as a pipeline gate rather than a document nobody reads.

import pandas as pd


def check_ranges(df: pd.DataFrame) -> None:
    allowed = {0, 1}
    bad = set(df["default"].unique()) - allowed
    if bad:
        raise AssertionError(f"default outside {allowed}: {bad}")
    # add a dti bound check here so dti=-5 raises
    return None


def main() -> None:
    frame = pd.DataFrame({"default": [0, 1], "dti": [22.0, -5.0]})
    check_ranges(frame)
    print("passed (it should not have)")


if __name__ == "__main__":
    main()

import pandas as pd


def check_ranges(df: pd.DataFrame) -> None:
    allowed = {0, 1}
    bad = set(df["default"].unique()) - allowed
    if bad:
        raise AssertionError(f"default outside {allowed}: {bad}")
    if ((df["dti"] < 0) | (df["dti"] > 100)).any():
        raise AssertionError("dti outside [0, 100]")


def main() -> None:
    frame = pd.DataFrame({"default": [0, 1], "dti": [22.0, -5.0]})
    try:
        check_ranges(frame)
    except AssertionError as err:
        print(f"caught: {err}")


if __name__ == "__main__":
    main()

The leakage test: assert the model cannot see the future

A default model validates at near-perfect ranking in CI and predicts at near-baseline in production. Offline it looks like the best model the team has ever trained, so the wrong hypothesis writes itself: validation accuracy is high, so the model is good, ship it. High validation accuracy is treated as the all-clear, and the diagnosis stops at the number that looks best. That number is the symptom, not the all-clear. The model is not good: it has been handed a feature that is known only after the outcome it is supposed to predict, and the validation split cannot tell the difference.

The most dangerous data bug is a feature correlated with the target only because it is recorded after the outcome. On the Lending Club default scorer the killer column is recoveries, the dollar amount recovered from a borrower after a default. It is non-zero almost only after a default, so a non-zero recoveries is a default; the model is not predicting, it is reading the answer. The split makes this invisible. A train/test split is a time-blind random shuffle, so a post-outcome feature leaks into both the training rows and the validation rows, and validation looks spectacular exactly when the model has learned nothing useful. In production those fields are zero or null at origination-time scoring, so the learned weights point at nothing and accuracy collapses toward baseline. The inflation is up offline and down in production for the same reason: the leak is present in validation and absent at serve time.

Here is that asymmetry as a number. The model trains and validates with recoveries in the matrix, then is asked to score the way production scores it, with recoveries zeroed out, because at origination it has not happened. Watch the validation score and the serve-time score diverge.

import numpy as np
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import roc_auc_score
from sklearn.model_selection import train_test_split


def main() -> None:
    rng = np.random.default_rng(0)
    n = 4000
    default = rng.integers(0, 2, size=n)
    annual_inc = rng.normal(60000, 15000, size=n) - default * 4000
    # recoveries: a post-outcome field, non-zero almost only after a default
    recoveries = default * rng.uniform(500, 3000, size=n)

    X = np.column_stack([annual_inc, recoveries])
    y = default
    X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.3, random_state=0)

    model = LogisticRegression(max_iter=1000).fit(X_tr, y_tr)

    val_auc = roc_auc_score(y_te, model.predict_proba(X_te)[:, 1])

    # serve time: recoveries is 0 at origination -- it has not happened yet
    X_serve = X_te.copy()
    X_serve[:, 1] = 0.0
    serve_auc = roc_auc_score(y_te, model.predict_proba(X_serve)[:, 1])

    print(f"validation AUC (recoveries present): {val_auc:.3f}")
    print(f"serve-time AUC (recoveries zeroed):  {serve_auc:.3f}")
    print(f"the model lost {val_auc - serve_auc:.3f} AUC crossing to production")


if __name__ == "__main__":
    main()

AUC (the ranking measure from the evaluation lesson, the probability a random defaulter is scored above a random non-defaulter) sits near the top offline and falls toward the 0.5 of a coin flip once recoveries is zeroed. The model did not get worse; it was always this weak, and the leaked column was carrying the score. This is why leakage is a data test, not a modelling mistake caught by eyeballing accuracy: eyeballing accuracy is exactly what makes a leaky model look like a triumph.

The correct gate is a structural audit of feature availability at serve time, not a threshold on validation accuracy. Because the leak is many columns and an upstream join can re-add one at any time, the assertion cannot be “ban recoveries.” On Lending Club the post-outcome set the data stage enforces is these seven fields (recoveries, collection_recovery_fee, total_pymnt, total_rec_prncp, last_pymnt_amnt, out_prncp, settlement_amount), all zero or null at loan origination and only populated during or after the loan’s life. The audit asks one question of every column, is this value known at serve time?, and fails the build if any post-outcome column is in the feature matrix. The diagram below is the structure that question enforces: two schemas over one timeline, split at the scoring instant.

package "Known at origination (serve time)" #2d4a3e {
  [annual_inc]
  [loan_amnt]
  [grade]
  [emp_length]
}
package "Known only after the outcome (post-outcome)" #4a2d2d {
  [total_pymnt]
  [out_prncp]
  [recoveries]
  [settlement_amount]
}
[feature matrix] as fm
[annual_inc] --> fm : allowed
[loan_amnt] --> fm : allowed
[recoveries] ..> fm : BANNED — known only after default
[settlement_amount] ..> fm : BANNED — post-outcome

Read the two packages as the serve-time-availability contract: the green set is what a scoring request can actually carry at origination, the red set is what only exists once the loan has run. The audit fails the build if any red-set column reaches the feature matrix. Here is that audit as the assertion that turns the leaky commit red in CI. Watch it reject the matrix that contains a post-outcome column and pass the one that does not.

POST_OUTCOME = {
    "recoveries",
    "collection_recovery_fee",
    "total_pymnt",
    "total_rec_prncp",
    "last_pymnt_amnt",
    "out_prncp",
    "settlement_amount",
}


def assert_no_leakage(feature_columns: list[str]) -> None:
    """Fail the build if any feature is known only after the outcome."""
    leaked = sorted(set(feature_columns) & POST_OUTCOME)
    if leaked:
        raise AssertionError(
            f"post-outcome columns in feature matrix: {leaked}. "
            f"These are known only after the loan's outcome and cannot be served."
        )


def main() -> None:
    safe = ["annual_inc", "loan_amnt", "grade", "emp_length"]
    assert_no_leakage(safe)
    print("safe feature set: passed")

    leaky = ["annual_inc", "loan_amnt", "recoveries"]
    try:
        assert_no_leakage(leaky)
    except AssertionError as err:
        print(f"leaky feature set: caught -> {err}")


if __name__ == "__main__":
    main()

The audit fails on the set difference, so a single re-added post-outcome column is enough to turn the commit red, which is the point, because an upstream join needs only one such column to recreate the incident. Why audit every column instead of validating the score against a threshold? Because a threshold on validation accuracy would pass the leaky model, whose validation number is excellent, and the structural audit encodes the one fact the data alone cannot state: when, relative to the prediction, is this value known? The named failure mode is the incident exactly: a default model validated at near-perfect ranking and predicted at near-baseline because recoveries had been re-added to the feature matrix by an upstream join. The model had been reading the outcome, not predicting it. This forward-links to the train/serve skew lesson at the end of the module: leakage is the training-time half of “the model saw something at fit time it will not see at serve time.”

POST_OUTCOME = {"recoveries", "total_pymnt", "out_prncp"}


def assert_no_leakage(feature_columns: list[str]) -> None:
    # this only catches recoveries by name -- total_pymnt sails through
    if "recoveries" in feature_columns:
        raise AssertionError("recoveries is leaked")
    return None


def main() -> None:
    leaky = ["annual_inc", "total_pymnt"]
    assert_no_leakage(leaky)
    print("passed (it should not have)")


if __name__ == "__main__":
    main()

POST_OUTCOME = {"recoveries", "total_pymnt", "out_prncp"}


def assert_no_leakage(feature_columns: list[str]) -> None:
    leaked = sorted(set(feature_columns) & POST_OUTCOME)
    if leaked:
        raise AssertionError(f"post-outcome columns leaked: {leaked}")


def main() -> None:
    leaky = ["annual_inc", "total_pymnt"]
    try:
        assert_no_leakage(leaky)
    except AssertionError as err:
        print(f"caught: {err}")


if __name__ == "__main__":
    main()

Summary

• A green code suite is a statement about the code, not the system: the served model is f(code, data), and CI only ever exercises the code half against a frozen fixture. The data half changes in production with no code diff and no failing test, which is why a corrupted feed decays the model while the pipeline stays green.
• A pandas column holds a numeric dtype only while every value parses as numeric; one non-numeric value forces a silent fall back to object, a column of Python pointers the model fits on without complaint. A dtype assertion at ingest is the cheapest thing in the data stage and the one check that catches this.
• The data stage runs structure, then statistics, then model, cheapest and most-common failure first. Structure (names, dtypes, nullability) and range (bounds, allowed sets) catch the boring, non-distributional failures a drift check is blind to, and they run on metadata before any model fit is spent.
• A range bound is a judgment about the domain, not a fact about the current sample: pinning it to the observed min and max bakes today’s data into tomorrow’s gate, so a legitimate range shift fails good data.
• Leakage is a data test, not a modelling mistake. A post-outcome feature like recoveries inflates validation and collapses at serve time; the gate is a structural audit of every column against a serve-time-availability contract, never a threshold on validation accuracy.

Check your understanding:

• Without looking back: a model’s accuracy drops over a week with zero code commits and a fully green CI suite. What is the first place you look, and why could the test suite never have caught it?
• Why does the range check have to be a separate layer from the dtype check, and what kind of bad value passes the dtype check and is caught only by range?
• A model validates at near-perfect AUC and serves at an AUC near a coin flip. What single class of bug produces exactly this up-offline, down-in-production asymmetry, and why does a threshold on validation accuracy fail to catch it?