preprocessing_pipeline.py

"""Build a leakage-safe preprocessing and classification pipeline."""

import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import classification_report
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler


def main() -> None:
    rng = np.random.default_rng(42)
    countries = np.array(["France", "Turkey", "United States"])
    frame = pd.DataFrame(
        {
            "country": rng.choice(countries, size=180),
            "height_cm": rng.normal(172, 10, size=180),
            "weight_kg": rng.normal(75, 14, size=180),
            "age": rng.integers(18, 65, size=180).astype(float),
        }
    )
    frame.loc[rng.choice(frame.index, 15, replace=False), "age"] = np.nan
    frame["target"] = (
        frame["weight_kg"] + frame["height_cm"] * 0.15 + rng.normal(0, 6, 180) > 101
    ).astype(int)

    X = frame.drop(columns="target")
    y = frame["target"]
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.25, random_state=42, stratify=y
    )

    numeric = ["height_cm", "weight_kg", "age"]
    categorical = ["country"]
    preprocessor = ColumnTransformer(
        [
            (
                "numeric",
                Pipeline(
                    [
                        ("imputer", SimpleImputer(strategy="median")),
                        ("scaler", StandardScaler()),
                    ]
                ),
                numeric,
            ),
            (
                "categorical",
                OneHotEncoder(handle_unknown="ignore"),
                categorical,
            ),
        ]
    )
    model = Pipeline(
        [("preprocessor", preprocessor), ("classifier", LogisticRegression(max_iter=1000))]
    )
    model.fit(X_train, y_train)
    predictions = model.predict(X_test)
    print(classification_report(y_test, predictions, digits=3))


if __name__ == "__main__":
    main()