import numpy as np
from sklearn.utils import check_array
from sklearn.utils.extmath import softmax
from sklearn.utils.validation import check_is_fitted
from sklearn.base import RegressorMixin, ClassifierMixin
from .base import MoE
from .. import MoCatBoostRegressor, MoCatBoostClassifier
from .. import MoLGBMRegressor, MoLGBMClassifier
from .. import MoXGBRegressor, MoXGBClassifier
from ..base import ModelBaseRegressor, ModelBaseClassifier
[docs]
class MoMoERegressor(RegressorMixin, MoE, ModelBaseRegressor):
"""
A Mixture of Experts (MoE) regressor that combines multiple expert models for regression tasks.
This regressor creates a weighted ensemble of expert models, where each expert specializes in different regions
of the input space. The final prediction is computed as a weighted sum of expert predictions, where weights
are determined by a gating network based on the input features.
Parameters
----------
name : str, default=None
Identifier for the model instance.
n_clusters : int, default=10
Number of expert models (clusters) to create.
cluster_method : {'ltc', 'kmeans'}, default='kmeans'
Which algorithm to use.
- 'kmeans': Use KMeans clustering algorithm directly.
- 'ltc': This method (Learning Trajectory Cluster; LTC) first fits a baseline xgboost model and calculates its absolute residuals; then fits a gradient boosting models between predictors and the absolute residuals. It extracts prediction trajectories during training, applies optional weighting schemes, performs PCA dimensionality reduction, and clusters samples based on their learning patterns.
centroids : np.ndarray, default=None
Pre-defined cluster centers of shape (n_clusters, n_features). If provided,
skips the clustering step and uses these centers directly.
calibration : bool, default=True
Whether to calibrate the gating network's probability estimates using cross-validation.
cv : int, cross-validation generator or iterable, default=3
Cross-validation strategy for probability calibration. Can be:
- int: number of folds for K-Fold cross-validation
- cross-validation generator: custom splitting strategy
- iterable: yields (train, test) splits as indices
feature_names : list or None, default=None
The list of feature names. If None, will use "X0", "X1", "X2", etc.
cluster_features : list or None, default=None
The list of feature names used for clustering. If None, will use all features.
gating_features : list or None, default=None
The list of feature names used for gating model fitting. If None, will use all features.
ltc_baseline_n_estimators : int, default=300
The number of estimators used for fitting the baseline Xgboost, used when cluster_method="ltc" only.
ltc_baseline_max_depth : int, default=2
The max depth used for fitting the baseline Xgboost, used when cluster_method="ltc" only.
ltc_n_estimators : int, default=300
The number of estimators used for fitting the gradient boosting models in trajectory analysis, used when cluster_method="ltc" only.
ltc_max_depth : int, default=2
The max depth used for fitting the gradient boosting models in trajectory analysis, used when cluster_method="ltc" only.
expert : {"xgboost", "lightgbm", "catboost"}, default="xgboost"
The backend estimator used for each cluster.
*args
Variable length argument list passed to the underlying expert model.
**kwargs
Arbitrary keyword arguments passed to the underlying expert model.
"""
def __init__(self,
name: str = None,
n_clusters: int = 10,
cluster_method: str = "kmeans",
centroids: np.ndarray = None,
calibration: bool = True,
cv=3,
feature_names: list = None,
cluster_features: list = None,
gating_features: list = None,
ltc_baseline_n_estimators: int = 300,
ltc_baseline_max_depth: int = 2,
ltc_n_estimators: int = 100,
ltc_max_depth: int = 5,
expert: str = "xgboost",
*args, **kwargs):
self.name = name
self.expert = expert
if expert == "xgboost":
estimator = MoXGBRegressor(*args, **kwargs)
elif expert == "lightgbm":
estimator = MoLGBMRegressor(*args, **kwargs)
elif expert == "catboost":
estimator = MoCatBoostRegressor(*args, **kwargs)
MoE.__init__(self,
estimator=estimator,
n_clusters=n_clusters,
cluster_method=cluster_method,
centroids=centroids,
calibration=calibration,
cv=cv,
feature_names=feature_names,
cluster_features=cluster_features,
gating_features=gating_features,
ltc_baseline_n_estimators=ltc_baseline_n_estimators,
ltc_baseline_max_depth=ltc_baseline_max_depth,
ltc_n_estimators=ltc_n_estimators,
ltc_max_depth=ltc_max_depth
)
def _predict(self, X):
"""
Predicts target values for the input samples.
Generates predictions by combining the weighted outputs of all expert models, where
weights are determined by the gating network based on the input features.
Parameters
----------
X : np.ndarray of shape (n_samples, n_features)
Input samples for which to generate predictions.
Returns
-------
pred : np.ndarray of shape (n_samples,)
Predicted target values for each input sample.
"""
X = check_array(X)
check_is_fitted(self)
pred = self.get_raw_output(X)
return pred
[docs]
class MoMoEClassifier(ClassifierMixin, MoE, ModelBaseClassifier):
"""
A Mixture of Experts (MoE) classifier that combines multiple expert models for classification tasks.
This classifier creates a weighted ensemble of expert models, where each expert specializes in different regions
of the input space. The final prediction is computed by combining expert predictions weighted by the gating
network's outputs.
Parameters
----------
name : str, default=None
Identifier for the model instance.
n_clusters : int, default=10
Number of expert models (clusters) to create.
cluster_method : {'ltc', 'kmeans'}, default='kmeans'
Which algorithm to use.
- 'kmeans': Use KMeans clustering algorithm directly.
- 'ltc': This method (Learning Trajectory Cluster; LTC) first fits a baseline xgboost model and calculates its absolute residuals; then fits a gradient boosting models between predictors and the absolute residuals. It extracts prediction trajectories during training, applies optional weighting schemes, performs PCA dimensionality reduction, and clusters samples based on their learning patterns.
centroids : np.ndarray, default=None
Pre-defined cluster centers of shape (n_clusters, n_features). If provided,
skips the clustering step and uses these centers directly.
calibration : bool, default=True
Whether to calibrate the gating network's probability estimates using cross-validation.
cv : int, cross-validation generator or iterable, default=3
Cross-validation strategy for probability calibration. Can be:
- int: number of folds for K-Fold cross-validation
- cross-validation generator: custom splitting strategy
- iterable: yields (train, test) splits as indices
feature_names : list or None, default=None
The list of feature names. If None, will use "X0", "X1", "X2", etc.
cluster_features : list or None, default=None
The list of feature names used for clustering. If None, will use all features.
gating_features : list or None, default=None
The list of feature names used for gating model fitting. If None, will use all features.
ltc_baseline_n_estimators : int, default=300
The number of estimators used for fitting the baseline Xgboost, used when cluster_method="ltc" only.
ltc_baseline_max_depth : int, default=2
The max depth used for fitting the baseline Xgboost, used when cluster_method="ltc" only.
ltc_n_estimators : int, default=300
The number of estimators used for fitting the gradient boosting models in trajectory analysis, used when cluster_method="ltc" only.
ltc_max_depth : int, default=2
The max depth used for fitting the gradient boosting models in trajectory analysis, used when cluster_method="ltc" only.
expert : {"xgboost", "lightgbm", "catboost"}, default="xgboost"
The backend estimator used for each cluster.
*args
Variable length argument list passed to the underlying expert model.
**kwargs
Arbitrary keyword arguments passed to the underlying expert model.
"""
def __init__(self,
name: str = None,
n_clusters: int = 10,
cluster_method: str = "kmeans",
centroids: np.ndarray = None,
calibration: bool = True,
cv=3,
feature_names: list = None,
cluster_features: list = None,
gating_features: list = None,
ltc_baseline_n_estimators: int = 300,
ltc_baseline_max_depth: int = 2,
ltc_n_estimators: int = 100,
ltc_max_depth: int = 5,
expert: str = "xgboost",
*args, **kwargs):
self.name = name
self.expert = expert
if expert == "xgboost":
estimator = MoXGBClassifier(*args, **kwargs)
elif expert == "lightgbm":
estimator = MoLGBMClassifier(*args, **kwargs)
elif expert == "catboost":
estimator = MoCatBoostClassifier(*args, **kwargs)
MoE.__init__(self,
estimator=estimator,
n_clusters=n_clusters,
cluster_method=cluster_method,
centroids=centroids,
calibration=calibration,
cv=cv,
feature_names=feature_names,
cluster_features=cluster_features,
gating_features=gating_features,
ltc_baseline_n_estimators=ltc_baseline_n_estimators,
ltc_baseline_max_depth=ltc_baseline_max_depth,
ltc_n_estimators=ltc_n_estimators,
ltc_max_depth=ltc_max_depth
)
def _decision_function(self, X):
"""Computes raw decision scores for samples.
Returns the raw decision values before sigmoid transformation, representing the
model's confidence in predicting the positive class. Higher positive values indicate
stronger confidence in class 1, while lower negative values indicate stronger
confidence in class 0.
Parameters
----------
X : array-like of shape (n_samples, n_features)
Input samples for prediction.
Returns
-------
array-like of shape (n_samples,)
Decision function values, where larger values indicate higher confidence
in the positive class.
"""
X = check_array(X)
check_is_fitted(self)
pred = self.get_raw_output(X)
return pred
def _predict_proba(self, X):
"""Predicts class probabilities for samples.
Computes probability estimates for each class by applying softmax transformation
to the decision function outputs. Returns probabilities for both classes, where
the probabilities sum to 1 for each sample.
Parameters
----------
X : array-like of shape (n_samples, n_features)
Input samples for prediction.
Returns
-------
array-like of shape (n_samples, 2)
Probability estimates for each class, where [:, 0] contains probabilities
for class 0 and [:, 1] contains probabilities for class 1.
"""
pred = self._decision_function(X)
pred_proba = softmax(np.vstack([-pred, pred]).T / 2, copy=False)
return pred_proba