Hands-on lab — Day 161: A Complete Classification Project

"""
Complete Classification Project reference library.
"""
import numpy as np
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.neighbors import KNeighborsClassifier
from sklearn.naive_bayes import GaussianNB
from sklearn.model_selection import StratifiedKFold
from sklearn.metrics import (
    precision_score, recall_score, f1_score, roc_auc_score,
    average_precision_score, matthews_corrcoef, confusion_matrix
)


class ClassificationProjectPipeline:
    """
    End-to-end disciplined classification project pipeline:
    1. Feature Standardization
    2. Stratified 5-Fold CV model benchmark (LogisticRegression, KNN, GaussianNB)
    3. Asymmetric cost threshold optimization
    4. Gated single test-set evaluation
    """
    def __init__(self, random_state: int = 42):
        self.random_state = random_state
        self.scaler = StandardScaler()
        self.candidates = {
            "logistic_l2": LogisticRegression(C=1.0, class_weight="balanced", random_state=random_state, max_iter=1000),
            "logistic_l1": LogisticRegression(C=0.5, penalty="l1", solver="liblinear", class_weight="balanced", random_state=random_state),
            "knn_5": KNeighborsClassifier(n_neighbors=5, weights="distance"),
            "gaussian_nb": GaussianNB(),
        }
        self.cv_results = {}
        self.best_model_name = None
        self.best_model = None
        self.optimal_threshold = 0.50
        self.test_evaluated = False

    def fit_and_select(self, X_train: np.ndarray, y_train: np.ndarray, cv_folds: int = 5) -> dict[str, float]:
        X_train_scaled = self.scaler.fit_transform(X_train)
        skf = StratifiedKFold(n_splits=cv_folds, shuffle=True, random_state=self.random_state)
        
        scores = {}
        for name, model in self.candidates.items():
            f1_list = []
            for train_idx, val_idx in skf.split(X_train_scaled, y_train):
                X_tr, X_val = X_train_scaled[train_idx], X_train_scaled[val_idx]
                y_tr, y_val = y_train[train_idx], y_train[val_idx]
                
                model.fit(X_tr, y_tr)
                preds = model.predict(X_val)
                f1_list.append(f1_score(y_val, preds, zero_division=0))
                
            scores[name] = float(np.mean(f1_list))
            
        self.cv_results = scores
        self.best_model_name = max(scores, key=scores.get)
        self.best_model = self.candidates[self.best_model_name]
        # Fit winner on full training set
        self.best_model.fit(X_train_scaled, y_train)
        return scores

    def calibrate_threshold(self, X_val: np.ndarray, y_val: np.ndarray, cost_fp: float = 10.0, cost_fn: float = 100.0) -> float:
        X_val_scaled = self.scaler.transform(X_val)
        probs = self.best_model.predict_proba(X_val_scaled)[:, 1]
        
        thresholds = np.linspace(0.01, 0.99, 100)
        best_tau = 0.50
        min_cost = float("inf")
        
        for tau in thresholds:
            preds = (probs >= tau).astype(int)
            fp = np.sum((y_val == 0) & (preds == 1))
            fn = np.sum((y_val == 1) & (preds == 0))
            cost = float(cost_fp * fp + cost_fn * fn)
            if cost < min_cost:
                min_cost = cost
                best_tau = float(tau)
                
        self.optimal_threshold = best_tau
        return best_tau

    def evaluate_test(self, X_test: np.ndarray, y_test: np.ndarray) -> dict[str, float]:
        if self.test_evaluated:
            raise RuntimeError("Test set evaluation can only be executed ONCE to prevent leakage!")
        self.test_evaluated = True
        
        X_test_scaled = self.scaler.transform(X_test)
        probs = self.best_model.predict_proba(X_test_scaled)[:, 1]
        preds = (probs >= self.optimal_threshold).astype(int)
        
        cm = confusion_matrix(y_test, preds)
        return {
            "accuracy": float(np.mean(preds == y_test)),
            "precision": float(precision_score(y_test, preds, zero_division=0)),
            "recall": float(recall_score(y_test, preds, zero_division=0)),
            "f1": float(f1_score(y_test, preds, zero_division=0)),
            "mcc": float(matthews_corrcoef(y_test, preds)),
            "roc_auc": float(roc_auc_score(y_test, probs)),
            "pr_auc": float(average_precision_score(y_test, probs)),
            "tn": int(cm[0, 0]),
            "fp": int(cm[0, 1]),
            "fn": int(cm[1, 0]),
            "tp": int(cm[1, 1]),
        }

"""
Tests for reference Complete Classification Project implementation.
"""
import pytest
import numpy as np
from sklearn.datasets import load_breast_cancer
from sklearn.model_selection import train_test_split
import project_lib as prj


def test_complete_pipeline_flow():
    cancer = load_breast_cancer()
    X, y = cancer.data, cancer.target
    
    # 3-way split: 60% Train, 20% Val, 20% Test
    X_train_val, X_test, y_train_val, y_test = train_test_split(X, y, test_size=0.20, stratify=y, random_state=42)
    X_train, X_val, y_train, y_val = train_test_split(X_train_val, y_train_val, test_size=0.25, stratify=y_train_val, random_state=42)
    
    pipe = prj.ClassificationProjectPipeline(random_state=42)
    cv_scores = pipe.fit_and_select(X_train, y_train, cv_folds=5)
    
    assert len(cv_scores) == 4
    assert pipe.best_model is not None
    
    # Calibrate threshold on Val
    tau = pipe.calibrate_threshold(X_val, y_val, cost_fp=10.0, cost_fn=100.0)
    assert 0.0 < tau < 1.0
    
    # Single Test Evaluation
    metrics = pipe.evaluate_test(X_test, y_test)
    assert metrics["f1"] >= 0.90
    assert metrics["roc_auc"] >= 0.95
    
    # Assert second test evaluation raises error
    with pytest.raises(RuntimeError):
        pipe.evaluate_test(X_test, y_test)

lesson_id: D161
day: 161
kind: end-to-end-project
languages:
  - python
setup_commands:
  - python3 -m venv .venv
  - .venv/bin/pip install -r requirements/requirements.txt
run_commands:
  - .venv/bin/python examples/project_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: 60
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 -> 1 passed. pytest starter -v -> 1 passed.
  Executed full 8-phase project lifecycle: stratified splitting, multi-model CV benchmark, threshold calibration, and gated single test evaluation.

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

"""
Complete Classification Project starter library.
"""
import numpy as np


class ClassificationProjectPipeline:
    def __init__(self, random_state: int = 42):
        self.random_state = random_state
        self.scaler = None
        self.best_model = None
        self.optimal_threshold = 0.50

    def fit_and_select(self, X_train: np.ndarray, y_train: np.ndarray) -> dict[str, float]:
        """Fit candidate models via Stratified 5-Fold CV and select top performer."""
        raise NotImplementedError("Implement fit_and_select")

    def calibrate_threshold(self, X_val: np.ndarray, y_val: np.ndarray, cost_fp: float = 1.0, cost_fn: float = 5.0) -> float:
        """Find optimal decision threshold minimizing asymmetric cost on validation split."""
        raise NotImplementedError("Implement calibrate_threshold")

    def evaluate_test(self, X_test: np.ndarray, y_test: np.ndarray) -> dict[str, float]:
        """Perform ONE final evaluation on held-out test data."""
        raise NotImplementedError("Implement evaluate_test")

"""
Tests for starter Complete Classification Project.
"""
import pytest
import numpy as np
import project_lib as prj


def test_pipeline_stub():
    pipe = prj.ClassificationProjectPipeline()
    with pytest.raises(NotImplementedError):
        pipe.fit_and_select(np.zeros((10, 2)), np.array([0, 1]*5))

#!/usr/bin/env bash
# Day 161 lab harness: "A Complete Classification Project"
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
from sklearn.datasets import load_breast_cancer
from sklearn.model_selection import train_test_split
import project_lib as prj

cancer = load_breast_cancer()
X_tr, X_te, y_tr, y_te = train_test_split(cancer.data, cancer.target, test_size=0.2, stratify=cancer.target, random_state=42)
pipe = prj.ClassificationProjectPipeline(random_state=42)
pipe.fit_and_select(X_tr, y_tr)
pipe.calibrate_threshold(X_tr[:50], y_tr[:50])
m = pipe.evaluate_test(X_te, y_te)

# Total test count matches confusion matrix sum
assert m["tn"] + m["fp"] + m["fn"] + m["tp"] == len(y_te)
print("MATH_OK")
PYEOF
)

if [ "$MATH_CHECK" = "MATH_OK" ]; then
  ok "Classification pipeline confusion matrix sum conservation 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