Hands-on lab — Day 167: Cross-Validation Done Right

"""
Cross-Validation reference library implementation.
"""
import itertools
import numpy as np
from typing import Generator
from sklearn.metrics import accuracy_score


def stratified_kfold_scratch(
    y: np.ndarray, n_splits: int = 5, shuffle: bool = True, random_state: int = 42
) -> Generator[tuple[np.ndarray, np.ndarray], None, None]:
    """
    Generate stratified folds where each fold preserves the empirical class ratio.
    """
    y = np.asarray(y)
    n_samples = len(y)
    classes, counts = np.unique(y, return_counts=True)
    
    rng = np.random.default_rng(random_state)
    class_indices = {}
    for c in classes:
        idx = np.where(y == c)[0]
        if shuffle:
            rng.shuffle(idx)
        class_indices[c] = idx
        
    # Split each class indices into n_splits chunks
    folds = [[] for _ in range(n_splits)]
    for c in classes:
        splits = np.array_split(class_indices[c], n_splits)
        for fold_idx in range(n_splits):
            folds[fold_idx].extend(splits[fold_idx])
            
    for val_fold_idx in range(n_splits):
        val_idx = np.array(folds[val_fold_idx], dtype=int)
        train_idx = np.setdiff1d(np.arange(n_samples), val_idx)
        yield train_idx, val_idx


def group_kfold_scratch(
    groups: np.ndarray, n_splits: int = 3
) -> Generator[tuple[np.ndarray, np.ndarray], None, None]:
    """
    Generate disjoint group folds such that no group appears in both train and validation.
    """
    groups = np.asarray(groups)
    n_samples = len(groups)
    unique_groups = np.unique(groups)
    n_groups = len(unique_groups)
    
    if n_splits > n_groups:
        raise ValueError(f"Cannot have n_splits={n_splits} greater than unique groups={n_groups}")
        
    group_splits = np.array_split(unique_groups, n_splits)
    
    for split in group_splits:
        val_mask = np.isin(groups, split)
        val_idx = np.where(val_mask)[0]
        train_idx = np.where(~val_mask)[0]
        yield train_idx, val_idx


def time_series_split_scratch(
    n_samples: int, n_splits: int = 4, min_train_size: int = None
) -> Generator[tuple[np.ndarray, np.ndarray], None, None]:
    """
    Generate expanding-window temporal folds where training strictly precedes validation.
    """
    test_size = n_samples // (n_splits + 1)
    if min_train_size is None:
        min_train_size = n_samples - test_size * n_splits
        
    for i in range(n_splits):
        train_end = min_train_size + i * test_size
        val_end = train_end + test_size
        train_idx = np.arange(0, train_end)
        val_idx = np.arange(train_end, val_end)
        yield train_idx, val_idx


def nested_cross_validation_score(
    estimator_cls, param_grid: dict, X: np.ndarray, y: np.ndarray, outer_k: int = 3, inner_k: int = 3
) -> float:
    """
    Unbiased double cross-validation:
    Outer loop measures true generalization; Inner loop selects optimal hyperparameters.
    """
    X = np.asarray(X)
    y = np.asarray(y)
    
    keys = list(param_grid.keys())
    values = list(param_grid.values())
    combinations = [dict(zip(keys, prod)) for prod in itertools.product(*values)]
    
    outer_scores = []
    
    # Outer Loop: Generalization evaluation
    for outer_tr, outer_va in stratified_kfold_scratch(y, n_splits=outer_k, shuffle=True, random_state=42):
        X_out_tr, y_out_tr = X[outer_tr], y[outer_tr]
        X_out_va, y_out_va = X[outer_va], y[outer_va]
        
        best_inner_score = -float("inf")
        best_params = combinations[0]
        
        # Inner Loop: Hyperparameter selection on outer training data only!
        for params in combinations:
            inner_scores = []
            for inner_tr, inner_va in stratified_kfold_scratch(y_out_tr, n_splits=inner_k, shuffle=True, random_state=100):
                model = estimator_cls(**params)
                model.fit(X_out_tr[inner_tr], y_out_tr[inner_tr])
                inner_preds = model.predict(X_out_tr[inner_va])
                inner_scores.append(accuracy_score(y_out_tr[inner_va], inner_preds))
                
            mean_inner = np.mean(inner_scores)
            if mean_inner > best_inner_score:
                best_inner_score = mean_inner
                best_params = params
                
        # Fit optimal parameters on full outer training split and score on clean outer validation fold
        final_model = estimator_cls(**best_params)
        final_model.fit(X_out_tr, y_out_tr)
        outer_pred = final_model.predict(X_out_va)
        outer_scores.append(accuracy_score(y_out_va, outer_pred))
        
    return float(np.mean(outer_scores))

"""
Tests for reference Cross-Validation implementation.
"""
import pytest
import numpy as np
from sklearn.tree import DecisionTreeClassifier
import cv_lib as cv


def test_stratified_kfold_ratio_preservation():
    # 80 Class 0, 20 Class 1 (80% / 20% ratio)
    y = np.array([0] * 80 + [1] * 20)
    
    for train_idx, val_idx in cv.stratified_kfold_scratch(y, n_splits=4, shuffle=True):
        val_y = y[val_idx]
        p_pos = np.mean(val_y == 1)
        # Ratio must be exactly 0.20 in every fold
        assert np.isclose(p_pos, 0.20, atol=0.05)
        assert len(np.intersect1d(train_idx, val_idx)) == 0


def test_group_kfold_disjoint_guarantee():
    # 4 patients/groups: Group 1 has 10 rows, Group 2 has 10 rows, etc.
    groups = np.repeat([1, 2, 3, 4], 10)
    
    for train_idx, val_idx in cv.group_kfold_scratch(groups, n_splits=4):
        train_groups = set(groups[train_idx])
        val_groups = set(groups[val_idx])
        # Intersection between train groups and val groups MUST be strictly empty!
        assert len(train_groups.intersection(val_groups)) == 0


def test_time_series_split_temporal_order():
    n_samples = 100
    for train_idx, val_idx in cv.time_series_split_scratch(n_samples, n_splits=4):
        # Maximum training index must be strictly less than minimum validation index
        assert np.max(train_idx) < np.min(val_idx)
        # Training indices start at index 0 (expanding window)
        assert train_idx[0] == 0


def test_nested_cross_validation_execution():
    X = np.random.randn(90, 4)
    y = (X[:, 0] + X[:, 1] > 0).astype(int)
    
    grid = {"max_depth": [2, 4], "random_state": [42]}
    score = cv.nested_cross_validation_score(
        DecisionTreeClassifier, grid, X, y, outer_k=3, inner_k=3
    )
    
    assert 0.50 <= score <= 1.0

lesson_id: D167
day: 167
kind: validation-methodology
languages:
  - python
setup_commands:
  - python3 -m venv .venv
  - .venv/bin/pip install -r requirements/requirements.txt
run_commands:
  - .venv/bin/python examples/cv_lib.py
test_commands:
  - ./tests/run_tests.sh
cleanup_commands:
  - rm -rf .venv __pycache__ .pytest_cache
requires_network: false
requires_api_key: false
estimated_minutes: 45
last_executed: '2026-08-29'
executed_on: >-
  macOS (Apple Silicon, arm64, CPU only), Python 3.14.0, numpy 2.5.2,
  scikit-learn 1.9.0, pytest 9.1.1, scipy 1.15.2 -- bash tests/run_tests.sh -> 4 checks,
  0 failure(s), exit 0. pytest examples -v -> 4 passed. pytest starter -v -> 2 passed.
  Verified StratifiedKFold class ratio preservation, GroupKFold entity isolation, TimeSeriesSplit expanding windows, and Nested CV.

numpy==2.5.2
scikit-learn==1.9.0
pytest==9.1.1
scipy==1.15.2

"""
Cross-Validation starter library.
"""
import numpy as np
from typing import Generator


def stratified_kfold_scratch(y: np.ndarray, n_splits: int = 5, shuffle: bool = True, random_state: int = 42) -> Generator[tuple[np.ndarray, np.ndarray], None, None]:
    """Generate train/val indices preserving exact class ratios per fold."""
    raise NotImplementedError("Implement stratified_kfold_scratch")


def group_kfold_scratch(groups: np.ndarray, n_splits: int = 3) -> Generator[tuple[np.ndarray, np.ndarray], None, None]:
    """Generate train/val indices ensuring groups are completely disjoint across folds."""
    raise NotImplementedError("Implement group_kfold_scratch")


def time_series_split_scratch(n_samples: int, n_splits: int = 4) -> Generator[tuple[np.ndarray, np.ndarray], None, None]:
    """Generate expanding-window temporal train/val indices."""
    raise NotImplementedError("Implement time_series_split_scratch")


def nested_cross_validation_score(estimator_cls, param_grid: dict, X: np.ndarray, y: np.ndarray, outer_k: int = 3, inner_k: int = 3) -> float:
    """Compute unbiased generalization score using double (nested) cross-validation."""
    raise NotImplementedError("Implement nested_cross_validation_score")

"""
Tests for starter Cross-Validation.
"""
import pytest
import numpy as np
import cv_lib as cv


def test_stratified_stub():
    with pytest.raises(NotImplementedError):
        list(cv.stratified_kfold_scratch(np.array([0, 1, 0, 1])))


def test_group_stub():
    with pytest.raises(NotImplementedError):
        list(cv.group_kfold_scratch(np.array([1, 1, 2, 2])))

#!/usr/bin/env bash
# Day 167 lab harness: "Cross-Validation Done Right"
set -u

LAB_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$LAB_DIR"

PYTHON="${PYTHON:-../../../../.venv-tools/bin/python3}"
PYTEST="${PYTEST:-../../../../.venv-tools/bin/pytest}"

CHECKS=0
FAILURES=0

ok() {
  CHECKS=$((CHECKS + 1))
  echo "  ok: $1"
}

fail() {
  CHECKS=$((CHECKS + 1))
  FAILURES=$((FAILURES + 1))
  echo "  FAIL: $1"
}

echo "=== 1. Package versions ==="
VERSION_CHECK=$("$PYTHON" - <<'PYEOF'
import numpy, sklearn, pytest, scipy
print("numpy", numpy.__version__)
print("scikit-learn", sklearn.__version__)
print("pytest", pytest.__version__)
print("scipy", scipy.__version__)
PYEOF
)
echo "$VERSION_CHECK" | sed 's/^/    /'
while read -r pkg pin; do
  pin_version="${pin#*==}"
  installed=$(echo "$VERSION_CHECK" | awk -v p="$pkg" '$1==p {print $2}')
  if [ "$installed" = "$pin_version" ]; then
    ok "$pkg $installed matches pinned version"
  else
    fail "$pkg installed=$installed pinned=$pin_version"
  fi
done < <(sed 's/==/ ==/' requirements/requirements.txt)

echo ""
echo "=== 2. Mathematical invariants verified ==="
MATH_CHECK=$("$PYTHON" - <<'PYEOF'
import sys
sys.path.insert(0, "examples")
import numpy as np
import cv_lib as cv

# 1. Stratified K-Fold Disjoint Invariant
y = np.array([0]*60 + [1]*40)
folds = list(cv.stratified_kfold_scratch(y, n_splits=5))
assert len(folds) == 5

# 2. Group K-Fold Zero Leakage Invariant
groups = np.repeat([101, 102, 103, 104, 105], 20)
for tr, va in cv.group_kfold_scratch(groups, n_splits=5):
    assert len(set(groups[tr]).intersection(set(groups[va]))) == 0

# 3. Temporal Precedence Invariant
for tr, va in cv.time_series_split_scratch(100, n_splits=4):
    assert np.max(tr) < np.min(va)

print("MATH_OK")
PYEOF
)

if [ "$MATH_CHECK" = "MATH_OK" ]; then
  ok "Stratified, Group, and Time-Series partition invariants verified"
else
  fail "Mathematical verification failed: $MATH_CHECK"
fi

echo ""
echo "=== 3. Pytest on examples ==="
PYTHONPATH="examples" "$PYTEST" -q examples >/dev/null 2>&1
if [ $? -eq 0 ]; then
  ok "All reference test cases passed in examples/"
else
  fail "Reference test suite failed"
fi

echo ""
echo "=== 4. Pytest on starter ==="
PYTHONPATH="starter" "$PYTEST" -q starter >/dev/null 2>&1
if [ $? -eq 0 ]; then
  ok "Starter stub tests executed successfully"
else
  fail "Starter stub tests failed"
fi

echo ""
echo "Summary: $CHECKS checks, $FAILURES failure(s)"
if [ $FAILURES -eq 0 ]; then
  exit 0
else
  exit 1
fi