How to Build a Data Quality Monitoring System With Pandas

What a data quality check covers

Every batch of incoming data should pass three checks before it enters your system:

• Null / missing values - which columns have gaps, and how many
• Duplicate rows - rows that carry identical values on the key columns
• Statistical anomalies - numeric values more than 3 standard deviations from the column mean (the classic 3-sigma rule)

The result is a report.json with counts for each category and a clean_data.csv with the bad rows removed.

Building the pipeline

import json
import pandas as pd
import numpy as np

def run_quality_check(input_file: str, output_file: str, report_file: str):
    df = pd.read_csv(input_file)
    total_rows = len(df)

    # --- Null detection ---
    null_counts = {c: int(df[c].isna().sum()) for c in df.columns}
    total_nulls = int(sum(null_counts.values()))

    # --- Duplicate detection ---
    # Ignore surrogate ID columns so near-duplicate rows don't slip through
    dedupe_cols = [c for c in df.columns if c.lower() != "id"]
    dup_mask = df.duplicated(subset=dedupe_cols, keep="first")
    duplicate_count = int(dup_mask.sum())

    # --- Anomaly detection (3-sigma) ---
    anomalies = []
    anomaly_mask = pd.Series(False, index=df.index)
    for col in df.select_dtypes(include=[np.number]).columns:
        col_vals = df[col].dropna()
        if col_vals.empty or col_vals.std() == 0:
            continue
        mean, std = col_vals.mean(), col_vals.std()
        mask = (df[col] - mean).abs() > 3 * std
        for idx in df.index[mask & df[col].notna()]:
            anomalies.append({"row": int(idx), "column": col, "value": float(df.at[idx, col])})
            anomaly_mask.at[idx] = True

    # --- Clean output ---
    clean = df[~dup_mask & ~anomaly_mask].dropna()
    clean.to_csv(output_file, index=False)

    report = {
        "total_rows": total_rows,
        "null_counts": null_counts,
        "total_nulls": total_nulls,
        "duplicate_count": duplicate_count,
        "anomaly_count": len(anomalies),
        "anomalies": anomalies,
        "clean_rows": len(clean),
    }
    with open(report_file, "w") as f:
        json.dump(report, f, indent=2)

Why 3 standard deviations?

The 3-sigma rule comes from the empirical rule for normally distributed data: ~99.7% of values fall within 3 standard deviations of the mean. Anything outside that window is statistically unusual enough to warrant inspection. In practice, you tune the threshold per column - sensor data might use 2 sigma, financial amounts might use 4 - but 3 is a safe starting point.

Making the report useful over time

A single report is a snapshot. The real value is trending the counts across batches:

• null_counts growing for a column means an upstream validator broke or a field was made optional
• duplicate_count spiking means a source system is double-writing (common after retries)
• anomaly_count trending up means the underlying distribution is shifting - investigate before it becomes a data drift problem

Store one report JSON per batch (named by timestamp or batch ID) and compare them in a dashboard or a simple diff to catch regressions early.