from __future__ import annotations
from typing import Any
import numpy as np
from ConfigSpace import ConfigurationSpace
from pyrfr import regression
from pyrfr.regression import binary_rss_forest as BinaryForest
from pyrfr.regression import default_data_container as DataContainer
from smac.constants import N_TREES, VERY_SMALL_NUMBER
from smac.model.random_forest import AbstractRandomForest
__copyright__ = "Copyright 2022, automl.org"
__license__ = "3-clause BSD"
[docs]
class RandomForest(AbstractRandomForest):
"""Random forest that takes instance features into account.
Parameters
----------
n_trees : int, defaults to `N_TREES`
The number of trees in the random forest.
n_points_per_tree : int, defaults to -1
Number of points per tree. If the value is smaller than 0, the number of samples will be used.
ratio_features : float, defaults to 5.0 / 6.0
The ratio of features that are considered for splitting.
min_samples_split : int, defaults to 3
The minimum number of data points to perform a split.
min_samples_leaf : int, defaults to 3
The minimum number of data points in a leaf.
max_depth : int, defaults to 2**20
The maximum depth of a single tree.
eps_purity : float, defaults to 1e-8
The minimum difference between two target values to be considered.
max_nodes : int, defaults to 2**20
The maximum total number of nodes in a tree.
bootstrapping : bool, defaults to True
Enables bootstrapping.
log_y: bool, defaults to False
The y values (passed to this random forest) are expected to be log(y) transformed.
This will be considered during predicting.
instance_features : dict[str, list[int | float]] | None, defaults to None
Features (list of int or floats) of the instances (str). The features are incorporated into the X data,
on which the model is trained on.
pca_components : float, defaults to 7
Number of components to keep when using PCA to reduce dimensionality of instance features.
seed : int
"""
def __init__(
self,
configspace: ConfigurationSpace,
n_trees: int = N_TREES,
n_points_per_tree: int = -1,
ratio_features: float = 5.0 / 6.0,
min_samples_split: int = 3,
min_samples_leaf: int = 3,
max_depth: int = 2**20,
eps_purity: float = 1e-8,
max_nodes: int = 2**20,
bootstrapping: bool = True,
log_y: bool = False,
instance_features: dict[str, list[int | float]] | None = None,
pca_components: int | None = 7,
seed: int = 0,
) -> None:
super().__init__(
configspace=configspace,
instance_features=instance_features,
pca_components=pca_components,
seed=seed,
)
max_features = 0 if ratio_features > 1.0 else max(1, int(len(self._types) * ratio_features))
self._rf_opts = regression.forest_opts()
self._rf_opts.num_trees = n_trees
self._rf_opts.do_bootstrapping = bootstrapping
self._rf_opts.tree_opts.max_features = max_features
self._rf_opts.tree_opts.min_samples_to_split = min_samples_split
self._rf_opts.tree_opts.min_samples_in_leaf = min_samples_leaf
self._rf_opts.tree_opts.max_depth = max_depth
self._rf_opts.tree_opts.epsilon_purity = eps_purity
self._rf_opts.tree_opts.max_num_nodes = max_nodes
self._rf_opts.compute_law_of_total_variance = False
self._rf: BinaryForest | None = None
self._log_y = log_y
# Case to `int` incase we get an `np.integer` type
self._rng = regression.default_random_engine(int(seed))
self._n_trees = n_trees
self._n_points_per_tree = n_points_per_tree
self._ratio_features = ratio_features
self._min_samples_split = min_samples_split
self._min_samples_leaf = min_samples_leaf
self._max_depth = max_depth
self._eps_purity = eps_purity
self._max_nodes = max_nodes
self._bootstrapping = bootstrapping
# This list well be read out by save_iteration() in the solver
# self._hypers = [
# n_trees,
# max_nodes,
# bootstrapping,
# n_points_per_tree,
# ratio_features,
# min_samples_split,
# min_samples_leaf,
# max_depth,
# eps_purity,
# self._seed,
# ]
@property
def meta(self) -> dict[str, Any]: # noqa: D102
meta = super().meta
meta.update(
{
"n_trees": self._n_trees,
"n_points_per_tree": self._n_points_per_tree,
"ratio_features": self._ratio_features,
"min_samples_split": self._min_samples_split,
"min_samples_leaf": self._min_samples_leaf,
"max_depth": self._max_depth,
"eps_purity": self._eps_purity,
"max_nodes": self._max_nodes,
"bootstrapping": self._bootstrapping,
"pca_components": self._pca_components,
}
)
return meta
def _train(self, X: np.ndarray, y: np.ndarray) -> RandomForest:
X = self._impute_inactive(X)
y = y.flatten()
# self.X = X
# self.y = y.flatten()
if self._n_points_per_tree <= 0:
self._rf_opts.num_data_points_per_tree = X.shape[0]
else:
self._rf_opts.num_data_points_per_tree = self._n_points_per_tree
self._rf = regression.binary_rss_forest()
self._rf.options = self._rf_opts
data = self._init_data_container(X, y)
self._rf.fit(data, rng=self._rng)
return self
def _init_data_container(self, X: np.ndarray, y: np.ndarray) -> DataContainer:
"""Fills a pyrfr default data container s.t. the forest knows categoricals and bounds for continous data.
Parameters
----------
X : np.ndarray [#samples, #hyperparameter + #features]
Input data points.
Y : np.ndarray [#samples, #objectives]
The corresponding target values.
Returns
-------
data : DataContainer
The filled data container that pyrfr can interpret.
"""
# Retrieve the types and the bounds from the ConfigSpace
data = regression.default_data_container(X.shape[1])
for i, (mn, mx) in enumerate(self._bounds):
if np.isnan(mx):
data.set_type_of_feature(i, mn)
else:
data.set_bounds_of_feature(i, mn, mx)
for row_X, row_y in zip(X, y):
data.add_data_point(row_X, row_y)
return data
def _predict(
self,
X: np.ndarray,
covariance_type: str | None = "diagonal",
) -> tuple[np.ndarray, np.ndarray | None]:
if len(X.shape) != 2:
raise ValueError("Expected 2d array, got %dd array!" % len(X.shape))
if X.shape[1] != len(self._types):
raise ValueError("Rows in X should have %d entries but have %d!" % (len(self._types), X.shape[1]))
if covariance_type != "diagonal":
raise ValueError("`covariance_type` can only take `diagonal` for this model.")
assert self._rf is not None
X = self._impute_inactive(X)
if self._log_y:
all_preds = []
third_dimension = 0
# Gather data in a list of 2d arrays and get statistics about the required size of the 3d array
for row_X in X:
preds_per_tree = self._rf.all_leaf_values(row_X)
all_preds.append(preds_per_tree)
max_num_leaf_data = max(map(len, preds_per_tree))
third_dimension = max(max_num_leaf_data, third_dimension)
# Transform list of 2d arrays into a 3d array
preds_as_array = np.zeros((X.shape[0], self._rf_opts.num_trees, third_dimension)) * np.nan
for i, preds_per_tree in enumerate(all_preds):
for j, pred in enumerate(preds_per_tree):
preds_as_array[i, j, : len(pred)] = pred
# Do all necessary computation with vectorized functions
preds_as_array = np.log(np.nanmean(np.exp(preds_as_array), axis=2) + VERY_SMALL_NUMBER)
# Compute the mean and the variance across the different trees
means = preds_as_array.mean(axis=1)
vars_ = preds_as_array.var(axis=1)
else:
means, vars_ = [], []
for row_X in X:
mean_, var = self._rf.predict_mean_var(row_X)
means.append(mean_)
vars_.append(var)
means = np.array(means)
vars_ = np.array(vars_)
return means.reshape((-1, 1)), vars_.reshape((-1, 1))
[docs]
def predict_marginalized(self, X: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
"""Predicts mean and variance marginalized over all instances.
Note
----
The method is random forest specific and follows the SMAC2 implementation. It requires
no distribution assumption to marginalize the uncertainty estimates.
Parameters
----------
X : np.ndarray [#samples, #hyperparameter + #features]
Input data points.
Returns
-------
means : np.ndarray [#samples, 1]
The predictive mean.
vars : np.ndarray [#samples, 1]
The predictive variance.
"""
if self._n_features == 0:
mean_, var = self.predict(X)
assert var is not None
var[var < self._var_threshold] = self._var_threshold
var[np.isnan(var)] = self._var_threshold
return mean_, var
assert self._instance_features is not None
if len(X.shape) != 2:
raise ValueError("Expected 2d array, got %dd array!" % len(X.shape))
if X.shape[1] != len(self._bounds):
raise ValueError("Rows in X should have %d entries but have %d!" % (len(self._bounds), X.shape[1]))
assert self._rf is not None
X = self._impute_inactive(X)
X_feat = list(self._instance_features.values())
dat_ = self._rf.predict_marginalized_over_instances_batch(X, X_feat, self._log_y)
dat_ = np.array(dat_)
# 3. compute statistics across trees
mean_ = dat_.mean(axis=1)
var = dat_.var(axis=1)
if var is None:
raise RuntimeError("The variance must not be none.")
var[var < self._var_threshold] = self._var_threshold
if len(mean_.shape) == 1:
mean_ = mean_.reshape((-1, 1))
if len(var.shape) == 1:
var = var.reshape((-1, 1))
return mean_, var