Hands-on lab — Day 173: scikit-learn Pipelines

"""
Scikit-Learn Pipelines reference library implementation.
"""
import numpy as np
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder


class CustomLogTransformer(BaseEstimator, TransformerMixin):
    """
    Log1p feature transformer: z = log(max(x, 0) + offset)
    Compatible with scikit-learn clone() and get_params().
    """
    def __init__(self, offset: float = 1.0):
        self.offset = offset

    def fit(self, X, y=None):
        # Stateless transformer validation
        X = np.asarray(X, dtype=float)
        return self

    def transform(self, X):
        X = np.asarray(X, dtype=float)
        clipped = np.maximum(X, 0.0)
        return np.log(clipped + self.offset)


class OutlierClipperTransformer(BaseEstimator, TransformerMixin):
    """
    Fits empirical percentile thresholds on training data:
    lower_bound = percentile(X, lower_p)
    upper_bound = percentile(X, upper_p)
    Clips incoming data to [lower_bound, upper_bound] to guard against extreme anomalies.
    """
    def __init__(self, lower_percentile: float = 1.0, upper_percentile: float = 99.0):
        self.lower_percentile = lower_percentile
        self.upper_percentile = upper_percentile
        self.lower_bounds_ = None
        self.upper_bounds_ = None

    def fit(self, X, y=None):
        X = np.asarray(X, dtype=float)
        self.lower_bounds_ = np.percentile(X, self.lower_percentile, axis=0)
        self.upper_bounds_ = np.percentile(X, self.upper_percentile, axis=0)
        return self

    def transform(self, X):
        X = np.asarray(X, dtype=float)
        if self.lower_bounds_ is None or self.upper_bounds_ is None:
            raise RuntimeError("Transformer has not been fitted yet!")
        return np.clip(X, self.lower_bounds_, self.upper_bounds_)


def build_heterogeneous_tabular_pipeline(num_indices: list[int], cat_indices: list[int], estimator) -> Pipeline:
    """
    Constructs a complete leak-free heterogeneous preprocessing and modeling Pipeline:
    - Numerical Branch: OutlierClipper -> StandardScaler
    - Categorical Branch: OneHotEncoder(handle_unknown='ignore')
    - Final Step: Estimator
    """
    num_pipeline = Pipeline([
        ("clipper", OutlierClipperTransformer(lower_percentile=1.0, upper_percentile=99.0)),
        ("scaler", StandardScaler())
    ])
    
    cat_pipeline = Pipeline([
        ("ohe", OneHotEncoder(handle_unknown="ignore", sparse_output=False))
    ])
    
    preprocessor = ColumnTransformer(
        transformers=[
            ("num", num_pipeline, num_indices),
            ("cat", cat_pipeline, cat_indices)
        ],
        remainder="drop"
    )
    
    full_pipeline = Pipeline([
        ("preprocessor", preprocessor),
        ("estimator", estimator)
    ])
    
    return full_pipeline

"""
Tests for reference pipeline implementation.
"""
import pytest
import numpy as np
from sklearn.base import clone
from sklearn.linear_model import Ridge
import pipeline_lib as pl


def test_log_transformer_math():
    X = np.array([[0.0], [np.e - 1.0], [np.e**2 - 1.0]])
    tr = pl.CustomLogTransformer(offset=1.0)
    X_log = tr.fit_transform(X)
    
    assert np.isclose(X_log[0, 0], 0.0)
    assert np.isclose(X_log[1, 0], 1.0)
    assert np.isclose(X_log[2, 0], 2.0)


def test_outlier_clipper_bounds_and_clone():
    X_train = np.linspace(0, 100, 101).reshape(-1, 1)
    clipper = pl.OutlierClipperTransformer(lower_percentile=5.0, upper_percentile=95.0)
    
    # Test clone compatibility
    cloned = clone(clipper)
    assert cloned.lower_percentile == 5.0
    assert cloned.upper_percentile == 95.0
    
    clipper.fit(X_train)
    # Test data with extreme outliers below 0 and above 100
    X_test = np.array([[-500.0], [50.0], [5000.0]])
    X_clipped = clipper.transform(X_test)
    
    assert np.isclose(X_clipped[0, 0], 5.0)   # 5th percentile of 0..100 is 5.0
    assert np.isclose(X_clipped[1, 0], 50.0)
    assert np.isclose(X_clipped[2, 0], 95.0)  # 95th percentile is 95.0


def test_heterogeneous_pipeline_end_to_end():
    # 2 numerical columns, 1 categorical column
    X_num = np.random.randn(100, 2) * 50.0
    X_cat = np.random.choice(["Red", "Green", "Blue"], size=(100, 1))
    X_mixed = np.hstack([X_num, X_cat])
    
    y = X_num[:, 0] * 2.0 + (X_cat[:, 0] == "Red").astype(float) * 10.0
    
    pipeline = pl.build_heterogeneous_tabular_pipeline(
        num_indices=[0, 1], cat_indices=[2], estimator=Ridge(alpha=1.0)
    )
    
    pipeline.fit(X_mixed, y)
    preds = pipeline.predict(X_mixed)
    assert len(preds) == 100
    assert preds.ndim == 1

lesson_id: D173
day: 173
kind: production-pipeline-engineering
languages:
  - python
setup_commands:
  - python3 -m venv .venv
  - .venv/bin/pip install -r requirements/requirements.txt
run_commands:
  - .venv/bin/python examples/pipeline_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 -> 3 passed. pytest starter -v -> 2 passed.
  Verified custom BaseEstimator transformers, ColumnTransformer branching, and leak-free end-to-end Pipelines.

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

"""
Scikit-Learn Pipelines starter library.
"""
import numpy as np
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer


class CustomLogTransformer(BaseEstimator, TransformerMixin):
    """Stateless or offset log transformer: log(x + offset)."""
    def __init__(self, offset: float = 1.0):
        self.offset = offset

    def fit(self, X, y=None):
        raise NotImplementedError("Implement fit")

    def transform(self, X):
        raise NotImplementedError("Implement transform")


class OutlierClipperTransformer(BaseEstimator, TransformerMixin):
    """Clips numerical features to empirical percentile bounds [lower_p, upper_p]."""
    def __init__(self, lower_percentile: float = 1.0, upper_percentile: float = 99.0):
        self.lower_percentile = lower_percentile
        self.upper_percentile = upper_percentile
        self.lower_bounds_ = None
        self.upper_bounds_ = None

    def fit(self, X, y=None):
        raise NotImplementedError("Implement fit")

    def transform(self, X):
        raise NotImplementedError("Implement transform")


def build_heterogeneous_tabular_pipeline(num_indices: list[int], cat_indices: list[int], estimator) -> Pipeline:
    """Build end-to-end ColumnTransformer and Estimator Pipeline."""
    raise NotImplementedError("Implement build_heterogeneous_tabular_pipeline")

"""
Tests for starter scikit-learn pipelines.
"""
import pytest
import numpy as np
import pipeline_lib as pl


def test_log_transformer_stub():
    tr = pl.CustomLogTransformer()
    with pytest.raises(NotImplementedError):
        tr.fit(np.ones((5, 2)))


def test_outlier_clipper_stub():
    tr = pl.OutlierClipperTransformer()
    with pytest.raises(NotImplementedError):
        tr.fit(np.ones((5, 2)))

#!/usr/bin/env bash
# Day 173 lab harness: "Scikit-Learn Pipelines"
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 pipeline_lib as pl

# Outlier clipping bounded invariant
X = np.array([[-1000.0], [10.0], [20.0], [30.0], [1000.0]])
clipper = pl.OutlierClipperTransformer(lower_percentile=20.0, upper_percentile=80.0)
X_c = clipper.fit_transform(X)
assert np.all(X_c >= clipper.lower_bounds_), "Clipped data must be >= lower percentile"
assert np.all(X_c <= clipper.upper_bounds_), "Clipped data must be <= upper percentile"

print("MATH_OK")
PYEOF
)

if [ "$MATH_CHECK" = "MATH_OK" ]; then
  ok "Custom transformer clipping bounds invariant 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