Expand source code
from copy import deepcopy

import numpy as np
from matplotlib import pyplot as plt
from sklearn import tree
from sklearn.base import BaseEstimator
from sklearn.linear_model import RidgeCV, RidgeClassifierCV
from sklearn.tree import plot_tree
from sklearn.utils import check_X_y

from imodels.tree.viz_utils import extract_sklearn_tree_from_figs


class Node:
    def __init__(self, feature: int = None, threshold: int = None,
                 value=None, idxs=None, is_root: bool = False, left=None,
                 impurity_reduction: float = None, tree_num: int = None,
                 right=None, split_or_linear='split', n_samples=0):
        """Node class for splitting
        """

        # split or linear
        self.is_root = is_root
        self.idxs = idxs
        self.tree_num = tree_num
        self.split_or_linear = split_or_linear
        self.feature = feature
        self.n_samples = n_samples
        self.impurity_reduction = impurity_reduction

        # different meanings
        self.value = value  # for split this is mean, for linear this is weight

        # split-specific (for linear these should all be None)
        self.threshold = threshold
        self.left = left
        self.right = right
        self.left_temp = None
        self.right_temp = None

    def update_values(self, X, y):
        self.value = y.mean()
        if self.threshold is not None:
            right_indicator = np.apply_along_axis(
                lambda x: x[self.feature] > self.threshold, 1, X)
            X_right = X[right_indicator, :]
            X_left = X[~right_indicator, :]
            y_right = y[right_indicator]
            y_left = y[~right_indicator]
            if self.left is not None:
                self.left.update_values(X_left, y_left)
            if self.right is not None:
                self.right.update_values(X_right, y_right)

    def shrink(self, reg_param, cum_sum=0):
        if self.is_root:
            cum_sum = self.value
        if self.left is None:  # if leaf node, change prediction
            self.value = cum_sum
        else:
            shrunk_diff = (self.left.value - self.value) / \
                (1 + reg_param / self.n_samples)
            self.left.shrink(reg_param, cum_sum + shrunk_diff)
            shrunk_diff = (self.right.value - self.value) / \
                (1 + reg_param / self.n_samples)
            self.right.shrink(reg_param, cum_sum + shrunk_diff)

    def setattrs(self, **kwargs):
        for k, v in kwargs.items():
            setattr(self, k, v)

    def __str__(self):
        if self.split_or_linear == 'linear':
            if self.is_root:
                return f'X_{self.feature} * {self.value:0.3f} (Tree #{self.tree_num} linear root)'
            else:
                return f'X_{self.feature} * {self.value:0.3f} (linear)'
        else:
            if self.is_root:
                return f'X_{self.feature} <= {self.threshold:0.3f} (Tree #{self.tree_num} root)'
            elif self.left is None and self.right is None:
                return f'Val: {self.value[0][0]:0.3f} (leaf)'
            else:
                return f'X_{self.feature} <= {self.threshold:0.3f} (split)'

    def __repr__(self):
        return self.__str__()


class FIGSExt(BaseEstimator):
    """FIGSExt (sum of trees) classifier.
    Fast Interpretable Greedy-Tree Sums (FIGS) is an algorithm for fitting concise rule-based models.
    Specifically, FIGS generalizes CART to simultaneously grow a flexible number of trees in a summation.
    The total number of splits across all the trees can be restricted by a pre-specified threshold, keeping the model interpretable.
    Experiments across a wide array of real-world datasets show that FIGS achieves state-of-the-art prediction performance when restricted to just a few splits (e.g. less than 20).
    https://arxiv.org/abs/2201.11931
    """

    def __init__(self, max_rules: int = None, posthoc_ridge: bool = False,
                 include_linear: bool = False,
                 max_features=None, min_impurity_decrease: float = 0.0,
                 k1: int = 0, k2: int = 0):
        """
        max_features
            The number of features to consider when looking for the best split
        k1: number of iterations of tree-prediction backfitting to do after making each split
        k2: number of iterations of tree-prediction backfitting to do after the end of the entire
            tree-growing phase
        """
        super().__init__()
        self.max_rules = max_rules
        self.posthoc_ridge = posthoc_ridge
        self.include_linear = include_linear
        self.max_features = max_features
        self.weighted_model_ = None  # set if using posthoc_ridge
        self.min_impurity_decrease = min_impurity_decrease
        self.k1 = k1
        self.k2 = k2
        self._init_prediction_task()  # decides between regressor and classifier

    def _init_prediction_task(self):
        """
        FIGSExtRegressor and FIGSExtClassifier override this method
        to alter the prediction task. When using this class directly,
        it is equivalent to FIGSExtRegressor
        """
        self.prediction_task = 'regression'

    def _construct_node_linear(self, X, y, idxs, tree_num=0, sample_weight=None):
        """This can be made a lot faster
        Assumes there are at least 5 points in node
        Doesn't currently support _sample_weight!
        """
        y_target = y[idxs]
        impurity_orig = np.mean(np.square(y_target)) * idxs.sum()

        # find best linear split
        best_impurity = impurity_orig
        best_linear_coef = None
        best_feature = None
        for feature_num in range(X.shape[1]):
            x = X[idxs, feature_num].reshape(-1, 1)
            m = RidgeCV(fit_intercept=False)
            m.fit(x, y_target)
            impurity = np.min(-m.best_score_) * idxs.sum()
            assert impurity >= 0, 'impurity should not be negative'
            if impurity < best_impurity:
                best_impurity = impurity
                best_linear_coef = m.coef_[0]
                best_feature = feature_num
        impurity_reduction = impurity_orig - best_impurity

        # no good linear fit found
        if impurity_reduction == 0:
            return Node(idxs=idxs, value=np.mean(y_target), tree_num=tree_num,
                        feature=None, threshold=None,
                        impurity_reduction=-1, split_or_linear='split')  # leaf node that just returns its value
        else:
            assert isinstance(best_linear_coef,
                              float), 'coef should be a float'
            return Node(idxs=idxs, value=best_linear_coef, tree_num=tree_num,
                        feature=best_feature, threshold=None,
                        impurity_reduction=impurity_reduction, split_or_linear='linear')

    def _construct_node_with_stump(self, X, y, idxs, tree_num, sample_weight=None, max_features=None):
        # array indices
        SPLIT = 0
        LEFT = 1
        RIGHT = 2

        # fit stump
        stump = tree.DecisionTreeRegressor(
            max_depth=1, max_features=max_features)
        if sample_weight is not None:
            sample_weight = sample_weight[idxs]
        stump.fit(X[idxs], y[idxs], sample_weight=sample_weight)

        # these are all arrays, arr[0] is split node
        # note: -2 is dummy
        feature = stump.tree_.feature
        threshold = stump.tree_.threshold

        impurity = stump.tree_.impurity
        n_node_samples = stump.tree_.n_node_samples
        value = stump.tree_.value

        # no split
        if len(feature) == 1:
            # print('no split found!', idxs.sum(), impurity, feature)
            return Node(idxs=idxs, value=value[SPLIT], tree_num=tree_num,
                        feature=feature[SPLIT], threshold=threshold[SPLIT],
                        impurity_reduction=-1, n_samples=n_node_samples)

        # split node
        impurity_reduction = (
            impurity[SPLIT] -
            impurity[LEFT] * n_node_samples[LEFT] / n_node_samples[SPLIT] -
            impurity[RIGHT] * n_node_samples[RIGHT] / n_node_samples[SPLIT]
        ) * idxs.sum()

        node_split = Node(idxs=idxs, value=value[SPLIT], tree_num=tree_num,
                          feature=feature[SPLIT], threshold=threshold[SPLIT],
                          impurity_reduction=impurity_reduction, n_samples=n_node_samples)
        # print('\t>>>', node_split, 'impurity', impurity, 'num_pts', idxs.sum(), 'imp_reduc', impurity_reduction)

        # manage children
        idxs_split = X[:, feature[SPLIT]] <= threshold[SPLIT]
        idxs_left = idxs_split & idxs
        idxs_right = ~idxs_split & idxs
        node_left = Node(idxs=idxs_left, value=value[LEFT], tree_num=tree_num)
        node_right = Node(
            idxs=idxs_right, value=value[RIGHT], tree_num=tree_num)
        node_split.setattrs(left_temp=node_left, right_temp=node_right, )
        return node_split

    def fit(self, X, y=None, feature_names=None, verbose=False, sample_weight=None):
        """
        Params
        ------
        _sample_weight: array-like of shape (n_samples,), default=None
            Sample weights. If None, then samples are equally weighted.
            Splits that would create child nodes with net zero or negative weight
            are ignored while searching for a split in each node.
        """

        if self.prediction_task == 'classification':
            self.classes_, y = np.unique(
                y, return_inverse=True)  # deals with str inputs
        X, y = check_X_y(X, y)
        y = y.astype(float)
        if feature_names is not None:
            self.feature_names_ = feature_names

        self.trees_ = []  # list of the root nodes of added trees
        self.complexity_ = 0  # tracks the number of rules in the model
        y_predictions_per_tree = {}  # predictions for each tree
        y_residuals_per_tree = {}  # based on predictions above

        def _update_tree_preds(n_iter):
            for k in range(n_iter):
                for tree_num_, tree_ in enumerate(self.trees_):
                    y_residuals_per_tree[tree_num_] = deepcopy(y)

                    # subtract predictions of all other trees
                    for tree_num_2_ in range(len(self.trees_)):
                        if not tree_num_2_ == tree_num_:
                            y_residuals_per_tree[tree_num_] -= y_predictions_per_tree[tree_num_2_]
                    tree_.update_values(X, y_residuals_per_tree[tree_num_])
                    y_predictions_per_tree[tree_num_] = self._predict_tree(self.trees_[
                                                                           tree_num_], X)

        # set up initial potential_splits
        # everything in potential_splits either is_root (so it can be added directly to self.trees_)
        # or it is a child of a root node that has already been added
        idxs = np.ones(X.shape[0], dtype=bool)
        node_init = self._construct_node_with_stump(X=X, y=y, idxs=idxs, tree_num=-1,
                                                    sample_weight=sample_weight, max_features=self.max_features)
        potential_splits = [node_init]
        if self.include_linear and idxs.sum() >= 5:
            node_init_linear = self._construct_node_linear(X=X, y=y, idxs=idxs, tree_num=-1,
                                                           sample_weight=sample_weight)
            potential_splits.append(node_init_linear)
        for node in potential_splits:
            node.setattrs(is_root=True)
        potential_splits = sorted(
            potential_splits, key=lambda x: x.impurity_reduction)

        # start the greedy fitting algorithm
        finished = False
        while len(potential_splits) > 0 and not finished:
            # print('potential_splits', [str(s) for s in potential_splits])
            # get node with max impurity_reduction (since it's sorted)
            split_node = potential_splits.pop()

            # don't split on node
            if split_node.impurity_reduction < self.min_impurity_decrease:
                finished = True
                break

            # split on node
            if verbose:
                print('\nadding ' + str(split_node))
            self.complexity_ += 1

            # if added a tree root
            if split_node.is_root:
                # start a new tree
                self.trees_.append(split_node)

                # update tree_num
                for node_ in [split_node, split_node.left_temp, split_node.right_temp]:
                    if node_ is not None:
                        node_.tree_num = len(self.trees_) - 1

                # add new root potential node
                node_new_root = Node(is_root=True, idxs=np.ones(X.shape[0], dtype=bool),
                                     tree_num=-1, split_or_linear=split_node.split_or_linear)
                potential_splits.append(node_new_root)

            # add children to potential splits (note this doesn't currently add linear potential splits)
            if split_node.split_or_linear == 'split':
                # assign left_temp, right_temp to be proper children
                # (basically adds them to tree in predict method)
                split_node.setattrs(left=split_node.left_temp,
                                    right=split_node.right_temp)

                # add children to potential_splits
                potential_splits.append(split_node.left)
                potential_splits.append(split_node.right)

            # update predictions for altered tree
            for tree_num_ in range(len(self.trees_)):
                y_predictions_per_tree[tree_num_] = self._predict_tree(self.trees_[
                                                                       tree_num_], X)
            # dummy 0 preds for possible new trees
            y_predictions_per_tree[-1] = np.zeros(X.shape[0])

            # update residuals for each tree
            # -1 is key for potential new tree
            for tree_num_ in list(range(len(self.trees_))) + [-1]:
                y_residuals_per_tree[tree_num_] = deepcopy(y)

                # subtract predictions of all other trees
                for tree_num_2_ in range(len(self.trees_)):
                    if not tree_num_2_ == tree_num_:
                        y_residuals_per_tree[tree_num_] -= y_predictions_per_tree[tree_num_2_]

            _update_tree_preds(self.k1)

            # recompute all impurities + update potential_split children
            potential_splits_new = []
            for potential_split in potential_splits:
                y_target = y_residuals_per_tree[potential_split.tree_num]

                if potential_split.split_or_linear == 'split':
                    # re-calculate the best split
                    potential_split_updated = self._construct_node_with_stump(X=X,
                                                                              y=y_target,
                                                                              idxs=potential_split.idxs,
                                                                              tree_num=potential_split.tree_num,
                                                                              sample_weight=sample_weight,
                                                                              max_features=self.max_features)

                    # need to preserve certain attributes from before (value at this split + is_root)
                    # value may change because residuals may have changed, but we want it to store the value from before
                    potential_split.setattrs(
                        feature=potential_split_updated.feature,
                        threshold=potential_split_updated.threshold,
                        impurity_reduction=potential_split_updated.impurity_reduction,
                        left_temp=potential_split_updated.left_temp,
                        right_temp=potential_split_updated.right_temp,
                    )
                elif potential_split.split_or_linear == 'linear':
                    assert potential_split.is_root, 'Currently, linear node only supported as root'
                    assert potential_split.idxs.sum(
                    ) == X.shape[0], 'Currently, linear node only supported as root'
                    potential_split_updated = self._construct_node_linear(idxs=potential_split.idxs,
                                                                          X=X,
                                                                          y=y_target,
                                                                          tree_num=potential_split.tree_num,
                                                                          sample_weight=sample_weight)

                    # don't need to retain anything from before (besides maybe is_root)
                    potential_split.setattrs(
                        feature=potential_split_updated.feature,
                        impurity_reduction=potential_split_updated.impurity_reduction,
                        value=potential_split_updated.value,
                    )

                # this is a valid split
                if potential_split.impurity_reduction is not None:
                    potential_splits_new.append(potential_split)

            # sort so largest impurity reduction comes last (should probs make this a heap later)
            potential_splits = sorted(
                potential_splits_new, key=lambda x: x.impurity_reduction)
            if verbose:
                print(self)
            if self.max_rules is not None and self.complexity_ >= self.max_rules:
                finished = True
                break

        _update_tree_preds(self.k2)

        # potentially fit linear model on the tree preds
        if self.posthoc_ridge:
            if self.prediction_task == 'regression':
                self.weighted_model_ = RidgeCV(
                    alphas=(0.01, 0.1, 0.5, 1.0, 5, 10))
            elif self.prediction_task == 'classification':
                self.weighted_model_ = RidgeClassifierCV(
                    alphas=(0.01, 0.1, 0.5, 1.0, 5, 10))
            X_feats = self._extract_tree_predictions(X)
            self.weighted_model_.fit(X_feats, y)
        return self

    def _tree_to_str(self, root: Node, prefix=''):
        if root is None:
            return ''
        elif root.split_or_linear == 'linear':
            return prefix + str(root)
        elif root.threshold is None:
            return ''
        pprefix = prefix + '\t'
        return prefix + str(root) + '\n' + self._tree_to_str(root.left, pprefix) + self._tree_to_str(root.right,
                                                                                                     pprefix)

    def __str__(self):
        s = '------------\n' + \
            '\n\t+\n'.join([self._tree_to_str(t) for t in self.trees_])
        if hasattr(self, 'feature_names_') and self.feature_names_ is not None:
            for i in range(len(self.feature_names_))[::-1]:
                s = s.replace(f'X_{i}', self.feature_names_[i])
        return s

    def predict(self, X):
        if self.posthoc_ridge and self.weighted_model_:  # note, during fitting don't use the weighted moel
            X_feats = self._extract_tree_predictions(X)
            return self.weighted_model_.predict(X_feats)
        preds = np.zeros(X.shape[0])
        for figs_tree in self.trees_:
            preds += self._predict_tree(figs_tree, X)
        if self.prediction_task == 'regression':
            return preds
        elif self.prediction_task == 'classification':
            return (preds > 0.5).astype(int)

    def predict_proba(self, X):
        if self.prediction_task == 'regression':
            return NotImplemented
        elif self.posthoc_ridge and self.weighted_model_:  # note, during fitting don't use the weighted moel
            X_feats = self._extract_tree_predictions(X)
            d = self.weighted_model_.decision_function(
                X_feats)  # for 2 classes, this (n_samples,)
            probs = np.exp(d) / (1 + np.exp(d))
            return np.vstack((1 - probs, probs)).transpose()
        else:
            preds = np.zeros(X.shape[0])
            for figs_tree in self.trees_:
                preds += self._predict_tree(figs_tree, X)
            # constrain to range of probabilities
            preds = np.clip(preds, a_min=0., a_max=1.)
            return np.vstack((1 - preds, preds)).transpose()

    def _extract_tree_predictions(self, X):
        """Extract predictions for all trees
        """
        X_feats = np.zeros((X.shape[0], len(self.trees_)))
        for tree_num_ in range(len(self.trees_)):
            preds_tree = self._predict_tree(self.trees_[tree_num_], X)
            X_feats[:, tree_num_] = preds_tree
        return X_feats

    def _predict_tree(self, root: Node, X):
        """Predict for a single tree
        This can be made way faster
        """

        def _predict_tree_single_point(root: Node, x):
            if root.split_or_linear == 'linear':
                return x[root.feature] * root.value
            elif root.left is None and root.right is None:
                return root.value
            left = x[root.feature] <= root.threshold
            if left:
                if root.left is None:  # we don't actually have to worry about this case
                    return root.value
                else:
                    return _predict_tree_single_point(root.left, x)
            else:
                if root.right is None:  # we don't actually have to worry about this case
                    return root.value
                else:
                    return _predict_tree_single_point(root.right, x)

        preds = np.zeros(X.shape[0])
        for i in range(X.shape[0]):
            # node values are stored as length-1 arrays; numpy >= 2 no longer
            # unwraps those implicitly when assigning into a scalar slot
            preds[i] = np.ravel(_predict_tree_single_point(root, X[i]))[0]
        return preds

    def plot(self, cols=2, feature_names=None, filename=None, label="all",
             impurity=False, tree_number=None, dpi=150, fig_size=None):
        is_single_tree = len(self.trees_) < 2 or tree_number is not None
        n_plots = int(len(self.trees_)) if tree_number is None else 1
        n_cols = 1 if is_single_tree else max(1, min(int(cols), n_plots))
        n_rows = int(np.ceil(n_plots / n_cols))
        fig, axs = plt.subplots(n_rows, n_cols, dpi=dpi, squeeze=False)
        if fig_size is not None:
            fig.set_size_inches(fig_size, fig_size)
        for ax in axs.flat[n_plots:]:
            ax.axis("off")
        n_classes = 1 if self.prediction_task == 'regression' else 2
        for i in range(n_plots):
            ax = axs.flat[i]
            try:
                dt = extract_sklearn_tree_from_figs(
                    self, i if tree_number is None else tree_number, n_classes)
                plot_tree(dt, ax=ax, feature_names=feature_names,
                          label=label, impurity=impurity)
            except IndexError:
                ax.axis('off')
                continue

            ax.set_title(f"Tree {i}")
        if filename is not None:
            plt.savefig(filename)
            return
        plt.show()


class FIGSExtRegressor(FIGSExt):
    def _init_prediction_task(self):
        self.prediction_task = 'regression'


class FIGSExtClassifier(FIGSExt):
    def _init_prediction_task(self):
        self.prediction_task = 'classification'

Classes

class FIGSExt (max_rules: int = None, posthoc_ridge: bool = False, include_linear: bool = False, max_features=None, min_impurity_decrease: float = 0.0, k1: int = 0, k2: int = 0)

FIGSExt (sum of trees) classifier. Fast Interpretable Greedy-Tree Sums (FIGS) is an algorithm for fitting concise rule-based models. Specifically, FIGS generalizes CART to simultaneously grow a flexible number of trees in a summation. The total number of splits across all the trees can be restricted by a pre-specified threshold, keeping the model interpretable. Experiments across a wide array of real-world datasets show that FIGS achieves state-of-the-art prediction performance when restricted to just a few splits (e.g. less than 20). https://arxiv.org/abs/2201.11931

max_features The number of features to consider when looking for the best split k1: number of iterations of tree-prediction backfitting to do after making each split k2: number of iterations of tree-prediction backfitting to do after the end of the entire tree-growing phase

Expand source code
class FIGSExt(BaseEstimator):
    """FIGSExt (sum of trees) classifier.
    Fast Interpretable Greedy-Tree Sums (FIGS) is an algorithm for fitting concise rule-based models.
    Specifically, FIGS generalizes CART to simultaneously grow a flexible number of trees in a summation.
    The total number of splits across all the trees can be restricted by a pre-specified threshold, keeping the model interpretable.
    Experiments across a wide array of real-world datasets show that FIGS achieves state-of-the-art prediction performance when restricted to just a few splits (e.g. less than 20).
    https://arxiv.org/abs/2201.11931
    """

    def __init__(self, max_rules: int = None, posthoc_ridge: bool = False,
                 include_linear: bool = False,
                 max_features=None, min_impurity_decrease: float = 0.0,
                 k1: int = 0, k2: int = 0):
        """
        max_features
            The number of features to consider when looking for the best split
        k1: number of iterations of tree-prediction backfitting to do after making each split
        k2: number of iterations of tree-prediction backfitting to do after the end of the entire
            tree-growing phase
        """
        super().__init__()
        self.max_rules = max_rules
        self.posthoc_ridge = posthoc_ridge
        self.include_linear = include_linear
        self.max_features = max_features
        self.weighted_model_ = None  # set if using posthoc_ridge
        self.min_impurity_decrease = min_impurity_decrease
        self.k1 = k1
        self.k2 = k2
        self._init_prediction_task()  # decides between regressor and classifier

    def _init_prediction_task(self):
        """
        FIGSExtRegressor and FIGSExtClassifier override this method
        to alter the prediction task. When using this class directly,
        it is equivalent to FIGSExtRegressor
        """
        self.prediction_task = 'regression'

    def _construct_node_linear(self, X, y, idxs, tree_num=0, sample_weight=None):
        """This can be made a lot faster
        Assumes there are at least 5 points in node
        Doesn't currently support _sample_weight!
        """
        y_target = y[idxs]
        impurity_orig = np.mean(np.square(y_target)) * idxs.sum()

        # find best linear split
        best_impurity = impurity_orig
        best_linear_coef = None
        best_feature = None
        for feature_num in range(X.shape[1]):
            x = X[idxs, feature_num].reshape(-1, 1)
            m = RidgeCV(fit_intercept=False)
            m.fit(x, y_target)
            impurity = np.min(-m.best_score_) * idxs.sum()
            assert impurity >= 0, 'impurity should not be negative'
            if impurity < best_impurity:
                best_impurity = impurity
                best_linear_coef = m.coef_[0]
                best_feature = feature_num
        impurity_reduction = impurity_orig - best_impurity

        # no good linear fit found
        if impurity_reduction == 0:
            return Node(idxs=idxs, value=np.mean(y_target), tree_num=tree_num,
                        feature=None, threshold=None,
                        impurity_reduction=-1, split_or_linear='split')  # leaf node that just returns its value
        else:
            assert isinstance(best_linear_coef,
                              float), 'coef should be a float'
            return Node(idxs=idxs, value=best_linear_coef, tree_num=tree_num,
                        feature=best_feature, threshold=None,
                        impurity_reduction=impurity_reduction, split_or_linear='linear')

    def _construct_node_with_stump(self, X, y, idxs, tree_num, sample_weight=None, max_features=None):
        # array indices
        SPLIT = 0
        LEFT = 1
        RIGHT = 2

        # fit stump
        stump = tree.DecisionTreeRegressor(
            max_depth=1, max_features=max_features)
        if sample_weight is not None:
            sample_weight = sample_weight[idxs]
        stump.fit(X[idxs], y[idxs], sample_weight=sample_weight)

        # these are all arrays, arr[0] is split node
        # note: -2 is dummy
        feature = stump.tree_.feature
        threshold = stump.tree_.threshold

        impurity = stump.tree_.impurity
        n_node_samples = stump.tree_.n_node_samples
        value = stump.tree_.value

        # no split
        if len(feature) == 1:
            # print('no split found!', idxs.sum(), impurity, feature)
            return Node(idxs=idxs, value=value[SPLIT], tree_num=tree_num,
                        feature=feature[SPLIT], threshold=threshold[SPLIT],
                        impurity_reduction=-1, n_samples=n_node_samples)

        # split node
        impurity_reduction = (
            impurity[SPLIT] -
            impurity[LEFT] * n_node_samples[LEFT] / n_node_samples[SPLIT] -
            impurity[RIGHT] * n_node_samples[RIGHT] / n_node_samples[SPLIT]
        ) * idxs.sum()

        node_split = Node(idxs=idxs, value=value[SPLIT], tree_num=tree_num,
                          feature=feature[SPLIT], threshold=threshold[SPLIT],
                          impurity_reduction=impurity_reduction, n_samples=n_node_samples)
        # print('\t>>>', node_split, 'impurity', impurity, 'num_pts', idxs.sum(), 'imp_reduc', impurity_reduction)

        # manage children
        idxs_split = X[:, feature[SPLIT]] <= threshold[SPLIT]
        idxs_left = idxs_split & idxs
        idxs_right = ~idxs_split & idxs
        node_left = Node(idxs=idxs_left, value=value[LEFT], tree_num=tree_num)
        node_right = Node(
            idxs=idxs_right, value=value[RIGHT], tree_num=tree_num)
        node_split.setattrs(left_temp=node_left, right_temp=node_right, )
        return node_split

    def fit(self, X, y=None, feature_names=None, verbose=False, sample_weight=None):
        """
        Params
        ------
        _sample_weight: array-like of shape (n_samples,), default=None
            Sample weights. If None, then samples are equally weighted.
            Splits that would create child nodes with net zero or negative weight
            are ignored while searching for a split in each node.
        """

        if self.prediction_task == 'classification':
            self.classes_, y = np.unique(
                y, return_inverse=True)  # deals with str inputs
        X, y = check_X_y(X, y)
        y = y.astype(float)
        if feature_names is not None:
            self.feature_names_ = feature_names

        self.trees_ = []  # list of the root nodes of added trees
        self.complexity_ = 0  # tracks the number of rules in the model
        y_predictions_per_tree = {}  # predictions for each tree
        y_residuals_per_tree = {}  # based on predictions above

        def _update_tree_preds(n_iter):
            for k in range(n_iter):
                for tree_num_, tree_ in enumerate(self.trees_):
                    y_residuals_per_tree[tree_num_] = deepcopy(y)

                    # subtract predictions of all other trees
                    for tree_num_2_ in range(len(self.trees_)):
                        if not tree_num_2_ == tree_num_:
                            y_residuals_per_tree[tree_num_] -= y_predictions_per_tree[tree_num_2_]
                    tree_.update_values(X, y_residuals_per_tree[tree_num_])
                    y_predictions_per_tree[tree_num_] = self._predict_tree(self.trees_[
                                                                           tree_num_], X)

        # set up initial potential_splits
        # everything in potential_splits either is_root (so it can be added directly to self.trees_)
        # or it is a child of a root node that has already been added
        idxs = np.ones(X.shape[0], dtype=bool)
        node_init = self._construct_node_with_stump(X=X, y=y, idxs=idxs, tree_num=-1,
                                                    sample_weight=sample_weight, max_features=self.max_features)
        potential_splits = [node_init]
        if self.include_linear and idxs.sum() >= 5:
            node_init_linear = self._construct_node_linear(X=X, y=y, idxs=idxs, tree_num=-1,
                                                           sample_weight=sample_weight)
            potential_splits.append(node_init_linear)
        for node in potential_splits:
            node.setattrs(is_root=True)
        potential_splits = sorted(
            potential_splits, key=lambda x: x.impurity_reduction)

        # start the greedy fitting algorithm
        finished = False
        while len(potential_splits) > 0 and not finished:
            # print('potential_splits', [str(s) for s in potential_splits])
            # get node with max impurity_reduction (since it's sorted)
            split_node = potential_splits.pop()

            # don't split on node
            if split_node.impurity_reduction < self.min_impurity_decrease:
                finished = True
                break

            # split on node
            if verbose:
                print('\nadding ' + str(split_node))
            self.complexity_ += 1

            # if added a tree root
            if split_node.is_root:
                # start a new tree
                self.trees_.append(split_node)

                # update tree_num
                for node_ in [split_node, split_node.left_temp, split_node.right_temp]:
                    if node_ is not None:
                        node_.tree_num = len(self.trees_) - 1

                # add new root potential node
                node_new_root = Node(is_root=True, idxs=np.ones(X.shape[0], dtype=bool),
                                     tree_num=-1, split_or_linear=split_node.split_or_linear)
                potential_splits.append(node_new_root)

            # add children to potential splits (note this doesn't currently add linear potential splits)
            if split_node.split_or_linear == 'split':
                # assign left_temp, right_temp to be proper children
                # (basically adds them to tree in predict method)
                split_node.setattrs(left=split_node.left_temp,
                                    right=split_node.right_temp)

                # add children to potential_splits
                potential_splits.append(split_node.left)
                potential_splits.append(split_node.right)

            # update predictions for altered tree
            for tree_num_ in range(len(self.trees_)):
                y_predictions_per_tree[tree_num_] = self._predict_tree(self.trees_[
                                                                       tree_num_], X)
            # dummy 0 preds for possible new trees
            y_predictions_per_tree[-1] = np.zeros(X.shape[0])

            # update residuals for each tree
            # -1 is key for potential new tree
            for tree_num_ in list(range(len(self.trees_))) + [-1]:
                y_residuals_per_tree[tree_num_] = deepcopy(y)

                # subtract predictions of all other trees
                for tree_num_2_ in range(len(self.trees_)):
                    if not tree_num_2_ == tree_num_:
                        y_residuals_per_tree[tree_num_] -= y_predictions_per_tree[tree_num_2_]

            _update_tree_preds(self.k1)

            # recompute all impurities + update potential_split children
            potential_splits_new = []
            for potential_split in potential_splits:
                y_target = y_residuals_per_tree[potential_split.tree_num]

                if potential_split.split_or_linear == 'split':
                    # re-calculate the best split
                    potential_split_updated = self._construct_node_with_stump(X=X,
                                                                              y=y_target,
                                                                              idxs=potential_split.idxs,
                                                                              tree_num=potential_split.tree_num,
                                                                              sample_weight=sample_weight,
                                                                              max_features=self.max_features)

                    # need to preserve certain attributes from before (value at this split + is_root)
                    # value may change because residuals may have changed, but we want it to store the value from before
                    potential_split.setattrs(
                        feature=potential_split_updated.feature,
                        threshold=potential_split_updated.threshold,
                        impurity_reduction=potential_split_updated.impurity_reduction,
                        left_temp=potential_split_updated.left_temp,
                        right_temp=potential_split_updated.right_temp,
                    )
                elif potential_split.split_or_linear == 'linear':
                    assert potential_split.is_root, 'Currently, linear node only supported as root'
                    assert potential_split.idxs.sum(
                    ) == X.shape[0], 'Currently, linear node only supported as root'
                    potential_split_updated = self._construct_node_linear(idxs=potential_split.idxs,
                                                                          X=X,
                                                                          y=y_target,
                                                                          tree_num=potential_split.tree_num,
                                                                          sample_weight=sample_weight)

                    # don't need to retain anything from before (besides maybe is_root)
                    potential_split.setattrs(
                        feature=potential_split_updated.feature,
                        impurity_reduction=potential_split_updated.impurity_reduction,
                        value=potential_split_updated.value,
                    )

                # this is a valid split
                if potential_split.impurity_reduction is not None:
                    potential_splits_new.append(potential_split)

            # sort so largest impurity reduction comes last (should probs make this a heap later)
            potential_splits = sorted(
                potential_splits_new, key=lambda x: x.impurity_reduction)
            if verbose:
                print(self)
            if self.max_rules is not None and self.complexity_ >= self.max_rules:
                finished = True
                break

        _update_tree_preds(self.k2)

        # potentially fit linear model on the tree preds
        if self.posthoc_ridge:
            if self.prediction_task == 'regression':
                self.weighted_model_ = RidgeCV(
                    alphas=(0.01, 0.1, 0.5, 1.0, 5, 10))
            elif self.prediction_task == 'classification':
                self.weighted_model_ = RidgeClassifierCV(
                    alphas=(0.01, 0.1, 0.5, 1.0, 5, 10))
            X_feats = self._extract_tree_predictions(X)
            self.weighted_model_.fit(X_feats, y)
        return self

    def _tree_to_str(self, root: Node, prefix=''):
        if root is None:
            return ''
        elif root.split_or_linear == 'linear':
            return prefix + str(root)
        elif root.threshold is None:
            return ''
        pprefix = prefix + '\t'
        return prefix + str(root) + '\n' + self._tree_to_str(root.left, pprefix) + self._tree_to_str(root.right,
                                                                                                     pprefix)

    def __str__(self):
        s = '------------\n' + \
            '\n\t+\n'.join([self._tree_to_str(t) for t in self.trees_])
        if hasattr(self, 'feature_names_') and self.feature_names_ is not None:
            for i in range(len(self.feature_names_))[::-1]:
                s = s.replace(f'X_{i}', self.feature_names_[i])
        return s

    def predict(self, X):
        if self.posthoc_ridge and self.weighted_model_:  # note, during fitting don't use the weighted moel
            X_feats = self._extract_tree_predictions(X)
            return self.weighted_model_.predict(X_feats)
        preds = np.zeros(X.shape[0])
        for figs_tree in self.trees_:
            preds += self._predict_tree(figs_tree, X)
        if self.prediction_task == 'regression':
            return preds
        elif self.prediction_task == 'classification':
            return (preds > 0.5).astype(int)

    def predict_proba(self, X):
        if self.prediction_task == 'regression':
            return NotImplemented
        elif self.posthoc_ridge and self.weighted_model_:  # note, during fitting don't use the weighted moel
            X_feats = self._extract_tree_predictions(X)
            d = self.weighted_model_.decision_function(
                X_feats)  # for 2 classes, this (n_samples,)
            probs = np.exp(d) / (1 + np.exp(d))
            return np.vstack((1 - probs, probs)).transpose()
        else:
            preds = np.zeros(X.shape[0])
            for figs_tree in self.trees_:
                preds += self._predict_tree(figs_tree, X)
            # constrain to range of probabilities
            preds = np.clip(preds, a_min=0., a_max=1.)
            return np.vstack((1 - preds, preds)).transpose()

    def _extract_tree_predictions(self, X):
        """Extract predictions for all trees
        """
        X_feats = np.zeros((X.shape[0], len(self.trees_)))
        for tree_num_ in range(len(self.trees_)):
            preds_tree = self._predict_tree(self.trees_[tree_num_], X)
            X_feats[:, tree_num_] = preds_tree
        return X_feats

    def _predict_tree(self, root: Node, X):
        """Predict for a single tree
        This can be made way faster
        """

        def _predict_tree_single_point(root: Node, x):
            if root.split_or_linear == 'linear':
                return x[root.feature] * root.value
            elif root.left is None and root.right is None:
                return root.value
            left = x[root.feature] <= root.threshold
            if left:
                if root.left is None:  # we don't actually have to worry about this case
                    return root.value
                else:
                    return _predict_tree_single_point(root.left, x)
            else:
                if root.right is None:  # we don't actually have to worry about this case
                    return root.value
                else:
                    return _predict_tree_single_point(root.right, x)

        preds = np.zeros(X.shape[0])
        for i in range(X.shape[0]):
            # node values are stored as length-1 arrays; numpy >= 2 no longer
            # unwraps those implicitly when assigning into a scalar slot
            preds[i] = np.ravel(_predict_tree_single_point(root, X[i]))[0]
        return preds

    def plot(self, cols=2, feature_names=None, filename=None, label="all",
             impurity=False, tree_number=None, dpi=150, fig_size=None):
        is_single_tree = len(self.trees_) < 2 or tree_number is not None
        n_plots = int(len(self.trees_)) if tree_number is None else 1
        n_cols = 1 if is_single_tree else max(1, min(int(cols), n_plots))
        n_rows = int(np.ceil(n_plots / n_cols))
        fig, axs = plt.subplots(n_rows, n_cols, dpi=dpi, squeeze=False)
        if fig_size is not None:
            fig.set_size_inches(fig_size, fig_size)
        for ax in axs.flat[n_plots:]:
            ax.axis("off")
        n_classes = 1 if self.prediction_task == 'regression' else 2
        for i in range(n_plots):
            ax = axs.flat[i]
            try:
                dt = extract_sklearn_tree_from_figs(
                    self, i if tree_number is None else tree_number, n_classes)
                plot_tree(dt, ax=ax, feature_names=feature_names,
                          label=label, impurity=impurity)
            except IndexError:
                ax.axis('off')
                continue

            ax.set_title(f"Tree {i}")
        if filename is not None:
            plt.savefig(filename)
            return
        plt.show()

Ancestors

  • sklearn.base.BaseEstimator
  • sklearn.utils._repr_html.base.ReprHTMLMixin
  • sklearn.utils._repr_html.base._HTMLDocumentationLinkMixin
  • sklearn.utils._metadata_requests._MetadataRequester

Subclasses

Methods

def fit(self, X, y=None, feature_names=None, verbose=False, sample_weight=None)

Params

_sample_weight: array-like of shape (n_samples,), default=None Sample weights. If None, then samples are equally weighted. Splits that would create child nodes with net zero or negative weight are ignored while searching for a split in each node.

Expand source code
def fit(self, X, y=None, feature_names=None, verbose=False, sample_weight=None):
    """
    Params
    ------
    _sample_weight: array-like of shape (n_samples,), default=None
        Sample weights. If None, then samples are equally weighted.
        Splits that would create child nodes with net zero or negative weight
        are ignored while searching for a split in each node.
    """

    if self.prediction_task == 'classification':
        self.classes_, y = np.unique(
            y, return_inverse=True)  # deals with str inputs
    X, y = check_X_y(X, y)
    y = y.astype(float)
    if feature_names is not None:
        self.feature_names_ = feature_names

    self.trees_ = []  # list of the root nodes of added trees
    self.complexity_ = 0  # tracks the number of rules in the model
    y_predictions_per_tree = {}  # predictions for each tree
    y_residuals_per_tree = {}  # based on predictions above

    def _update_tree_preds(n_iter):
        for k in range(n_iter):
            for tree_num_, tree_ in enumerate(self.trees_):
                y_residuals_per_tree[tree_num_] = deepcopy(y)

                # subtract predictions of all other trees
                for tree_num_2_ in range(len(self.trees_)):
                    if not tree_num_2_ == tree_num_:
                        y_residuals_per_tree[tree_num_] -= y_predictions_per_tree[tree_num_2_]
                tree_.update_values(X, y_residuals_per_tree[tree_num_])
                y_predictions_per_tree[tree_num_] = self._predict_tree(self.trees_[
                                                                       tree_num_], X)

    # set up initial potential_splits
    # everything in potential_splits either is_root (so it can be added directly to self.trees_)
    # or it is a child of a root node that has already been added
    idxs = np.ones(X.shape[0], dtype=bool)
    node_init = self._construct_node_with_stump(X=X, y=y, idxs=idxs, tree_num=-1,
                                                sample_weight=sample_weight, max_features=self.max_features)
    potential_splits = [node_init]
    if self.include_linear and idxs.sum() >= 5:
        node_init_linear = self._construct_node_linear(X=X, y=y, idxs=idxs, tree_num=-1,
                                                       sample_weight=sample_weight)
        potential_splits.append(node_init_linear)
    for node in potential_splits:
        node.setattrs(is_root=True)
    potential_splits = sorted(
        potential_splits, key=lambda x: x.impurity_reduction)

    # start the greedy fitting algorithm
    finished = False
    while len(potential_splits) > 0 and not finished:
        # print('potential_splits', [str(s) for s in potential_splits])
        # get node with max impurity_reduction (since it's sorted)
        split_node = potential_splits.pop()

        # don't split on node
        if split_node.impurity_reduction < self.min_impurity_decrease:
            finished = True
            break

        # split on node
        if verbose:
            print('\nadding ' + str(split_node))
        self.complexity_ += 1

        # if added a tree root
        if split_node.is_root:
            # start a new tree
            self.trees_.append(split_node)

            # update tree_num
            for node_ in [split_node, split_node.left_temp, split_node.right_temp]:
                if node_ is not None:
                    node_.tree_num = len(self.trees_) - 1

            # add new root potential node
            node_new_root = Node(is_root=True, idxs=np.ones(X.shape[0], dtype=bool),
                                 tree_num=-1, split_or_linear=split_node.split_or_linear)
            potential_splits.append(node_new_root)

        # add children to potential splits (note this doesn't currently add linear potential splits)
        if split_node.split_or_linear == 'split':
            # assign left_temp, right_temp to be proper children
            # (basically adds them to tree in predict method)
            split_node.setattrs(left=split_node.left_temp,
                                right=split_node.right_temp)

            # add children to potential_splits
            potential_splits.append(split_node.left)
            potential_splits.append(split_node.right)

        # update predictions for altered tree
        for tree_num_ in range(len(self.trees_)):
            y_predictions_per_tree[tree_num_] = self._predict_tree(self.trees_[
                                                                   tree_num_], X)
        # dummy 0 preds for possible new trees
        y_predictions_per_tree[-1] = np.zeros(X.shape[0])

        # update residuals for each tree
        # -1 is key for potential new tree
        for tree_num_ in list(range(len(self.trees_))) + [-1]:
            y_residuals_per_tree[tree_num_] = deepcopy(y)

            # subtract predictions of all other trees
            for tree_num_2_ in range(len(self.trees_)):
                if not tree_num_2_ == tree_num_:
                    y_residuals_per_tree[tree_num_] -= y_predictions_per_tree[tree_num_2_]

        _update_tree_preds(self.k1)

        # recompute all impurities + update potential_split children
        potential_splits_new = []
        for potential_split in potential_splits:
            y_target = y_residuals_per_tree[potential_split.tree_num]

            if potential_split.split_or_linear == 'split':
                # re-calculate the best split
                potential_split_updated = self._construct_node_with_stump(X=X,
                                                                          y=y_target,
                                                                          idxs=potential_split.idxs,
                                                                          tree_num=potential_split.tree_num,
                                                                          sample_weight=sample_weight,
                                                                          max_features=self.max_features)

                # need to preserve certain attributes from before (value at this split + is_root)
                # value may change because residuals may have changed, but we want it to store the value from before
                potential_split.setattrs(
                    feature=potential_split_updated.feature,
                    threshold=potential_split_updated.threshold,
                    impurity_reduction=potential_split_updated.impurity_reduction,
                    left_temp=potential_split_updated.left_temp,
                    right_temp=potential_split_updated.right_temp,
                )
            elif potential_split.split_or_linear == 'linear':
                assert potential_split.is_root, 'Currently, linear node only supported as root'
                assert potential_split.idxs.sum(
                ) == X.shape[0], 'Currently, linear node only supported as root'
                potential_split_updated = self._construct_node_linear(idxs=potential_split.idxs,
                                                                      X=X,
                                                                      y=y_target,
                                                                      tree_num=potential_split.tree_num,
                                                                      sample_weight=sample_weight)

                # don't need to retain anything from before (besides maybe is_root)
                potential_split.setattrs(
                    feature=potential_split_updated.feature,
                    impurity_reduction=potential_split_updated.impurity_reduction,
                    value=potential_split_updated.value,
                )

            # this is a valid split
            if potential_split.impurity_reduction is not None:
                potential_splits_new.append(potential_split)

        # sort so largest impurity reduction comes last (should probs make this a heap later)
        potential_splits = sorted(
            potential_splits_new, key=lambda x: x.impurity_reduction)
        if verbose:
            print(self)
        if self.max_rules is not None and self.complexity_ >= self.max_rules:
            finished = True
            break

    _update_tree_preds(self.k2)

    # potentially fit linear model on the tree preds
    if self.posthoc_ridge:
        if self.prediction_task == 'regression':
            self.weighted_model_ = RidgeCV(
                alphas=(0.01, 0.1, 0.5, 1.0, 5, 10))
        elif self.prediction_task == 'classification':
            self.weighted_model_ = RidgeClassifierCV(
                alphas=(0.01, 0.1, 0.5, 1.0, 5, 10))
        X_feats = self._extract_tree_predictions(X)
        self.weighted_model_.fit(X_feats, y)
    return self
def plot(self, cols=2, feature_names=None, filename=None, label='all', impurity=False, tree_number=None, dpi=150, fig_size=None)
Expand source code
def plot(self, cols=2, feature_names=None, filename=None, label="all",
         impurity=False, tree_number=None, dpi=150, fig_size=None):
    is_single_tree = len(self.trees_) < 2 or tree_number is not None
    n_plots = int(len(self.trees_)) if tree_number is None else 1
    n_cols = 1 if is_single_tree else max(1, min(int(cols), n_plots))
    n_rows = int(np.ceil(n_plots / n_cols))
    fig, axs = plt.subplots(n_rows, n_cols, dpi=dpi, squeeze=False)
    if fig_size is not None:
        fig.set_size_inches(fig_size, fig_size)
    for ax in axs.flat[n_plots:]:
        ax.axis("off")
    n_classes = 1 if self.prediction_task == 'regression' else 2
    for i in range(n_plots):
        ax = axs.flat[i]
        try:
            dt = extract_sklearn_tree_from_figs(
                self, i if tree_number is None else tree_number, n_classes)
            plot_tree(dt, ax=ax, feature_names=feature_names,
                      label=label, impurity=impurity)
        except IndexError:
            ax.axis('off')
            continue

        ax.set_title(f"Tree {i}")
    if filename is not None:
        plt.savefig(filename)
        return
    plt.show()
def predict(self, X)
Expand source code
def predict(self, X):
    if self.posthoc_ridge and self.weighted_model_:  # note, during fitting don't use the weighted moel
        X_feats = self._extract_tree_predictions(X)
        return self.weighted_model_.predict(X_feats)
    preds = np.zeros(X.shape[0])
    for figs_tree in self.trees_:
        preds += self._predict_tree(figs_tree, X)
    if self.prediction_task == 'regression':
        return preds
    elif self.prediction_task == 'classification':
        return (preds > 0.5).astype(int)
def predict_proba(self, X)
Expand source code
def predict_proba(self, X):
    if self.prediction_task == 'regression':
        return NotImplemented
    elif self.posthoc_ridge and self.weighted_model_:  # note, during fitting don't use the weighted moel
        X_feats = self._extract_tree_predictions(X)
        d = self.weighted_model_.decision_function(
            X_feats)  # for 2 classes, this (n_samples,)
        probs = np.exp(d) / (1 + np.exp(d))
        return np.vstack((1 - probs, probs)).transpose()
    else:
        preds = np.zeros(X.shape[0])
        for figs_tree in self.trees_:
            preds += self._predict_tree(figs_tree, X)
        # constrain to range of probabilities
        preds = np.clip(preds, a_min=0., a_max=1.)
        return np.vstack((1 - preds, preds)).transpose()
def set_fit_request(self: FIGSExt, *, feature_names: bool | str | None = '$UNCHANGED$', sample_weight: bool | str | None = '$UNCHANGED$', verbose: bool | str | None = '$UNCHANGED$') ‑> FIGSExt

Configure whether metadata should be requested to be passed to the fit method.

Note that this method is only relevant when this estimator is used as a sub-estimator within a :term:meta-estimator and metadata routing is enabled with enable_metadata_routing=True (see :func:sklearn.set_config). Please check the :ref:User Guide <metadata_routing> on how the routing mechanism works.

The options for each parameter are:

  • True: metadata is requested, and passed to fit if provided. The request is ignored if metadata is not provided.

  • False: metadata is not requested and the meta-estimator will not pass it to fit.

  • None: metadata is not requested, and the meta-estimator will raise an error if the user provides it.

  • str: metadata should be passed to the meta-estimator with this given alias instead of the original name.

The default (sklearn.utils.metadata_routing.UNCHANGED) retains the existing request. This allows you to change the request for some parameters and not others.

Added in version: 1.3

Parameters

feature_names : str, True, False, or None, default=sklearn.utils.metadata_routing.UNCHANGED
Metadata routing for feature_names parameter in fit.
sample_weight : str, True, False, or None, default=sklearn.utils.metadata_routing.UNCHANGED
Metadata routing for sample_weight parameter in fit.
verbose : str, True, False, or None, default=sklearn.utils.metadata_routing.UNCHANGED
Metadata routing for verbose parameter in fit.

Returns

self : object
The updated object.
Expand source code
def func(*args, **kw):
    """Updates the `_metadata_request` attribute of the consumer (`instance`)
    for the parameters provided as `**kw`.

    This docstring is overwritten below.
    See REQUESTER_DOC for expected functionality.
    """
    if not _routing_enabled():
        raise RuntimeError(
            "This method is only available when metadata routing is enabled."
            " You can enable it using"
            " sklearn.set_config(enable_metadata_routing=True)."
        )

    if self.validate_keys and (set(kw) - set(self.keys)):
        raise TypeError(
            f"Unexpected args: {set(kw) - set(self.keys)} in {self.name}. "
            f"Accepted arguments are: {set(self.keys)}"
        )

    # This makes it possible to use the decorated method as an unbound method,
    # for instance when monkeypatching.
    # https://github.com/scikit-learn/scikit-learn/issues/28632
    if instance is None:
        _instance = args[0]
        args = args[1:]
    else:
        _instance = instance

    # Replicating python's behavior when positional args are given other than
    # `self`, and `self` is only allowed if this method is unbound.
    if args:
        raise TypeError(
            f"set_{self.name}_request() takes 0 positional argument but"
            f" {len(args)} were given"
        )

    requests = _instance._get_metadata_request()
    method_metadata_request = getattr(requests, self.name)

    for prop, alias in kw.items():
        if alias is not UNCHANGED:
            method_metadata_request.add_request(param=prop, alias=alias)
    _instance._metadata_request = requests

    return _instance
class FIGSExtClassifier (max_rules: int = None, posthoc_ridge: bool = False, include_linear: bool = False, max_features=None, min_impurity_decrease: float = 0.0, k1: int = 0, k2: int = 0)

FIGSExt (sum of trees) classifier. Fast Interpretable Greedy-Tree Sums (FIGS) is an algorithm for fitting concise rule-based models. Specifically, FIGS generalizes CART to simultaneously grow a flexible number of trees in a summation. The total number of splits across all the trees can be restricted by a pre-specified threshold, keeping the model interpretable. Experiments across a wide array of real-world datasets show that FIGS achieves state-of-the-art prediction performance when restricted to just a few splits (e.g. less than 20). https://arxiv.org/abs/2201.11931

max_features The number of features to consider when looking for the best split k1: number of iterations of tree-prediction backfitting to do after making each split k2: number of iterations of tree-prediction backfitting to do after the end of the entire tree-growing phase

Expand source code
class FIGSExtClassifier(FIGSExt):
    def _init_prediction_task(self):
        self.prediction_task = 'classification'

Ancestors

  • FIGSExt
  • sklearn.base.BaseEstimator
  • sklearn.utils._repr_html.base.ReprHTMLMixin
  • sklearn.utils._repr_html.base._HTMLDocumentationLinkMixin
  • sklearn.utils._metadata_requests._MetadataRequester

Inherited members

class FIGSExtRegressor (max_rules: int = None, posthoc_ridge: bool = False, include_linear: bool = False, max_features=None, min_impurity_decrease: float = 0.0, k1: int = 0, k2: int = 0)

FIGSExt (sum of trees) classifier. Fast Interpretable Greedy-Tree Sums (FIGS) is an algorithm for fitting concise rule-based models. Specifically, FIGS generalizes CART to simultaneously grow a flexible number of trees in a summation. The total number of splits across all the trees can be restricted by a pre-specified threshold, keeping the model interpretable. Experiments across a wide array of real-world datasets show that FIGS achieves state-of-the-art prediction performance when restricted to just a few splits (e.g. less than 20). https://arxiv.org/abs/2201.11931

max_features The number of features to consider when looking for the best split k1: number of iterations of tree-prediction backfitting to do after making each split k2: number of iterations of tree-prediction backfitting to do after the end of the entire tree-growing phase

Expand source code
class FIGSExtRegressor(FIGSExt):
    def _init_prediction_task(self):
        self.prediction_task = 'regression'

Ancestors

  • FIGSExt
  • sklearn.base.BaseEstimator
  • sklearn.utils._repr_html.base.ReprHTMLMixin
  • sklearn.utils._repr_html.base._HTMLDocumentationLinkMixin
  • sklearn.utils._metadata_requests._MetadataRequester

Inherited members

class Node (feature: int = None, threshold: int = None, value=None, idxs=None, is_root: bool = False, left=None, impurity_reduction: float = None, tree_num: int = None, right=None, split_or_linear='split', n_samples=0)

Node class for splitting

Expand source code
class Node:
    def __init__(self, feature: int = None, threshold: int = None,
                 value=None, idxs=None, is_root: bool = False, left=None,
                 impurity_reduction: float = None, tree_num: int = None,
                 right=None, split_or_linear='split', n_samples=0):
        """Node class for splitting
        """

        # split or linear
        self.is_root = is_root
        self.idxs = idxs
        self.tree_num = tree_num
        self.split_or_linear = split_or_linear
        self.feature = feature
        self.n_samples = n_samples
        self.impurity_reduction = impurity_reduction

        # different meanings
        self.value = value  # for split this is mean, for linear this is weight

        # split-specific (for linear these should all be None)
        self.threshold = threshold
        self.left = left
        self.right = right
        self.left_temp = None
        self.right_temp = None

    def update_values(self, X, y):
        self.value = y.mean()
        if self.threshold is not None:
            right_indicator = np.apply_along_axis(
                lambda x: x[self.feature] > self.threshold, 1, X)
            X_right = X[right_indicator, :]
            X_left = X[~right_indicator, :]
            y_right = y[right_indicator]
            y_left = y[~right_indicator]
            if self.left is not None:
                self.left.update_values(X_left, y_left)
            if self.right is not None:
                self.right.update_values(X_right, y_right)

    def shrink(self, reg_param, cum_sum=0):
        if self.is_root:
            cum_sum = self.value
        if self.left is None:  # if leaf node, change prediction
            self.value = cum_sum
        else:
            shrunk_diff = (self.left.value - self.value) / \
                (1 + reg_param / self.n_samples)
            self.left.shrink(reg_param, cum_sum + shrunk_diff)
            shrunk_diff = (self.right.value - self.value) / \
                (1 + reg_param / self.n_samples)
            self.right.shrink(reg_param, cum_sum + shrunk_diff)

    def setattrs(self, **kwargs):
        for k, v in kwargs.items():
            setattr(self, k, v)

    def __str__(self):
        if self.split_or_linear == 'linear':
            if self.is_root:
                return f'X_{self.feature} * {self.value:0.3f} (Tree #{self.tree_num} linear root)'
            else:
                return f'X_{self.feature} * {self.value:0.3f} (linear)'
        else:
            if self.is_root:
                return f'X_{self.feature} <= {self.threshold:0.3f} (Tree #{self.tree_num} root)'
            elif self.left is None and self.right is None:
                return f'Val: {self.value[0][0]:0.3f} (leaf)'
            else:
                return f'X_{self.feature} <= {self.threshold:0.3f} (split)'

    def __repr__(self):
        return self.__str__()

Methods

def setattrs(self, **kwargs)
Expand source code
def setattrs(self, **kwargs):
    for k, v in kwargs.items():
        setattr(self, k, v)
def shrink(self, reg_param, cum_sum=0)
Expand source code
def shrink(self, reg_param, cum_sum=0):
    if self.is_root:
        cum_sum = self.value
    if self.left is None:  # if leaf node, change prediction
        self.value = cum_sum
    else:
        shrunk_diff = (self.left.value - self.value) / \
            (1 + reg_param / self.n_samples)
        self.left.shrink(reg_param, cum_sum + shrunk_diff)
        shrunk_diff = (self.right.value - self.value) / \
            (1 + reg_param / self.n_samples)
        self.right.shrink(reg_param, cum_sum + shrunk_diff)
def update_values(self, X, y)
Expand source code
def update_values(self, X, y):
    self.value = y.mean()
    if self.threshold is not None:
        right_indicator = np.apply_along_axis(
            lambda x: x[self.feature] > self.threshold, 1, X)
        X_right = X[right_indicator, :]
        X_left = X[~right_indicator, :]
        y_right = y[right_indicator]
        y_left = y[~right_indicator]
        if self.left is not None:
            self.left.update_values(X_left, y_left)
        if self.right is not None:
            self.right.update_values(X_right, y_right)