Cross-Validation

import numpy as np
from sklearn.datasets import load_iris, fetch_california_housing
from sklearn.model_selection import (
    train_test_split, cross_val_score, cross_validate,
    KFold, StratifiedKFold, TimeSeriesSplit, GroupKFold,
    GridSearchCV, RandomizedSearchCV,
)
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.linear_model import LogisticRegression, LinearRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, classification_report

# 1) Why cross-validation?
# A single train/test split gives one noisy estimate. CV reduces variance + lets you tune
# hyperparameters without contaminating the held-out test set.

# 2) Basic K-fold (regression)
X, y = fetch_california_housing(return_X_y=True)
pipe = Pipeline([
    ('scaler', StandardScaler()),
    ('lr',     LinearRegression()),
])
scores = cross_val_score(pipe, X, y, cv=5, scoring='neg_mean_absolute_error')
print('MAE:', -scores.mean(), '±', scores.std())

# 3) Stratified K-fold (classification) — preserves class proportions
X, y = load_iris(return_X_y=True)
clf = Pipeline([('scaler', StandardScaler()), ('lr', LogisticRegression(max_iter=1000))])

skf = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
scores = cross_val_score(clf, X, y, cv=skf, scoring='accuracy')
print(f'acc: {scores.mean():.3f} ± {scores.std():.3f}')

# 4) cross_validate — multiple metrics + train scores + fit times
results = cross_validate(
    clf, X, y,
    cv = skf,
    scoring = ['accuracy', 'f1_macro', 'roc_auc_ovr'],
    return_train_score = True,
    n_jobs = -1,
)
print(results['test_accuracy'].mean())
print(results['fit_time'].sum())

# 5) Time-series split — no future leakage
# Standard K-fold shuffles → trains on future, tests on past. Use TimeSeriesSplit.
tscv = TimeSeriesSplit(n_splits=5, test_size=100, gap=0)
for train_idx, test_idx in tscv.split(X):
    print(f'train: {train_idx.min()}..{train_idx.max()}, test: {test_idx.min()}..{test_idx.max()}')

scores = cross_val_score(model, X, y, cv=tscv, scoring='neg_mean_absolute_error')

# 6) Group K-fold — when rows aren't independent (multiple readings per patient)
groups = patient_ids        # one group per patient
gkf = GroupKFold(n_splits=5)
for train_idx, test_idx in gkf.split(X, y, groups):
    # Same patient never appears in both train and test
    pass

scores = cross_val_score(model, X, y, groups=groups, cv=gkf, scoring='accuracy')

# 7) GridSearchCV — tune hyperparameters via nested CV
param_grid = {
    'lr__C':       [0.01, 0.1, 1, 10],
    'lr__penalty': ['l2'],
}
grid = GridSearchCV(
    clf, param_grid,
    cv = StratifiedKFold(5, shuffle=True, random_state=42),
    scoring = 'f1_macro',
    n_jobs = -1,
    return_train_score = True,
)
grid.fit(X, y)
print(grid.best_params_, grid.best_score_)
print(grid.cv_results_['mean_test_score'])

# 8) RandomizedSearchCV — when the grid is huge
from scipy.stats import uniform, randint
rand = RandomizedSearchCV(
    RandomForestClassifier(random_state=42),
    param_distributions = {
        'n_estimators':    randint(100, 1000),
        'max_depth':       randint(3, 30),
        'min_samples_leaf': randint(1, 20),
        'max_features':    uniform(0.1, 0.9),
    },
    n_iter = 50,
    cv = 5,
    scoring = 'f1_macro',
    n_jobs = -1,
    random_state = 42,
)
rand.fit(X, y)

# 9) Avoid leakage — fit preprocessing INSIDE the pipeline, NOT before CV
# WRONG
X_scaled = StandardScaler().fit_transform(X)        # uses TEST statistics → leakage
cross_val_score(LogisticRegression(), X_scaled, y, cv=5)

# RIGHT
pipe = Pipeline([('scaler', StandardScaler()), ('lr', LogisticRegression())])
cross_val_score(pipe, X, y, cv=5)
# The scaler is re-fit on each fold's train data.

# 10) Nested CV — unbiased estimate of tuned model performance
# Outer loop = evaluate; inner loop = tune
outer = StratifiedKFold(5, shuffle=True, random_state=42)
inner = StratifiedKFold(3, shuffle=True, random_state=43)

estimator = GridSearchCV(
    pipe, param_grid, cv=inner, scoring='accuracy', n_jobs=-1,
)
scores = cross_val_score(estimator, X, y, cv=outer, scoring='accuracy', n_jobs=-1)
print(scores.mean(), scores.std())
# This is the FAIR estimate; grid.best_score_ alone is optimistic.

# 11) Stratification for regression — by binned target
from sklearn.model_selection import StratifiedKFold
bins = pd.qcut(y, q=10, labels=False)
for train_idx, test_idx in StratifiedKFold(5, shuffle=True, random_state=42).split(X, bins):
    # Each fold has similar y distribution
    pass

# 12) When NOT to use K-fold
#   • Tiny dataset (< 100 samples) → use leave-one-out (LeaveOneOut())
#   • Time series → TimeSeriesSplit
#   • Grouped / repeated measurements → GroupKFold
#   • Highly imbalanced → StratifiedKFold + custom split logic

# 13) Scoring options worth knowing
# Classification: 'accuracy', 'f1', 'f1_macro', 'roc_auc', 'roc_auc_ovr', 'average_precision', 'neg_log_loss'
# Regression:    'neg_mean_absolute_error', 'neg_mean_squared_error', 'r2'
# Multi-output / multilabel: append '_samples' / '_weighted'

# Custom scorer
from sklearn.metrics import make_scorer
def profit(y_true, y_pred):
    return (y_pred * y_true).sum() - 0.1 * (y_pred * (1 - y_true)).sum()

cross_val_score(model, X, y, scoring=make_scorer(profit), cv=5)

# 14) Best practices
#   • Always shuffle for I.I.D. data; never for time series
#   • Use Pipelines — preprocessing inside, never outside CV
#   • Stratify for classification (especially imbalanced)
#   • Hold a separate test set after all CV / tuning — never touch until final
#   • cv=5 is the practical default; cv=10 for small / noisy data
#   • Set random_state for reproducible splits

# 15) Common pitfalls
#   • Tuning + reporting on the same CV → optimistic score (use nested CV)
#   • Scaling outside the pipeline → test leakage
#   • K-fold on time series → future-into-past leakage
#   • Stratification by class only, ignoring groups → person appears in both folds

from sklearn.model_selection import cross_val_score
scores = cross_val_score(pipe, X, y, cv=5, scoring='f1_macro')
print('mean F1:', scores.mean(), '±', scores.std())