Feature Engineering Pipelines

Real-World Scenario

A credit scoring model uses customer financial data: income (numerical), employment type (categorical with nulls), account age (date), and transaction history (time series). Raw features fed directly to a model cause errors and underperform. A proper feature engineering pipeline handles missing values, encodes categoricals, creates interaction features, and runs safely inside cross-validation.

The Leakage Problem

import numpy as np
from sklearn.datasets import make_classification
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_score
from sklearn.pipeline import Pipeline

X, y = make_classification(n_samples=1000, n_features=20, random_state=42)

# ── WRONG: fit scaler on all data BEFORE cross-validation ─────────────
scaler       = StandardScaler()
X_scaled_all = scaler.fit_transform(X)   # uses test fold info — LEAKAGE

bad_scores = cross_val_score(
    LogisticRegression(), X_scaled_all, y, cv=5, scoring="roc_auc"
)
print(f"Leaky CV AUC:  {bad_scores.mean():.4f}  (overly optimistic)")

# ── CORRECT: scaler inside the pipeline ────────────────────────────────
correct_pipe = Pipeline([
    ("scaler", StandardScaler()),
    ("model",  LogisticRegression()),
])

good_scores = cross_val_score(correct_pipe, X, y, cv=5, scoring="roc_auc")
print(f"Correct CV AUC: {good_scores.mean():.4f}  (honest estimate)")

ColumnTransformer for Mixed Data Types

import pandas as pd
import numpy as np
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder, OrdinalEncoder
from sklearn.impute import SimpleImputer, KNNImputer
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.model_selection import train_test_split, cross_val_score

# Realistic mixed-type dataset
rng = np.random.default_rng(42)
n   = 2000

df = pd.DataFrame({
    "income":          rng.lognormal(10, 1, n),           # continuous, right-skewed
    "age":             rng.integers(18, 75, n),            # integer
    "credit_score":    rng.normal(680, 80, n).clip(300, 850),
    "debt_ratio":      rng.beta(2, 5, n),
    "employment_type": rng.choice(
        ["salaried", "self-employed", "unemployed", None],
        n, p=[0.6, 0.25, 0.1, 0.05]
    ),
    "loan_purpose":    rng.choice(
        ["home", "auto", "education", "personal"], n
    ),
    "credit_history":  rng.choice(
        ["excellent", "good", "fair", "poor"], n,
        p=[0.2, 0.4, 0.3, 0.1]
    ),
})
# Inject some nulls in numerical columns
df.loc[rng.choice(n, 100), "income"]       = np.nan
df.loc[rng.choice(n, 80),  "credit_score"] = np.nan

# Target: default (1) or not (0)
df["default"] = (
    (df["debt_ratio"] > 0.5) |
    (df["credit_history"].isin(["poor", "fair"])) |
    (df["employment_type"] == "unemployed")
).astype(int)

X = df.drop("default", axis=1)
y = df["default"]

# Column groups by transformation needed
num_features = ["income", "age", "credit_score", "debt_ratio"]
cat_ohe      = ["employment_type", "loan_purpose"]   # low cardinality → one-hot
cat_ord      = ["credit_history"]                     # ordinal → ordered integers

# Preprocessing pipelines per column type
num_pipe = Pipeline([
    ("imputer", KNNImputer(n_neighbors=5)),    # KNN imputation for numerical
    ("scaler",  StandardScaler()),
])

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

cat_ord_pipe = Pipeline([
    ("imputer", SimpleImputer(strategy="most_frequent")),
    ("encoder", OrdinalEncoder(
        categories=[["poor", "fair", "good", "excellent"]],
        handle_unknown="use_encoded_value", unknown_value=-1,
    )),
])

preprocessor = ColumnTransformer([
    ("numerical",   num_pipe,       num_features),
    ("categorical", cat_ohe_pipe,   cat_ohe),
    ("ordinal",     cat_ord_pipe,   cat_ord),
], remainder="drop")

full_pipeline = Pipeline([
    ("preprocess", preprocessor),
    ("model",      HistGradientBoostingClassifier(max_iter=200, random_state=42)),
])

# Cross-validate the entire pipeline (no leakage)
scores = cross_val_score(full_pipeline, X, y, cv=5, scoring="roc_auc", n_jobs=-1)
print(f"CV AUC: {scores.mean():.4f} ± {scores.std():.4f}")

Custom Feature Engineering Transformers

import pandas as pd
import numpy as np
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.pipeline import Pipeline

class DateFeatureExtractor(BaseEstimator, TransformerMixin):
    """Extract temporal features from a date column."""

    def __init__(self, date_col: str, drop_original: bool = True):
        self.date_col      = date_col
        self.drop_original = drop_original

    def fit(self, X: pd.DataFrame, y=None):
        return self   # stateless transformer

    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
        X = X.copy()
        dates = pd.to_datetime(X[self.date_col], errors="coerce")
        X[f"{self.date_col}_year"]         = dates.dt.year
        X[f"{self.date_col}_month"]        = dates.dt.month
        X[f"{self.date_col}_dayofweek"]    = dates.dt.dayofweek   # 0=Mon, 6=Sun
        X[f"{self.date_col}_is_weekend"]   = (dates.dt.dayofweek >= 5).astype(int)
        X[f"{self.date_col}_quarter"]      = dates.dt.quarter
        X[f"{self.date_col}_days_since"]   = (
            pd.Timestamp.now() - dates
        ).dt.days.fillna(-1).astype(int)

        if self.drop_original:
            X = X.drop(columns=[self.date_col])
        return X


class InteractionFeatures(BaseEstimator, TransformerMixin):
    """Create pairwise ratio and product features."""

    def __init__(self, pairs: list[tuple[str, str]], operations: list[str] = ("ratio", "product")):
        self.pairs      = pairs
        self.operations = operations

    def fit(self, X, y=None):
        return self

    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
        X = X.copy()
        for col_a, col_b in self.pairs:
            if "ratio" in self.operations:
                X[f"{col_a}_div_{col_b}"] = X[col_a] / (X[col_b].replace(0, np.nan)).fillna(0)
            if "product" in self.operations:
                X[f"{col_a}_x_{col_b}"] = X[col_a] * X[col_b]
        return X


class TargetEncoder(BaseEstimator, TransformerMixin):
    """Replace categorical values with their target mean — fit-only on training data."""

    def __init__(self, cols: list[str], smoothing: float = 10.0):
        self.cols      = cols
        self.smoothing = smoothing
        self._maps     = {}
        self._global_mean = 0.0

    def fit(self, X: pd.DataFrame, y: pd.Series) -> "TargetEncoder":
        self._global_mean = float(y.mean())
        for col in self.cols:
            stats = pd.DataFrame({"y": y.values, "x": X[col].values})
            agg   = stats.groupby("x")["y"].agg(["mean", "count"])
            # Smoothed target encoding: blend category mean with global mean
            agg["smoothed"] = (
                agg["mean"] * agg["count"] + self._global_mean * self.smoothing
            ) / (agg["count"] + self.smoothing)
            self._maps[col] = agg["smoothed"].to_dict()
        return self

    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
        X = X.copy()
        for col in self.cols:
            X[f"{col}_te"] = X[col].map(self._maps[col]).fillna(self._global_mean)
        return X


# Build a pipeline with custom transformers
rng = np.random.default_rng(42)
n   = 1000
df  = pd.DataFrame({
    "signup_date":  pd.date_range("2020-01-01", periods=n, freq="D").astype(str),
    "income":       rng.lognormal(10, 1, n),
    "loan_amount":  rng.lognormal(9, 1, n),
    "region":       rng.choice(["north", "south", "east", "west"], n),
})
y = (rng.uniform(0, 1, n) > 0.7).astype(int)

from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_score

# Note: custom transformers that need column names must work on DataFrames
pipeline = Pipeline([
    ("dates",        DateFeatureExtractor("signup_date")),
    ("interactions", InteractionFeatures([("income", "loan_amount")])),
    ("target_enc",   TargetEncoder(["region"])),
    ("num_prep",     ColumnTransformer([
        ("scale", StandardScaler(), slice(None))   # scale all remaining columns
    ], remainder="passthrough")),
    ("model", LogisticRegression(max_iter=1000)),
])

# Must use cross_val_score with a DataFrame to preserve column names
from sklearn.model_selection import StratifiedKFold
cv     = StratifiedKFold(5, shuffle=True, random_state=42)
scores = cross_val_score(pipeline, df, y, cv=cv, scoring="roc_auc")
print(f"CV AUC: {scores.mean():.4f} ± {scores.std():.4f}")