Automating pandas Data Cleaning with Custom Validation Rules, Error Logs, and a Corrected Output File

Introduction

You've inherited a monthly sales report process. Every month, someone downloads a CSV from the CRM, opens it in Excel, manually hunts for obvious problems — negative quantities, missing customer IDs, dates in the wrong format — fixes what they can, and prays the rest is clean enough to not blow up the downstream reports. This takes three hours, introduces new errors, and leaves no audit trail. When the quarterly number looks wrong six weeks later, nobody can reconstruct what was "cleaned" and what wasn't.

This is the problem a data validation pipeline solves. Not just detecting bad data, but catching it systematically, logging exactly what broke and where, applying corrections where possible, and writing a clean output file that any downstream consumer can trust — all without a human in the loop. By the end of this lesson, you'll have built that pipeline from scratch. You'll understand how to define validation rules as reusable code, accumulate structured error logs that are themselves a pandas DataFrame, apply corrections in a safe and traceable way, and write both a cleaned output file and a human-readable error report.

What you'll learn:

• How to architect a custom data validation framework using pure pandas and Python
• How to write rule functions that return structured, actionable error records rather than just True/False
• How to build an error log DataFrame that captures row index, column, rule name, bad value, and corrective action
• How to apply auto-corrections safely without silently masking data quality problems
• How to export a corrected output file alongside a validation report your stakeholders can actually read
• How to structure this as a reusable, schedulable script rather than a one-time notebook

---

Prerequisites

You should be comfortable with:

• Cleaning Messy Data with pandas: Missing Values, Duplicates, and Data Types — we'll build on the foundations covered there
• Selecting and Filtering Data in pandas: loc, iloc, and Boolean Masks — boolean masks are central to how validation rules work
• Basic Python functions and list/dict comprehensions — see Python Basics for Excel Users: Variables, Lists, Dictionaries, and Loops if you need a refresher

You should have pandas, openpyxl, and Python 3.9+ installed. If you're setting up fresh, Setting Up Python for Data Analysis: Python, VS Code, Jupyter, and Virtual Environments has you covered.

---

The Dataset: What We're Working With

Let's make this concrete. We're working with a sales transaction CSV that a CRM system exports monthly. It looks clean at first glance, but experience tells you it isn't. Here's the schema:

Column	Expected Type	Business Rules
transaction_id	string	Required, unique, format TXN-XXXXXX
customer_id	string	Required, no nulls
sale_date	date	Required, not in the future, not before 2020-01-01
product_sku	string	Must match SKU-\d{4} pattern
quantity	integer	Must be positive, ≤ 500
unit_price	float	Must be positive, ≤ 10,000
discount_pct	float	0.0 to 0.5 (0%–50%)
sales_rep_id	string	Must exist in a reference table
region	string	One of: North, South, East, West

Let's create a synthetic dirty dataset to work with throughout this lesson:

import pandas as pd
import numpy as np
from datetime import date, timedelta
import random
import re

random.seed(42)
np.random.seed(42)

# Clean base data
n = 200
regions = ["North", "South", "East", "West"]
rep_ids = [f"REP-{i:03d}" for i in range(1, 11)]  # REP-001 through REP-010

df = pd.DataFrame({
    "transaction_id": [f"TXN-{i:06d}" for i in range(1, n + 1)],
    "customer_id": [f"CUST-{random.randint(1000, 9999)}" for _ in range(n)],
    "sale_date": pd.date_range("2023-01-01", periods=n, freq="D").strftime("%Y-%m-%d").tolist(),
    "product_sku": [f"SKU-{random.randint(1000, 9999)}" for _ in range(n)],
    "quantity": np.random.randint(1, 100, size=n).tolist(),
    "unit_price": np.round(np.random.uniform(10, 500, size=n), 2).tolist(),
    "discount_pct": np.round(np.random.uniform(0, 0.4, size=n), 2).tolist(),
    "sales_rep_id": [random.choice(rep_ids) for _ in range(n)],
    "region": [random.choice(regions) for _ in range(n)],
})

# Inject known errors
df.loc[5, "customer_id"] = None
df.loc[12, "quantity"] = -15
df.loc[22, "unit_price"] = 0
df.loc[33, "discount_pct"] = 0.75  # Over 50%
df.loc[44, "sale_date"] = "2019-06-15"  # Before cutoff
df.loc[55, "sale_date"] = "2099-01-01"  # Future date
df.loc[66, "transaction_id"] = "TXN-000001"  # Duplicate
df.loc[77, "region"] = "Northwest"  # Invalid region
df.loc[88, "product_sku"] = "BADSKU"  # Wrong format
df.loc[99, "sales_rep_id"] = "REP-999"  # Not in reference list
df.loc[110, "quantity"] = 750  # Exceeds max
df.loc[120, "unit_price"] = -50  # Negative price

df.to_csv("sales_raw.csv", index=False)
print(f"Created sales_raw.csv with {len(df)} rows and {df['transaction_id'].duplicated().sum()} duplicate transaction IDs")

This gives us a realistic dirty file with 12 distinct problems spread across different columns and rule types.

---

Designing the Validation Architecture

Before writing a single validation check, it's worth pausing to design the system. The naive approach is a series of if statements that print warnings. The problem: that approach doesn't scale, isn't reusable, and produces no artifact you can share with stakeholders or store for audit.

The better approach treats each validation rule as a function that takes the DataFrame and returns a list of structured error records. Every error record contains the same fields:

{
    "row_index": 12,
    "column": "quantity",
    "rule": "quantity_positive",
    "bad_value": -15,
    "severity": "error",        # "error" or "warning"
    "corrective_action": "set_to_null",
    "corrected_value": None
}

This design decision — errors as data — is what separates a toy script from a production-grade pipeline. When errors are records in a DataFrame, you can aggregate them by rule, filter by severity, join them back to the original data for reporting, and export them to Excel for your ops team.

The pipeline has four stages:

1. Load — read the raw file and parse types where possible
2. Validate — run all rules, collect error records into a log
3. Correct — apply corrections where you've defined them; mark un-correctable rows
4. Export — write the corrected DataFrame and the error log

Let's build each stage.

---

Stage 1: Loading and Type-Parsing the Raw File

Robust loading means being explicit about types and not letting pandas guess:

import pandas as pd
import numpy as np
from datetime import date
import re

REFERENCE_DATA = {
    "valid_rep_ids": {f"REP-{i:03d}" for i in range(1, 11)},
    "valid_regions": {"North", "South", "East", "West"},
    "min_sale_date": pd.Timestamp("2020-01-01"),
    "max_sale_date": pd.Timestamp(date.today()),
    "max_quantity": 500,
    "max_unit_price": 10_000.0,
    "max_discount_pct": 0.5,
}

def load_raw(filepath: str) -> pd.DataFrame:
    """Load the raw CSV with explicit dtype handling."""
    df = pd.read_csv(
        filepath,
        dtype={
            "transaction_id": "string",
            "customer_id": "string",
            "product_sku": "string",
            "sales_rep_id": "string",
            "region": "string",
        },
        parse_dates=["sale_date"],
    )
    # Coerce numeric columns — invalid entries become NaN rather than crashing
    df["quantity"] = pd.to_numeric(df["quantity"], errors="coerce")
    df["unit_price"] = pd.to_numeric(df["unit_price"], errors="coerce")
    df["discount_pct"] = pd.to_numeric(df["discount_pct"], errors="coerce")
    
    print(f"Loaded {len(df):,} rows from {filepath}")
    return df

Notice the use of errors="coerce" on numeric columns. This converts garbage string values like "N/A" or "$15.00" to NaN rather than raising an exception. We'll catch those NaNs as validation errors rather than letting them crash the pipeline.

---

Stage 2: Writing Validation Rules

Each rule is a Python function with a consistent signature:

def rule_name(df: pd.DataFrame, ref: dict) -> list[dict]:
    # Returns a list of error record dicts (empty if no violations)
    ...

This consistency is critical — it means you can store all your rules in a list and loop over them without knowing anything about what each rule checks. Let's build them:

def check_missing_customer_id(df, ref):
    """customer_id must not be null."""
    mask = df["customer_id"].isna()
    errors = []
    for idx in df[mask].index:
        errors.append({
            "row_index": idx,
            "column": "customer_id",
            "rule": "required_field",
            "bad_value": None,
            "severity": "error",
            "corrective_action": "flag_for_manual_review",
            "corrected_value": None,
        })
    return errors


def check_transaction_id_unique(df, ref):
    """transaction_id must be unique across the file."""
    duplicated_mask = df["transaction_id"].duplicated(keep="first")
    errors = []
    for idx in df[duplicated_mask].index:
        errors.append({
            "row_index": idx,
            "column": "transaction_id",
            "rule": "unique_id",
            "bad_value": df.at[idx, "transaction_id"],
            "severity": "error",
            "corrective_action": "flag_for_manual_review",
            "corrected_value": None,
        })
    return errors


def check_transaction_id_format(df, ref):
    """transaction_id must match TXN-XXXXXX."""
    pattern = re.compile(r"^TXN-\d{6}$")
    # isna() check first to avoid applying regex to NaN
    mask = df["transaction_id"].notna() & ~df["transaction_id"].str.match(pattern)
    errors = []
    for idx in df[mask].index:
        errors.append({
            "row_index": idx,
            "column": "transaction_id",
            "rule": "id_format",
            "bad_value": df.at[idx, "transaction_id"],
            "severity": "error",
            "corrective_action": "flag_for_manual_review",
            "corrected_value": None,
        })
    return errors


def check_sale_date_range(df, ref):
    """sale_date must be between min_sale_date and today."""
    too_old = df["sale_date"] < ref["min_sale_date"]
    in_future = df["sale_date"] > ref["max_sale_date"]
    errors = []
    for idx in df[too_old | in_future].index:
        bad_date = df.at[idx, "sale_date"]
        if pd.isna(bad_date):
            reason = "missing_date"
        elif bad_date < ref["min_sale_date"]:
            reason = "date_before_cutoff"
        else:
            reason = "date_in_future"
        errors.append({
            "row_index": idx,
            "column": "sale_date",
            "rule": "date_range",
            "bad_value": str(bad_date),
            "severity": "error",
            "corrective_action": "flag_for_manual_review",
            "corrected_value": None,
        })
    return errors


def check_quantity(df, ref):
    """quantity must be a positive integer, max 500."""
    missing = df["quantity"].isna()
    negative = df["quantity"] <= 0
    too_large = df["quantity"] > ref["max_quantity"]
    
    errors = []
    for idx in df[missing | negative | too_large].index:
        val = df.at[idx, "quantity"]
        if pd.isna(val):
            action, corrected = "set_to_null", None
        elif val <= 0:
            action, corrected = "set_to_null", None  # Can't guess correct quantity
        else:
            # Exceeds max — this might be a data entry error (e.g., 750 entered instead of 75)
            action, corrected = "cap_at_max", ref["max_quantity"]
        errors.append({
            "row_index": idx,
            "column": "quantity",
            "rule": "quantity_range",
            "bad_value": val,
            "severity": "error",
            "corrective_action": action,
            "corrected_value": corrected,
        })
    return errors


def check_unit_price(df, ref):
    """unit_price must be > 0 and <= 10,000."""
    missing = df["unit_price"].isna()
    not_positive = df["unit_price"] <= 0
    too_high = df["unit_price"] > ref["max_unit_price"]
    
    errors = []
    for idx in df[missing | not_positive | too_high].index:
        val = df.at[idx, "unit_price"]
        errors.append({
            "row_index": idx,
            "column": "unit_price",
            "rule": "price_range",
            "bad_value": val,
            "severity": "error",
            "corrective_action": "set_to_null",
            "corrected_value": None,
        })
    return errors


def check_discount_pct(df, ref):
    """discount_pct must be between 0.0 and 0.5."""
    out_of_range = (df["discount_pct"] < 0) | (df["discount_pct"] > ref["max_discount_pct"])
    errors = []
    for idx in df[out_of_range].index:
        val = df.at[idx, "discount_pct"]
        errors.append({
            "row_index": idx,
            "column": "discount_pct",
            "rule": "discount_range",
            "bad_value": val,
            "severity": "warning",  # Warning, not error — might be a legitimate override
            "corrective_action": "cap_at_max",
            "corrected_value": ref["max_discount_pct"],
        })
    return errors


def check_product_sku_format(df, ref):
    """product_sku must match SKU-DDDD."""
    pattern = re.compile(r"^SKU-\d{4}$")
    mask = df["product_sku"].notna() & ~df["product_sku"].str.match(pattern)
    errors = []
    for idx in df[mask].index:
        errors.append({
            "row_index": idx,
            "column": "product_sku",
            "rule": "sku_format",
            "bad_value": df.at[idx, "product_sku"],
            "severity": "error",
            "corrective_action": "flag_for_manual_review",
            "corrected_value": None,
        })
    return errors


def check_sales_rep_id(df, ref):
    """sales_rep_id must exist in the reference set."""
    invalid_mask = ~df["sales_rep_id"].isin(ref["valid_rep_ids"])
    errors = []
    for idx in df[invalid_mask].index:
        errors.append({
            "row_index": idx,
            "column": "sales_rep_id",
            "rule": "referential_integrity",
            "bad_value": df.at[idx, "sales_rep_id"],
            "severity": "error",
            "corrective_action": "set_to_null",
            "corrected_value": None,
        })
    return errors


def check_region(df, ref):
    """region must be one of the valid values."""
    invalid_mask = ~df["region"].isin(ref["valid_regions"])
    errors = []
    for idx in df[invalid_mask].index:
        errors.append({
            "row_index": idx,
            "column": "region",
            "rule": "allowed_values",
            "bad_value": df.at[idx, "region"],
            "severity": "error",
            "corrective_action": "set_to_null",
            "corrected_value": None,
        })
    return errors

A critical design point: notice that some rules return "warning" severity rather than "error". The distinction matters. Errors mean the row should not enter the clean output without correction or review. Warnings mean the value is suspicious but the row might still be useful. Building this distinction into the log gives downstream consumers the ability to make informed decisions.

---

Stage 3: Running the Validation Engine

Now we assemble all the rules into a runner:

VALIDATION_RULES = [
    check_missing_customer_id,
    check_transaction_id_unique,
    check_transaction_id_format,
    check_sale_date_range,
    check_quantity,
    check_unit_price,
    check_discount_pct,
    check_product_sku_format,
    check_sales_rep_id,
    check_region,
]

def run_validation(df: pd.DataFrame, ref: dict) -> pd.DataFrame:
    """
    Run all validation rules and return a structured error log DataFrame.
    """
    all_errors = []
    for rule_fn in VALIDATION_RULES:
        rule_errors = rule_fn(df, ref)
        all_errors.extend(rule_errors)
        if rule_errors:
            print(f"  [{rule_fn.__name__}] found {len(rule_errors)} violation(s)")
    
    if not all_errors:
        print("✓ No validation errors found.")
        return pd.DataFrame(columns=[
            "row_index", "column", "rule", "bad_value",
            "severity", "corrective_action", "corrected_value"
        ])
    
    error_log = pd.DataFrame(all_errors)
    error_log = error_log.sort_values(["row_index", "column"]).reset_index(drop=True)
    
    print(f"\nValidation complete: {len(error_log)} total violations")
    print(error_log.groupby(["severity", "rule"])["row_index"].count()
          .rename("count")
          .to_string())
    
    return error_log

Running this produces output like:

Loaded 200 rows from sales_raw.csv
  [check_missing_customer_id] found 1 violation(s)
  [check_transaction_id_unique] found 1 violation(s)
  [check_sale_date_range] found 2 violation(s)
  [check_quantity] found 2 violation(s)
  [check_unit_price] found 2 violation(s)
  [check_discount_pct] found 1 violation(s)
  [check_product_sku_format] found 1 violation(s)
  [check_sales_rep_id] found 1 violation(s)
  [check_region] found 1 violation(s)

Validation complete: 12 total violations
severity  rule
error     allowed_values         1
          date_range             2
          id_format              0
          price_range            2
          quantity_range         2
          referential_integrity  1
          required_field         1
          sku_format             1
          unique_id              1
warning   discount_range         1

---

Stage 4: Applying Corrections

Now we apply corrections to a copy of the DataFrame (never mutate the original in place — you want to be able to compare before and after):

def apply_corrections(df: pd.DataFrame, error_log: pd.DataFrame) -> pd.DataFrame:
    """
    Apply auto-corrections defined in the error log.
    Returns a corrected copy of df.
    """
    corrected = df.copy()
    
    # Add a column to track correction status
    corrected["_validation_status"] = "ok"
    corrected["_correction_notes"] = ""
    
    for _, err in error_log.iterrows():
        idx = err["row_index"]
        col = err["column"]
        action = err["corrective_action"]
        corrected_val = err["corrected_value"]
        rule = err["rule"]
        
        if action == "set_to_null":
            corrected.at[idx, col] = np.nan
            corrected.at[idx, "_validation_status"] = "corrected"
            note = corrected.at[idx, "_correction_notes"]
            corrected.at[idx, "_correction_notes"] = (
                f"{note}; {col}={err['bad_value']} set to null [{rule}]"
            ).lstrip("; ")
        
        elif action == "cap_at_max":
            corrected.at[idx, col] = corrected_val
            corrected.at[idx, "_validation_status"] = "corrected"
            note = corrected.at[idx, "_correction_notes"]
            corrected.at[idx, "_correction_notes"] = (
                f"{note}; {col}={err['bad_value']} capped at {corrected_val} [{rule}]"
            ).lstrip("; ")
        
        elif action == "flag_for_manual_review":
            if corrected.at[idx, "_validation_status"] != "corrected":
                corrected.at[idx, "_validation_status"] = "needs_review"
            note = corrected.at[idx, "_correction_notes"]
            corrected.at[idx, "_correction_notes"] = (
                f"{note}; {col}={err['bad_value']} flagged [{rule}]"
            ).lstrip("; ")
    
    # Summary
    status_counts = corrected["_validation_status"].value_counts()
    print("\nCorrection summary:")
    print(status_counts.to_string())
    
    return corrected

This gives every row a _validation_status of "ok", "corrected", or "needs_review", plus a human-readable _correction_notes string that documents exactly what happened. When an ops analyst opens the output file, they can filter on _validation_status = "needs_review" and see the ten rows that need a human decision.

---

Stage 5: Exporting the Corrected File and Error Report

The final stage writes two artifacts:

1. sales_corrected.csv — the full corrected DataFrame (all 200 rows, with _validation_status and _correction_notes columns added)
2. sales_validation_report.xlsx — a multi-sheet Excel workbook with the error log, a summary by rule, and the rows needing manual review

from datetime import datetime

def export_outputs(
    corrected_df: pd.DataFrame,
    error_log: pd.DataFrame,
    corrected_path: str = "sales_corrected.csv",
    report_path: str = "sales_validation_report.xlsx",
):
    """
    Write the corrected CSV and a structured validation report Excel file.
    """
    # 1. Write corrected CSV
    corrected_df.to_csv(corrected_path, index=False)
    print(f"✓ Corrected file written: {corrected_path} ({len(corrected_df):,} rows)")
    
    # 2. Prepare report sheets
    run_timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    
    # Summary sheet
    summary_data = {
        "Run timestamp": [run_timestamp],
        "Total rows processed": [len(corrected_df)],
        "Rows OK": [(corrected_df["_validation_status"] == "ok").sum()],
        "Rows auto-corrected": [(corrected_df["_validation_status"] == "corrected").sum()],
        "Rows needing review": [(corrected_df["_validation_status"] == "needs_review").sum()],
        "Total violations": [len(error_log)],
        "Errors": [(error_log["severity"] == "error").sum()],
        "Warnings": [(error_log["severity"] == "warning").sum()],
    }
    summary_df = pd.DataFrame(summary_data).T.rename(columns={0: "Value"})
    
    # Rule breakdown
    if not error_log.empty:
        rule_breakdown = (
            error_log.groupby(["rule", "severity", "corrective_action"])["row_index"]
            .count()
            .rename("violation_count")
            .reset_index()
            .sort_values("violation_count", ascending=False)
        )
    else:
        rule_breakdown = pd.DataFrame(
            columns=["rule", "severity", "corrective_action", "violation_count"]
        )
    
    # Rows needing manual review
    needs_review = corrected_df[
        corrected_df["_validation_status"] == "needs_review"
    ].copy()
    
    # Write to Excel
    with pd.ExcelWriter(report_path, engine="openpyxl") as writer:
        summary_df.to_excel(writer, sheet_name="Summary", index=True)
        rule_breakdown.to_excel(writer, sheet_name="Rule Breakdown", index=False)
        if not error_log.empty:
            error_log.to_excel(writer, sheet_name="All Violations", index=False)
        if not needs_review.empty:
            needs_review.to_excel(writer, sheet_name="Needs Manual Review", index=False)
    
    print(f"✓ Validation report written: {report_path}")
    print(f"  Sheets: Summary, Rule Breakdown, All Violations, Needs Manual Review")

For more control over Excel formatting — adding colors, column widths, header styles — see Automating Excel Reports with pandas and openpyxl: Formatted Workbooks Without Manual Work.

---

Assembling the Full Pipeline

Now we stitch everything together into one callable function that you can invoke from a script or scheduler:

def run_cleaning_pipeline(
    input_path: str,
    corrected_path: str = "sales_corrected.csv",
    report_path: str = "sales_validation_report.xlsx",
    ref: dict = None,
) -> tuple[pd.DataFrame, pd.DataFrame]:
    """
    Full data cleaning pipeline:
      1. Load raw data
      2. Run validation rules
      3. Apply corrections
      4. Export corrected file and error report
    
    Returns: (corrected_df, error_log)
    """
    if ref is None:
        ref = REFERENCE_DATA
    
    print("=" * 60)
    print(f"Starting validation pipeline: {input_path}")
    print("=" * 60)
    
    # Stage 1: Load
    raw_df = load_raw(input_path)
    
    # Stage 2: Validate
    print("\n--- Validation ---")
    error_log = run_validation(raw_df, ref)
    
    # Stage 3: Correct
    print("\n--- Applying Corrections ---")
    corrected_df = apply_corrections(raw_df, error_log)
    
    # Stage 4: Export
    print("\n--- Exporting Outputs ---")
    export_outputs(corrected_df, error_log, corrected_path, report_path)
    
    print("\n✓ Pipeline complete.")
    print("=" * 60)
    return corrected_df, error_log


# Run it
if __name__ == "__main__":
    corrected, log = run_cleaning_pipeline("sales_raw.csv")

---

Making Rules Configurable Without Rewriting Code

Hardcoding max_quantity = 500 directly in rule logic is a smell. What happens when the business changes the threshold? You want the reference data — thresholds, allowed value sets, reference tables — to be separable from the rule logic.

The ref dictionary we've been passing around is already most of the way there. But you can take it further by loading reference data from a config file or a database:

import json

def load_ref_from_config(config_path: str) -> dict:
    """Load reference data from a JSON config file."""
    with open(config_path) as f:
        config = json.load(f)
    
    return {
        "valid_rep_ids": set(config["valid_rep_ids"]),
        "valid_regions": set(config["valid_regions"]),
        "min_sale_date": pd.Timestamp(config["min_sale_date"]),
        "max_sale_date": pd.Timestamp(date.today()),
        "max_quantity": config["max_quantity"],
        "max_unit_price": config["max_unit_price"],
        "max_discount_pct": config["max_discount_pct"],
    }

With a validation_config.json like:

{
    "valid_rep_ids": ["REP-001", "REP-002", "REP-003", "REP-004", "REP-005",
                      "REP-006", "REP-007", "REP-008", "REP-009", "REP-010"],
    "valid_regions": ["North", "South", "East", "West"],
    "min_sale_date": "2020-01-01",
    "max_quantity": 500,
    "max_unit_price": 10000.0,
    "max_discount_pct": 0.5
}

Now your pipeline can also load reference sales_rep_id values from a live database — for instance, querying a sales_reps table — rather than a static list. If you're connecting to SQL, see Reading from SQL Databases into pandas with SQLAlchemy for the pattern.

---

Advanced: Cross-Column and Multi-Row Rules

The rules we've built so far check a single column in isolation. Real data quality problems are often relational: a discount of 0.45 is fine on its own, but if quantity is also null, the whole row is effectively useless. And sometimes the problem is at the group level — for example, a customer with 50 transactions in a single day is suspicious even if each row individually passes all checks.

Cross-column rule example

def check_revenue_sanity(df, ref):
    """
    Flag rows where quantity * unit_price suggests an implausibly large transaction.
    Revenue over $250,000 per line item triggers a warning.
    """
    max_revenue = 250_000
    # Only evaluate rows where both values are present
    calculable = df["quantity"].notna() & df["unit_price"].notna()
    revenue = df.loc[calculable, "quantity"] * df.loc[calculable, "unit_price"]
    high_revenue_mask = revenue > max_revenue
    high_revenue_indices = revenue[high_revenue_mask].index
    
    errors = []
    for idx in high_revenue_indices:
        rev = df.at[idx, "quantity"] * df.at[idx, "unit_price"]
        errors.append({
            "row_index": idx,
            "column": "quantity,unit_price",
            "rule": "revenue_sanity",
            "bad_value": f"qty={df.at[idx,'quantity']}, price={df.at[idx,'unit_price']}, revenue={rev:,.0f}",
            "severity": "warning",
            "corrective_action": "flag_for_manual_review",
            "corrected_value": None,
        })
    return errors

Group-level rule example

def check_customer_daily_transaction_volume(df, ref):
    """
    Flag customers with more than 10 transactions on a single day — likely a data entry issue.
    """
    MAX_DAILY_TXN = 10
    daily_counts = (
        df.groupby(["customer_id", "sale_date"])["transaction_id"]
        .count()
        .reset_index()
        .rename(columns={"transaction_id": "daily_count"})
    )
    high_volume = daily_counts[daily_counts["daily_count"] > MAX_DAILY_TXN]
    
    errors = []
    if high_volume.empty:
        return errors
    
    for _, group_row in high_volume.iterrows():
        cid = group_row["customer_id"]
        dt = group_row["sale_date"]
        # Find all row indices for this customer+date
        matching = df[(df["customer_id"] == cid) & (df["sale_date"] == dt)].index
        for idx in matching:
            errors.append({
                "row_index": idx,
                "column": "customer_id,sale_date",
                "rule": "daily_transaction_volume",
                "bad_value": f"customer {cid} has {group_row['daily_count']} txns on {dt}",
                "severity": "warning",
                "corrective_action": "flag_for_manual_review",
                "corrected_value": None,
            })
    return errors

Notice that the column field now contains "customer_id,sale_date" — a multi-column reference. This is a deliberate choice to keep the error log schema consistent (one column field) while still being descriptive. Some teams prefer to create one error record per column involved; either approach works as long as you're consistent.

---

Making the Pipeline Schedulable

A one-time notebook is a liability. The real value of this pipeline comes when it runs automatically every month against the new export, writes the corrected file to a shared drive, and emails the validation report to the ops team. For a full treatment of scheduling patterns, see Building and Automating Recurring Reports with pandas: Scheduling Scripts to Run Without You.

For now, add command-line argument support to your script so it can be invoked from cron or Task Scheduler:

import argparse
import sys

def main():
    parser = argparse.ArgumentParser(
        description="Validate and clean a sales transaction CSV."
    )
    parser.add_argument("input", help="Path to raw input CSV")
    parser.add_argument(
        "--output", default="sales_corrected.csv", help="Path for corrected CSV"
    )
    parser.add_argument(
        "--report", default="sales_validation_report.xlsx", help="Path for Excel report"
    )
    parser.add_argument(
        "--config", default="validation_config.json", help="Path to reference config JSON"
    )
    args = parser.parse_args()
    
    try:
        ref = load_ref_from_config(args.config)
    except FileNotFoundError:
        print(f"Config file not found: {args.config}. Using defaults.")
        ref = REFERENCE_DATA
    
    corrected, log = run_cleaning_pipeline(
        input_path=args.input,
        corrected_path=args.output,
        report_path=args.report,
        ref=ref,
    )
    
    error_count = (log["severity"] == "error").sum() if not log.empty else 0
    sys.exit(1 if error_count > 0 else 0)

if __name__ == "__main__":
    main()

The sys.exit(1) on errors is important for CI/CD and cron integration. If the pipeline exits with code 1, any orchestration tool (cron, Airflow, GitHub Actions) can detect failure automatically. Exit code 0 means clean data, exit code 1 means errors found. Warnings alone don't fail the pipeline.

---

Hands-On Exercise

Now it's your turn. Using the framework we've built, extend the pipeline with the following additions:

Exercise 1 — New validation rule: Write a rule called check_duplicate_customer_on_same_sku that flags cases where the same customer_id appears more than 3 times with the same product_sku. This could indicate a batch import error. The rule should return "warning" severity with "flag_for_manual_review" action.

Exercise 2 — Smarter correction: The current check_region rule sets invalid regions to null. Improve it: if the bad region value starts with a valid region name (e.g., "Northwest" starts with "North"), attempt to correct it to the matching valid region. If no match is found, fall back to null. Update both the rule and the apply_corrections function to handle a new action type called "fuzzy_correct".

Exercise 3 — Tiered exit codes: Currently sys.exit(1) fires for any error. Modify main() to return exit code 2 if there are more than 10 error-severity violations (suggesting catastrophic data quality), exit code 1 for 1–10 errors, and exit code 0 only when there are zero errors (warnings are OK).

Exercise 4 — Historical log: Modify export_outputs to append the error log to a persistent validation_history.csv file rather than overwriting it each run. Add a run_timestamp and input_filename column so you can track data quality trends over time. Then use pandas groupby and aggregation to analyze which rules fire most often across runs — a quality trend report.

---

Common Mistakes & Troubleshooting

"My rule fires on rows that already have NaN from a previous correction"

When you apply corrections that set values to NaN, and then a later rule also checks that column, you might accumulate spurious errors. The fix: run all validation checks against the original raw DataFrame, not the in-progress corrected one. Our architecture already handles this correctly — run_validation runs against raw_df, and apply_corrections runs afterward. If you're tempted to run validation in multiple passes, resist it unless you explicitly intend to validate the corrections themselves.

"iterrows() over my full DataFrame is painfully slow"

If you're iterating over the full DataFrame inside a rule function (rather than over the filtered error subset), stop. Use boolean masks to identify failing rows, then iterate only over those indices. With 100,000 rows and 2% error rate, you want 2,000 iterations, not 100,000. For very high-cardinality rule checks like regex matching, consider vectorized string methods — Text Cleanup at Scale with pandas String Methods and Regular Expressions covers the performance-optimal patterns.

"The error log has duplicate entries for the same row"

This can happen if a row fails multiple rules that both touch the same column. This is expected behavior — one row can generate multiple error records, and that's fine. The _correction_notes column accumulates all of them. If you want to deduplicate at the row level, use error_log.groupby("row_index")["rule"].apply(list) to see all rules that fired per row.

"My date rule isn't catching NaN values"

Pandas date comparison with NaN returns False, not True. So df["sale_date"] < ref["min_sale_date"] will be False for null dates — they slip through. Always add an explicit df["sale_date"].isna() check in date rules and decide separately how to handle missing dates.

# Correct pattern: check for NaN first
missing_dates = df["sale_date"].isna()
too_old = df["sale_date"].notna() & (df["sale_date"] < ref["min_sale_date"])
in_future = df["sale_date"].notna() & (df["sale_date"] > ref["max_sale_date"])

"The corrected CSV has the wrong dtypes"

When you write to CSV and read back, all those carefully preserved numeric types get inferred again on reload. If the corrected file is consumed by another pandas script, either write a companion schema file or use Parquet format (corrected_df.to_parquet("sales_corrected.parquet")) which preserves dtype metadata exactly. See Exporting and Sharing Analysis Results: Writing CSV, Excel, and JSON Files from pandas for a full comparison of output formats.

"A rule that was passing last month is failing now — but the data looks the same"

Check your reference data. If valid_rep_ids is loaded from a database and a rep was deleted, previously-valid rep IDs will now fail the referential integrity check. This is correct behavior — but it can be surprising. Build a "reference data change log" by persisting the reference data used each run alongside the error log. When a new failure pattern appears, you can diagnose whether it's data quality or reference data drift.

---

Performance Considerations at Scale

For files up to about 500,000 rows, the pattern in this article works fine. Beyond that, a few changes become necessary:

Chunked reading: If the raw file is too large to hold in memory, read it in chunks and accumulate error records. The tricky part is rules like check_transaction_id_unique that need to see the full file. For those, consider a two-pass approach: first pass collects IDs into a set, second pass runs remaining rules in chunks. See Handling Large Datasets in Python: Chunked Reading, Efficient dtypes, and When to Use Polars for the full pattern.

Vectorized rules: Replace the inner for idx in df[mask].index: loop with a vectorized construction using df[mask].apply() or numpy operations. For most rules, this is a 10–100x speedup. See Writing Fast pandas Code: Vectorization Instead of apply and Loops for the specifics.

Parallel rule execution: Rules are embarrassingly parallel since each reads from the same immutable DataFrame. Use concurrent.futures.ThreadPoolExecutor (or ProcessPoolExecutor for GIL-bound work) to run rules concurrently and merge the results.

---

Summary & Next Steps

You've built a complete, production-oriented data validation pipeline. Let's review what you now have:

• Rule functions that return structured error records — not print statements, not exceptions, but data
• An error log DataFrame that you can aggregate, filter, and export with the same pandas skills you already have
• A correction engine that applies traceable, documented fixes and flags rows that need human review
• An export layer that writes both a corrected CSV and a multi-sheet Excel validation report
• A CLI wrapper that makes the pipeline schedulable and integration-friendly
• A configurable reference data layer that separates business rules from thresholds

The architecture decisions here — validation as data, corrections as documented transformations, severity levels, error accumulation rather than fail-fast — are patterns you'll use regardless of the domain. The same structure applies to financial reconciliation, HR data validation, supply chain feeds, or any situation where data quality is a recurring operational concern.

To deepen this further, consider these directions:

• Schema-driven rules: Define your validation rules in a YAML or JSON schema and generate the rule functions programmatically. This makes adding new columns trivial.
• Statistical outlier detection: Add rules based on rolling means and standard deviations to catch values that are individually valid but statistically anomalous — a different category of data quality problem than structural violations.
• Great Expectations or Pandera integration: Libraries like Pandera let you define a DataFrame schema as a class with type annotations and constraint decorators, then validate against it. For complex pipelines, they're worth the dependency.
• Audit trail persistence: Store the error log to a database table (rather than overwriting an Excel file) so you can run trend analysis on data quality over time — which rules are improving, which are getting worse, and whether a particular data source has systemic issues.

For structuring this pipeline into a maintainable, importable module rather than a script, see Structuring a Reusable Data Analysis Project: Functions, Modules, Notebooks, and Scripts.