"""StackingTransformer. Scikit-learn compatible API for stacking.

def __init__(self,

estimators=None,

regression=True,

transform_target=None,

transform_pred=None,

variant='A',

needs_proba=False,

metric=None,

n_folds=4,

stratified=False,

shuffle=False,

random_state=0,

verbose=0):

self.estimators = estimators

self.regression = regression

self.transform_target = transform_target

self.transform_pred = transform_pred

self.variant = variant

self.needs_proba = needs_proba

self.metric = metric

self.n_folds = n_folds

self.stratified = stratified

self.shuffle = shuffle

self.random_state = random_state

self.verbose = verbose

# -------------------------------------------------------------------------

# -------------------------------------------------------------------------

def __sklearn_tags__(self):

tags = super().__sklearn_tags__()

tags.estimator_type = 'transformer'

tags.transformer_tags.preserves_dtype = []

tags.target_tags.required = True

tags.input_tags.sparse = True

return tags

# -------------------------------------------------------------------------

# -------------------------------------------------------------------------

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

"""Fit all base estimators.

# ---------------------------------------------------------------------

# Validation

# ---------------------------------------------------------------------

# ---------------------------------------------------------------------

# Check input data

# ---------------------------------------------------------------------

# Check data and set `self.n_features_in_` and `self.feature_names_in_`

X, y = validate_data(self, X, y,

reset=True, # default: True, if True will set `self.n_features_in_` and `self.feature_names_in_`

validate_separately=False, # default: False, if False will use `check_X_y`

skip_check_array=False, # default: False, if False will NOT skip checks

accept_sparse=['csr'],

ensure_all_finite=True,

multi_output=False)

# Legacy check included in `validate_data`

# Check X and y

# ``check_estimator`` does not allow ``ensure_all_finite=False``

# X, y = check_X_y(X, y,

# accept_sparse=['csr'], # allow csr, cast all others to csr

# ensure_all_finite=True, # do not allow nan and inf

# multi_output=False) # allow only one column in y_train

# Check X and sample_weight

# X is alredy checked, but we need it to compare length of sample_weight

if sample_weight is not None:

X, sample_weight = check_X_y(X, sample_weight,

accept_sparse=['csr'],

ensure_all_finite=True,

multi_output=False)

# ---------------------------------------------------------------------

# Check ``estimators``

# ---------------------------------------------------------------------

if self.estimators is None:

if self.regression:

self.estimators_ = [('dumregr', DummyRegressor(strategy='constant', constant=5.5))]

else:

self.estimators_ = [('dumclf', DummyClassifier(strategy='constant', constant=1))]

# warnings.warn('No estimators were specified. '

# 'Using single dummy estimator as demo.', UserWarning)

else:

if 0 == len(self.estimators):

raise ValueError('List of estimators is empty')

else:

# Clone

self.estimators_ = [(name, clone(estim)) for name, estim in self.estimators]

# Check names of estimators

names, estims = zip(*self.estimators_)

self._validate_names(names)

# Check if all estimators support ``sample_weight``

if sample_weight is not None:

for name, estim in self.estimators_:

if not has_fit_parameter(estim, 'sample_weight'):

raise ValueError('Underlying estimator [%s] does not '

'support sample weights.' % name)

# ---------------------------------------------------------------------

# Check other StackingTransformer parameters

# ---------------------------------------------------------------------

# ``variant``

if self.variant not in ['A', 'B']:

raise ValueError('Parameter ``variant`` must be set properly')

# ``n_folds``

if not isinstance(self.n_folds, int):

raise ValueError('Parameter ``n_folds`` must be integer')

if not self.n_folds > 1:

raise ValueError('Parameter ``n_folds`` must be not less than 2')

# ``verbose``

if self.verbose not in [0, 1, 2]:

raise ValueError('Parameter ``verbose`` must be 0, 1, or 2')

# Additional check for inapplicable parameter combinations

# If ``regression=True`` we ignore classification-specific

# parameters and issue user warning

if self.regression and (self.needs_proba or self.stratified):

warn_str = ('This is regression task hence classification-specific'

'parameters set to ``True`` were ignored:')

if self.needs_proba:

self.needs_proba = False

warn_str += ' ``needs_proba``'

if self.stratified:

self.stratified = False

warn_str += ' ``stratified``'

warnings.warn(warn_str, UserWarning)

# To comply with sklearn requirement

if not self.shuffle:

random_state_internal = None

else:

random_state_internal = self.random_state

# ---------------------------------------------------------------------

# Compute attributes (basic properties of data, number of estimators, etc.)

# ---------------------------------------------------------------------

self.train_shape_ = X.shape

self.n_train_examples_ = X.shape[0]

self.n_features_ = X.shape[1]

if not self.regression:

self.n_classes_ = len(np.unique(y))

else:

self.n_classes_ = None

self.n_estimators_ = len(self.estimators_)

self.train_footprint_ = self._get_footprint(X)

# ---------------------------------------------------------------------

# Specify default metric

# ---------------------------------------------------------------------

if self.metric is None and self.regression:

self.metric_ = mean_absolute_error

elif self.metric is None and not self.regression:

if self.needs_proba:

self.metric_ = log_loss

else:

self.metric_ = accuracy_score

else:

self.metric_ = self.metric

# ---------------------------------------------------------------------

# Create report header strings and print report header

# ---------------------------------------------------------------------

if self.verbose > 0:

if self.regression:

task_str = 'task: [regression]'

else:

task_str = 'task: [classification]'

n_classes_str = 'n_classes: [%d]' % self.n_classes_

metric_str = 'metric: [%s]' % self.metric_.__name__

variant_str = 'variant: [%s]' % self.variant

n_estimators_str = 'n_estimators: [%d]' % self.n_estimators_

print(task_str)

if not self.regression:

print(n_classes_str)

print(metric_str)

print(variant_str)

print(n_estimators_str + '\n')

# ---------------------------------------------------------------------

# Initialize cross-validation split

# Stratified can be used only for classification

# ---------------------------------------------------------------------

if not self.regression and self.stratified:

self.kf_ = StratifiedKFold(n_splits=self.n_folds,

shuffle=self.shuffle,

random_state=random_state_internal)

# Save target to be able to create stratified split in ``transform`` method

# This is more efficient than to save split indices

self._y_ = y.copy()

else:

self.kf_ = KFold(n_splits=self.n_folds,

shuffle=self.shuffle,

random_state=random_state_internal)

self._y_ = None

# ---------------------------------------------------------------------

# Compute implicit number of classes to create appropriate empty arrays.

# !!! Important. In order to unify array creation

# variable ``n_classes_implicit_`` is always equal to 1, except the case

# when we performing classification task with ``needs_proba=True``

# ---------------------------------------------------------------------

if not self.regression and self.needs_proba:

self.n_classes_implicit_ = len(np.unique(y))

self.action_ = 'predict_proba'

else:

self.n_classes_implicit_ = 1

self.action_ = 'predict'

# ---------------------------------------------------------------------

# Create empty numpy array for train predictions (OOF)

# !!! Important. We have to implicitly predict during fit

# in order to compute CV scores, because

# the most reasonable place to print out CV scores is fit method

# ---------------------------------------------------------------------

S_train = np.zeros((X.shape[0], self.n_estimators_ * self.n_classes_implicit_))

# ---------------------------------------------------------------------

# Prepare (clone) estmators for fitting and storing

# We need models_A_ for both variant A and varian B

# We need models_B_ for varian B only (in variant A attribute models_B_ is None)

# ---------------------------------------------------------------------

self.models_A_ = []

self.models_B_ = None

for n, est in self.estimators_:

self.models_A_.append([clone(est) for _ in range(self.n_folds)])

if self.variant in ['B']:

self.models_B_ = [clone(est) for n, est in self.estimators_]

# ---------------------------------------------------------------------

# Create empty numpy array to store scores for each estimator and each fold

# ---------------------------------------------------------------------

self.scores_ = np.zeros((self.n_estimators_, self.n_folds))

# ---------------------------------------------------------------------

# Create empty list to store name, mean and std for each estimator

# ---------------------------------------------------------------------

self.mean_std_ = []

# ---------------------------------------------------------------------

# MAIN FIT PROCEDURE

# ---------------------------------------------------------------------

# Loop across estimators

# ---------------------------------------------------------------------

for estimator_counter, (name, estimator) in enumerate(self.estimators_):

if self.verbose > 0:

estimator_str = 'estimator %2d: [%s: %s]' % (estimator_counter, name, estimator.__class__.__name__)

print(estimator_str)

# -----------------------------------------------------------------

# Loop across folds

# -----------------------------------------------------------------

for fold_counter, (tr_index, te_index) in enumerate(self.kf_.split(X, y)):

# Split data and target

X_tr = X[tr_index]

y_tr = y[tr_index]

X_te = X[te_index]

y_te = y[te_index]

# Split sample weights accordingly (if passed)

if sample_weight is not None:

sample_weight_tr = sample_weight[tr_index]

# sample_weight_te = sample_weight[te_index]

else:

sample_weight_tr = None

# sample_weight_te = None

# Fit estimator

_ = self._estimator_action(self.models_A_[estimator_counter][fold_counter],

X_tr, y_tr, None,

sample_weight=sample_weight_tr,

action='fit',

transform=self.transform_target)

# Predict out-of-fold part of train set

if 'predict_proba' == self.action_:

col_slice_estimator = slice(estimator_counter * self.n_classes_implicit_,

estimator_counter * self.n_classes_implicit_ + self.n_classes_implicit_)

else:

col_slice_estimator = estimator_counter

S_train[te_index, col_slice_estimator] = self._estimator_action(self.models_A_[estimator_counter][fold_counter],

None, None,

X_te, action=self.action_,

transform=self.transform_pred)

# Compute score

score = self.metric_(y_te, S_train[te_index, col_slice_estimator])

self.scores_[estimator_counter, fold_counter] = score

# Print fold score

if self.verbose > 1:

fold_str = ' fold %2d: [%.8f]' % (fold_counter, score)

print(fold_str)

# Compute mean and std and save in dict

estim_name = self.estimators_[estimator_counter][0]

estim_mean = np.mean(self.scores_[estimator_counter])

estim_std = np.std(self.scores_[estimator_counter])

self.mean_std_.append((estim_name, estim_mean, estim_std))

if self.verbose > 1:

sep_str = ' ----'

print(sep_str)

# Compute mean + std (and full)

if self.verbose > 0:

mean_str = ' MEAN: [%.8f] + [%.8f]\n' % (estim_mean, estim_std)

print(mean_str)

# Fit estimator on full train set

if self.variant in ['B']:

if self.verbose > 0:

print(' Fitting on full train set...\n')

_ = self._estimator_action(self.models_B_[estimator_counter],

X, y, None,

sample_weight=sample_weight,

action='fit',

transform=self.transform_target)

# ---------------------------------------------------------------------

# ---------------------------------------------------------------------

# Return fitted StackingTransformer instance

return self

# -------------------------------------------------------------------------

# -------------------------------------------------------------------------

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

"""Fit all base estimators and transform (predict) train set.

# ---------------------------------------------------------------------

# All validation and procedures are done inside corresponding methods

# fit and transform

# ---------------------------------------------------------------------

return self.fit(X, y, sample_weight).transform(X)

# -------------------------------------------------------------------------

# -------------------------------------------------------------------------

def transform(self, X, is_train_set=None):

"""Transform (predict) given data set.

# Check if fitted

check_is_fitted(self, ['models_A_'])

# Check data without resetting `self.n_features_in_` and `self.feature_names_in_`

X = validate_data(self, X,

reset=False, # default: True, if True will set `self.n_features_in_` and `self.feature_names_in_`

validate_separately=False, # default: False, if False will use `check_X_y`

skip_check_array=False, # default: False, if False will NOT skip checks

accept_sparse=['csr'],

ensure_all_finite=True) # no need for `multi_output`, because no `y`

# Legacy check included in `validate_data`

# Input validation

# ``check_estimator`` does not allow ``ensure_all_finite=False``

# X = check_array(X, accept_sparse=['csr'], ensure_all_finite=True)

# *********************************************************************

# Fitted StackingTransformer instance is bound to train set used for fitting.

# So during transformation we have different actions for train set

# and all other sets

# *********************************************************************

if is_train_set is None:

is_train_set = self._check_identity(X)

# Print

if self.verbose > 0:

if is_train_set:

print('Train set was detected.')

print('Transforming...\n')

# *********************************************************************

# Transform train set

# *********************************************************************

if is_train_set:

# In case if user directly tells that it is train set but shape is different

if self.train_shape_ != X.shape:

raise ValueError('Train set must have the same shape '

'in order to be transformed.')

# Create empty numpy array for train predictions (OOF)

S_train = np.zeros((X.shape[0], self.n_estimators_ * self.n_classes_implicit_))

# -----------------------------------------------------------------

# MAIN TRANSFORM (PREDICT) PROCEDURE for train set

# -----------------------------------------------------------------

# Loop across estimators

# -----------------------------------------------------------------

for estimator_counter, (name, estimator) in enumerate(self.estimators_):

if self.verbose > 0:

estimator_str = 'estimator %2d: [%s: %s]' % (estimator_counter, name, estimator.__class__.__name__)

print(estimator_str)

# -------------------------------------------------------------

# Loop across folds

# -------------------------------------------------------------

for fold_counter, (tr_index, te_index) in enumerate(self.kf_.split(X, self._y_)):

# Split data

# X_tr = X[tr_index]

X_te = X[te_index]

# Predict out-of-fold part of train set

if 'predict_proba' == self.action_:

col_slice_estimator = slice(estimator_counter * self.n_classes_implicit_,

estimator_counter * self.n_classes_implicit_ + self.n_classes_implicit_)

else:

col_slice_estimator = estimator_counter

S_train[te_index, col_slice_estimator] = self._estimator_action(self.models_A_[estimator_counter][fold_counter],

None, None,

X_te, action=self.action_,

transform=self.transform_pred)

# Print

if self.verbose > 1:

fold_str = ' model from fold %2d: done' % fold_counter

print(fold_str)

if self.verbose > 1:

sep_str = ' ----'

print(sep_str)

if self.verbose > 0:

done_str = ' DONE\n'

print(done_str)

# -----------------------------------------------------------------

# Cast class labels to int

# -----------------------------------------------------------------

if not self.regression and not self.needs_proba:

S_train = S_train.astype(int)

# Return transformed data (OOF)

return S_train # X_transformed

# *********************************************************************

# Transform any other set

# *********************************************************************

else:

# Legacy check included in `validate_data`

# Check n_features

# if X.shape[1] != self.n_features_:

# raise ValueError('Inconsistent number of features.')

# Create empty numpy array for test predictions

S_test = np.zeros((X.shape[0], self.n_estimators_ * self.n_classes_implicit_))

# ---------------------------------------------------------------------

# MAIN TRANSFORM (PREDICT) PROCEDURE for any other set

# -----------------------------------------------------------------

# Loop across estimators

# -----------------------------------------------------------------

for estimator_counter, (name, estimator) in enumerate(self.estimators_):

if self.verbose > 0:

estimator_str = 'estimator %2d: [%s: %s]' % (estimator_counter, name, estimator.__class__.__name__)

print(estimator_str)

# -------------------------------------------------------------

# Variant A

# -------------------------------------------------------------

if self.variant in ['A']:

# Create empty numpy array, which will contain temporary predictions

# for test set made in each fold

S_test_temp = np.zeros((X.shape[0], self.n_folds * self.n_classes_implicit_))

# ---------------------------------------------------------

# Loop across fitted models (it is the same as loop across folds)

# ---------------------------------------------------------

for fold_counter, model in enumerate(self.models_A_[estimator_counter]):

# Predict test set in each fold

if 'predict_proba' == self.action_:

col_slice_fold = slice(fold_counter * self.n_classes_implicit_,

fold_counter * self.n_classes_implicit_ + self.n_classes_implicit_)

else:

col_slice_fold = fold_counter

S_test_temp[:, col_slice_fold] = self._estimator_action(model, None, None, X,

action=self.action_,

transform=self.transform_pred)

# Print

if self.verbose > 1:

fold_str = ' model from fold %2d: done' % fold_counter

print(fold_str)

if self.verbose > 1:

sep_str = ' ----'

print(sep_str)

# ---------------------------------------------------------

# Compute mean or mode (majority voting) of predictions for test set

# ---------------------------------------------------------

if 'predict_proba' == self.action_:

# Here we copute means of probabilirties for each class

for class_id in range(self.n_classes_implicit_):

S_test[:, estimator_counter * self.n_classes_implicit_ + class_id] = np.mean(S_test_temp[:, class_id::self.n_classes_implicit_], axis=1)

else:

if self.regression:

S_test[:, estimator_counter] = np.mean(S_test_temp, axis=1)

else:

S_test[:, estimator_counter] = st.mode(S_test_temp, axis=1)[0].ravel()

if self.verbose > 0:

done_str = ' DONE\n'

print(done_str)

# -------------------------------------------------------------

# Variant B

# -------------------------------------------------------------

else:

if 'predict_proba' == self.action_:

col_slice_estimator = slice(estimator_counter * self.n_classes_implicit_,

estimator_counter * self.n_classes_implicit_ + self.n_classes_implicit_)

else:

col_slice_estimator = estimator_counter

S_test[:, col_slice_estimator] = self._estimator_action(self.models_B_[estimator_counter],

None, None, X,

action=self.action_,

transform=self.transform_pred)

if self.verbose > 0:

done_str = ' DONE\n'

print(done_str)

# ---------------------------------------------------------------------

# Cast class labels to int

# ---------------------------------------------------------------------

if not self.regression and not self.needs_proba:

S_test = S_test.astype(int)

return S_test # X_transformed

# -------------------------------------------------------------------------

# -------------------------------------------------------------------------

# -------------------------------------------------------------------------

# -------------------------------------------------------------------------

# -------------------------------------------------------------------------

def _transformer(self, y, func=None):

"""Transforms target variable and prediction

if func is None:

return y

else:

return func(y)

# -------------------------------------------------------------------------

# -------------------------------------------------------------------------

def _estimator_action(self, estimator, X_train, y_train, X_test,

sample_weight=None, action=None,

transform=None):

"""Performs estimator action.

if 'fit' == action:

# We use following condition, because some estimators (e.g. Lars)

# may not have ``sample_weight`` parameter of ``fit`` method

if sample_weight is not None:

return estimator.fit(X_train, self._transformer(y_train, func=transform),

sample_weight=sample_weight)

else:

return estimator.fit(X_train, self._transformer(y_train, func=transform))

elif 'predict' == action:

return self._transformer(estimator.predict(X_test), func=transform)

elif 'predict_proba' == action:

return self._transformer(estimator.predict_proba(X_test), func=transform)

else:

raise ValueError('Parameter action must be set properly')

# -------------------------------------------------------------------------

# -------------------------------------------------------------------------

def _random_choice(self, n, size, bound=2**30):

"""

try:

if n < size:

raise ValueError('Drawing without replacement: '

'``n`` cannot be less than ``size``')

ids = []

while len(ids) < size:

rnd = np.random.randint(min(bound, n))

if rnd not in ids:

ids.append(rnd)

return np.array(ids, dtype=np.int32)

except Exception:

raise ValueError('Internal error. '

'Please save traceback and inform developers.')

# -------------------------------------------------------------------------

# -------------------------------------------------------------------------

def _get_footprint(self, X, n_items=1000):

"""Selects ``n_items`` random elements from 2d numpy array or

try:

footprint = []

r, c = X.shape

n = r * c

# np.random.seed(0) # for development

# OOM with large arrays (see #29)

# ids = np.random.choice(n, min(n_items, n), replace=False)

ids = self._random_choice(n, min(n_items, n))

for i in ids:

row = i // c

col = i - row * c

footprint.append((row, col, X[row, col]))

return footprint

except Exception:

raise ValueError('Internal error. '

'Please save traceback and inform developers.')

# -------------------------------------------------------------------------

# -------------------------------------------------------------------------

def _check_identity(self, X,

rtol=1e-05, atol=1e-08,

equal_nan=False):

"""Checks 2d numpy array or sparse matrix identity

try:

# Check shape

if X.shape != self.train_shape_:

return False

# Check footprint

try:

for coo in self.train_footprint_:

assert np.isclose(X[coo[0], coo[1]], coo[2], rtol=rtol, atol=atol, equal_nan=equal_nan)

return True

except AssertionError:

return False

except Exception:

raise ValueError('Internal error. '

'Please save traceback and inform developers.')

# -------------------------------------------------------------------------

# -------------------------------------------------------------------------

def _get_params(self, attr, deep=True):

"""

out = super().get_params(deep=deep)

if not deep:

return out

estimators = getattr(self, attr)

try:

out.update(estimators)

except (TypeError, ValueError):

# Ignore TypeError for cases where estimators is not a list of