Source code for modeva.models.gaminet.api

import numpy as np
import torch
from sklearn.base import RegressorMixin, ClassifierMixin
from sklearn.model_selection import train_test_split
from sklearn.utils import check_array
from sklearn.utils.extmath import softmax
from sklearn.utils.validation import check_is_fitted

from .base import GAMINet
from ..base import ModelBaseRegressor, ModelBaseClassifier
from ..wrappers.builtin.lightgbm import MoLGBMRegressor


[docs] class MoGAMINetRegressor(RegressorMixin, GAMINet, ModelBaseRegressor): """ Generalized additive model with pairwise interaction regressor. Parameters ---------- name : str, default=None The name of the model. feature_names : list or None, default=None The list of feature names. If None, will use "X0", "X1", "X2", etc. feature_types : list or None, default=None The list of feature types. Available types include "numerical" and "categorical". If None, will use numerical for all features. interact_num : int, default=10 The max number of interactions to be included in the second stage training. subnet_size_main_effect : tuple of int, default=(20, ) The hidden layer architecture of each subnetwork in the main effect block. subnet_size_interaction : tuple of int, default=(20, 20) The hidden layer architecture of each subnetwork in the interaction block. activation_func : {"ReLU", "Sigmoid", "Tanh"}, default="ReLU" The name of the activation function. max_epochs : tuple of THREE int, default=(1000, 1000, 1000) The max number of epochs in the first (main effect training), second (interaction training), and third (fine-tuning) stages, respectively. learning_rates : tuple of THREE float, default=(1e-3, 1e-3, 1e-4) The initial learning rates of Adam optimizer in the first (main effect training), second (interaction training), and third (fine-tuning) stages, respectively. early_stop_thres : tuple of THREE int or "auto", default=["auto", "auto", "auto"] The early stopping threshold in the first (main effect training), second (interaction training), and third (fine-tuning) stages, respectively. In auto mode, the value is set to max(5, min(5000 * n_features / (max_iter_per_epoch * batch_size), 100)). batch_size : int, default=1000 The batch size. Note that it should not be larger than the training size * (1 - validation ratio). batch_size_inference : int, default=10000 The batch size used in the inference stage. It is imposed to avoid out-of-memory issue when dealing very large dataset. max_iter_per_epoch : int, default=100 The max number of iterations per epoch. In the init stage of model fit, its value will be clipped by min(max_iter_per_epoch, int(sample_size / batch_size)). For each epoch, the data would be reshuffled and only the first "max_iter_per_epoch" batches would be used for training. It is imposed to make the training scalable for very large dataset. val_ratio : float, default=0.2 The validation ratio, should be greater than 0 and smaller than 1. warm_start : bool, default=True Initialize the network by fitting a rough LGBM model. The initialization is performed by, 1) fit a LGBM model as teacher model, 2) generate random samples from the teacher model, 3) fit each subnetwork using the generated samples. And it is used for both main effect and interaction subnetwork initialization. gam_sample_size : int, default=5000 The sub-sample size for GAM fitting as warm_start=True. mlp_sample_size : int, default=1000 The generated sample size for individual subnetwork fitting as warm_start=True. heredity : bool, default=True Whether to perform interaction screening subject to heredity constraint. loss_threshold : float, default=0.01 The loss tolerance threshold for selecting fewer main effects or interactions, according to the validation performance. For instance, assume the best validation performance is achieved when using 10 main effects; if only use the top 5 main effects also gives similar validation performance, we could prune the last 5 by setting this parameter to be positive. reg_clarity : float, default=0.1 The regularization strength of marginal clarity constraint. reg_mono : float, default=0.1 The regularization strength of monotonicity constraint. mono_sample_size : int, default=1000 As monotonicity constraint is used, we would generate some data points uniformly within the feature space per epoch, to impose the monotonicity regularization in addition to original training samples. mono_increasing_list : tuple of str, default=() The feature name tuple subject to monotonic increasing constraint. mono_decreasing_list : tuple of str, default=() The feature name tuple subject to monotonic decreasing constraint. include_interaction_list : tuple of (str, str), default=() The tuple of interaction to be included for fitting, each interaction is expressed by (feature_name1, feature_name2). boundary_clip : bool, default=True In the inference stage, whether to clip the feature values by their min and max values in the training data. normalize : bool, default=True Whether to normalize the data before inputting to the network. verbose : bool, default=False Whether to output the training logs. n_jobs : int, default=10 The number of cpu cores for parallel computing. -1 means all the available cpus will be used. device : string, default=None The hardware device name used for training. random_state : int, default=0 The random seed. Attributes ---------- net_ : torch network object The fitted GAMI-Net module. interaction_list_ : list of tuples The list of feature index pairs (tuple) for each fitted interaction. active_main_effect_index_ : list of int The selected main effect index. active_interaction_index_ : list of int The selected interaction index. main_effect_val_loss_ : list of float The validation loss as the most important main effects are sequentially added. interaction_val_loss_ : list of float The validation loss as the most important interactions are sequentially added. time_cost_ : list of tuple The time cost of each stage. n_interactions_ : int The actual number of interactions used in the fitting stage. It is greater or equal to the number of active interactions. """ def __init__(self, name: str = None, feature_names=None, feature_types=None, interact_num=10, subnet_size_main_effect=(20,), subnet_size_interaction=(20, 20), activation_func="ReLU", max_epochs=(1000, 1000, 1000), learning_rates=(1e-3, 1e-3, 1e-4), early_stop_thres=("auto", "auto", "auto"), batch_size=1000, batch_size_inference=10000, max_iter_per_epoch=100, val_ratio=0.2, warm_start=True, gam_sample_size=5000, mlp_sample_size=1000, heredity=True, reg_clarity=0.1, loss_threshold=0.01, reg_mono=0.1, mono_increasing_list=(), mono_decreasing_list=(), mono_sample_size=1000, include_interaction_list=(), boundary_clip=True, normalize=True, verbose=False, n_jobs=10, device=None, random_state=0): self.name = name super().__init__(loss_fn=torch.nn.MSELoss(reduction="none"), feature_names=feature_names, feature_types=feature_types, interact_num=interact_num, subnet_size_main_effect=subnet_size_main_effect, subnet_size_interaction=subnet_size_interaction, activation_func=activation_func, max_epochs=max_epochs, learning_rates=learning_rates, early_stop_thres=early_stop_thres, batch_size=batch_size, batch_size_inference=batch_size_inference, max_iter_per_epoch=max_iter_per_epoch, val_ratio=val_ratio, warm_start=warm_start, gam_sample_size=gam_sample_size, mlp_sample_size=mlp_sample_size, heredity=heredity, reg_clarity=reg_clarity, loss_threshold=loss_threshold, reg_mono=reg_mono, mono_sample_size=mono_sample_size, mono_increasing_list=mono_increasing_list, mono_decreasing_list=mono_decreasing_list, include_interaction_list=include_interaction_list, boundary_clip=boundary_clip, normalize=normalize, verbose=verbose, n_jobs=n_jobs, device=device, random_state=random_state) def _more_tags(self): """ Internal function for skipping some sklearn estimator checks. """ return {"_xfail_checks": {"check_sample_weights_invariance": ("zero sample_weight is not equivalent to removing samples",)}} def _build_teacher_main_effect(self): """ Internal function for fitting a spline based additive interaction model. It works as follows. 1) Subsample at most self.gam_sample_size data from training set. 2) Get the residual with respect to the fitted main effect networks, for classification case, the residual is y_label - pred_proba. 3) Fit a tensor-product spline GAM for selected interactions, to make it scalable for large number of interactions, the number of knots in spline is adaptively adjusted from 10 to 2, according to the number of interactions. 4) Wrap the partial function of each effect and intercept. Returns ------- surrogate_estimator : object List of wrapped functions, each element is a fitted effect. intercept : float Fitted intercept. """ x = self.training_generator_.tensors[0].cpu().numpy() y = self.training_generator_.tensors[1].cpu().numpy() sw = self.training_generator_.tensors[2].cpu().numpy() if self.gam_sample_size >= x.shape[0]: xx, yy, swsw = x, y, sw else: _, xx, _, yy, _, swsw = train_test_split(x, y, sw, test_size=self.gam_sample_size, random_state=self.random_state) lgbm = MoLGBMRegressor(linear_trees=True, max_depth=1, verbose=-1) lgbm.fit((xx - self.mu_list_.cpu().numpy()) / self.std_list_.cpu().numpy(), yy, sample_weight=swsw) lgbm.extract_model_info(X=(xx - self.mu_list_.cpu().numpy()) / self.std_list_.cpu().numpy(), feature_names=self.feature_names_, feature_types=self.feature_types_) def marginal_effect(fidx): fn = self.feature_names_[fidx] if fn in lgbm.modeva_effects_["main_effect"]: return lambda inputs: lgbm.predict_effect(fidx=fidx, X=inputs) else: return lambda inputs: np.zeros((inputs.shape[0],)) intercept = lgbm.modeva_intercept_ surrogate_estimator = [marginal_effect(i) for i in range(self.n_features_in_)] return surrogate_estimator, intercept def _build_teacher_interaction(self): """ Internal function for fitting a spline based additive interaction model. It works as follows. 1) Subsample at most self.gam_sample_size data from training set. 2) Get the residual with respect to the fitted main effect networks, for classification case, the residual is y_label - pred_proba. 3) Fit a tensor-product spline GAM for selected interactions, to make it scalable for large number of interactions, the number of knots in spline is adaptively adjusted from 10 to 2, according to the number of interactions. 4) Wrap the partial function of each effect and intercept. Returns ------- surrogate_estimator : object List of wrapped functions, each element is a fitted effect. intercept : float Fitted intercept. """ x = self.training_generator_.tensors[0].cpu().numpy() y = self.training_generator_.tensors[1].cpu().numpy() sw = self.training_generator_.tensors[2].cpu().numpy() if self.gam_sample_size >= x.shape[0]: xx, yy, swsw = x, y, sw else: _, xx, _, yy, _, swsw = train_test_split(x, y, sw, test_size=self.gam_sample_size, random_state=self.random_state) residual = yy - self.get_raw_output(xx, main_effect=True, interaction=False).detach().cpu().numpy().ravel() lgbm = MoLGBMRegressor(linear_trees=True, max_depth=2, verbose=-1) lgbm.fit((xx - self.mu_list_.cpu().numpy()) / self.std_list_.cpu().numpy(), residual, sample_weight=swsw) lgbm.extract_model_info(X=(xx - self.mu_list_.cpu().numpy()) / self.std_list_.cpu().numpy(), feature_names=self.feature_names_, feature_types=self.feature_types_) def marginal_effect(feature_idx): fn1 = self.feature_names_[feature_idx[0]] fn2 = self.feature_names_[feature_idx[1]] key = fn1 + " & " + fn2 if key in lgbm.modeva_effects_["interaction"]: return lambda inputs: lgbm.predict_effect(fidx=feature_idx, X=inputs) else: return lambda inputs: np.zeros((inputs.shape[0],)) intercept = lgbm.modeva_intercept_ surrogate_estimator = [marginal_effect(fidx) for fidx in self.interaction_list_] return surrogate_estimator, intercept
[docs] def fit(self, X, y, sample_weight=None): """ Fits a GAMINet regression model to the training data. This method trains the model in three stages: main effects training, interaction training, and fine-tuning. It handles data preprocessing, model initialization, and the complete training pipeline. Parameters ---------- X : np.ndarray of shape (n_samples, n_features) Training data features. y : np.ndarray of shape (n_samples,) Target values for regression. sample_weight : np.ndarray of shape (n_samples,), default=None Individual weights for each sample. If None, all samples are weighted equally. Returns ------- self : object Returns the fitted estimator. """ self._init_fit(X, y, sample_weight) return self._fit()
def _predict(self, X): """ Returns numpy array of predicted values. Parameters ---------- X : np.ndarray of shape (n_samples, n_features) Data features. Returns ------- pred: np.ndarray of shape (n_samples, ) numpy array of predicted values. """ X = check_array(X) check_is_fitted(self) pred = self.get_raw_output(X).detach().cpu().numpy().ravel() return pred
[docs] class MoGAMINetClassifier(ClassifierMixin, GAMINet, ModelBaseClassifier): """ Generalized additive model with pairwise interaction classifier. Parameters ---------- name : str, default=None The name of the model. feature_names : list or None, default=None The list of feature names. If None, will use "X0", "X1", "X2", etc. feature_types : list or None, default=None The list of feature types. Available types include "numerical" and "categorical". If None, will use numerical for all features. interact_num : int, default=10 The max number of interactions to be included in the second stage training. subnet_size_main_effect : tuple of int, default=(20, ) The hidden layer architecture of each subnetwork in the main effect block. subnet_size_interaction : tuple of int, default=(20, 20) The hidden layer architecture of each subnetwork in the interaction block. activation_func : {"ReLU", "Sigmoid", "Tanh"}, default="ReLU" The name of the activation function. max_epochs : tuple of THREE int, default=(1000, 1000, 1000) The max number of epochs in the first (main effect training), second (interaction training), and third (fine-tuning) stages, respectively. learning_rates : tuple of THREE float, default=(1e-3, 1e-3, 1e-4) The initial learning rates of Adam optimizer in the first (main effect training), second (interaction training), and third (fine-tuning) stages, respectively. early_stop_thres : tuple of THREE int or "auto", default=["auto", "auto", "auto"] The early stopping threshold in the first (main effect training), second (interaction training), and third (fine-tuning) stages, respectively. In auto mode, the value is set to max(5, min(5000 * n_features / (max_iter_per_epoch * batch_size), 100)). batch_size : int, default=1000 The batch size. Note that it should not be larger than the training size * (1 - validation ratio). batch_size_inference : int, default=10000 The batch size used in the inference stage. It is imposed to avoid out-of-memory issue when dealing very large dataset. max_iter_per_epoch : int, default=100 The max number of iterations per epoch. In the init stage of model fit, its value will be clipped by min(max_iter_per_epoch, int(sample_size / batch_size)). For each epoch, the data would be reshuffled and only the first "max_iter_per_epoch" batches would be used for training. It is imposed to make the training scalable for very large dataset. val_ratio : float, default=0.2 The validation ratio, should be greater than 0 and smaller than 1. warm_start : bool, default=True Initialize the network by fitting a rough LGBM model. The initialization is performed by, 1) fit a LGBM model as teacher model, 2) generate random samples from the teacher model, 3) fit each subnetwork using the generated samples. And it is used for both main effect and interaction subnetwork initialization. gam_sample_size : int, default=5000 The sub-sample size for GAM fitting as warm_start=True. mlp_sample_size : int, default=1000 The generated sample size for individual subnetwork fitting as warm_start=True. heredity : bool, default=True Whether to perform interaction screening subject to heredity constraint. loss_threshold : float, default=0.01 The loss tolerance threshold for selecting fewer main effects or interactions, according to the validation performance. For instance, assume the best validation performance is achieved when using 10 main effects; if only use the top 5 main effects also gives similar validation performance, we could prune the last 5 by setting this parameter to be positive. reg_clarity : float, default=0.1 The regularization strength of marginal clarity constraint. reg_mono : float, default=0.1 The regularization strength of monotonicity constraint. mono_sample_size : int, default=1000 As monotonicity constraint is used, we would generate some data points uniformly within the feature space per epoch, to impose the monotonicity regularization in addition to original training samples. mono_increasing_list : tuple of str, default=() The feature name tuple subject to monotonic increasing constraint. mono_decreasing_list : tuple of str, default=() The feature name tuple subject to monotonic decreasing constraint. include_interaction_list : tuple of (str, str), default=() The tuple of interaction to be included for fitting, each interaction is expressed by (feature_name1, feature_name2). boundary_clip : bool, default=True In the inference stage, whether to clip the feature values by their min and max values in the training data. normalize : bool, default=True Whether to normalize the data before inputting to the network. verbose : bool, default=False Whether to output the training logs. n_jobs : int, default=10 The number of cpu cores for parallel computing. -1 means all the available cpus will be used. device : string, default=None The hardware device name used for training. random_state : int, default=0 The random seed. Attributes ---------- net_ : torch network object The fitted GAMI-Net module. interaction_list_ : list of tuples The list of feature index pairs (tuple) for each fitted interaction. active_main_effect_index_ : list of int The selected main effect index. active_interaction_index_ : list of int The selected interaction index. main_effect_val_loss_ : list of float The validation loss as the most important main effects are sequentially added. interaction_val_loss_ : list of float The validation loss as the most important interactions are sequentially added. time_cost_ : list of tuple The time cost of each stage. n_interactions_ : int The actual number of interactions used in the fitting stage. It is greater or equal to the number of active interactions. """ def __init__(self, name: str = None, feature_names=None, feature_types=None, interact_num=10, subnet_size_main_effect=(20,), subnet_size_interaction=(20, 20), activation_func="ReLU", max_epochs=(1000, 1000, 1000), learning_rates=(1e-3, 1e-3, 1e-4), early_stop_thres=("auto", "auto", "auto"), batch_size=1000, batch_size_inference=10000, max_iter_per_epoch=100, val_ratio=0.2, warm_start=True, gam_sample_size=5000, mlp_sample_size=1000, heredity=True, reg_clarity=0.1, loss_threshold=0.01, reg_mono=0.1, mono_increasing_list=(), mono_decreasing_list=(), mono_sample_size=1000, include_interaction_list=(), boundary_clip=True, normalize=True, verbose=False, n_jobs=10, device=None, random_state=0): self.name = name super().__init__(loss_fn=torch.nn.BCEWithLogitsLoss(reduction="none"), feature_names=feature_names, feature_types=feature_types, interact_num=interact_num, subnet_size_main_effect=subnet_size_main_effect, subnet_size_interaction=subnet_size_interaction, activation_func=activation_func, max_epochs=max_epochs, learning_rates=learning_rates, early_stop_thres=early_stop_thres, batch_size=batch_size, batch_size_inference=batch_size_inference, max_iter_per_epoch=max_iter_per_epoch, val_ratio=val_ratio, warm_start=warm_start, gam_sample_size=gam_sample_size, mlp_sample_size=mlp_sample_size, heredity=heredity, reg_clarity=reg_clarity, loss_threshold=loss_threshold, reg_mono=reg_mono, mono_sample_size=mono_sample_size, mono_increasing_list=mono_increasing_list, mono_decreasing_list=mono_decreasing_list, include_interaction_list=include_interaction_list, boundary_clip=boundary_clip, normalize=normalize, verbose=verbose, n_jobs=n_jobs, device=device, random_state=random_state) def _more_tags(self): """ Internal function for skipping some sklearn estimator checks. """ return {"binary_only": True, "_xfail_checks": {"check_sample_weights_invariance": ("zero sample_weight is not equivalent to removing samples",)}} def _build_teacher_main_effect(self): """ Internal function for fitting a spline based additive model. It works as follows. 1) Subsample at most self.gam_sample_size data from training set. 2) Fit a LGBM model using all input features. 3) Wrap the partial function of each effect and intercept. Returns ------- surrogate_estimator : object List of wrapped functions, each element is a fitted effect. intercept : float Fitted intercept. """ x = self.training_generator_.tensors[0].cpu().numpy() y = self.training_generator_.tensors[1].cpu().numpy() * 4 - 2 sw = self.training_generator_.tensors[2].cpu().numpy() if self.gam_sample_size >= x.shape[0]: xx, yy, swsw = x, y, sw else: _, xx, _, yy, _, swsw = train_test_split(x, y, sw, test_size=self.gam_sample_size, stratify=y, random_state=self.random_state) lgbm = MoLGBMRegressor(linear_trees=True, max_depth=1, verbose=-1) lgbm.fit((xx - self.mu_list_.cpu().numpy()) / self.std_list_.cpu().numpy(), yy, sample_weight=swsw) lgbm.extract_model_info(X=(xx - self.mu_list_.cpu().numpy()) / self.std_list_.cpu().numpy(), feature_names=self.feature_names_, feature_types=self.feature_types_) def marginal_effect(fidx): fn = self.feature_names_[fidx] if fn in lgbm.modeva_effects_["main_effect"]: return lambda inputs: lgbm.predict_effect(fidx=fidx, X=inputs) else: return lambda inputs: np.zeros((inputs.shape[0],)) intercept = lgbm.modeva_intercept_ surrogate_estimator = [marginal_effect(i) for i in range(self.n_features_in_)] return surrogate_estimator, intercept def _build_teacher_interaction(self): """ Internal function for fitting a spline based additive interaction model. It works as follows. 1) Subsample at most self.gam_sample_size data from training set. 2) Get the residual with respect to the fitted main effect networks, for classification case, the residual is y_label - pred_proba. 3) Fit a tensor-product spline GAM for selected interactions, to make it scalable for large number of interactions, the number of knots in spline is adaptively adjusted from 10 to 2, according to the number of interactions. 4) Wrap the partial function of each effect and intercept. Returns ------- surrogate_estimator : object List of wrapped functions, each element is a fitted effect. intercept : float Fitted intercept. """ x = self.training_generator_.tensors[0].cpu().numpy() y = self.training_generator_.tensors[1].cpu().numpy() sw = self.training_generator_.tensors[2].cpu().numpy() if self.gam_sample_size >= x.shape[0]: xx, yy, swsw = x, y, sw else: _, xx, _, yy, _, swsw = train_test_split(x, y, sw, test_size=self.gam_sample_size, stratify=y, random_state=self.random_state) pred = self.get_raw_output(xx, main_effect=True, interaction=False).detach().cpu().numpy().ravel() pred_proba = softmax(np.vstack([-pred, pred]).T / 2, copy=False)[:, 1] residual = yy - pred_proba lgbm = MoLGBMRegressor(linear_trees=True, max_depth=2, verbose=-1) lgbm.fit((xx - self.mu_list_.cpu().numpy()) / self.std_list_.cpu().numpy(), residual, sample_weight=swsw) lgbm.extract_model_info(X=(xx - self.mu_list_.cpu().numpy()) / self.std_list_.cpu().numpy(), feature_names=self.feature_names_, feature_types=self.feature_types_) def marginal_effect(feature_idx): fn1 = self.feature_names_[feature_idx[0]] fn2 = self.feature_names_[feature_idx[1]] key = fn1 + " & " + fn2 if key in lgbm.modeva_effects_["interaction"]: return lambda inputs: lgbm.predict_effect(fidx=feature_idx, X=inputs) else: return lambda inputs: np.zeros((inputs.shape[0],)) intercept = lgbm.modeva_intercept_ surrogate_estimator = [marginal_effect(fidx) for fidx in self.interaction_list_] return surrogate_estimator, intercept
[docs] def fit(self, X, y, sample_weight=None): """ Fit GAMINetClassifier model. Parameters ---------- X : np.ndarray of shape (n_samples, n_features) Data features. y : np.ndarray of shape (n_samples, ) Target response. sample_weight : np.ndarray of shape (n_samples, ) Sample weight. Returns ------- self : object Fitted Estimator. """ self._init_fit(X, y, sample_weight, stratified=True) return self._fit()
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).detach().cpu().numpy().ravel() 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