Train / validation / test and leakage

import pandas as pd
from sklearn.model_selection import train_test_split

df = pd.read_csv(DATA / "uy_ijara.csv")
X = df.drop(columns=["ijaraga_berildi_30k", "narx_usd", "narx_matn", "elon_id", "agent_komissiyasi"])
y = df["ijaraga_berildi_30k"]

X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.3, random_state=42, stratify=y)
print(f"total : {len(X)} rows")
print(f"train : {len(X_tr)} rows  ({len(X_tr)/len(X):.0%})  -> the model learns HERE")
print(f"test  : {len(X_te)} rows  ({len(X_te)/len(X):.0%})  -> the model NEVER SEES this")

import numpy as np

print(f"{'random_state':>13} {'no stratify':>14} {'stratify':>14}")
print("-" * 45)
for rs in [0, 1, 42, 7, 99]:
    _, _, _, y1 = train_test_split(X, y, test_size=0.3, random_state=rs)
    _, _, _, y2 = train_test_split(X, y, test_size=0.3, random_state=rs, stratify=y)
    print(f"{rs:>13} {y1.mean():>13.1%} {y2.mean():>14.1%}")
print()
print(f"True ratio: {y.mean():.1%}")

from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.impute import SimpleImputer
from sklearn.metrics import accuracy_score

num = ["xonalar", "maydon_m2", "qavat", "qavatlar_soni", "metro_km", "mebel", "qurilgan_yil"]

# ===== WRONG: the scaler learns from ALL the data =====
imp_bad = SimpleImputer(strategy="median").fit(X[num])              # <-- all of X
scale_bad = StandardScaler().fit(imp_bad.transform(X[num]))         # <-- all of X
Xs = scale_bad.transform(imp_bad.transform(X[num]))
Xs = pd.DataFrame(Xs, columns=num, index=X.index)
m_bad = LogisticRegression(max_iter=2000).fit(Xs.loc[X_tr.index], y_tr)
acc_bad = accuracy_score(y_te, m_bad.predict(Xs.loc[X_te.index]))

# ===== RIGHT: the scaler learns from train ONLY =====
imp = SimpleImputer(strategy="median").fit(X_tr[num])
scaler = StandardScaler().fit(imp.transform(X_tr[num]))
m_good = LogisticRegression(max_iter=2000).fit(scaler.transform(imp.transform(X_tr[num])), y_tr)
acc_good = accuracy_score(y_te, m_good.predict(scaler.transform(imp.transform(X_te[num]))))

print(f"WRONG (scaler on all the data) : {acc_bad:.4f}")
print(f"RIGHT (scaler on train only)   : {acc_good:.4f}")
print(f"Difference                     : {acc_bad - acc_good:+.4f}")

show(pd.DataFrame([
    {"Step": "StandardScaler", "What it 'sees'": "Mean and standard deviation",
     "Damage": "Small, barely visible in this example"},
    {"Step": "SimpleImputer(median)", "What it 'sees'": "The median of the test rows",
     "Damage": "Small to medium"},
    {"Step": "Feature selection (on all data)", "What it 'sees'": "Which columns relate to the target",
     "Damage": "LARGE, the most common disaster"},
    {"Step": "Target encoding (on all data)", "What it 'sees'": "The target mean of each category",
     "Damage": "FATAL: gives 0.99, collapses on the server"},
    {"Step": "SMOTE / oversampling (before the split)", "What it 'sees'": "Copies of test rows land in train",
     "Damage": "FATAL"},
]))

from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import OneHotEncoder

pipe = Pipeline([
    ("prep", ColumnTransformer([
        ("num", Pipeline([("impute", SimpleImputer(strategy="median")),
                          ("scale", StandardScaler())]), num),
        ("cat", OneHotEncoder(handle_unknown="ignore"), ["tuman"]),
    ])),
    ("model", LogisticRegression(max_iter=2000)),
])

pipe.fit(X_tr, y_tr)
print(f"Pipeline result: {accuracy_score(y_te, pipe.predict(X_te)):.4f}")
print()
print("Why this is safe:")
print("  pipe.fit(X_tr, y_tr)  -> every step INSIDE sees ONLY X_tr")
print("  pipe.predict(X_te)    -> transform only, NO refitting")

show(pd.DataFrame([
    {"Set": "train", "Share": "~60%", "What for": "The model learns here",
     "How many times you use it": "Unlimited"},
    {"Set": "validation", "Share": "~20%", "What for": "CHOOSING models and parameters",
     "How many times you use it": "Unlimited, decisions are made here"},
    {"Set": "test", "Share": "~20%", "What for": "The FINAL score",
     "How many times you use it": "ONCE, at the very end"},
]))

# Split in two stages: first set the test aside, then split what is left
X_rest, X_test, y_rest, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y)
X_train, X_val, y_train, y_val = train_test_split(
    X_rest, y_rest, test_size=0.25, random_state=42, stratify=y_rest)

print(f"train      : {len(X_train):4d} rows  ({len(X_train)/len(X):.0%})")
print(f"validation : {len(X_val):4d} rows  ({len(X_val)/len(X):.0%})")
print(f"test       : {len(X_test):4d} rows  ({len(X_test)/len(X):.0%})  <-- IN THE SAFE")

days = pd.date_range("2026-01-01", periods=10, freq="D")
print("RANDOM split (WRONG):")
print("  train: days 1, 3, 4, 7, 9, 10")
print("  test : days 2, 5, 6, 8         <-- train has seen the future")
print()
print("Split BY TIME (RIGHT):")
print("  train: days 1-7")
print("  test : days 8-10               <-- learns only from the past")
print()
print("In sklearn: TimeSeriesSplit (Chapter 10)")

show(pd.DataFrame([
    {"Rule": "Split first", "Detail": "BEFORE cleaning, before charts, before everything"},
    {"Rule": "stratify=y", "Detail": "Mandatory in classification"},
    {"Rule": "fit on train only", "Detail": "Scaler, imputer, encoder, selector: all of them"},
    {"Rule": "Use a Pipeline", "Detail": "Don't rely on memory, enforce the rule by structure"},
    {"Rule": "Test once", "Detail": "Decisions are made on validation or in CV"},
    {"Rule": "No shuffling when there is time", "Detail": "TimeSeriesSplit"},
]))