<commit_before>"""
This module is for both in-fold and out-of-fold generation of
features that are derived in the following manner:
1) group target variable by a particular initial feature;
2) apply aggregating function to each group;
3) for each object, use corresponding aggregated value
   as a new feature.

@author: <NAME>
"""


from typing import List, Callable

import numpy as np

from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.model_selection import KFold


class TargetBasedFeaturesCreator(BaseEstimator, TransformerMixin):
    """
    This class can augment datasets with new features such that
    each of them is conditional value of an aggregating function
    applied to target variable where conditioning is taken over
    a specified initial feature.

    For example, this class can replace a categorical feature
    (unsuitable for lots of machine learning algorithms) with
    within-group mean of target variable where grouping is by
    this categorical feature.

    In order to handle rare values of initial source features,
    smoothing towards unconditional (i.e., population, general)
    aggregates can be applied.

    Also this class allows avoiding overfitting by generating
    new features based only on out-of-fold values of target.

    :param aggregators: functions that compute aggregates
    :param smoothing_strength: strength of smoothing towards
                               unconditional aggregates
    :param min_frequency: minimal number of occurrences of a feature's
                          value (if value occurs less times than this
                          parameter, this value is mapped to
                          unconditional aggregate)
    """

    def __init__(
            self,
            aggregators: List[Callable] = None,
            smoothing_strength: float = 0,
            min_frequency: int = 1
            ):
        self.aggregators = [np.mean] if aggregators is None else aggregators
        self.smoothing_strength = smoothing_strength
        self.min_frequency = min_frequency
        self.mappings_ = dict()

    def __compute_aggregate(
            self,
            mask: np.array,
            target: np.array,
            aggregator: Callable
            ) -> float:
        n_occurrences = sum(mask)
        if n_occurrences < self.min_frequency:
            return aggregator(target)
        else:
            numerator = (
                n_occurrences * aggregator(target[mask]) +
                self.smoothing_strength * aggregator(target)
            )
            denominator = n_occurrences + self.smoothing_strength
            return numerator / denominator

    def fit(
            self,
            X: np.ndarray,
            y: np.ndarray,
            source_positions: List[int]
            ) -> 'TargetBasedFeaturesCreator':
        """
        Fit to a whole dataset of `X` and `y`.
        In other words, memorize mappings from initial values
        of selected columns to conditional aggregates.

        :param X: features
        :param y: target
        :param source_positions: indices of initial features to be
                                 used as conditions
        :return: fitted instance
        """
        for position in source_positions:
            feature = X[:, position]
            self.mappings_[position] = {
                k: [self.__compute_aggregate(feature == k, y, agg)
                    for agg in self.aggregators]
                for k in np.unique(feature)
            }
            # `None` is a reserved key for unseen values.
            self.mappings_[position][None] = [
                agg(feature) for agg in self.aggregators
            ]
        return self

    def transform(
            self,
            X: np.ndarray,
            drop_source_features: bool = True
            ) -> np.ndarray:
        """
        Augment `X` with learnt conditional aggregates.

        :param X: feature representation to be augmented
        :param drop_source_features: drop or keep those of initial
                                     features that are used for
                                     conditioning over them
        :return: transformed data
        """
        transformed_X = X.copy()
        for position, mappings in self.mappings_.items():
            feature = X[:, position]
            n_new = len(next(iter(mappings.values())))
            new_features = np.full((X.shape[0], n_new), np.nan)<commit_msg>bug: Aggregate target, not initial feature<commit_after>"""
This module is for both in-fold and out-of-fold generation of
features that are derived in the following manner:
1) group target variable by a particular initial feature;
2) apply aggregating function to each group;
3) for each object, use corresponding aggregated value
   as a new feature.

@author: <NAME>
"""


from typing import List, Callable

import numpy as np

from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.model_selection import KFold


class TargetBasedFeaturesCreator(BaseEstimator, TransformerMixin):
    """
    This class can augment datasets with new features such that
    each of them is conditional value of an aggregating function
    applied to target variable where conditioning is taken over
    a specified initial feature.

    For example, this class can replace a categorical feature
    (unsuitable for lots of machine learning algorithms) with
    within-group mean of target variable where grouping is by
    this categorical feature.

    In order to handle rare values of initial source features,
    smoothing towards unconditional (i.e., population, general)
    aggregates can be applied.

    Also this class allows avoiding overfitting by generating
    new features based only on out-of-fold values of target.

    :param aggregators: functions that compute aggregates
    :param smoothing_strength: strength of smoothing towards
                               unconditional aggregates
    :param min_frequency: minimal number of occurrences of a feature's
                          value (if value occurs less times than this
                          parameter, this value is mapped to
                          unconditional aggregate)
    """

    def __init__(
            self,
            aggregators: List[Callable] = None,
            smoothing_strength: float = 0,
            min_frequency: int = 1
            ):
        self.aggregators = [np.mean] if aggregators is None else aggregators
        self.smoothing_strength = smoothing_strength
        self.min_frequency = min_frequency
        self.mappings_ = dict()

    def __compute_aggregate(
            self,
            mask: np.array,
            target: np.array,
            aggregator: Callable
            ) -> float:
        n_occurrences = sum(mask)
        if n_occurrences < self.min_frequency:
            return aggregator(target)
        else:
            numerator = (
                n_occurrences * aggregator(target[mask]) +
                self.smoothing_strength * aggregator(target)
            )
            denominator = n_occurrences + self.smoothing_strength
            return numerator / denominator

    def fit(
            self,
            X: np.ndarray,
            y: np.ndarray,
            source_positions: List[int]
            ) -> 'TargetBasedFeaturesCreator':
        """
        Fit to a whole dataset of `X` and `y`.
        In other words, memorize mappings from initial values
        of selected columns to conditional aggregates.

        :param X: features
        :param y: target
        :param source_positions: indices of initial features to be
                                 used as conditions
        :return: fitted instance
        """
        for position in source_positions:
            feature = X[:, position]
            self.mappings_[position] = {
                k: [self.__compute_aggregate(feature == k, y, agg)
                    for agg in self.aggregators]
                for k in np.unique(feature)
            }
            # `None` is a reserved key for unseen values.
            self.mappings_[position][None] = [
                agg(y) for agg in self.aggregators
            ]
        return self

    def transform(
            self,
            X: np.ndarray,
            drop_source_features: bool = True
            ) -> np.ndarray:
        """
        Augment `X` with learnt conditional aggregates.

        :param X: feature representation to be augmented
        :param drop_source_features: drop or keep those of initial
                                     features that are used for
                                     conditioning over them
        :return: transformed data
        """
        transformed_X = X.copy()
        for position, mappings in self.mappings_.items():
            feature = X[:, position]
            n_new = len(next(iter(mappings.values())))
            new_features = np.full((X.shape[0], n_new), np.nan)
{'fraction_non_alphanumeric': 0.06105006105006105, 'fraction_numerical': 0.001332001332001332, 'mean_word_length': 1.9329427083333333, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3510574', 'n_tokens_mistral': 2280, 'n_tokens_neox': 2104, 'n_words': 824}