Table of Contents

    Implement tested preprocessing utilities and visualize feature distributions

    Rumman Ansari November 07, 2026 1 views Subject Details
    GUIDED IMPLEMENTATION • 60 MIN

    Implement Tested Preprocessing Utilities and Visualize Feature Distributions

    Build reusable Python preprocessing utilities, verify them with pytest, and create diagnostic plots that reveal missingness, skew, outliers and train-versus-held-out distribution differences.

    Learning objective: By the end of this guided implementation, you will have runnable and documented Python code for deterministic tabular-data cleaning, schema and finiteness validation, train-only preprocessing, automated tests, and feature-distribution visualization.

    1. What You Will Build

    This implementation separates exploratory work from reusable code. The preprocessing utilities will live in a Python module, automated tests will verify their behavior, and a runnable script will load a small CSV, create a held-out dataset, fit preprocessing on training data only, and save diagnostic plots.

    IMPLEMENTATION FLOW
    Load → Validate → Clean → Split → Visualize → Fit train only → Test
    FilePurpose
    preprocessing_utils.pyReusable cleaning, validation, preprocessing and plotting utilities
    test_preprocessing_utils.pyAutomated pytest tests
    run_preprocessing.pyEnd-to-end runnable workflow
    customer_features.csvSmall practice dataset
    artifacts/Saved plots, reports and fitted preprocessor

    2. Prerequisites

    Required Setup

    • Python 3.9 or later is recommended.
    • Basic knowledge of pandas DataFrames and NumPy arrays.
    • Basic understanding of train and held-out datasets.
    • Use pytest for automated tests.
    • Use Matplotlib for feature-distribution plots.
    python -m pip install numpy pandas matplotlib scikit-learn pytest joblib

    3. Recommended Project Structure

    scientific_preprocessing/
├── customer_features.csv
├── preprocessing_utils.py
├── run_preprocessing.py
├── test_preprocessing_utils.py
└── artifacts/

    Keeping reusable functions outside a notebook makes them easier to test, import and reuse. A notebook may call these utilities for exploration, but the core preprocessing logic should not depend on hidden notebook state.

    4. Create the Practice Dataset

    Run the following code once to create a small CSV containing missing numerical values, inconsistent categories, an invalid numeric token and an invalid date.

    from pathlib import Path
import pandas as pd


data = pd.DataFrame({
    "customer_id": list(range(1001, 1025)),
    "age": [
        22, 25, 29, None, 35, 41, 47, 52,
        31, 28, 38, 44, 50, 26, None, 33,
        36, 42, 55, 30, 27, 49, 40, 34
    ],
    "monthly_income": [
        28000, 32000, 39000, 46000, None, 62000, 71000, 83000,
        48000, 37000, 56000, "unknown", 76000, 35000, 43000, 51000,
        59000, 65000, 90000, 45000, 34000, 74000, 61000, 53000
    ],
    "city": [
        "Kolkata", "Delhi", "Mumbai", " kolkata ", "Pune", "Delhi",
        "Mumbai", "Kolkata", None, "Pune", "Delhi", "Mumbai",
        "Kolkata", "Pune", "Delhi", "Mumbai", "Kolkata", "Delhi",
        "Pune", "Mumbai", "Kolkata", "Delhi", "Pune", "Bengaluru"
    ],
    "segment": [
        "Standard", "Standard", "Premium", "Standard", None, "Premium",
        "Premium", "Premium", "Standard", "Standard", "Premium", "Premium",
        "Premium", "Standard", "Standard", "Premium", "Standard", "Premium",
        "Premium", "Standard", "Standard", "Premium", "Premium", "Standard"
    ],
    "signup_date": [
        "2025-01-05", "2025-01-18", "2025-02-03", "2025-02-20",
        "2025-03-09", "2025-03-22", "invalid", "2025-04-15",
        "2025-05-01", "2025-05-19", "2025-06-02", "2025-06-18",
        "2025-07-04", None, "2025-08-01", "2025-08-17",
        "2025-09-05", "2025-09-21", "2025-10-03", "2025-10-19",
        "2025-11-02", "2025-11-16", "2025-12-01", "2025-12-18"
    ],
    "purchased": [
        0, 0, 1, 0, 0, 1, 1, 1,
        0, 0, 1, 1, 1, 0, 0, 1,
        0, 1, 1, 0, 0, 1, 1, 0
    ]
})

path = Path("customer_features.csv")
data.to_csv(path, index=False)

print("Created:", path.resolve())
print("Shape:", data.shape)
    RAW SHAPE
    \[ (24,7) \]

    5. Implement the Utility Module

    Create preprocessing_utils.py. The module contains deterministic cleaning, schema validation, missingness reporting, feature engineering, train-only preprocessing and plot-generation utilities.

    from __future__ import annotations

from pathlib import Path
from typing import Iterable

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler


MISSING_TOKENS = {
    "": np.nan,
    " ": np.nan,
    "na": np.nan,
    "n/a": np.nan,
    "none": np.nan,
    "null": np.nan,
    "unknown": np.nan,
    "?": np.nan,
}


def load_csv(path: str | Path) -> pd.DataFrame:
    """Load a non-empty CSV with unique column names."""
    csv_path = Path(path)

    if not csv_path.exists():
        raise FileNotFoundError(f"CSV not found: {csv_path.resolve()}")

    if csv_path.suffix.lower() != ".csv":
        raise ValueError("Expected a .csv file")

    df = pd.read_csv(csv_path)

    if df.empty:
        raise ValueError("CSV contains no data rows")

    if not df.columns.is_unique:
        raise ValueError("CSV contains duplicate column names")

    return df


def normalize_column_names(columns: Iterable[str]) -> list[str]:
    """Return normalized snake_case-like column names."""
    series = pd.Series(list(columns), dtype="string")
    normalized = (
        series.str.strip()
        .str.lower()
        .str.replace(r"[^a-z0-9]+", "_", regex=True)
        .str.strip("_")
    )
    return normalized.tolist()


def clean_structure(df: pd.DataFrame) -> pd.DataFrame:
    """Apply deterministic cleaning without fitting dataset statistics."""
    cleaned = df.copy()
    cleaned.columns = normalize_column_names(cleaned.columns)

    # Normalize known missing tokens without changing legitimate zero values.
    cleaned = cleaned.replace(MISSING_TOKENS)

    for column in ["age", "monthly_income"]:
        if column in cleaned.columns:
            cleaned[column] = pd.to_numeric(cleaned[column], errors="coerce")

    for column in ["city", "segment"]:
        if column in cleaned.columns:
            cleaned[column] = (
                cleaned[column]
                .astype("string")
                .str.strip()
                .str.title()
            )

    if "signup_date" in cleaned.columns:
        cleaned["signup_date"] = pd.to_datetime(
            cleaned["signup_date"],
            errors="coerce"
        )

    return cleaned


def validate_schema(
    df: pd.DataFrame,
    required_columns: Iterable[str],
    target_column: str | None = None
) -> None:
    """Validate columns, duplicate names and optional target completeness."""
    required = set(required_columns)
    missing = required.difference(df.columns)

    if missing:
        raise ValueError(f"Missing required columns: {sorted(missing)}")

    if not df.columns.is_unique:
        raise ValueError("Duplicate column names detected")

    if target_column is not None and df[target_column].isna().any():
        raise ValueError(f"Target contains missing values: {target_column}")


def validate_finite(df: pd.DataFrame, columns: Iterable[str]) -> None:
    """Raise an error when selected numerical columns contain infinity."""
    selected = df[list(columns)].to_numpy(dtype=float)

    # NaN may be intentional before imputation, but infinity is never accepted.
    if np.isinf(selected).any():
        bad_positions = np.argwhere(np.isinf(selected))
        raise ValueError(
            "Infinite values detected at positions "
            f"{bad_positions[:5].tolist()}"
        )


def missing_value_report(df: pd.DataFrame) -> pd.DataFrame:
    """Return dtype, missing count and missing percentage per column."""
    report = pd.DataFrame({
        "dtype": df.dtypes.astype(str),
        "missing_count": df.isna().sum(),
        "missing_percent": df.isna().mean().mul(100).round(2),
    })

    return report.sort_values(
        ["missing_percent", "missing_count"],
        ascending=False
    )


def build_model_features(df: pd.DataFrame) -> pd.DataFrame:
    """Create deterministic model features from cleaned source fields."""
    features = df.copy()
    features = features.drop(columns=["customer_id"], errors="ignore")

    if "signup_date" in features.columns:
        features["signup_year"] = features["signup_date"].dt.year
        features["signup_month"] = features["signup_date"].dt.month
        features["signup_day_of_week"] = features["signup_date"].dt.dayofweek
        features = features.drop(columns=["signup_date"])

    return features


def infer_feature_groups(
    df: pd.DataFrame
) -> tuple[list[str], list[str]]:
    """Infer numerical and categorical feature names from training data."""
    numeric = df.select_dtypes(include=["number"]).columns.tolist()
    categorical = df.select_dtypes(
        include=["object", "string", "category", "bool"]
    ).columns.tolist()

    unsupported = [
        column for column in df.columns
        if column not in numeric and column not in categorical
    ]

    if unsupported:
        raise TypeError(f"Unsupported feature dtypes: {unsupported}")

    if not numeric and not categorical:
        raise ValueError("No supported model features found")

    return numeric, categorical


def build_preprocessor(
    numeric_features: list[str],
    categorical_features: list[str]
) -> ColumnTransformer:
    """Create an unfitted preprocessing graph."""
    numeric_pipeline = Pipeline(steps=[
        ("imputer", SimpleImputer(strategy="median", add_indicator=True)),
        ("scaler", StandardScaler()),
    ])

    categorical_pipeline = Pipeline(steps=[
        ("imputer", SimpleImputer(strategy="most_frequent")),
        (
            "encoder",
            OneHotEncoder(handle_unknown="ignore", sparse_output=False),
        ),
    ])

    return ColumnTransformer(
        transformers=[
            ("numeric", numeric_pipeline, numeric_features),
            ("categorical", categorical_pipeline, categorical_features),
        ],
        remainder="drop",
        verbose_feature_names_out=False,
    )


def transform_to_dataframe(
    preprocessor: ColumnTransformer,
    df: pd.DataFrame,
    *,
    fit: bool = False
) -> pd.DataFrame:
    """Fit optionally, transform data and return a labeled DataFrame."""
    array = (
        preprocessor.fit_transform(df)
        if fit
        else preprocessor.transform(df)
    )

    names = preprocessor.get_feature_names_out()
    return pd.DataFrame(array, columns=names, index=df.index)


def save_numeric_distributions(
    df: pd.DataFrame,
    output_path: str | Path,
    columns: Iterable[str] | None = None,
    bins: int = 15,
) -> Path:
    """Save a grid of numerical histograms."""
    output_path = Path(output_path)
    output_path.parent.mkdir(parents=True, exist_ok=True)

    if columns is None:
        selected = df.select_dtypes(include=["number"]).columns.tolist()
    else:
        selected = list(columns)

    if not selected:
        raise ValueError("No numerical columns supplied for plotting")

    n_columns = 2
    n_rows = int(np.ceil(len(selected) / n_columns))

    fig, axes = plt.subplots(
        n_rows,
        n_columns,
        figsize=(12, 4 * n_rows),
        squeeze=False,
    )

    for axis, column in zip(axes.ravel(), selected):
        values = df[column].dropna()
        axis.hist(values, bins=bins)
        axis.set_title(f"Distribution of {column}")
        axis.set_xlabel(column)
        axis.set_ylabel("Frequency")
        axis.grid(True)

    for axis in axes.ravel()[len(selected):]:
        axis.set_visible(False)

    fig.tight_layout()
    fig.savefig(output_path, dpi=150, bbox_inches="tight")
    plt.close(fig)
    return output_path


def save_categorical_distributions(
    df: pd.DataFrame,
    output_path: str | Path,
    columns: Iterable[str] | None = None,
) -> Path:
    """Save category-frequency bar charts, including missing values."""
    output_path = Path(output_path)
    output_path.parent.mkdir(parents=True, exist_ok=True)

    if columns is None:
        selected = df.select_dtypes(
            include=["object", "string", "category", "bool"]
        ).columns.tolist()
    else:
        selected = list(columns)

    if not selected:
        raise ValueError("No categorical columns supplied for plotting")

    fig, axes = plt.subplots(
        len(selected),
        1,
        figsize=(12, 4 * len(selected)),
        squeeze=False,
    )

    for axis, column in zip(axes.ravel(), selected):
        counts = df[column].fillna("Missing").value_counts(dropna=False)
        axis.bar(counts.index.astype(str), counts.values)
        axis.set_title(f"Category counts for {column}")
        axis.set_xlabel(column)
        axis.set_ylabel("Count")
        axis.tick_params(axis="x", rotation=30)
        axis.grid(True, axis="y")

    fig.tight_layout()
    fig.savefig(output_path, dpi=150, bbox_inches="tight")
    plt.close(fig)
    return output_path


def save_train_holdout_histograms(
    train_df: pd.DataFrame,
    holdout_df: pd.DataFrame,
    columns: Iterable[str],
    output_path: str | Path,
    bins: int = 12,
) -> Path:
    """Save overlaid histograms for train-versus-held-out diagnosis."""
    selected = list(columns)
    if not selected:
        raise ValueError("At least one comparison column is required")

    output_path = Path(output_path)
    output_path.parent.mkdir(parents=True, exist_ok=True)

    fig, axes = plt.subplots(
        len(selected),
        1,
        figsize=(12, 4 * len(selected)),
        squeeze=False,
    )

    for axis, column in zip(axes.ravel(), selected):
        axis.hist(
            train_df[column].dropna(),
            bins=bins,
            alpha=0.65,
            label="Training",
            density=True,
        )
        axis.hist(
            holdout_df[column].dropna(),
            bins=bins,
            alpha=0.65,
            label="Held-out",
            density=True,
        )
        axis.set_title(f"Training vs held-out: {column}")
        axis.set_xlabel(column)
        axis.set_ylabel("Density")
        axis.legend()
        axis.grid(True)

    fig.tight_layout()
    fig.savefig(output_path, dpi=150, bbox_inches="tight")
    plt.close(fig)
    return output_path

    6. Why the Utilities Are Separated

    Utility typeExampleDesign reason
    Deterministic transformationTrim city names or parse datesDoes not learn population statistics
    ValidationCheck schema or infinityFails early with a useful diagnostic
    Learned preprocessingMedian imputation and scalingMust be fitted on training data only
    VisualizationHistogram or category count plotProvides diagnostic evidence without altering data

    7. Missing-Value and Finiteness Policies

    MISSING VALUES
    Numeric features use the training median. Categorical features use the most frequent training category. Numeric missing indicators preserve evidence that values were absent.
    NON-FINITE VALUES
    NaN may represent planned missingness before imputation. Positive and negative infinity are rejected because they usually indicate invalid input or numerical failure.

    8. Implement the Automated Tests

    Create test_preprocessing_utils.py. These tests verify normal behavior, shape preservation, error handling, train-only transformations and plot creation.

    from pathlib import Path

import numpy as np
import pandas as pd
import pytest

from preprocessing_utils import (
    build_model_features,
    build_preprocessor,
    clean_structure,
    infer_feature_groups,
    missing_value_report,
    normalize_column_names,
    save_categorical_distributions,
    save_numeric_distributions,
    transform_to_dataframe,
    validate_finite,
    validate_schema,
)


@pytest.fixture
def raw_frame():
    return pd.DataFrame({
        " Customer ID ": [1, 2, 3, 4],
        "Age": [20, None, 40, 30],
        "Monthly Income": [30000, "unknown", 60000, 45000],
        "City": [" kolkata ", "DELHI", None, "Mumbai"],
        "Segment": ["Standard", "Premium", "Standard", None],
        "Signup Date": ["2025-01-01", "invalid", "2025-02-01", None],
        "Purchased": [0, 1, 1, 0],
    })


def test_normalize_column_names():
    result = normalize_column_names([" Customer ID ", "Monthly Income"])
    assert result == ["customer_id", "monthly_income"]


def test_clean_structure_preserves_shape(raw_frame):
    cleaned = clean_structure(raw_frame)
    assert cleaned.shape == raw_frame.shape


def test_clean_structure_converts_types(raw_frame):
    cleaned = clean_structure(raw_frame)

    assert pd.api.types.is_numeric_dtype(cleaned["monthly_income"])
    assert pd.api.types.is_datetime64_any_dtype(cleaned["signup_date"])
    assert pd.isna(cleaned.loc[1, "monthly_income"])
    assert pd.isna(cleaned.loc[1, "signup_date"])
    assert cleaned.loc[0, "city"] == "Kolkata"
    assert cleaned.loc[1, "city"] == "Delhi"


def test_validate_schema_rejects_missing_column(raw_frame):
    cleaned = clean_structure(raw_frame).drop(columns=["city"])

    with pytest.raises(ValueError, match="Missing required columns"):
        validate_schema(cleaned, required_columns=["city", "age"])


def test_validate_schema_rejects_missing_target(raw_frame):
    cleaned = clean_structure(raw_frame)
    cleaned.loc[0, "purchased"] = np.nan

    with pytest.raises(ValueError, match="Target contains missing"):
        validate_schema(
            cleaned,
            required_columns=["purchased"],
            target_column="purchased",
        )


def test_validate_finite_allows_nan_but_rejects_infinity():
    valid_missing = pd.DataFrame({"x": [1.0, np.nan, 3.0]})
    validate_finite(valid_missing, ["x"])

    invalid = pd.DataFrame({"x": [1.0, np.inf, 3.0]})
    with pytest.raises(ValueError, match="Infinite values"):
        validate_finite(invalid, ["x"])


def test_missing_report_contains_counts(raw_frame):
    cleaned = clean_structure(raw_frame)
    report = missing_value_report(cleaned)

    assert "missing_count" in report.columns
    assert "missing_percent" in report.columns
    assert report.loc["age", "missing_count"] == 1


def test_build_model_features_removes_identifier_and_date(raw_frame):
    cleaned = clean_structure(raw_frame)
    X = cleaned.drop(columns=["purchased"])
    features = build_model_features(X)

    assert "customer_id" not in features.columns
    assert "signup_date" not in features.columns
    assert "signup_year" in features.columns
    assert "signup_month" in features.columns
    assert "signup_day_of_week" in features.columns


def test_preprocessor_produces_matching_train_holdout_columns(raw_frame):
    cleaned = clean_structure(raw_frame)
    X = build_model_features(cleaned.drop(columns=["purchased"]))

    train = X.iloc[:3].copy()
    holdout = X.iloc[3:].copy()

    numeric, categorical = infer_feature_groups(train)
    preprocessor = build_preprocessor(numeric, categorical)

    train_result = transform_to_dataframe(preprocessor, train, fit=True)
    holdout_result = transform_to_dataframe(preprocessor, holdout, fit=False)

    assert list(train_result.columns) == list(holdout_result.columns)
    assert train_result.shape[0] == 3
    assert holdout_result.shape[0] == 1
    assert not train_result.isna().any().any()
    assert not holdout_result.isna().any().any()


def test_unknown_holdout_category_does_not_fail(raw_frame):
    cleaned = clean_structure(raw_frame)
    X = build_model_features(cleaned.drop(columns=["purchased"]))

    train = X.iloc[:3].copy()
    holdout = X.iloc[[3]].copy()
    holdout.loc[holdout.index[0], "city"] = "Bengaluru"

    numeric, categorical = infer_feature_groups(train)
    preprocessor = build_preprocessor(numeric, categorical)

    transform_to_dataframe(preprocessor, train, fit=True)
    result = transform_to_dataframe(preprocessor, holdout, fit=False)

    assert result.shape[0] == 1
    assert np.isfinite(result.to_numpy()).all()


def test_plot_functions_create_nonempty_files(raw_frame, tmp_path):
    cleaned = clean_structure(raw_frame)

    numeric_path = tmp_path / "numeric.png"
    categorical_path = tmp_path / "categorical.png"

    save_numeric_distributions(
        cleaned,
        numeric_path,
        columns=["age", "monthly_income"],
    )
    save_categorical_distributions(
        cleaned,
        categorical_path,
        columns=["city", "segment"],
    )

    assert numeric_path.exists()
    assert numeric_path.stat().st_size > 0
    assert categorical_path.exists()
    assert categorical_path.stat().st_size > 0

    9. Run the Tests

    pytest -q

    For more detail:

    pytest -v
    Testing principle: Verify values, shapes, dtypes, failure behavior and generated artifacts. A test that merely confirms a function executes may miss scientifically important defects.

    10. Implement the End-to-End Runner

    Create run_preprocessing.py. This script performs a stratified split, saves raw distribution plots, fits preprocessing only on training data, validates the outputs and saves the fitted artifact.

    from pathlib import Path
import logging

import joblib
import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split

from preprocessing_utils import (
    build_model_features,
    build_preprocessor,
    clean_structure,
    infer_feature_groups,
    load_csv,
    missing_value_report,
    save_categorical_distributions,
    save_numeric_distributions,
    save_train_holdout_histograms,
    transform_to_dataframe,
    validate_finite,
    validate_schema,
)


logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
)
logger = logging.getLogger(__name__)

CSV_PATH = Path("customer_features.csv")
ARTIFACT_DIR = Path("artifacts")
TARGET = "purchased"
SEED = 42


def main() -> None:
    ARTIFACT_DIR.mkdir(parents=True, exist_ok=True)

    raw_df = load_csv(CSV_PATH)
    clean_df = clean_structure(raw_df)

    required_columns = [
        "customer_id",
        "age",
        "monthly_income",
        "city",
        "segment",
        "signup_date",
        TARGET,
    ]
    validate_schema(clean_df, required_columns, target_column=TARGET)
    validate_finite(clean_df, ["age", "monthly_income"])

    logger.info("Raw shape: %s", raw_df.shape)
    logger.info("Clean shape: %s", clean_df.shape)

    report = missing_value_report(clean_df)
    report.to_csv(ARTIFACT_DIR / "missing_value_report.csv")
    logger.info("Saved missing-value report")

    X = clean_df.drop(columns=[TARGET])
    y = clean_df[TARGET].astype("int64")

    X_train, X_holdout, y_train, y_holdout = train_test_split(
        X,
        y,
        test_size=0.25,
        random_state=SEED,
        stratify=y,
    )

    logger.info("X_train shape: %s", X_train.shape)
    logger.info("X_holdout shape: %s", X_holdout.shape)

    # Save diagnostic plots before learned preprocessing.
    save_numeric_distributions(
        X_train,
        ARTIFACT_DIR / "training_numeric_distributions.png",
        columns=["age", "monthly_income"],
    )
    save_categorical_distributions(
        X_train,
        ARTIFACT_DIR / "training_categorical_distributions.png",
        columns=["city", "segment"],
    )
    save_train_holdout_histograms(
        X_train,
        X_holdout,
        columns=["age", "monthly_income"],
        output_path=ARTIFACT_DIR / "train_holdout_numeric_comparison.png",
    )

    X_train_features = build_model_features(X_train)
    X_holdout_features = build_model_features(X_holdout)

    numeric_features, categorical_features = infer_feature_groups(
        X_train_features
    )

    logger.info("Numeric features: %s", numeric_features)
    logger.info("Categorical features: %s", categorical_features)

    preprocessor = build_preprocessor(
        numeric_features,
        categorical_features,
    )

    # The leakage boundary: fit only on training data.
    train_processed = transform_to_dataframe(
        preprocessor,
        X_train_features,
        fit=True,
    )
    holdout_processed = transform_to_dataframe(
        preprocessor,
        X_holdout_features,
        fit=False,
    )

    assert train_processed.shape[0] == y_train.shape[0]
    assert holdout_processed.shape[0] == y_holdout.shape[0]
    assert list(train_processed.columns) == list(holdout_processed.columns)
    assert np.isfinite(train_processed.to_numpy()).all()
    assert np.isfinite(holdout_processed.to_numpy()).all()

    logger.info("Processed train shape: %s", train_processed.shape)
    logger.info("Processed holdout shape: %s", holdout_processed.shape)

    train_output = train_processed.copy()
    train_output[TARGET] = y_train.to_numpy()
    train_output.to_csv(
        ARTIFACT_DIR / "training_processed.csv",
        index=False,
    )

    holdout_output = holdout_processed.copy()
    holdout_output[TARGET] = y_holdout.to_numpy()
    holdout_output.to_csv(
        ARTIFACT_DIR / "held_out_processed.csv",
        index=False,
    )

    joblib.dump(
        preprocessor,
        ARTIFACT_DIR / "preprocessor.joblib",
    )

    shape_report = pd.DataFrame([
        {"stage": "raw", "shape": str(raw_df.shape)},
        {"stage": "clean", "shape": str(clean_df.shape)},
        {"stage": "X_train", "shape": str(X_train.shape)},
        {"stage": "X_holdout", "shape": str(X_holdout.shape)},
        {"stage": "train_features", "shape": str(X_train_features.shape)},
        {"stage": "holdout_features", "shape": str(X_holdout_features.shape)},
        {"stage": "train_processed", "shape": str(train_processed.shape)},
        {"stage": "holdout_processed", "shape": str(holdout_processed.shape)},
    ])
    shape_report.to_csv(ARTIFACT_DIR / "shape_report.csv", index=False)

    logger.info("Workflow completed successfully")


if __name__ == "__main__":
    main()

    11. Run the Workflow

    python run_preprocessing.py

    The artifacts directory should contain:

    artifacts/
├── held_out_processed.csv
├── missing_value_report.csv
├── preprocessor.joblib
├── shape_report.csv
├── train_holdout_numeric_comparison.png
├── training_categorical_distributions.png
├── training_numeric_distributions.png
└── training_processed.csv

    12. Understand the Shape Changes

    The raw dataset has 24 rows and 7 columns. Separating the target produces six source feature columns and one target vector:

    RAW SEPARATION
    \[ (24,7) \rightarrow X:(24,6) + y:(24,) \]

    A 25% held-out split produces 18 training rows and 6 held-out rows:

    SPLIT
    \[ X_{train}:(18,6),\qquad X_{holdout}:(6,6) \]

    Feature engineering removes the identifier and replaces one date with three calendar features:

    FEATURE ENGINEERING
    \[ (18,6) \rightarrow (18,7) \] \[ (6,6) \rightarrow (6,7) \]

    After one-hot encoding and missing indicators, the output has \(k\) columns:

    PROCESSED SHAPES
    \[ X_{train}^{processed}:(18,k) \] \[ X_{holdout}^{processed}:(6,k) \]

    The exact value of \(k\) depends on categories and missing indicators learned from the training subset. Both processed datasets must have the same \(k\) columns in the same order.

    13. Interpret Numerical Histograms

    Inspect Each Numerical Feature

    • Look for impossible values, such as negative age or income.
    • Check whether the distribution is strongly skewed.
    • Identify isolated extreme values that may require domain review.
    • Observe gaps or suspicious spikes caused by data-entry rules.
    • Compare training and held-out ranges for possible distribution shift.

    14. Interpret Categorical Bar Charts

    Inspect Category Counts

    • Check whether spelling and capitalization variations were normalized.
    • Identify rare categories with very few observations.
    • Confirm that missing values are visible as a separate diagnostic count.
    • Look for severe category imbalance.
    • Confirm whether held-out data contains categories absent from training.

    15. Why Visualize Before Scaling?

    Standard scaling changes location and scale. Plotting source values before scaling preserves interpretability in real-world units such as years or income. You may additionally visualize transformed features, but transformed plots do not replace source-data inspection.

    Before preprocessing

    • Inspect values in domain units.
    • Find missingness and outliers.
    • Review category quality.

    After preprocessing

    • Verify finite model inputs.
    • Confirm training and held-out schemas match.
    • Inspect transformed feature ranges.

    16. Why Train-Only Fitting Is Tested

    The preprocessor learns medians, scaling parameters and category vocabularies. These values must come only from training data.

    TRAINING STATISTICS
    \[ \tilde{x}_{train}=median\{x_i:i\in train\} \] \[ z=\frac{x-\mu_{train}}{\sigma_{train}} \]
    Correct transform_to_dataframe(preprocessor, X_train_features, fit=True)
    transform_to_dataframe(preprocessor, X_holdout_features, fit=False)
    Data leakage Calling fit=True for held-out data would recalculate preprocessing rules from evaluation records.

    17. Add a Test for a Previously Discovered Bug

    Whenever a preprocessing defect is found, create a regression test before or while fixing it. For example, if lowercase "unknown" was converted to missing but uppercase "UNKNOWN" was not, first normalize string tokens or expand the policy, then lock the expected behavior into a test.

    def test_unknown_income_token_becomes_missing():
    raw = pd.DataFrame({
        "monthly_income": ["UNKNOWN", "45000"]
    })

    # Adapt clean_structure if uppercase token support is required
    # by the documented input contract.
    cleaned = clean_structure(raw)

    assert pd.isna(cleaned.loc[0, "monthly_income"])
    assert cleaned.loc[1, "monthly_income"] == 45000
    Regression-testing principle: A fixed bug without a test can silently return. A focused test preserves the discovered requirement.

    18. Debug Plotting Failures

    SymptomDiagnostic action
    Blank histogramCheck whether the selected column is entirely missing after conversion.
    Categories overlap on the x-axisRotate labels, enlarge the figure or display the most frequent categories.
    One extreme value compresses the plotVerify the value, then consider a log scale or a separate outlier view.
    Training and held-out charts look very differentCheck split design, small sample size, time effects and population drift.
    Figure files are not createdCreate the output directory and test that saved files exist and are non-empty.

    19. Optional TensorBoard Image Logging

    If the workflow is part of a PyTorch experiment, a saved diagnostic plot can be recorded as an experiment artifact. Keep Matplotlib plot creation independent so it remains testable and useful outside the training framework.

    from pathlib import Path
from PIL import Image
import numpy as np
from torch.utils.tensorboard import SummaryWriter


writer = SummaryWriter("runs/preprocessing_debug")
image_path = Path("artifacts/training_numeric_distributions.png")

image = np.asarray(Image.open(image_path).convert("RGB"))
image = np.moveaxis(image, -1, 0)  # HWC to CHW

writer.add_image("Data/training_numeric_distributions", image, 0)
writer.close()

    20. Production Improvements

    Extend the Utilities Carefully

    • Define ranges for fields such as age, quantity and monetary values.
    • Add schema versioning and explicit expected dtypes.
    • Use grouped or chronological splitting when random splitting is inappropriate.
    • Persist the fitted preprocessor with the matching trained model.
    • Record code revision, dataset version, seed and package versions.
    • Monitor production missingness, unknown categories and distribution shift.
    • Use continuous integration to run pytest whenever utilities change.

    21. Guided Verification Checklist

    Confirm the Implementation

    • The CSV loads and the raw shape is recorded.
    • Column names and known text values are normalized deterministically.
    • Invalid numeric and date values become measurable missing values.
    • Infinity is rejected explicitly.
    • The target is separated before feature preprocessing.
    • The held-out dataset is created before fitting learned preprocessing.
    • The preprocessor is fitted only on training features.
    • The processed training and held-out schemas match.
    • Numerical and categorical distribution plots are saved.
    • Train-versus-held-out numerical plots are generated.
    • pytest verifies transformations, errors, shapes and plot artifacts.
    • The fitted preprocessor and evidence reports are saved.

    22. Practice Extensions

    Extend the Guided Implementation

    • Add an interquartile-range report without automatically deleting outliers.
    • Add a function that plots missing-value percentages by column.
    • Add tests for empty input, duplicate columns and unsupported dtypes.
    • Add a chronological split using signup_date.
    • Add a test proving that an unseen held-out city does not change the output schema.
    • Add model training through a single scikit-learn Pipeline.
    • Log distribution images and selected preprocessing statistics to TensorBoard.

    Key Takeaway

    Reliable preprocessing requires more than a sequence of DataFrame edits. Implement deterministic utilities, validate assumptions, fit learned transformations on training data only, test normal and failure cases, and visualize source distributions before trusting model-ready output. The result should be runnable, observable, reusable and protected by automated evidence.

    Orientation, Python and Scientific Computing • Scientific Workflow and Debugging • Guided Implementation

    Get Neural Networks and Deep Learning updates by email

    Short, practical lessons and course updates from Rumman Ansari. No spam.