test-driven-data-science

Test-Driven Data Science (TDDS)

Write and run three layers of data assertions before every training step. If any CRITICAL assertion fails, STOP — do not train. Fix the data, then re-run all three layers.

Why three layers: Schema checks catch type errors. Business rule checks catch impossible domain values. Statistical checks catch silent distribution shifts that break trained models. One layer is not enough.

Iron Laws

• NO TRAINING WITHOUT PASSING ALL THREE ASSERTION LAYERS
• NO SILENT FAILURES — every failed assertion must be named and reported
• NEVER skip Statistical layer because "it's the same data source"

When to Use

Data arrives / pipeline produces new features
           ↓
Run Layer 1: Physical (schema)    ← always first
           ↓ pass
Run Layer 2: Logical (business)   ← domain rules
           ↓ pass
Run Layer 3: Statistical (drift)  ← distribution stability
           ↓ pass
Proceed to training

Trigger this skill whenever:

• New data arrives from an upstream pipeline
• A feature engineering step produces a new dataset
• You are about to call model.fit() for the first time or after any data change

Layer 1 — Physical (Schema) Assertions

Check that the DataFrame matches the declared schema. Use Pandera.

import pandera as pa
from pandera import Column, DataFrameSchema, Check
import pandas as pd

def build_physical_schema(expected_columns: dict) -> DataFrameSchema:
    """
    expected_columns: {col_name: {"dtype": pa.Float, "nullable": False, "checks": [...]}}
    """
    return DataFrameSchema(
        columns={
            name: Column(
                spec["dtype"],
                nullable=spec.get("nullable", False),
                checks=spec.get("checks", []),
            )
            for name, spec in expected_columns.items()
        },
        strict=True,  # fail on unexpected columns
    )

# Example schema
schema = build_physical_schema({
    "age":          {"dtype": pa.Float, "nullable": False, "checks": [Check.between(0, 120)]},
    "income":       {"dtype": pa.Float, "nullable": True,  "checks": [Check.greater_than_or_equal_to(0)]},
    "churn_label":  {"dtype": pa.Int,   "nullable": False, "checks": [Check.isin([0, 1])]},
})

def run_physical_layer(df: pd.DataFrame, schema: DataFrameSchema) -> dict:
    try:
        schema.validate(df, lazy=True)
        return {"layer": "physical", "status": "PASS", "failures": []}
    except pa.errors.SchemaErrors as e:
        failures = e.failure_cases[["column", "check", "failure_case"]].to_dict("records")
        return {"layer": "physical", "status": "FAIL", "failures": failures}