Spaces:
Running
Running
File size: 4,231 Bytes
f381be8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | """
src.models.ensemble.stacking
=============================
Ensemble methods for combining multiple model predictions.
1. Stacking ensemble — Level-0 base models + Level-1 meta-learner
2. Weighted averaging — Optimize weights via L-BFGS on held-out MAE
"""
from __future__ import annotations
from typing import Any, Callable
import numpy as np
from scipy.optimize import minimize
from sklearn.linear_model import Ridge
from sklearn.model_selection import KFold
from src.evaluation.metrics import regression_metrics
class StackingEnsemble:
"""Stacking ensemble with out-of-fold predictions for level-1 training.
Level-0: list of (name, predict_fn) base learners that are already trained
Level-1: Ridge regression meta-learner
"""
def __init__(self, base_learners: list[tuple[str, Callable]], alpha: float = 1.0):
"""
Parameters
----------
base_learners : list of (name, predict_fn)
Each predict_fn accepts X and returns predictions array.
alpha : float
Ridge regression regularization.
"""
self.base_learners = base_learners
self.meta_learner = Ridge(alpha=alpha)
self.is_fitted = False
def _get_meta_features(self, X: np.ndarray) -> np.ndarray:
"""Generate meta-features from base learner predictions."""
preds = []
for name, pred_fn in self.base_learners:
p = pred_fn(X)
if p.ndim == 1:
p = p.reshape(-1, 1)
preds.append(p)
return np.hstack(preds)
def fit(self, X_train: np.ndarray, y_train: np.ndarray, n_folds: int = 5) -> None:
"""Fit the meta-learner using out-of-fold predictions."""
kf = KFold(n_splits=n_folds, shuffle=True, random_state=42)
oof_preds = np.zeros((len(X_train), len(self.base_learners)))
for fold_idx, (train_idx, val_idx) in enumerate(kf.split(X_train)):
X_val_fold = X_train[val_idx]
meta = self._get_meta_features(X_val_fold)
oof_preds[val_idx] = meta
self.meta_learner.fit(oof_preds, y_train)
self.is_fitted = True
def predict(self, X: np.ndarray) -> np.ndarray:
"""Predict using the stacking ensemble."""
meta = self._get_meta_features(X)
return self.meta_learner.predict(meta)
def evaluate(self, X_test: np.ndarray, y_test: np.ndarray) -> dict[str, float]:
preds = self.predict(X_test)
return regression_metrics(y_test, preds, prefix="ensemble")
class WeightedAverageEnsemble:
"""Optimized weighted averaging of base model predictions.
Weights are found by minimizing MAE on a validation set via L-BFGS-B.
"""
def __init__(self, base_learners: list[tuple[str, Callable]]):
self.base_learners = base_learners
self.weights: np.ndarray | None = None
def fit(self, X_val: np.ndarray, y_val: np.ndarray) -> None:
"""Find optimal weights on validation set."""
n = len(self.base_learners)
preds_list = []
for name, pred_fn in self.base_learners:
preds_list.append(pred_fn(X_val))
preds_matrix = np.column_stack(preds_list) # (N, n_models)
def objective(w):
w_norm = w / w.sum()
combined = preds_matrix @ w_norm
return np.mean(np.abs(combined - y_val))
# Constrain weights to be non-negative
result = minimize(
objective,
x0=np.ones(n) / n,
method="L-BFGS-B",
bounds=[(0.0, 1.0)] * n,
)
self.weights = result.x / result.x.sum()
def predict(self, X: np.ndarray) -> np.ndarray:
preds_list = []
for name, pred_fn in self.base_learners:
preds_list.append(pred_fn(X))
preds_matrix = np.column_stack(preds_list)
return preds_matrix @ self.weights
def evaluate(self, X_test: np.ndarray, y_test: np.ndarray) -> dict[str, float]:
preds = self.predict(X_test)
return regression_metrics(y_test, preds, prefix="weighted_avg")
def get_weights_dict(self) -> dict[str, float]:
return {name: float(w) for (name, _), w in zip(self.base_learners, self.weights)}
|