S084 · The ML Mental Model — Features, Labels, Train/Val/Test

# ml_scaffold.py — the mental model in ~120 lines of runnable code.
import numpy as np
import pandas as pd
from sklearn.datasets import fetch_california_housing
from sklearn.model_selection import train_test_split, KFold, cross_val_score
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LinearRegression
from sklearn.dummy import DummyRegressor
from sklearn.metrics import mean_squared_error, r2_score
from sklearn.pipeline import Pipeline
 
np.random.seed(42)
 
# --- 1. Load raw data → (X, y) ---
data = fetch_california_housing(as_frame=True)
X: pd.DataFrame = data.frame.drop(columns=["MedHouseVal"])
y: pd.Series = data.frame["MedHouseVal"]
print(f"Shape: X={X.shape}, y={y.shape}")
print(f"Features: {list(X.columns)}")
print(f"Label stats: min={y.min():.2f}, max={y.max():.2f}, mean={y.mean():.2f}")
 
# --- 2. Split BEFORE any fitting or scaling ---
X_trainval, X_test, y_trainval, y_test = train_test_split(
    X, y, test_size=0.15, random_state=42
)
X_train, X_val, y_train, y_val = train_test_split(
    X_trainval, y_trainval, test_size=0.15, random_state=42
)
print(f"Split → train={len(X_train)}, val={len(X_val)}, test={len(X_test)}")
 
# --- 3. Baseline — always start here ---
baseline = DummyRegressor(strategy="mean")
baseline.fit(X_train, y_train)
rmse_baseline = mean_squared_error(y_val, baseline.predict(X_val), squared=False)
print(f"[baseline · predict mean] val RMSE = {rmse_baseline:.3f}")
 
# --- 4. Real model in a pipeline — scaler fit ONLY on train ---
pipe = Pipeline([
    ("scaler", StandardScaler()),   # fit on train, transform val/test
    ("model", LinearRegression()),
])
pipe.fit(X_train, y_train)
rmse_val = mean_squared_error(y_val, pipe.predict(X_val), squared=False)
r2_val = r2_score(y_val, pipe.predict(X_val))
print(f"[linear · pipeline]     val RMSE = {rmse_val:.3f}, R² = {r2_val:.3f}")
 
# --- 5. Cross-validation on train+val (proper hyperparam search) ---
cv_rmse = -cross_val_score(
    pipe, X_trainval, y_trainval,
    scoring="neg_root_mean_squared_error", cv=KFold(5, shuffle=True, random_state=42)
).mean()
print(f"[linear · 5-fold CV]    mean RMSE = {cv_rmse:.3f}")
 
# --- 6. FINAL test-set evaluation · ONE TIME ONLY ---
rmse_test = mean_squared_error(y_test, pipe.predict(X_test), squared=False)
r2_test = r2_score(y_test, pipe.predict(X_test))
print(f"[linear · TEST · one-shot] RMSE = {rmse_test:.3f}, R² = {r2_test:.3f}")
 
# --- 7. DEMO — data leakage via scaling on the whole set (WRONG) ---
print("\n=== LEAKAGE DEMO — do NOT do this ===")
scaler_wrong = StandardScaler().fit(X)          # fit on ALL data (leak!)
X_wrong = scaler_wrong.transform(X)
X_wtrain, X_wtest, y_wtrain, y_wtest = train_test_split(
    X_wrong, y, test_size=0.15, random_state=42
)
m_wrong = LinearRegression().fit(X_wtrain, y_wtrain)
rmse_wrong = mean_squared_error(y_wtest, m_wrong.predict(X_wtest), squared=False)
print(f"[leaky scaler]  test RMSE = {rmse_wrong:.3f}  (looks similar, but the mean/std saw the test set)")
print("On this small dataset the effect is tiny; on real projects the leaky number can be 20-40% too optimistic.")
 
# --- 8. DEMO — target leakage (WRONG) ---
print("\n=== LEAKAGE DEMO 2 — target leakage ===")
X_leaky = X.copy()
X_leaky["cheat_col"] = y + np.random.normal(0, 0.01, size=len(y))  # near-perfect proxy for target
X_lt, X_lte, y_lt, y_lte = train_test_split(X_leaky, y, test_size=0.15, random_state=42)
m_leak = LinearRegression().fit(X_lt, y_lt)
print(f"[target-leak]   test R² = {r2_score(y_lte, m_leak.predict(X_lte)):.4f}  ← if you see 0.99 in your first model, something is wrong")