A pipeline is the object that makes a model reproducible: one thing to fit, one thing to save, one thing to call. Tuning on top of it is how you choose hyperparameters without quietly spending the test set.
Everything in one object
import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler
rng = np.random.default_rng(3)
n = 3000
frame = pd.DataFrame({
'income': rng.lognormal(10.2, 0.5, n).round(-2),
'loan': rng.lognormal(9.0, 0.6, n).round(-2),
'years': rng.gamma(2.2, 2.4, n).round(1),
'purpose': rng.choice(['car', 'home', 'business'], n),
'region': rng.choice(['North', 'South', 'East', 'West'], n),
})
frame.loc[rng.choice(n, 140, replace=False), 'years'] = np.nan
frame['default'] = ((3.0 * frame['loan'] / frame['income']
- 0.08 * frame['years'].fillna(4) + rng.normal(0, 0.8, n)) > 1.2).astype(int)
NUMERIC = ['income', 'loan', 'years']
CATEGORICAL = ['purpose', 'region']
model = Pipeline([
('prepare', ColumnTransformer([
('num', Pipeline([('impute', SimpleImputer(strategy='median')),
('scale', StandardScaler())]), NUMERIC),
('cat', Pipeline([('impute', SimpleImputer(strategy='most_frequent')),
('encode', OneHotEncoder(handle_unknown='ignore'))]), CATEGORICAL),
])),
('model', HistGradientBoostingClassifier(random_state=0)),
])
model.fit(frame[NUMERIC + CATEGORICAL], frame['default'])
print('fitted:', type(model.named_steps['model']).__name__)
print('steps: ', [name for name, _ in model.steps])
print('features out of the preparer:',
len(model.named_steps['prepare'].get_feature_names_out()))
# One call at inference, with the same preprocessing by construction.
new_row = pd.DataFrame([{'income': 42_000, 'loan': 18_000, 'years': np.nan,
'purpose': 'car', 'region': 'Nowhere'}])
print('prediction for an unseen region and a missing value:',
model.predict_proba(new_row)[0, 1].round(3))
Open this example in the editor and change it. Nothing you type is saved.
The last two lines are the argument. A new region the encoder has never seen and a missing value both pass through without special handling, because the pipeline carries the imputer's median and the encoder's vocabulary with it.
Three failures the pipeline prevents outright:
Preprocessing fitted on the test set, which inflates every score. Forgetting a step at inference, which produces predictions that are quietly wrong. Train and serve diverging, because there is one object rather than two code paths.
Tuning
Hyperparameters must be chosen on data the final score is not computed from. GridSearchCV and RandomizedSearchCV do that by cross-validating inside the training set.
import numpy as np
import pandas as pd
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.model_selection import GridSearchCV, StratifiedKFold, train_test_split
from sklearn.metrics import roc_auc_score
rng = np.random.default_rng(7)
n = 4000
X = rng.normal(size=(n, 8))
y = ((1.1 * X[:, 0] + 1.3 * (X[:, 1] > 0) * (X[:, 2] > 0)
+ rng.normal(0, 1.1, n)) > 0.8).astype(int)
X_rest, X_test, y_rest, y_test = train_test_split(
X, y, test_size=0.2, random_state=0, stratify=y)
grid = {
'learning_rate': [0.03, 0.1, 0.3],
'max_leaf_nodes': [7, 15, 31],
'min_samples_leaf': [5, 20],
}
search = GridSearchCV(
HistGradientBoostingClassifier(max_iter=200, early_stopping=False, random_state=0),
grid, cv=StratifiedKFold(4, shuffle=True, random_state=0),
scoring='roc_auc', n_jobs=1, refit=True).fit(X_rest, y_rest)
print(f'{len(search.cv_results_["params"])} combinations x 4 folds = '
f'{len(search.cv_results_["params"]) * 4} fits')
print(f'best parameters: {search.best_params_}')
print(f'best cross-validated AUC: {search.best_score_:.4f}')
print()
results = pd.DataFrame(search.cv_results_)[
['param_learning_rate', 'param_max_leaf_nodes', 'param_min_samples_leaf',
'mean_test_score', 'std_test_score']].sort_values('mean_test_score', ascending=False)
print(results.head(5).round(4).to_string(index=False))
print()
print(f'TEST AUC with the chosen settings: '
f'{roc_auc_score(y_test, search.predict_proba(X_test)[:, 1]):.4f}')
Open this example in the editor and change it. Nothing you type is saved.
Note the gap between best_score_ and the test score. The first is optimistic, because it is the maximum over eighteen combinations - the winner got there partly on luck. The test set, used once, is the number to publish.
Random search beats grid search when there are many hyperparameters, because a grid spends most of its budget varying parameters that do not matter:
import numpy as np
import pandas as pd
from scipy_free_uniform import _ # placeholder, replaced below
Open this example in the editor and change it. Nothing you type is saved.
import numpy as np
import pandas as pd
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.model_selection import RandomizedSearchCV, StratifiedKFold, train_test_split
rng = np.random.default_rng(11)
n = 4000
X = rng.normal(size=(n, 8))
y = ((1.1 * X[:, 0] + 1.3 * (X[:, 1] > 0) * (X[:, 2] > 0)
+ rng.normal(0, 1.1, n)) > 0.8).astype(int)
X_rest, X_test, y_rest, y_test = train_test_split(
X, y, test_size=0.2, random_state=0, stratify=y)
# Distributions, not lists: a random search can land between the
# values a grid would have tried.
distributions = {
'learning_rate': [0.02, 0.04, 0.06, 0.08, 0.1, 0.15, 0.2, 0.3],
'max_leaf_nodes': [5, 7, 10, 15, 20, 31, 45, 63],
'min_samples_leaf': [2, 5, 10, 20, 40, 80],
'l2_regularisation' if False else 'l2_regularization': [0.0, 0.1, 1.0, 10.0],
}
search = RandomizedSearchCV(
HistGradientBoostingClassifier(max_iter=200, early_stopping=False, random_state=0),
distributions, n_iter=20, cv=StratifiedKFold(4, shuffle=True, random_state=0),
scoring='roc_auc', random_state=0).fit(X_rest, y_rest)
print(f'grid would need {8 * 8 * 6 * 4} combinations; this tried 20')
print(f'best cross-validated AUC: {search.best_score_:.4f}')
print(f'best parameters: {search.best_params_}')
Open this example in the editor and change it. Nothing you type is saved.
Tuning the pipeline, not just the model, uses the double-underscore notation:
import numpy as np
from sklearn.compose import ColumnTransformer
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.impute import SimpleImputer
from sklearn.model_selection import GridSearchCV, StratifiedKFold
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
import pandas as pd
rng = np.random.default_rng(13)
n = 2000
frame = pd.DataFrame(rng.normal(size=(n, 4)), columns=list('abcd'))
frame.loc[rng.choice(n, 150, replace=False), 'a'] = np.nan
y = ((frame['b'].fillna(0) + frame['a'].fillna(0) + rng.normal(0, 1, n)) > 0).astype(int)
pipeline = Pipeline([
('prepare', ColumnTransformer([
('num', Pipeline([('impute', SimpleImputer()), ('scale', StandardScaler())]),
list('abcd'))])),
('model', HistGradientBoostingClassifier(random_state=0)),
])
grid = {
'prepare__num__impute__strategy': ['mean', 'median', 'most_frequent'],
'model__max_leaf_nodes': [7, 31],
}
search = GridSearchCV(pipeline, grid, cv=StratifiedKFold(3, shuffle=True, random_state=0),
scoring='roc_auc').fit(frame, y)
print(f'best: {search.best_params_}')
print(f'AUC: {search.best_score_:.4f}')
print()
print('The imputation strategy is now a tuned hyperparameter, chosen')
print('per fold - which is the only way to choose it honestly.')
Open this example in the editor and change it. Nothing you type is saved.
The number you report
import numpy as np
import pandas as pd
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.model_selection import (GridSearchCV, StratifiedKFold, cross_val_score,
train_test_split)
from sklearn.metrics import roc_auc_score
rng = np.random.default_rng(17)
n = 2500
X = rng.normal(size=(n, 6))
y = ((X[:, 0] + 0.8 * X[:, 1] + rng.normal(0, 1.2, n)) > 0.4).astype(int)
X_rest, X_test, y_rest, y_test = train_test_split(
X, y, test_size=0.2, random_state=0, stratify=y)
grid = {'max_leaf_nodes': [5, 15, 31, 63], 'learning_rate': [0.05, 0.1, 0.2]}
inner = StratifiedKFold(4, shuffle=True, random_state=0)
search = GridSearchCV(HistGradientBoostingClassifier(max_iter=150, early_stopping=False,
random_state=0),
grid, cv=inner, scoring='roc_auc').fit(X_rest, y_rest)
# Nested cross-validation: the whole search is treated as the model.
outer = StratifiedKFold(4, shuffle=True, random_state=1)
nested = cross_val_score(
GridSearchCV(HistGradientBoostingClassifier(max_iter=150, early_stopping=False,
random_state=0),
grid, cv=inner, scoring='roc_auc'),
X_rest, y_rest, cv=outer, scoring='roc_auc')
print(f'best_score_ (max over the grid): {search.best_score_:.4f} <- optimistic')
print(f'nested cross-validation: {nested.mean():.4f} +/- {nested.std():.4f}')
print(f'held-out test set (used once): '
f'{roc_auc_score(y_test, search.predict_proba(X_test)[:, 1]):.4f}')
print()
print('best_score_ is the winner of a competition and is biased upwards.')
print('Nested CV and the held-out test set agree, and either is the')
print('number to publish.')
Open this example in the editor and change it. Nothing you type is saved.
Reproducibility
import numpy as np
import pandas as pd
import sklearn
import sys
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import roc_auc_score
SEED = 20260407
rng = np.random.default_rng(SEED)
n = 2000
X = rng.normal(size=(n, 5))
y = ((X[:, 0] + rng.normal(0, 1, n)) > 0).astype(int)
def run(seed):
X_tr, X_te, y_tr, y_te = train_test_split(
X, y, test_size=0.3, random_state=seed, stratify=y)
model = HistGradientBoostingClassifier(random_state=seed).fit(X_tr, y_tr)
return roc_auc_score(y_te, model.predict_proba(X_te)[:, 1])
print(f'same seed twice: {run(SEED):.6f} and {run(SEED):.6f}')
print(f'identical: {run(SEED) == run(SEED)}')
print(f'different seed: {run(SEED + 1):.6f}')
print()
print('MANIFEST')
for key, value in {
'python': sys.version.split()[0],
'numpy': np.__version__,
'pandas': pd.__version__,
'scikit-learn': sklearn.__version__,
'seed': SEED,
'rows': n,
}.items():
print(f' {key:<14} {value}')
Open this example in the editor and change it. Nothing you type is saved.
Every random_state must be set: in the splitter, in the model, in the search. Missing one makes the result irreproducible, and it is always the one you forgot.
Four things to record with every trained model:
- The versions of Python and the libraries.
- The seed, in every place one is used.
- A hash of the training data, so "the data changed" is a checkable claim.
- The chosen hyperparameters and the score, with the split that produced it.
Saving and serving
import numpy as np
import pandas as pd
import tempfile
import os
import pickle
from sklearn.compose import ColumnTransformer
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, OneHotEncoder
rng = np.random.default_rng(19)
n = 1500
frame = pd.DataFrame({
'amount': rng.gamma(3, 40, n).round(2),
'region': rng.choice(['North', 'South'], n),
})
y = ((frame['amount'] > 120) ^ (rng.random(n) < 0.2)).astype(int)
pipeline = Pipeline([
('prepare', ColumnTransformer([
('num', StandardScaler(), ['amount']),
('cat', OneHotEncoder(handle_unknown='ignore'), ['region'])])),
('model', HistGradientBoostingClassifier(random_state=0)),
]).fit(frame, y)
with tempfile.TemporaryDirectory() as directory:
path = os.path.join(directory, 'model.pkl')
with open(path, 'wb') as handle:
pickle.dump({'pipeline': pipeline, 'features': list(frame.columns),
'trained_rows': len(frame), 'seed': 0}, handle)
size = os.path.getsize(path)
with open(path, 'rb') as handle:
loaded = pickle.load(handle)
before = pipeline.predict_proba(frame.head(5))[:, 1]
after = loaded['pipeline'].predict_proba(frame.head(5))[:, 1]
print(f'saved {size:,} bytes')
print(f'identical predictions after reload: {bool(np.allclose(before, after))}')
print(f'metadata saved alongside: {[k for k in loaded if k != "pipeline"]}')
Open this example in the editor and change it. Nothing you type is saved.
Two warnings about pickle, and both have bitten real deployments. A pickle is tied to the library version - unpickling under a different scikit-learn may fail or, worse, behave differently. Record the version and check it on load. And a pickle executes code on load, so never load one from an untrusted source.
When it goes wrong
| Symptom | Cause |
|---|---|
| Cross-validation beats the test set badly | Preprocessing fitted outside the pipeline |
| An unseen category crashed inference | handle_unknown='ignore' |
best_score_ did not hold up |
It is the max over a grid; use nested CV |
| Results differ between runs | A missing random_state somewhere |
| A grid search takes hours | Use RandomizedSearchCV, or fewer parameters |
| The loaded model predicts differently | A library version mismatch |
| Serving code diverged from training | Two code paths; save the pipeline |
| Tuning the preparer is awkward | Use step__substep__param in the grid |
A check you can run
Take your training script and search it for fit( and fit_transform(.
Every one of those calls must be inside a pipeline, or inside a fold. Any fit_transform on the full dataset before the split is leakage, and the score it produced is too high by an amount nobody can estimate afterwards.