Data Cleaning & Imputation

⚠ Common Data Quality Problems

Real-world data is never clean. Before any model training, you must understand what is wrong with your data, how much of it is affected, and what impact each problem will have on downstream learning. Rushing past this audit step is the single most common source of silent model failures.

❓ Handling Missing Values

Not all missing data is the same. The mechanism by which values go missing determines which handling strategy is appropriate. Using the wrong strategy can introduce systematic bias — even worse than leaving the data alone.

Imputation Strategies

import pandas as pd
import numpy as np
from sklearn.impute import SimpleImputer, KNNImputer
from sklearn.experimental import enable_iterative_imputer  # noqa
from sklearn.impute import IterativeImputer

df = pd.read_csv("dataset.csv")

# ---- Step 1: Audit missingness ----
missing_report = df.isnull().mean().sort_values(ascending=False)
print(missing_report[missing_report > 0])

# Add "was_missing" indicators BEFORE imputation (captures MNAR signal)
for col in df.columns[df.isnull().any()]:
    df[f"{col}_was_missing"] = df[col].isnull().astype(int)

# ---- Step 2: Simple imputation (baseline) ----
simple_imp = SimpleImputer(strategy="median")  # or "mean", "most_frequent", "constant"
df[["age", "income"]] = simple_imp.fit_transform(df[["age", "income"]])

# ---- Step 3: KNN imputation (preserves feature correlations) ----
# Note: KNNImputer requires scaled data — normalise first
knn_imp = KNNImputer(n_neighbors=5, weights="distance")
df[["bmi", "glucose"]] = knn_imp.fit_transform(df[["bmi", "glucose"]])

# ---- Step 4: MICE / Iterative imputation (best for MAR) ----
iter_imp = IterativeImputer(
    max_iter=10,
    random_state=42,
    initial_strategy="median",
    estimator=None  # defaults to BayesianRidge; swap for RandomForestRegressor for non-linear
)
numeric_cols = df.select_dtypes(include=np.number).columns
df[numeric_cols] = iter_imp.fit_transform(df[numeric_cols])

# IMPORTANT: fit imputers on train set only; transform train + test
# store imputer objects and call .transform() on test data

💥 Outlier Detection & Treatment

Outliers are data points that deviate significantly from the rest. They can be genuine (a real billionaire in income data) or errors (age = 999). The distinction is critical — removing genuine outliers destroys real-world signal; keeping erroneous ones corrupts your model.

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest

df = pd.read_csv("dataset.csv")

# ---- IQR capping (winsorization) ----
def cap_outliers_iqr(series, multiplier=1.5):
    Q1, Q3 = series.quantile(0.25), series.quantile(0.75)
    IQR = Q3 - Q1
    lower, upper = Q1 - multiplier * IQR, Q3 + multiplier * IQR
    return series.clip(lower=lower, upper=upper), (lower, upper)

df["age"], age_bounds = cap_outliers_iqr(df["age"])
print(f"Age capped at {age_bounds}")

# ---- Z-score filtering (removes rows — use with caution) ----
from scipy import stats
z_scores = np.abs(stats.zscore(df[["income", "spending"]]))
df_clean = df[(z_scores < 3).all(axis=1)]
print(f"Removed {len(df) - len(df_clean)} rows via z-score filter")

# ---- Isolation Forest (multivariate) ----
iso_forest = IsolationForest(contamination=0.05, random_state=42)  # 5% assumed outlier rate
feature_cols = ["age", "income", "spending", "credit_score"]
df["outlier_score"] = iso_forest.fit_predict(df[feature_cols])
# -1 = outlier, 1 = inlier
outliers = df[df["outlier_score"] == -1]
print(f"Isolation Forest flagged {len(outliers)} outliers ({len(outliers)/len(df)*100:.1f}%)")

# Treatment options (pick based on domain knowledge):
# Option A: Remove            df = df[df["outlier_score"] == 1]
# Option B: Cap/clip          (use cap_outliers_iqr above)
# Option C: Transform         df["income"] = np.log1p(df["income"])
# Option D: Keep + flag       (already done with outlier_score column)

🧹 Duplicate & Inconsistency Removal

Duplicates and format inconsistencies silently inflate your dataset and create spurious category splits. A model that learns "New York" and "new york" are different cities will fail systematically on any input that doesn't match training-time formatting.

import pandas as pd
from rapidfuzz import fuzz, process

df = pd.read_csv("customers.csv")

# ---- Exact duplicate removal ----
n_before = len(df)
df = df.drop_duplicates(subset=["email", "phone"], keep="first")
print(f"Removed {n_before - len(df)} exact duplicates")

# ---- String standardisation ----
df["city"] = df["city"].str.lower().str.strip()
df["city"] = df["city"].str.replace(r"[^\w\s]", "", regex=True)  # remove punctuation

# ---- Canonical category mapping ----
city_map = {
    "ny":        "new york",
    "n.y.":      "new york",
    "new york city": "new york",
    "nyc":       "new york",
    "la":        "los angeles",
    "l.a.":      "los angeles",
}
df["city"] = df["city"].map(city_map).fillna(df["city"])

# ---- Fuzzy deduplication for near-duplicates ----
# Given a reference list of canonical city names:
canonical_cities = ["new york", "los angeles", "chicago", "houston", "phoenix"]

def fuzzy_standardise(value, choices, threshold=85):
    if pd.isna(value):
        return value
    match, score, _ = process.extractOne(str(value), choices, scorer=fuzz.token_sort_ratio)
    return match if score >= threshold else value

df["city"] = df["city"].apply(lambda x: fuzzy_standardise(x, canonical_cities))

# ---- Category harmonisation ----
df["gender"] = df["gender"].str.lower().map({
    "m": "male", "man": "male", "male": "male",
    "f": "female", "woman": "female", "female": "female",
}).fillna("unknown")

print(df["city"].value_counts().head(10))

🔧 Cleaning Pipelines

Ad-hoc cleaning scripts are technical debt. The moment you need to apply the same cleaning logic to new data (a new batch, a test set, production inference), scattered notebook code becomes a liability. Building cleaning as a pipeline ensures reproducibility and prevents data leakage.

import pandas as pd
import numpy as np
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import FunctionTransformer
from sklearn.impute import SimpleImputer
import logging

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

# ---- Custom cleaning transformers ----
def remove_duplicates(X):
    n_before = len(X)
    X = X.drop_duplicates()
    log.info(f"Removed {n_before - len(X)} duplicate rows")
    return X

def cap_outliers(X, cols=None, multiplier=1.5):
    X = X.copy()
    cols = cols or X.select_dtypes(include=np.number).columns
    for col in cols:
        Q1, Q3 = X[col].quantile(0.25), X[col].quantile(0.75)
        IQR = Q3 - Q1
        lower, upper = Q1 - multiplier * IQR, Q3 + multiplier * IQR
        n_capped = ((X[col] < lower) | (X[col] > upper)).sum()
        X[col] = X[col].clip(lower, upper)
        log.info(f"Capped {n_capped} outliers in '{col}' to [{lower:.2f}, {upper:.2f}]")
    return X

def standardise_strings(X, cols):
    X = X.copy()
    for col in cols:
        X[col] = X[col].str.lower().str.strip()
    return X

# ---- Build the pipeline ----
# Note: sklearn Pipelines work with estimators; for pandas DataFrames,
# use FunctionTransformer or a custom BaseEstimator + TransformerMixin

from sklearn.base import BaseEstimator, TransformerMixin

class DataCleaner(BaseEstimator, TransformerMixin):
    def __init__(self, numeric_cols, string_cols):
        self.numeric_cols = numeric_cols
        self.string_cols   = string_cols
        self.imputer_      = SimpleImputer(strategy="median")
        self.iqr_bounds_   = {}

    def fit(self, X, y=None):
        # Compute statistics on TRAIN data only
        for col in self.numeric_cols:
            Q1, Q3 = X[col].quantile(0.25), X[col].quantile(0.75)
            IQR = Q3 - Q1
            self.iqr_bounds_[col] = (Q1 - 1.5*IQR, Q3 + 1.5*IQR)
        self.imputer_.fit(X[self.numeric_cols])
        return self

    def transform(self, X):
        X = X.copy()
        X = remove_duplicates(X)
        X = standardise_strings(X, self.string_cols)
        for col, (lo, hi) in self.iqr_bounds_.items():
            X[col] = X[col].clip(lo, hi)
        X[self.numeric_cols] = self.imputer_.transform(X[self.numeric_cols])
        return X

# Usage
cleaner = DataCleaner(
    numeric_cols=["age", "income", "balance"],
    string_cols=["city", "country"]
)
X_train_clean = cleaner.fit_transform(X_train)   # fit + transform train
X_test_clean  = cleaner.transform(X_test)         # transform test with TRAIN statistics