Source code for modeva.models.glmtree.neural_tree

from copy import deepcopy

import numpy as np
import torch
from sklearn.base import BaseEstimator, 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 .glmtreeboost import MoGLMTreeBoostRegressor, MoGLMTreeBoostClassifier
from ..base import ModelBaseRegressor, ModelBaseClassifier
from ..utils import FastTensorDataLoader, get_torch_device, set_seed
from ...auth import auth
from ...testsuite.interpret.fanova.base import InterpretFANOVA
from ...testsuite.interpret.fanova.tree_linear.extractor.neural_tree import InterpretNeuralTree
from ...utils.helper import limit_function_for_trial_license


class BoostGLMTreeModule(torch.nn.Module):

    def __init__(self,
                 n_features,
                 n_estimators,
                 split_feature_indice_list,
                 split_threshold_list,
                 coefs_list,
                 intercepts_list,
                 lr_list,
                 activation_func,
                 temperature,
                 mono_increasing_list,
                 mono_decreasing_list,
                 device):

        super().__init__()

        auth.run()
        self.n_features = n_features
        self.n_estimators = n_estimators
        self.activation_func = activation_func
        self.mono_increasing_list = mono_increasing_list
        self.mono_decreasing_list = mono_decreasing_list

        self.max_pred = torch.tensor(float("inf"), dtype=torch.float, requires_grad=False, device=device)
        self.min_pred = torch.tensor(float("-inf"), dtype=torch.float, requires_grad=False, device=device)

        self.split_weight = torch.zeros((n_features, n_estimators), dtype=torch.float, requires_grad=False,
                                        device=device)
        self.split_weight[split_feature_indice_list, torch.arange(n_estimators)] = 1
        self.split_threshold = torch.nn.Parameter(torch.tensor(split_threshold_list, dtype=torch.float,
                                                               requires_grad=True, device=device))
        self.w = torch.nn.Parameter(
            torch.tensor(np.hstack(coefs_list), dtype=torch.float, requires_grad=True, device=device))
        self.b = torch.nn.Parameter(torch.tensor(np.vstack(intercepts_list).ravel(), dtype=torch.float,
                                                 requires_grad=True, device=device))
        self.temperature = torch.tensor(temperature, dtype=torch.float, requires_grad=False, device=device)

        self.output_layer = torch.nn.Linear(n_estimators, 1, bias=False, device=device)
        self.output_layer.weight.data = torch.tensor(lr_list, dtype=torch.float, device=device).reshape(1, -1)

    def get_segment_indicator(self, inputs):
        xb = torch.matmul(inputs, self.split_weight) - self.split_threshold
        xb_temperature = xb / self.temperature
        segment_indicator = self.activation_func(torch.dstack([xb_temperature, - xb_temperature]))
        segment_indicator = segment_indicator.flatten().reshape(-1, 2 * self.n_estimators)
        return segment_indicator

    def update_pred_minmax(self, inputs):
        segment_indicator = self.get_segment_indicator(inputs)
        local_linear_output = torch.matmul(inputs, self.w) + self.b
        glmtree_output = local_linear_output * segment_indicator
        self.max_pred.data = glmtree_output.max(0).values
        self.min_pred.data = glmtree_output.min(0).values

    def get_mono_loss(self, inputs, outputs=None, monotonicity=False, sample_weight=None):

        mono_loss = torch.tensor(0.0, requires_grad=True)
        if not monotonicity:
            return mono_loss

        grad = torch.autograd.grad(outputs=torch.sum(outputs),
                                   inputs=inputs, create_graph=True)[0]

        if sample_weight is not None:
            if len(self.mono_increasing_list) > 0:
                mono_loss = mono_loss + torch.mean(torch.nn.ReLU()(
                    -grad[:, self.mono_increasing_list]) * sample_weight.reshape(-1, 1))
            if len(self.mono_decreasing_list) > 0:
                mono_loss = mono_loss + torch.mean(torch.nn.ReLU()(
                    grad[:, self.mono_decreasing_list]) * sample_weight.reshape(-1, 1))
        else:
            if len(self.mono_increasing_list) > 0:
                mono_loss = mono_loss + torch.mean(torch.nn.ReLU()(
                    -grad[:, self.mono_increasing_list]))
            if len(self.mono_decreasing_list) > 0:
                mono_loss = mono_loss + torch.mean(torch.nn.ReLU()(
                    grad[:, self.mono_decreasing_list]))
        return mono_loss

    def forward(self, inputs, monotonicity=False, sample_weight=None):
        inputs.requires_grad = True
        segment_indicator = self.get_segment_indicator(inputs)
        local_linear_output = torch.clip(torch.matmul(inputs, self.w) + self.b, self.min_pred, self.max_pred)
        glmtree_output = (local_linear_output * segment_indicator).reshape(-1, self.n_estimators, 2).sum(-1)
        outputs = self.output_layer(glmtree_output)
        self.mono_loss = self.get_mono_loss(inputs, outputs, monotonicity, sample_weight)
        return outputs


class NeuralTree(BaseEstimator):

    def __init__(self,
                 loss_fn,
                 estimator,
                 feature_names,
                 nn_temperature,
                 nn_lr,
                 nn_max_epochs,
                 nn_n_epoch_no_change,
                 nn_batch_size,
                 reg_mono,
                 mono_sample_size,
                 mono_increasing_list,
                 mono_decreasing_list,
                 val_ratio,
                 verbose,
                 device,
                 random_state):

        super().__init__()

        self.loss_fn = loss_fn
        self.estimator = estimator
        self.feature_names = feature_names
        self.nn_temperature = nn_temperature

        self.nn_lr = nn_lr
        self.nn_batch_size = nn_batch_size
        self.reg_mono = reg_mono
        self.mono_sample_size = mono_sample_size
        self.mono_increasing_list = mono_increasing_list
        self.mono_decreasing_list = mono_decreasing_list
        self.nn_max_epochs = nn_max_epochs
        self.nn_n_epoch_no_change = nn_n_epoch_no_change
        self.val_ratio = val_ratio
        self.verbose = verbose
        self.device = get_torch_device(device)
        self.random_state = random_state

    def _evaluate(self):
        """
        Internal function for evaluating the validation loss.

        Returns
        -------
        mean_loss : float
            Average validation loss.
        """
        pred = self.get_aggregate_output(self.validation_generator_.tensors[0])
        loss = self.loss_fn(pred.ravel(), self.validation_generator_.tensors[1].ravel())
        mean_loss = torch.mean(loss * self.validation_generator_.tensors[2].ravel()).cpu().detach().numpy().ravel()[0]
        return mean_loss

    def _init_fit(self, X, y, sample_weight=None):

        X, y, sample_weight = self._validate_fit_inputs(X, y, sample_weight)
        if self.feature_names is None:
            self.feature_names_ = ["X" + str(idx) for idx in range(self.n_features_in_)]
        else:
            self.feature_names_ = deepcopy(self.feature_names)

        self.min_value_ = torch.tensor(np.min(X, axis=0), dtype=torch.float32, device=self.device)
        self.max_value_ = torch.tensor(np.max(X, axis=0), dtype=torch.float32, device=self.device)

        if len(list(self.mono_increasing_list) + list(self.mono_decreasing_list)) > 0 and self.reg_mono > 0:
            self.monotonicity_ = True
        else:
            self.monotonicity_ = False
        self.mono_increasing_list_index_ = [self.feature_names_.index(fn) for fn in self.mono_increasing_list]
        self.mono_decreasing_list_index_ = [self.feature_names_.index(fn) for fn in self.mono_decreasing_list]

        try:
            check_is_fitted(self.estimator)
            if self.verbose:
                print("#### #### MoNeuralTree Training Stage 1: Use Fitted MoGLMTreeBoost ####")
        except:
            if self.verbose:
                print("#### MoNeuralTree Training Stage 1: Train MoGLMTreeBoost ####")
            self.estimator.fit(X, y, sample_weight)

        tr_x, val_x, tr_y, val_y, tr_sw, val_sw = train_test_split(
            X, y, sample_weight, test_size=self.val_ratio, random_state=self.random_state)

        batch_size = min(self.nn_batch_size, tr_x.shape[0])
        self.training_generator_ = FastTensorDataLoader(
            torch.tensor(tr_x, dtype=torch.float32, device=self.device),
            torch.tensor(tr_y, dtype=torch.float32, device=self.device),
            torch.tensor(tr_sw, dtype=torch.float32, device=self.device),
            batch_size=batch_size, shuffle=True)
        self.validation_generator_ = FastTensorDataLoader(
            torch.tensor(val_x, dtype=torch.float32, device=self.device),
            torch.tensor(val_y, dtype=torch.float32, device=self.device),
            torch.tensor(val_sw, dtype=torch.float32, device=self.device),
            batch_size=self.nn_batch_size * 10, shuffle=False)

        # the seed may not work for data loader. this needs to be checked
        set_seed(self.random_state)

        self.coefs_list_ = []
        self.intercepts_list_ = []
        self.split_threshold_list_ = []
        self.split_feature_indice_list_ = []
        self.n_estimators_ = len(self.estimator.estimators_)
        self.lr_list_ = self.estimator.learning_rates_[:self.n_estimators_]
        for glmtree_est in self.estimator.estimators_:
            if len(glmtree_est.tree_) == 3:
                self.split_feature_indice_list_.append(glmtree_est.tree_[1]["feature"])
                self.split_threshold_list_.append(glmtree_est.tree_[1]["threshold"])
                self.coefs_list_.append(np.array([est.coef_.flatten()
                                                  for nodeid, est in glmtree_est.leaf_estimators_.items() if
                                                  est is not None]).T)
                self.intercepts_list_.append(np.array([est.intercept_.flatten()
                                                       for nodeid, est in glmtree_est.leaf_estimators_.items() if
                                                       est is not None]))
            else:
                self.split_feature_indice_list_.append(0)
                self.split_threshold_list_.append(0)
                self.coefs_list_.append(np.array([glmtree_est.leaf_estimators_[1].coef_.ravel(),
                                                  glmtree_est.leaf_estimators_[1].coef_.ravel()]).T)
                self.intercepts_list_.append(np.array([glmtree_est.leaf_estimators_[1].intercept_.flatten(),
                                                       glmtree_est.leaf_estimators_[1].intercept_.flatten()]))

        self.net_ = BoostGLMTreeModule(self.n_features_in_,
                                       self.n_estimators_,
                                       self.split_feature_indice_list_,
                                       self.split_threshold_list_,
                                       self.coefs_list_,
                                       self.intercepts_list_,
                                       self.lr_list_,
                                       activation_func=torch.sigmoid,
                                       temperature=self.nn_temperature,
                                       mono_increasing_list=self.mono_increasing_list_index_,
                                       mono_decreasing_list=self.mono_decreasing_list_index_,
                                       device=self.device)
        self.best_loss_ = np.inf
        self.no_improved_count_ = 0
        self.train_epoch_loss_ = []
        self.valid_epoch_loss_ = []

    def _check_improvement(self):
        """Checks whether there is an improvement in validation loss or not."""
        if self.best_loss_ > self.valid_epoch_loss_[-1]:
            self.best_loss_ = self.valid_epoch_loss_[-1]
            self.no_improved_count_ = 0
        else:
            self.no_improved_count_ += 1

    def get_mono_loss(self, x, sample_weight=None):
        """
        Returns monotonicity loss of given samples.

        Parameters
        ----------
        x : np.ndarray of shape (n_samples, n_features)
            Data features.
        sample_weight : np.ndarray of shape (n_samples, ), default=None
            Sample weight.

        Returns
        -------
        mono_loss : float
            monotonicity loss.
        """
        mono_loss = 0
        self.net_.eval()
        x = np.asarray(x).reshape(-1, self.n_features_in_)
        xx = x if torch.is_tensor(x) else torch.from_numpy(x.astype(np.float32)).to(self.device)
        batch_size = int(np.minimum(self.nn_batch_size, x.shape[0]))
        data_generator = FastTensorDataLoader(xx, batch_size=batch_size, shuffle=False)
        for batch_no, batch_data in enumerate(data_generator):
            batch_xx = batch_data[0]
            self.net_(batch_xx, sample_weight=sample_weight, monotonicity=True)
            mono_loss += len(batch_data[0]) * self.net_.mono_loss.detach().cpu().numpy()
        mono_loss = mono_loss / data_generator.dataset_len
        return mono_loss

    def certify_mono(self, n_samples=10000):
        """
        Verify if the model satisfies specified monotonicity constraints.

        Generates random samples within the feature space and checks if the model's 
        predictions respect the monotonicity constraints specified during training.

        Parameters
        ----------
        n_samples : int, default=10000
            Number of random samples to use for verification.

        Returns
        -------
        mono_status : bool
            True if monotonicity constraints are satisfied, False otherwise.
        """
        rng = np.random.default_rng(self.random_state)
        x = rng.uniform(self.min_value_.cpu().numpy(),
                        self.max_value_.cpu().numpy(), size=(n_samples, self.n_features_in_))
        mono_loss = self.get_mono_loss(x)
        mono_status = mono_loss <= 0
        return mono_status

    def fit(self, X, y, sample_weight=None):

        self._init_fit(X, y, sample_weight)
        opt = torch.optim.Adam(self.net_.parameters(), lr=self.nn_lr)
        if self.verbose:
            train_pred = self.get_aggregate_output(self.training_generator_.tensors[0]).ravel()
            train_loss = torch.mean(self.loss_fn(train_pred, self.training_generator_.tensors[1])
                                    * self.training_generator_.tensors[2]).cpu().detach().numpy().ravel()[0]
            val_pred = self.get_aggregate_output(self.validation_generator_.tensors[0]).ravel()
            val_loss = torch.mean(self.loss_fn(val_pred, self.validation_generator_.tensors[1])
                                  * self.validation_generator_.tensors[2]).cpu().detach().numpy().ravel()[0]
            print("#### MoNeuralTree Training Stage 2: Fine-tuning via Gradient Descent ####")
            print("Initial training and validation loss: %0.4f and %0.4f" % (
                np.round(train_loss, 4), np.round(val_loss, 4)))

        for epoch in range(self.nn_max_epochs):
            self.net_.train()
            accumulated_size = 0.0
            accumulated_loss = 0.0
            for batch_no, batch_data in enumerate(self.training_generator_):
                opt.zero_grad(set_to_none=True)
                batch_xx = batch_data[0].to(self.device)
                batch_yy = batch_data[1].to(self.device).ravel()
                batch_sw = batch_data[2].to(self.device).ravel()
                pred = self.net_(batch_xx,
                                 sample_weight=batch_sw,
                                 monotonicity=self.monotonicity_
                                 ).ravel()

                if self.monotonicity_:
                    if self.random_state is not None:
                        rng = np.random.default_rng(self.random_state * epoch + batch_no)
                    else:
                        rng = np.random.default_rng(self.random_state)
                    mono_loss_reg = self.reg_mono * self.net_.mono_loss
                    simu_inputs = rng.uniform(self.min_value_.cpu().numpy(),
                                              self.max_value_.cpu().numpy(),
                                              size=(self.mono_sample_size, len(self.max_value_)))
                    simu_inputs = torch.tensor(simu_inputs, dtype=torch.float32, device=self.device)
                    self.net_(simu_inputs,
                              monotonicity=self.monotonicity_).ravel()
                    mono_loss_reg = mono_loss_reg + self.reg_mono * self.net_.mono_loss
                    mono_loss_reg.backward(retain_graph=True)

                loss = torch.mean(self.loss_fn(pred, batch_yy) * batch_sw)
                loss.backward()
                opt.step()
                accumulated_size += batch_xx.shape[0]
                accumulated_loss += (loss * batch_xx.shape[0]).cpu().detach().numpy()

            val_loss = self._evaluate()
            train_loss = accumulated_loss / accumulated_size
            self.train_epoch_loss_.append(train_loss)
            self.valid_epoch_loss_.append(val_loss)
            self._check_improvement()
            if self.verbose:
                if self.monotonicity_:
                    print(
                        """Epoch %d: Train loss %0.4f, Validation loss %0.4f, Monotonicity loss %0.4f"""
                        % (epoch, train_loss, val_loss, mono_loss_reg)
                    )
                else:
                    print(
                        """Epoch %d: Train loss %0.4f, Validation loss %0.4f"""
                        % (epoch, train_loss, val_loss)
                    )
            if self.no_improved_count_ > self.nn_n_epoch_no_change:
                if self.verbose:
                    print(
                        """Training is terminated as validation loss stops decreasing."""
                    )
                if self.monotonicity_:
                    if self.certify_mono():
                        break
                    else:
                        self.reg_mono = min(self.reg_mono * 1.2, 1e5)
                else:
                    break
        else:
            if self.verbose:
                print("""Training is terminated as max_epoch is reached.""")
        return self

    @torch.no_grad()
    def get_aggregate_output(self, X):
        """
        Returns numpy array of raw prediction.

        Parameters
        ----------
        X : np.ndarray of shape (n_samples, n_features)
            Data features.

        Returns
        -------
        pred : np.ndarray of shape (n_samples, 1)
            numpy array of raw prediction.
        """
        pred = []
        self.net_.eval()
        XX = X if torch.is_tensor(X) else torch.from_numpy(X.astype(np.float32)).to(self.device)
        batch_size = int(np.minimum(self.nn_batch_size * 10, X.shape[0]))
        data_generator = FastTensorDataLoader(XX, batch_size=batch_size, shuffle=False)
        for batch_no, batch_data in enumerate(data_generator):
            batch_xx = batch_data[0]
            pred.append(self.net_(batch_xx).detach())
        pred = torch.vstack(pred)
        return pred

    def get_sparsity(self, x):

        self.net_.eval()
        inputs = torch.tensor(x, dtype=torch.float).to(self.device)
        segment_indicator = self.net_.get_segment_indicator(inputs).cpu().detach().numpy()
        return np.mean((segment_indicator == 0) | (segment_indicator == 1))

    def extract_model_info(self, X, feature_names, feature_types):
        test_obj = InterpretNeuralTree(model=self,
                                       feature_names=feature_names,
                                       feature_types=feature_types,
                                       purification=True)
        test_obj.fit(X)
        self.modeva_effects_ = test_obj.effects_
        self.modeva_intercept_ = test_obj.intercept_
        self.predict_effect = test_obj._predict_effect

    def interpret(self, dataset):

        if not dataset.is_built_in():
            limit_function_for_trial_license()
        self.extract_model_info(X=dataset.train_x,
                                feature_names=dataset.feature_names,
                                feature_types=dataset.feature_types)
        return InterpretFANOVA(model=self, dataset=dataset)


[docs] class MoNeuralTreeRegressor(RegressorMixin, NeuralTree, ModelBaseRegressor): """ A neural network-based regression model that combines GLM trees with monotonicity constraints. This model first fits a depth-1 boosted GLMTree, converts it to an equivalent neural network, and then fine-tunes the network parameters. It supports monotonicity constraints and provides interpretability through feature effects analysis. Parameters ---------- name : str, default=None Custom identifier for the model instance. estimator : MoGLMTreeBoostRegressor, default=None Pre-fitted or unfitted GLMTree regressor for initialization. If None, creates a new instance. feature_names : list of str, default=None Names of input features for interpretability. val_ratio : float, default=0.2 Proportion of data used for validation during training (0 to 1). device : str, default=None Computing device for training ('cpu', 'cuda', etc.). verbose : bool, default=False If True, prints training progress and statistics. random_state : int, default=0 Seed for reproducible random operations. nn_temperature : float, default=0.0001 Smoothing parameter for neural network activation. nn_lr : float, default=0.001 Learning rate for neural network optimization. nn_max_epochs : int, default=200 Maximum number of training epochs. nn_batch_size : int, default=200 Number of samples per training batch. nn_n_epoch_no_change : int, default=10 Early stopping patience - number of epochs without improvement. reg_mono : float, default=0.1 Strength of monotonicity regularization. mono_sample_size : int, default=1000 Number of random samples for monotonicity regularization. mono_increasing_list : tuple of str, default=() Features that should have monotonically increasing relationships. mono_decreasing_list : tuple of str, default=() Features that should have monotonically decreasing relationships. **kwargs Additional parameters passed to MoGLMTreeBoostRegressor. Attributes ---------- net_ : object The internal Pytorch network object. """ def __init__(self, name: str = None, estimator: MoGLMTreeBoostRegressor = None, val_ratio=0.2, verbose=False, device=None, random_state=0, feature_names=None, nn_temperature=0.0001, nn_lr=0.0001, nn_max_epochs=200, nn_n_epoch_no_change=10, nn_batch_size=200, reg_mono=0.1, mono_sample_size=1000, mono_increasing_list=(), mono_decreasing_list=(), **kwargs): self.name = name if estimator is None: estimator = MoGLMTreeBoostRegressor(max_depth=1, val_ratio=val_ratio, verbose=verbose, random_state=random_state, **kwargs) else: if not isinstance(estimator, MoGLMTreeBoostRegressor): raise ValueError("estimator must be an instance of MoGLMTreeBoostRegressor.") super().__init__(estimator=estimator, loss_fn=torch.nn.MSELoss(reduction="none"), feature_names=feature_names, nn_temperature=nn_temperature, nn_lr=nn_lr, nn_max_epochs=nn_max_epochs, nn_n_epoch_no_change=nn_n_epoch_no_change, nn_batch_size=nn_batch_size, reg_mono=reg_mono, mono_sample_size=mono_sample_size, mono_increasing_list=mono_increasing_list, mono_decreasing_list=mono_decreasing_list, val_ratio=val_ratio, verbose=verbose, 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"}}
[docs] def set_params(self, **params): for key, value in params.items(): if key in ["name", "estimator", "val_ratio", "verbose", "device", "random_state", "nn_temperature", "nn_lr", "nn_max_epochs", "nn_n_epoch_no_change", "nn_batch_size"]: setattr(self, key, value) else: self.estimator.set_params(**{key: value}) return self
[docs] def get_params(self, deep=True): """ Get parameters for this estimator. Parameters ---------- deep : bool, default=True If True, will return the parameters for this estimator and contained subobjects that are estimators. Returns ------- params : mapping of string to any Parameter names mapped to their values. """ out = dict() for key in ["name", "estimator", "val_ratio", "verbose", "device", "random_state", "nn_temperature", "nn_lr", "nn_max_epochs", "nn_n_epoch_no_change", "nn_batch_size"]: try: value = getattr(self, key) except AttributeError: value = None out[key] = value out.update(self.estimator.get_params()) return out
def _predict(self, X): """output prediction for given samples Parameters --------- X : array-like of shape (n_samples, n_features) containing the input dataset Returns ------- np.ndarray of shape (n_samples,) containing prediction """ X = check_array(X) check_is_fitted(self) pred = self.get_aggregate_output(X).detach().cpu().numpy().ravel() return pred
[docs] class MoNeuralTreeClassifier(ClassifierMixin, NeuralTree, ModelBaseClassifier): """ A neural network-based classification model that combines GLM trees with monotonicity constraints. This model first fits a depth-1 boosted GLMTree, converts it to an equivalent neural network, and then fine-tunes the network parameters. It supports monotonicity constraints and provides interpretability through feature effects analysis. Parameters ---------- name : str, default=None Custom identifier for the model instance. estimator : MoGLMTreeBoostRegressor, default=None Pre-fitted or unfitted GLMTree regressor for initialization. If None, creates a new instance. feature_names : list of str, default=None Names of input features for interpretability. val_ratio : float, default=0.2 Proportion of data used for validation during training (0 to 1). device : str, default=None Computing device for training ('cpu', 'cuda', etc.). verbose : bool, default=False If True, prints training progress and statistics. random_state : int, default=0 Seed for reproducible random operations. nn_temperature : float, default=0.0001 Smoothing parameter for neural network activation. nn_lr : float, default=0.001 Learning rate for neural network optimization. nn_max_epochs : int, default=200 Maximum number of training epochs. nn_batch_size : int, default=200 Number of samples per training batch. nn_n_epoch_no_change : int, default=10 Early stopping patience - number of epochs without improvement. reg_mono : float, default=0.1 Strength of monotonicity regularization. mono_sample_size : int, default=1000 Number of random samples for monotonicity regularization. mono_increasing_list : tuple of str, default=() Features that should have monotonically increasing relationships. mono_decreasing_list : tuple of str, default=() Features that should have monotonically decreasing relationships. **kwargs Additional parameters passed to MoGLMTreeBoostRegressor. Attributes ---------- net_ : object The internal Pytorch network object. """ def __init__(self, name: str = None, estimator: MoGLMTreeBoostClassifier = None, feature_names=None, val_ratio=0.2, verbose=False, device=None, random_state=0, nn_temperature=0.0001, nn_lr=0.0001, nn_max_epochs=200, nn_n_epoch_no_change=10, nn_batch_size=200, reg_mono=0.1, mono_sample_size=1000, mono_increasing_list=(), mono_decreasing_list=(), **kwargs): self.name = name if estimator is None: estimator = MoGLMTreeBoostClassifier(max_depth=1, val_ratio=val_ratio, verbose=verbose, random_state=random_state, **kwargs) else: if not isinstance(estimator, MoGLMTreeBoostClassifier): raise ValueError("estimator must be an instance of MoGLMTreeBoostClassifier.") super().__init__(estimator=estimator, loss_fn=torch.nn.BCEWithLogitsLoss(reduction="none"), feature_names=feature_names, nn_temperature=nn_temperature, nn_lr=nn_lr, nn_max_epochs=nn_max_epochs, nn_n_epoch_no_change=nn_n_epoch_no_change, nn_batch_size=nn_batch_size, reg_mono=reg_mono, mono_sample_size=mono_sample_size, mono_increasing_list=mono_increasing_list, mono_decreasing_list=mono_decreasing_list, val_ratio=val_ratio, verbose=verbose, 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"}}
[docs] def set_params(self, **params): for key, value in params.items(): if key in ["name", "estimator", "val_ratio", "verbose", "device", "random_state", "nn_temperature", "nn_lr", "nn_max_epochs", "nn_n_epoch_no_change", "nn_batch_size"]: setattr(self, key, value) else: self.estimator.set_params(**{key: value}) return self
[docs] def get_params(self, deep=True): """ Get parameters for this estimator. Parameters ---------- deep : bool, default=True If True, will return the parameters for this estimator and contained subobjects that are estimators. Returns ------- params : mapping of string to any Parameter names mapped to their values. """ out = dict() for key in ["name", "estimator", "val_ratio", "verbose", "device", "random_state", "nn_temperature", "nn_lr", "nn_max_epochs", "nn_n_epoch_no_change", "nn_batch_size"]: try: value = getattr(self, key) except AttributeError: value = None out[key] = value out.update(self.estimator.get_params()) return out
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_aggregate_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