Scikit-Learn Pipeline Explained

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_score
import numpy as np

# io.thecodeforge: Atomic Pipeline Implementation
def build_forge_pipeline():
    """Build a pipeline that prevents data leakage during CV.

    Key contract:
    - Intermediate steps: must implement fit() and transform()
    - Final step: must implement fit() and predict()
    - Pipeline.fit() calls fit_transform() on intermediates, fit() on final
    - Pipeline.predict() calls transform() on intermediates, predict() on final

    What this guarantees:
    - No transformer ever sees test-fold data during fit
    - The same transformation graph is applied identically at inference time
    - Serializing the Pipeline serializes all learned parameters atomically
    """
    steps = [
        # Step 0: Handle missing values using training-fold median only
        ('imputer', SimpleImputer(strategy='median')),
        # Step 1: Standardize using training-fold mean and std only
        ('scaler', StandardScaler()),
        # Step 2: Final estimator — only step that receives y during fit
        ('classifier', LogisticRegression(max_iter=1000, random_state=42))
    ]

    pipeline = Pipeline(
        steps=steps,
        # memory='/tmp/forge_cache'  # Uncomment for GridSearchCV caching
        verbose=False
    )
    return pipeline


def demonstrate_leakage_prevention(X, y):
    """Show explicitly that the scaler never sees test data."""
    from sklearn.model_selection import StratifiedKFold

    pipeline = build_forge_pipeline()
    skf = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)

    fold_scaler_means = []

    for fold_idx, (train_idx, test_idx) in enumerate(skf.split(X, y)):
        X_train_fold, X_test_fold = X[train_idx], X[test_idx]
        y_train_fold = y[train_idx]

        # fit only touches train_idx — test_idx data is never seen by transformers
        pipeline.fit(X_train_fold, y_train_fold)

        # Inspect the scaler's learned mean — computed from train fold only
        fold_mean = pipeline.named_steps['scaler'].mean_
        fold_scaler_means.append(fold_mean)
        print(f"Fold {fold_idx + 1} scaler mean[0]: {fold_mean[0]:.6f}")

    # Different folds produce different scaler parameters — proof of no leakage
    # If all folds produced identical means, you would have a leakage problem
    means_array = np.array([m[0] for m in fold_scaler_means])
    print(f"\nScaler mean variance across folds: {means_array.var():.6f}")
    print("Non-zero variance confirms each fold trained on different data.")


# Correct usage: cross_val_score handles CV splits internally
# Each fold: fit_transform imputer+scaler on train, fit classifier on train
#            transform imputer+scaler on test, predict classifier on test
# forge_pipe = build_forge_pipeline()
# scores = cross_val_score(forge_pipe, X, y, cv=5, scoring='accuracy')
# print(f"Mean CV accuracy: {scores.mean():.4f} (+/- {scores.std():.4f})")

# io.thecodeforge: Standardized ML Inference Image
# Built for reproducibility — same artifact, same behavior, every environment
FROM python:3.11-slim

WORKDIR /app

# libgomp1 is required for joblib's parallel processing in scikit-learn
# Without it, certain operations silently fall back to single-threaded execution
RUN apt-get update && apt-get install -y \
    libgomp1 \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# The Pipeline artifact is the deployment unit — not the model file alone
# This single file contains all preprocessing parameters AND model weights
COPY forge_pipeline.joblib .
COPY serve.py .

# Health check ensures the Pipeline loads cleanly before accepting traffic
# Catches corrupted artifacts at startup rather than at first prediction
HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
    CMD python -c "import joblib; p = joblib.load('forge_pipeline.joblib'); print('Pipeline healthy:', type(p).__name__)"

EXPOSE 8080
CMD ["python", "serve.py"]

-- io.thecodeforge: Pipeline version registry
-- Tracks every trained Pipeline: configuration, performance, artifact location
-- The training_data_hash is your drift detection signal

-- Create the registry table if it does not exist
CREATE TABLE IF NOT EXISTS io_thecodeforge.pipeline_metadata (
    pipeline_id         VARCHAR(128) PRIMARY KEY,
    steps_config        JSONB          NOT NULL,
    training_accuracy   DECIMAL(6, 4)  NOT NULL,
    cv_accuracy_mean    DECIMAL(6, 4),
    cv_accuracy_std     DECIMAL(6, 4),
    artifact_path       TEXT           NOT NULL,
    training_data_hash  VARCHAR(64),   -- SHA-256 of training feature distribution
    git_commit_hash     VARCHAR(40),   -- Links model to exact code version
    created_at          TIMESTAMPTZ    NOT NULL DEFAULT NOW(),
    deployed_at         TIMESTAMPTZ,
    retired_at          TIMESTAMPTZ
);

-- Register a newly trained Pipeline version
INSERT INTO io_thecodeforge.pipeline_metadata (
    pipeline_id,
    steps_config,
    training_accuracy,
    cv_accuracy_mean,
    cv_accuracy_std,
    artifact_path,
    training_data_hash,
    git_commit_hash
) VALUES (
    'customer_churn_v2',
    '["imputer", "scaler", "random_forest"]'::JSONB,
    0.8942,
    0.8817,
    0.0124,
    's3://thecodeforge-artifacts/pipelines/churn_v2.joblib',
    'a3f8c2d1e4b7f9a2c5d8e1f4a7b0c3d6e9f2a5b8c1d4e7f0a3b6c9d2e5f8a1b4',
    'f3a8c2d1'
);

-- Query to detect model drift: find pipelines where live accuracy
-- dropped more than 5 points below CV accuracy
SELECT
    pipeline_id,
    cv_accuracy_mean,
    training_accuracy,
    (cv_accuracy_mean - training_accuracy) AS accuracy_gap,
    deployed_at
FROM io_thecodeforge.pipeline_metadata
WHERE deployed_at IS NOT NULL
  AND retired_at IS NULL
ORDER BY deployed_at DESC;

# io.thecodeforge: Common Pipeline mistakes and their correct counterparts
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
import numpy as np


def inspect_forge_pipeline(pipeline):
    """Correctly inspect Pipeline step attributes after fitting.

    Pattern: always fit before inspecting learned attributes.
    Use get_params() for pre-fit configuration inspection.
    """
    try:
        # Accessing the scaler's calculated mean after fitting
        means = pipeline.named_steps['scaler'].mean_
        stds = pipeline.named_steps['scaler'].scale_
        print(f"Scaler learned from training data:")
        print(f"  Feature means: {means.round(4)}")
        print(f"  Feature stds:  {stds.round(4)}")

        # Access classifier coefficients
        coefs = pipeline.named_steps['classifier'].coef_
        print(f"  Classifier coef shape: {coefs.shape}")

    except AttributeError as e:
        print(f"AttributeError: {e}")
        print("Pipeline must be fitted before inspecting learned attributes.")
        print("For pre-fit inspection, use: pipeline.get_params()")


# ============================================================
# WRONG: Manual preprocessing outside Pipeline (data leakage)
# ============================================================
# scaler = StandardScaler()
# X_scaled = scaler.fit_transform(X)          # Scaler sees ALL data including test
# X_train, X_test = train_test_split(X_scaled)# Split AFTER scaling — leakage locked in
# model = LogisticRegression()
# model.fit(X_train, y_train)                  # Trains on leaked data
# score = model.score(X_test, y_test)          # Inflated score

# ============================================================
# WRONG: Estimator as intermediate step (crashes during fit)
# ============================================================
# broken_pipeline = Pipeline([
#     ('scaler', StandardScaler()),
#     ('intermediate_model', LogisticRegression()),  # No .transform() method
#     ('final_model', LogisticRegression())           # Pipeline cannot pass data forward
# ])
# broken_pipeline.fit(X_train, y_train)  # Raises: AttributeError: no attribute 'transform'

# ============================================================
# WRONG: Inspecting attributes before fit
# ============================================================
# pipeline = Pipeline([('scaler', StandardScaler()), ('clf', LogisticRegression())])
# means = pipeline.named_steps['scaler'].mean_  # AttributeError: mean_ does not exist yet

# ============================================================
# RIGHT: Pipeline handles everything atomically
# ============================================================
def correct_pipeline_usage(X, y):
    """The correct pattern: split raw data, then fit the full Pipeline."""
    # Split raw, untransformed data first
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42, stratify=y
    )

    pipeline = Pipeline([
        ('scaler', StandardScaler()),
        ('classifier', LogisticRegression(max_iter=1000, random_state=42))
    ])

    # Pipeline.fit: calls scaler.fit_transform(X_train), then clf.fit(X_train_scaled, y_train)
    # The scaler NEVER sees X_test at this point
    pipeline.fit(X_train, y_train)

    # Now safe to inspect learned attributes
    inspect_forge_pipeline(pipeline)

    # Pipeline.score: calls scaler.transform(X_test) then clf.score(X_test_scaled, y_test)
    # Scaler uses parameters learned from X_train only
    test_score = pipeline.score(X_test, y_test)
    print(f"\nTest accuracy (no leakage): {test_score:.4f}")
    return pipeline