S093 · Feature Engineering — Encoding, Scaling, Missing

"""feature_lab.py — a leakage-proof feature engineering pipeline on Titanic.
 
Compares:
  A. Naive: drop non-numeric columns, forget scaling, no imputation.
  B. Engineered: ColumnTransformer with imputers, scaler, one-hot.
 
Same model in both cases; the delta is entirely from feature engineering.
"""
from __future__ import annotations
import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_score, StratifiedKFold
from sklearn.datasets import fetch_openml
 
RNG = 42
 
# ---------------- data ----------------
def load_titanic() -> tuple[pd.DataFrame, pd.Series]:
    ds = fetch_openml("titanic", version=1, as_frame=True, parser="auto")
    df = ds.frame.copy()
    y = (df["survived"].astype(int))
    X = df.drop(columns=["survived", "name", "ticket", "cabin", "boat", "body", "home.dest"])
    # Ensure clean dtypes
    X["age"] = pd.to_numeric(X["age"], errors="coerce")
    X["fare"] = pd.to_numeric(X["fare"], errors="coerce")
    return X, y
 
 
# ---------------- naive baseline ----------------
def naive_score(X: pd.DataFrame, y: pd.Series) -> float:
    """Drop non-numeric, drop NaN rows, no scaling. The Bad Way."""
    Xn = X.select_dtypes(include=[np.number]).dropna()
    yn = y.loc[Xn.index]
    clf = LogisticRegression(max_iter=2000)
    cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=RNG)
    return cross_val_score(clf, Xn, yn, cv=cv, scoring="roc_auc").mean()
 
 
# ---------------- engineered pipeline ----------------
def engineered_pipeline() -> Pipeline:
    numeric_cols = ["age", "fare", "sibsp", "parch"]
    categorical_cols = ["sex", "embarked", "pclass"]
 
    numeric_pipe = Pipeline(steps=[
        ("impute", SimpleImputer(strategy="median")),
        ("scale", StandardScaler()),
    ])
 
    categorical_pipe = Pipeline(steps=[
        ("impute", SimpleImputer(strategy="constant", fill_value="__MISSING__")),
        ("onehot", OneHotEncoder(handle_unknown="ignore", sparse_output=False)),
    ])
 
    pre = ColumnTransformer(transformers=[
        ("num", numeric_pipe, numeric_cols),
        ("cat", categorical_pipe, categorical_cols),
    ], remainder="drop", verbose_feature_names_out=False)
 
    return Pipeline(steps=[
        ("pre", pre),
        ("clf", LogisticRegression(max_iter=2000)),
    ])
 
 
def engineered_score(X: pd.DataFrame, y: pd.Series) -> float:
    pipe = engineered_pipeline()
    cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=RNG)
    return cross_val_score(pipe, X, y, cv=cv, scoring="roc_auc").mean()
 
 
# ---------------- inspect what came out of the transformer ----------------
def inspect_features(X: pd.DataFrame, y: pd.Series) -> None:
    pipe = engineered_pipeline()
    pipe.fit(X, y)
    pre = pipe.named_steps["pre"]
    feat_names = pre.get_feature_names_out()
    Xt = pre.transform(X.head(3))
    df_view = pd.DataFrame(Xt, columns=feat_names, index=X.head(3).index)
    print("\nFirst 3 rows after preprocessing:")
    print(df_view.round(2).to_string())
 
 
if __name__ == "__main__":
    X, y = load_titanic()
    print(f"Loaded: {X.shape[0]} rows, {X.shape[1]} raw columns, missing per col:")
    print(X.isna().sum().to_string())
 
    naive = naive_score(X, y)
    eng = engineered_score(X, y)
    print(f"\n5-fold ROC-AUC:")
    print(f"  naive       : {naive:.3f}")
    print(f"  engineered  : {eng:.3f}")
    print(f"  Δ           : {eng - naive:+.3f}")
 
    inspect_features(X, y)