Scikit-learn Interface

Think of sklearn like a set of standardised power tools from the same brand. A drill, a sander, and a circular saw all look different and do different things. But they all have the same battery pack, the same on/off button location, and the same safety mechanism. Once you know how to use one tool in the set, picking up a new one takes two minutes — not two hours.

sklearn's "battery pack" is the estimator interface: every model is an object, .fit() trains it, .predict() uses it, .transform() processes data with it. Same pattern, every time.

import numpy as np
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LinearRegression
from sklearn.model_selection import train_test_split

np.random.seed(42)
n = 1000

# DoorDash delivery time dataset
distance  = np.abs(np.random.normal(4.0, 2.0, n)).clip(0.5, 15)
traffic   = np.random.randint(1, 11, n).astype(float)
prep_time = np.abs(np.random.normal(15, 5, n)).clip(5, 35)
delivery  = (8.6 + 7.3*distance + 0.8*prep_time + 1.5*traffic
             + np.random.normal(0, 4, n)).clip(10, 120)

X = np.column_stack([distance, traffic, prep_time])
y = delivery

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

# ── Step 1: Create the object ──────────────────────────────────────────
# Just instantiation — nothing is learned yet
scaler = StandardScaler()
model  = LinearRegression()

# ── Step 2: fit() — the learning step ─────────────────────────────────
# ONLY on training data
scaler.fit(X_train)      # learns mean and std of X_train
model.fit(X_train, y_train)  # nope — needs scaled input, see Pipeline below

# What did scaler learn and store?
print("What StandardScaler learned from X_train:")
print(f"  mean_:   {scaler.mean_.round(2)}")   # stored mean per feature
print(f"  scale_:  {scaler.scale_.round(2)}")  # stored std per feature

# ── Step 3: transform() — apply learned stats ────────────────────────
# Uses STORED stats — does NOT refit on test data
X_train_scaled = scaler.transform(X_train)
X_test_scaled  = scaler.transform(X_test)   # same mean/std as training

# ── fit_transform() shortcut — ONLY for training ─────────────────────
scaler2        = StandardScaler()
X_train_scaled = scaler2.fit_transform(X_train)   # fit + transform in one call
X_test_scaled  = scaler2.transform(X_test)        # transform only — no refitting!

# ── WRONG: calling fit_transform on test (leakage!) ──────────────────
# X_test_WRONG = scaler2.fit_transform(X_test)   # NEVER DO THIS
# This would compute NEW mean/std from test data — leaking test stats into your pipeline

# ── Train and predict with the model ─────────────────────────────────
model.fit(X_train_scaled, y_train)

y_pred_train = model.predict(X_train_scaled)
y_pred_test  = model.predict(X_test_scaled)

from sklearn.metrics import mean_absolute_error
print(f"
Linear Regression:")
print(f"  Train MAE: {mean_absolute_error(y_train, y_pred_train):.2f} min")
print(f"  Test MAE:  {mean_absolute_error(y_test,  y_pred_test):.2f} min")

from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LinearRegression, LogisticRegression
from sklearn.tree import DecisionTreeClassifier
import numpy as np

np.random.seed(42)
n = 500
X_num = np.column_stack([
    np.abs(np.random.normal(4, 2, n)),    # distance
    np.random.randint(1, 11, n).astype(float),  # traffic
])
y_reg = 20 + 5*X_num[:,0] + 2*X_num[:,1] + np.random.randn(n)*3
y_cls = (y_reg > 35).astype(int)

# ── StandardScaler attributes after fit() ─────────────────────────────
scaler = StandardScaler().fit(X_num)
print("StandardScaler attributes:")
print(f"  .mean_    (learned mean):    {scaler.mean_.round(3)}")
print(f"  .scale_   (learned std):     {scaler.scale_.round(3)}")
print(f"  .var_     (learned variance):{scaler.var_.round(3)}")
print(f"  .n_features_in_ (n features):{scaler.n_features_in_}")
print(f"  .n_samples_seen_:            {scaler.n_samples_seen_}")

# ── LinearRegression attributes after fit() ───────────────────────────
X_sc = scaler.transform(X_num)
lr   = LinearRegression().fit(X_sc, y_reg)
print("
LinearRegression attributes:")
print(f"  .coef_      (learned weights): {lr.coef_.round(3)}")
print(f"  .intercept_ (learned bias):    {lr.intercept_:.3f}")
print(f"  .n_features_in_:               {lr.n_features_in_}")

# ── DecisionTreeClassifier attributes after fit() ─────────────────────
dt = DecisionTreeClassifier(max_depth=3).fit(X_sc, y_cls)
print("
DecisionTreeClassifier attributes:")
print(f"  .n_features_in_:       {dt.n_features_in_}")
print(f"  .n_classes_:           {dt.n_classes_}")
print(f"  .classes_:             {dt.classes_}")
print(f"  .feature_importances_: {dt.feature_importances_.round(3)}")
print(f"  .max_features_:        {dt.max_features_}")

# ── OneHotEncoder attributes after fit() ─────────────────────────────
restaurants = np.array(['Pizza Hut','KFC','Dominos','Pizza Hut','KFC']).reshape(-1,1)
ohe = OneHotEncoder(sparse_output=False).fit(restaurants)
print("
OneHotEncoder attributes:")
print(f"  .categories_:      {ohe.categories_}")
print(f"  .feature_names_out: {ohe.get_feature_names_out(['restaurant'])}") 

import numpy as np
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LinearRegression, Ridge, Lasso
from sklearn.tree import DecisionTreeRegressor
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.neighbors import KNeighborsRegressor
from sklearn.metrics import mean_absolute_error
import warnings
warnings.filterwarnings('ignore')

np.random.seed(42)
n = 2000
distance  = np.abs(np.random.normal(4.0, 2.0, n)).clip(0.5, 15)
traffic   = np.random.randint(1, 11, n).astype(float)
prep      = np.abs(np.random.normal(15, 5, n)).clip(5, 35)
value     = np.abs(np.random.normal(350, 150, n)).clip(50, 1200)
delivery  = (8.6 + 7.3*distance + 0.8*prep + 1.5*traffic
             + np.random.normal(0, 4, n)).clip(10, 120)

X = np.column_stack([distance, traffic, prep, value])
y = delivery

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)
scaler     = StandardScaler()
X_train_sc = scaler.fit_transform(X_train)
X_test_sc  = scaler.transform(X_test)

# ── Every algorithm has the same interface ────────────────────────────
# Change only the model object — everything else stays the same
models = {
    'Linear Regression':      LinearRegression(),
    'Ridge Regression':       Ridge(alpha=1.0),
    'Lasso Regression':       Lasso(alpha=0.1),
    'Decision Tree':          DecisionTreeRegressor(max_depth=5, random_state=42),
    'Random Forest':          RandomForestRegressor(n_estimators=100, random_state=42, n_jobs=-1),
    'Gradient Boosting':      GradientBoostingRegressor(n_estimators=100, random_state=42),
    'K-Nearest Neighbours':   KNeighborsRegressor(n_neighbors=10),
}

print(f"{'Algorithm':<25} {'Train MAE':>10} {'Test MAE':>10} {'CV MAE (5-fold)':>16}")
print("─" * 68)

for name, model in models.items():
    # ── SAME THREE LINES for every single algorithm ─────────────────
    model.fit(X_train_sc, y_train)                    # 1. train
    y_pred = model.predict(X_test_sc)                 # 2. predict
    test_mae = mean_absolute_error(y_test, y_pred)    # 3. evaluate
    # ────────────────────────────────────────────────────────────────

    train_mae = mean_absolute_error(y_train, model.predict(X_train_sc))
    cv_scores = cross_val_score(model, X_train_sc, y_train,
                                 cv=5, scoring='neg_mean_absolute_error')
    cv_mae = -cv_scores.mean()

    print(f"  {name:<23} {train_mae:>10.2f} {test_mae:>10.2f} {cv_mae:>14.2f}")

# ── score() method — quick accuracy check ────────────────────────────
# For regressors: returns R² score
# For classifiers: returns accuracy
print(f"
R² scores (model.score()):")
for name, model in list(models.items())[:3]:
    r2 = model.score(X_test_sc, y_test)
    print(f"  {name:<25}: R² = {r2:.4f}")

A Pipeline is like an assembly line in a factory. Raw materials (data) enter at one end. Each station performs one operation — wash, cut, assemble, paint. The finished product (predictions) comes out at the other end. The assembly line has a fixed order. Each station knows exactly what state the material is in when it arrives.

Without Pipeline you are doing each factory step manually and carrying the half-finished product between stations yourself — error-prone, slow, and easy to do in the wrong order.

import numpy as np
import pandas as pd
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.linear_model import Ridge
from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import cross_val_score, train_test_split
from sklearn.metrics import mean_absolute_error

np.random.seed(42)
n = 2000
restaurants = ['Pizza Hut','KFC','Dominos','Biryani Blues',"McDonald's",'Subway']
cities      = ['Seattle','New York','Denver','Austin','Boston']

df = pd.DataFrame({
    'distance_km':    np.abs(np.random.normal(4, 2, n)).clip(0.5, 15),
    'traffic_score':  np.random.randint(1, 11, n).astype(float),
    'restaurant_prep': np.abs(np.random.normal(15, 5, n)).clip(5, 35),
    'order_value':    np.abs(np.random.normal(350, 150, n)).clip(50, 1200),
    'restaurant':     np.random.choice(restaurants, n),
    'city':           np.random.choice(cities, n),
})
y = (8.6 + 7.3*df['distance_km'] + 0.8*df['restaurant_prep']
     + 1.5*df['traffic_score'] + np.random.normal(0, 4, n)).clip(10, 120)

# Introduce some missing values
df.loc[np.random.choice(n, 80, replace=False),  'restaurant_prep'] = np.nan
df.loc[np.random.choice(n, 40, replace=False),  'traffic_score']   = np.nan

X_train, X_test, y_train, y_test = train_test_split(
    df, y, test_size=0.2, random_state=42
)

NUM_COLS = ['distance_km', 'traffic_score', 'restaurant_prep', 'order_value']
CAT_COLS = ['restaurant', 'city']

# ── Step 1: Build sub-pipelines for each column type ──────────────────
numeric_pipeline = Pipeline([
    ('imputer', SimpleImputer(strategy='median')),   # fills NaN with median
    ('scaler',  StandardScaler()),                    # standardise to mean=0, std=1
])

categorical_pipeline = Pipeline([
    ('imputer', SimpleImputer(strategy='most_frequent')),   # fills NaN with mode
    ('encoder', OneHotEncoder(handle_unknown='ignore',      # unknown → all zeros
                               sparse_output=False,
                               drop='first')),
])

# ── Step 2: ColumnTransformer — apply different pipelines to different columns
preprocessor = ColumnTransformer([
    ('num', numeric_pipeline,    NUM_COLS),
    ('cat', categorical_pipeline, CAT_COLS),
])

# ── Step 3: Full pipeline — preprocessor + model ──────────────────────
pipeline = Pipeline([
    ('preprocessor', preprocessor),
    ('model',        Ridge(alpha=1.0)),
])

# ── The pipeline is now ONE sklearn object ─────────────────────────────
# fit() trains the entire chain on training data only
pipeline.fit(X_train, y_train)

# predict() applies the entire chain to new data
y_pred = pipeline.predict(X_test)
print(f"Pipeline (Ridge) Test MAE: {mean_absolute_error(y_test, y_pred):.2f} min")

# ── Cross-validation with pipeline — leakage-proof ────────────────────
# The pipeline is refit from scratch in each fold
# Scaler/encoder statistics are never contaminated by validation data
cv_scores = cross_val_score(
    pipeline, df, y, cv=5,
    scoring='neg_mean_absolute_error'
)
print(f"5-fold CV MAE: {-cv_scores.mean():.2f} ± {cv_scores.std():.2f} min")

# ── Switching the model — change ONE word ─────────────────────────────
pipeline_rf = Pipeline([
    ('preprocessor', preprocessor),
    ('model',        RandomForestRegressor(n_estimators=100, random_state=42, n_jobs=-1)),
])
pipeline_rf.fit(X_train, y_train)
y_pred_rf = pipeline_rf.predict(X_test)
print(f"Pipeline (RF)   Test MAE: {mean_absolute_error(y_test, y_pred_rf):.2f} min")

# After fitting, you can inspect any step inside the pipeline

# Access by name
fitted_scaler  = pipeline.named_steps['preprocessor']                          .named_transformers_['num']                          .named_steps['scaler']
fitted_encoder = pipeline.named_steps['preprocessor']                          .named_transformers_['cat']                          .named_steps['encoder']
fitted_model   = pipeline.named_steps['model']

print("Stored scaler mean (from training data):")
print(f"  {fitted_scaler.mean_.round(2)}")

print("
Stored encoder categories:")
print(f"  {fitted_encoder.categories_}")

print("
Model coefficients:")
print(f"  {fitted_model.coef_.round(3)}")

# set_params() — change hyperparameters without rebuilding the pipeline
# Uses double underscore __ to navigate into nested steps
pipeline.set_params(model__alpha=10.0)   # change Ridge alpha
pipeline.fit(X_train, y_train)
y_pred_new = pipeline.predict(X_test)
print(f"
After changing alpha to 10: MAE={mean_absolute_error(y_test, y_pred_new):.2f}")

You define named transformers as a list of tuples: (name, transformer, columns). Each transformer processes its assigned columns independently. The results are concatenated horizontally into one output matrix.

remainder='drop' (default) — columns not listed are dropped. remainder='passthrough' — unlisted columns pass through unchanged.

import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import (StandardScaler, MinMaxScaler,
                                    OneHotEncoder, OrdinalEncoder)
from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline

np.random.seed(42)
n = 500

df = pd.DataFrame({
    # Numeric — need scaling
    'distance_km':    np.abs(np.random.normal(4, 2, n)).clip(0.5, 15),
    'order_value':    np.abs(np.random.normal(350, 150, n)).clip(50, 1200),
    # Ordinal categorical — has a natural order
    'traffic_level':  np.random.choice(['low','medium','high'], n),
    # Nominal categorical — no order
    'city':           np.random.choice(['Seattle','New York','Denver'], n),
    'restaurant':     np.random.choice(['KFC','Dominos','Pizza Hut'], n),
    # Binary
    'is_weekend':     np.random.randint(0, 2, n).astype(float),
})

# ── Build a ColumnTransformer for mixed types ─────────────────────────
ct = ColumnTransformer([
    # Numeric columns → StandardScaler (after imputing median for NaN)
    ('num_standard', Pipeline([
        ('imp',   SimpleImputer(strategy='median')),
        ('scale', StandardScaler()),
    ]), ['distance_km', 'order_value']),

    # Ordinal → OrdinalEncoder with explicit order
    ('ordinal', OrdinalEncoder(
        categories=[['low', 'medium', 'high']],
        handle_unknown='use_encoded_value', unknown_value=-1,
    ), ['traffic_level']),

    # Nominal → OneHotEncoder
    ('nominal', OneHotEncoder(
        sparse_output=False, handle_unknown='ignore', drop='first'
    ), ['city', 'restaurant']),

    # Binary → pass through as-is (no transformation needed)
    ('binary', 'passthrough', ['is_weekend']),
])

# fit on training data
ct.fit(df)
X_transformed = ct.transform(df)

print(f"Original shape:    {df.shape}")
print(f"Transformed shape: {X_transformed.shape}")

# What columns does each transformer produce?
print(f"
Output feature names:")
for name in ct.get_feature_names_out():
    print(f"  {name}")

# ── make_column_selector — select columns by dtype automatically ───────
from sklearn.compose import make_column_selector

# Instead of listing column names manually, select by dtype
ct_auto = ColumnTransformer([
    ('num', StandardScaler(),
     make_column_selector(dtype_include=np.number)),
    ('cat', OneHotEncoder(sparse_output=False, handle_unknown='ignore'),
     make_column_selector(dtype_include=object)),
])

# This auto-detects numeric and categorical columns
df_typed = df.copy()
df_typed['city']       = df_typed['city'].astype('category')
df_typed['restaurant'] = df_typed['restaurant'].astype('category')
# make_column_selector with dtype_include='category' would pick these up

import numpy as np
import pandas as pd
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.ensemble import RandomForestRegressor
from sklearn.linear_model import Ridge
from sklearn.model_selection import (
    cross_val_score, KFold, StratifiedKFold,
    GridSearchCV, RandomizedSearchCV,
)
from sklearn.metrics import mean_absolute_error

np.random.seed(42)
n = 1500
df = pd.DataFrame({
    'distance_km':    np.abs(np.random.normal(4, 2, n)).clip(0.5, 15),
    'traffic_score':  np.random.randint(1, 11, n).astype(float),
    'restaurant_prep': np.abs(np.random.normal(15, 5, n)).clip(5, 35),
    'order_value':    np.abs(np.random.normal(350, 150, n)).clip(50, 1200),
    'city':           np.random.choice(['Seattle','New York','Denver','Austin'], n),
    'restaurant':     np.random.choice(['KFC','Dominos','Pizza Hut','Subway'], n),
})
y = (8.6 + 7.3*df['distance_km'] + 0.8*df['restaurant_prep']
     + 1.5*df['traffic_score'] + np.random.normal(0, 4, n)).clip(10, 120)

NUM_COLS = ['distance_km', 'traffic_score', 'restaurant_prep', 'order_value']
CAT_COLS = ['city', 'restaurant']

preprocessor = ColumnTransformer([
    ('num', Pipeline([('imp', SimpleImputer(strategy='median')),
                      ('sc',  StandardScaler())]), NUM_COLS),
    ('cat', OneHotEncoder(sparse_output=False, handle_unknown='ignore',
                           drop='first'), CAT_COLS),
])
pipeline = Pipeline([
    ('preprocessor', preprocessor),
    ('model', Ridge(alpha=1.0)),
])

# ── cross_val_score — most common evaluation ──────────────────────────
scores = cross_val_score(
    pipeline, df, y,
    cv=5,                                    # 5 folds
    scoring='neg_mean_absolute_error',       # sklearn uses negative for minimisation
    n_jobs=-1,                               # parallel across all CPU cores
)
print(f"5-fold CV MAE: {-scores.mean():.2f} ± {scores.std():.2f} min")
print(f"Per-fold scores: {(-scores).round(2)}")

# ── GridSearchCV — exhaustive hyperparameter search ───────────────────
# Syntax: step_name__parameter_name (double underscore)
param_grid = {
    'model__alpha': [0.01, 0.1, 1.0, 10.0, 100.0],
}

grid_search = GridSearchCV(
    pipeline,
    param_grid,
    cv=5,
    scoring='neg_mean_absolute_error',
    n_jobs=-1,
    verbose=0,
)
grid_search.fit(df, y)

print(f"
GridSearchCV results:")
print(f"  Best alpha:  {grid_search.best_params_}")
print(f"  Best CV MAE: {-grid_search.best_score_:.2f} min")
print(f"  Best model:  {grid_search.best_estimator_}")

# ── RandomizedSearchCV — faster for large search spaces ───────────────
from scipy.stats import loguniform

pipeline_rf = Pipeline([
    ('preprocessor', preprocessor),
    ('model', RandomForestRegressor(random_state=42, n_jobs=-1)),
])

param_dist = {
    'model__n_estimators':     [50, 100, 200, 300],
    'model__max_depth':        [None, 5, 10, 20],
    'model__min_samples_leaf': [1, 5, 10, 20],
    'model__max_features':     ['sqrt', 'log2', 0.3],
}

random_search = RandomizedSearchCV(
    pipeline_rf,
    param_dist,
    n_iter=20,          # try 20 random combinations (not all 192)
    cv=3,
    scoring='neg_mean_absolute_error',
    random_state=42,
    n_jobs=-1,
)
random_search.fit(df, y)

print(f"
RandomizedSearchCV results:")
print(f"  Best params: {random_search.best_params_}")
print(f"  Best CV MAE: {-random_search.best_score_:.2f} min")

import numpy as np
import pandas as pd
import joblib
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.ensemble import RandomForestRegressor

np.random.seed(42)
n = 2000
df = pd.DataFrame({
    'distance_km':     np.abs(np.random.normal(4, 2, n)).clip(0.5, 15),
    'traffic_score':   np.random.randint(1, 11, n).astype(float),
    'restaurant_prep': np.abs(np.random.normal(15, 5, n)).clip(5, 35),
    'order_value':     np.abs(np.random.normal(350, 150, n)).clip(50, 1200),
    'restaurant':      np.random.choice(['Pizza Hut', 'KFC', 'Dominos'], n),
    'city':            np.random.choice(['Seattle', 'Denver', 'Boston'], n),
})
y = (8.6 + 7.3*df['distance_km'] + 0.8*df['restaurant_prep']
     + 1.5*df['traffic_score'] + np.random.normal(0, 4, n)).clip(10, 120)

from sklearn.model_selection import train_test_split
X_train, X_test, y_train, y_test = train_test_split(df, y, test_size=0.2, random_state=42)

# ── Training job — runs on a schedule, e.g. every Monday morning ──────
NUM_COLS = ['distance_km', 'traffic_score', 'restaurant_prep', 'order_value']
CAT_COLS = ['restaurant', 'city']

preprocessor = ColumnTransformer([
    ('num', Pipeline([('imp', SimpleImputer(strategy='median')),
                      ('sc',  StandardScaler())]), NUM_COLS),
    ('cat', OneHotEncoder(handle_unknown='ignore', sparse_output=False), CAT_COLS),
])

pipeline = Pipeline([
    ('preprocessor', preprocessor),
    ('model', RandomForestRegressor(n_estimators=200, random_state=42, n_jobs=-1)),
])

pipeline.fit(X_train, y_train)   # fits imputer, scaler, encoder AND model together

# ── The entire chain — preprocessing steps AND model — is one artifact ─
joblib.dump(pipeline, '/tmp/models/delivery_time_v14.pkl')

# ── Inference service — a completely different process, possibly ─────
# ── a different machine, days or weeks later ──────────────────────────
loaded_pipeline = joblib.load('/tmp/models/delivery_time_v14.pkl')

# One call handles imputation, scaling, one-hot encoding, AND prediction —
# using the EXACT statistics learned during the training job above.
# There is no separate scaler.pkl and encoder.pkl to keep in sync,
# and no way to accidentally call .fit() on live traffic by mistake —
# the inference code only ever calls .predict(), never .fit().
new_order_df = X_test.iloc[[0]]   # one incoming order, same columns as training
prediction = loaded_pipeline.predict(new_order_df)

The method names are identical, but the assumptions behind them are not. Swapping a scaled linear model for a tree ensemble often means scaling is no longer necessary at all, since trees split on raw thresholds regardless of feature magnitude. Swapping in a model that supports class_weight opens up an imbalance-handling option a previous model did not have. The unified interface means less code has to change when you swap algorithms — as Section 4 demonstrates directly — not that literally nothing about the surrounding pipeline needs reconsidering.

fit_transform() does not just apply a transformation — the fit half genuinely learns new statistics from whatever data it is called on. Calling it on test data recomputes mean, standard deviation, or category lists directly from the test set, which both leaks test information into what is supposed to be an unbiased evaluation and can silently produce different scaling entirely if the test distribution differs even slightly from training. Test data must only ever see .transform(), which reuses the statistics fit() already learned from training data — never its own fit_transform() call.

Almost every real feature-engineering step that is not a plain column-level transformation — a business-specific ratio, a domain-informed outlier clip like the OutlierClipper pattern in the Python-for-ML module, target encoding that needs careful cross-fitting to avoid leakage — ends up as a small BaseEstimator and TransformerMixin subclass specifically so it can live inside a Pipeline. That gets it the same leakage protection, the same get_params()/set_params() support GridSearchCV relies on, and the same serialisation behaviour as every built-in transformer, for a few lines of extra code.

Manual step-by-step code is exactly what creates the classic leakage bug: a scaler fit once on the full dataset before the train/test split, or accidentally refit on validation data inside a cross-validation loop written by hand. Pipeline's real job is removing that opportunity structurally — inside cross_val_score(pipeline, ...), every fold refits the entire chain from scratch using only that fold's training portion, which manually-sequenced code has to painstakingly reproduce and very often does not.

The method name is identical everywhere, but what it computes depends entirely on the estimator type: for a regressor .score() returns R², for a classifier it returns plain accuracy. Accuracy alone is close to meaningless on an imbalanced classification problem — a model that always predicts the majority class can score above 90 percent while being useless. Knowing what a specific estimator's default .score() actually measures, and reaching for precision, recall, or AUC when accuracy would mislead, is part of using the interface correctly rather than trusting the number by name alone.

Because a Pipeline is itself a single estimator, tools like cross_val_score and GridSearchCV treat it as one unit and refit the entire thing — imputer, scaler, encoder, and model — from scratch on only each fold's training portion. Manual code that scales the full dataset once before splitting, or that reuses one fitted scaler across every fold, lets statistics from validation or test data quietly influence training. Pipeline does not add a new capability so much as remove the opportunity to make that mistake, by making "fit only on this fold's training data" the only thing that happens automatically.

I would inherit from BaseEstimator, which gives get_params() and set_params() for free so the transformer works with GridSearchCV, and from TransformerMixin, which gives a default fit_transform() implementation for free. Then I implement three methods myself: __init__ to store hyperparameters only, fit(X, y=None) to compute and store any statistics needed from the training data — using a trailing underscore naming convention for what gets learned — and transform(X) to apply those stored statistics to new data. The OutlierClipper example in the Python-for-ML module is exactly this pattern in nine lines.

fit() only learns and stores statistics from the data it is given — it changes nothing and returns no transformed output. transform() applies previously stored statistics to data without learning anything new from it. fit_transform() is a convenience that does both in one call, sometimes with a small performance benefit, but only for the data the model should learn from. fit() and fit_transform() are the wrong call on test or validation data every time, because both would learn new statistics from data that is supposed to be evaluated, not trained on — transform() is the only correct call there.

I would navigate through the named steps: pipeline.named_steps['preprocessor'] gets the ColumnTransformer, .named_transformers_['num'] gets the numeric sub-pipeline by the name it was registered under, and .named_steps['scaler'] gets the actual StandardScaler instance inside that sub-pipeline. From there .mean_ is the learned attribute. The general pattern — chaining named_steps and named_transformers_ — works for inspecting or debugging any nested step inside a fitted Pipeline, which comes up constantly when something inside a production pipeline needs to be audited.

If the feature is a simple deterministic function of columns already present in a single row — like dividing one column by another — placement usually does not introduce leakage, since no information from other rows is involved. But the same question matters enormously for anything that involves an aggregate: a feature like "average order value for this restaurant" must be computed only from training data and supplied via transform() at prediction time, never recomputed including the test row itself, or the model gets to see a statistic that secretly includes its own answer. Building it as a pipeline step forces that computation to happen correctly, fold by fold, instead of once over the whole dataset by habit.