Ensemble Methods in ML: Bagging, Boosting and Stacking Explained

// TheCodeForge — Ensemble Methods in ML example
// Always use meaningful names, not x or n
public class ForgeExample {
    public static void main(String[] args) {
        String topic = "Ensemble Methods in ML";
        System.out.println("Learning: " + topic + " 🔥");
    }
}

import numpy as np
from sklearn.tree import DecisionTreeClassifier
from sklearn.model_selection import train_test_split
from sklearn.datasets import make_classification

# Simulate bagging manually for understanding
def manual_bagging(X_train, y_train, n_estimators=10, seed=42):
    np.random.seed(seed)
    n_samples = X_train.shape[0]
    models = []
    for i in range(n_estimators):
        # bootstrap sample (with replacement)
        indices = np.random.choice(n_samples, size=n_samples, replace=True)
        X_boot = X_train[indices]
        y_boot = y_train[indices]
        tree = DecisionTreeClassifier(max_depth=5)
        tree.fit(X_boot, y_boot)
        models.append(tree)
    return models

# Predict by majority vote
def bagging_predict(models, X_test):
    predictions = np.array([m.predict(X_test) for m in models])
    # majority vote per sample
    return np.apply_along_axis(lambda x: np.bincount(x).argmax(), axis=0, arr=predictions)

# --- Example usage ---
X, y = make_classification(n_samples=500, n_features=10, random_state=0)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3)
models = manual_bagging(X_train, y_train, n_estimators=50)
y_pred = bagging_predict(models, X_test)
acc = np.mean(y_pred == y_test)
print(f"Bagging accuracy: {acc:.3f}")

import numpy as np
from sklearn.tree import DecisionTreeClassifier

def adaboost(X, y, n_estimators=50):
    n = X.shape[0]
    w = np.ones(n) / n  # initial uniform weights
    models = []
    alphas = []
    for t in range(n_estimators):
        stump = DecisionTreeClassifier(max_depth=1)  # weak learner
        stump.fit(X, y, sample_weight=w)
        y_pred = stump.predict(X)
        err = np.sum(w * (y_pred != y)) / np.sum(w)
        if err > 0.5:
            break
        alpha = 0.5 * np.log((1 - err) / (err + 1e-10))
        # update weights
        w *= np.exp(-alpha * (2 * y_pred - 1) * (2 * y - 1))
        w /= np.sum(w)
        models.append(stump)
        alphas.append(alpha)
    return models, alphas

def adaboost_predict(models, alphas, X_test):
    predictions = np.array([m.predict(X_test) for m in models])
    # weighted vote
    weighted = np.dot(alphas, predictions)
    return np.sign(weighted)

from sklearn.model_selection import StratifiedKFold
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.svm import SVC
import numpy as np

def create_meta_features(base_models, X, y, n_folds=5):
    skf = StratifiedKFold(n_splits=n_folds, shuffle=True, random_state=42)
    meta_features = np.zeros((X.shape[0], len(base_models)))
    for i, model in enumerate(base_models):
        for train_idx, val_idx in skf.split(X, y):
            model_clone = model.__class__(**model.get_params())
            model_clone.fit(X[train_idx], y[train_idx])
            meta_features[val_idx, i] = model_clone.predict(X[val_idx])
    return meta_features

# Example usage
X_train, X_test, y_train, y_test = ...
models = [RandomForestClassifier(n_estimators=100),
          SVC(kernel='rbf', probability=True),
          LogisticRegression(max_iter=1000)]
meta_features_train = create_meta_features(models, X_train, y_train)
meta_model = LogisticRegression()
meta_model.fit(meta_features_train, y_train)

# For test set, retrain models on full training data
for model in models:
    model.fit(X_train, y_train)
meta_features_test = np.column_stack([m.predict(X_test) for m in models])
y_pred = meta_model.predict(meta_features_test)

import joblib
from sklearn.metrics import accuracy_score

def check_drift(base_models, meta_model, X_new, y_new, threshold=0.95):
    """Alert if any base model accuracy drops below threshold."""
    for i, model in enumerate(base_models):
        y_pred = model.predict(X_new)
        acc = accuracy_score(y_new, y_pred)
        if acc < threshold:
            print(f"Model {i} accuracy {acc:.3f} — retrain needed")
    meta_pred = meta_model.predict(X_new)
    meta_acc = accuracy_score(y_new, meta_pred)
    if meta_acc < threshold:
        print(f"Meta-model accuracy {meta_acc:.3f} — consider refitting")

# Usage — run every month on fresh labeled data
# base_models, meta_model = joblib.load('ensemble.pkl')
# X_new, y_new = get_batch_from_production()
# check_drift(base_models, meta_model, X_new, y_new)