Source code for onetick.ml.impl.data_pipelines.preprocessors

import copy
import math
import numpy as np
import pandas as pd
from typing import Optional, Union
import onetick.py as otp

from sklearn import preprocessing
from onetick.ml.interfaces import BasePreprocess


class SKLearnPreprocessor(BasePreprocess):
    """Data preprocessing class that uses sklearn preprocessor.
    Could use only preprocessors that do not change the number of columns.

    Parameters
    ----------
    preprocessor_class : sklearn.preprocessing class
        Class of sklearn preprocessor to be used.
    columns : list, str
        List of columns to which preprocessing will be applied.
        String "__all__" means that all columns will be used.
        Default is "__all__".
    fit_on_train : bool
        If True, then preprocessor will be fitted only on train data.
        Default is False.
    params : dict
        Keyword arguments to be passed to sklearn preprocessor constructor.
    suffix : str
        Suffix to be added to column name after preprocessing.
        Default is None, which means that suffix will be equal to
        preprocessor class name in upper case.
    """
    def __init__(self,
                 preprocessor_class,
                 columns: Union[list, str] = "__all__",
                 params: Optional[dict] = None,
                 fit_on_train: bool = False,
                 group_by_column: bool = False,
                 suffix: str = ""):
        super().__init__(preprocessor_class=preprocessor_class,
                         columns=columns,
                         params=params,
                         fit_on_train=fit_on_train,
                         group_by_column=group_by_column,
                         suffix=suffix)
        self.preprocessor_class = preprocessor_class
        self.fit_on_train = fit_on_train
        self.params = params or {}
        self.preprocessors = {}

    def fit_pandas(self, df: pd.DataFrame):
        """Fits selected preprocessor on data without transforming it.

        Parameters
        ----------
        df : pd.DataFrame
            Dataframe to be preprocessed.
        columns : list, optional
            List of columns to which preprocessing will be applied.
            If set to None, then all columns will be used.
            By default None.
        """
        for column in self.column_names(df):
            preprocessor = self.preprocessor_class(**self.params)
            if self.fit_on_train and "__SPLIT" in df.columns:
                data = df[df["__SPLIT"] == "TRAIN"][column]
            else:
                data = df[column]
            preprocessor.fit(data.values.reshape(-1, 1))
            self.preprocessors[column] = preprocessor
        return df

    def transform_pandas(self, df: pd.DataFrame):
        """Transform data using selected sklearn preprocessor.

        Parameters
        ----------
        df : pd.DataFrame
            Dataframe to be preprocessed.
        columns : list, optional
            List of columns to which preprocessing will be applied.
            If set to None, then all columns will be used.
            By default None.
        """
        for new_column, column in self.iter_column_pairs(df):
            preprocessor = self.preprocessors[column]
            column_data = df[column].values.reshape(-1, 1)
            df[new_column] = preprocessor.transform(column_data)
        return df

    def inverse_transform(self, prediction_df: pd.DataFrame = None) -> pd.DataFrame:
        """Deprocess data using sklearn preprocessor.

        Parameters
        ----------
        df : pd.DataFrame
            Dataframe to be preprocessed.

        Returns
        -------
        pd.DataFrame
            Deprocessed dataframe.
        """
        if prediction_df is None:
            return None

        _df = prediction_df.copy(deep=True)
        for column, preprocessor in self.preprocessors.items():
            new_column = self.get_new_column_name(column)
            if new_column not in _df.columns:
                continue
            column_data = _df[new_column].values.reshape(-1, 1)
            _df[column] = preprocessor.inverse_transform(column_data)
        return _df


class ApplyLog(BasePreprocess):
    """
    Data preprocessing class that logarithms data.

    Parameters
    ----------
    columns : list
        List of columns to which preprocessing will be applied
    base : float
        Base of the logarithm.
        Default: math.e
    suffix : str
        Suffix to be added to column name after preprocessing.
        Default is "", which means that preprocessing will not create new column
        and apply logarithm to the original column.
    """

    refit_on_predict = True
    deprocessable = True

    def __init__(self,
                 base: Optional[float] = None,
                 columns: Union[list, str] = "__all__",
                 suffix: str = "",
                 ):
        super().__init__(columns=columns, base=base, suffix=suffix)
        self.base = math.e if base is None else base
        self.log_base = np.log(self.base)

    def transform_ot(self, src: otp.Source):
        self._columns = {}
        for new_column, column in self.iter_column_pairs(src):
            src[new_column] = otp.math.ln(src[column]) / otp.math.ln(self.base)
            self._columns[column] = self.base
        return src

    def transform_pandas(self, df: pd.DataFrame):
        """Preprocess data by applying logarithm.

        Parameters
        ----------
        df : pd.DataFrame
            Dataframe to be preprocessed.
        """
        self._columns = {}

        new_columns, columns = zip(*self.iter_column_pairs(df))
        df[list(new_columns)] = np.log(df[list(columns)]) / self.log_base

        self._columns.update(zip(columns, [self.base] * len(columns)))

        return df

    def inverse_transform(self,
                          prediction_df: Optional[pd.DataFrame] = None) -> pd.DataFrame:
        """Reverse process data by applying the exponential function.

        Parameters
        ----------
        prediction_df : pd.DataFrame, optional
            Dataframe with predictions, by default None

        Returns
        -------
        pd.DataFrame or None
            Reverse processed dataframe with predictions or None if `prediction_df` is None.
        """
        if prediction_df is None:
            return None

        _df = prediction_df.copy(deep=True)
        # todo: which is less: _df.columns or self._columns_map?
        columns = [column for column in _df.columns if column in self._columns_map]
        original_columns = [self._columns_map[column] for column in columns]
        base_series = pd.Series(self._columns)
        _df[original_columns] = base_series[original_columns].values ** _df[columns].values

        return _df


class LimitOutliers(BasePreprocess):
    """
    Data preprocessing class that limits outliers by using standard deviations.
    The maximum and minimum allowable values are calculated using the formula: `mean ± std_num * std`,
    where `mean` and `std` are the mean value and standard deviation calculated on the training set.

    Parameters
    ----------
    columns : list
        List of columns to which preprocessing will be applied
    std_num : float
        The number of standard deviations used to limit outliers.
        Default: 3
    """

    refit_on_predict = False
    deprocessable = False

    def __init__(self,
                 std_num: float = 3,
                 columns: Union[list, str] = "__all__",
                 suffix: str = "",):
        super().__init__(std_num=std_num, columns=columns, suffix=suffix)
        self.std_num = std_num
        self.fitted = False
        self._columns = {}

    def fit_pandas(self, df: pd.DataFrame):
        """Fits selected preprocessor on data without transforming it.

        Parameters
        ----------
        df : pd.DataFrame
            Dataframe to be preprocessed.
        """
        self._columns = {}
        if "__SPLIT" in df.columns:
            _df = df[df["__SPLIT"] == "TRAIN"]
        else:
            _df = df

        columns = self.column_names(_df)
        means = _df[columns].mean(numeric_only=True)
        stds = _df[columns].std(numeric_only=True)
        for column in columns:
            self._columns[column] = {
                'up_border': means[column] + self.std_num * stds[column],
                'down_border': means[column] - self.std_num * stds[column]
            }
        self.fitted = True
        return df

    def transform_pandas(self, df: pd.DataFrame):
        """Process data by limiting (capping) outliers.

        Parameters
        ----------
        df : pd.DataFrame
            Dataframe to be preprocessed.
        columns : list, optional
            List of columns to which preprocessing will be applied.
            If set to None, then all columns will be used.
            By default None.
        """
        for new_column, column in self.iter_column_pairs(df):
            if column not in self._columns:
                raise ValueError(f'Column {column} is not fitted by LimitOutliers. Please fit data before transform.')
            up_border = self._columns[column]['up_border']
            down_border = self._columns[column]['down_border']
            df[new_column] = df[column].clip(lower=down_border, upper=up_border)
        return df

    def transform_ot(self, src: otp.Source):
        columns = self.column_names(src)

        agg_dict = {}
        for column in columns:
            agg_dict[f"__{column}_MEAN"] = otp.agg.mean(column)
            agg_dict[f"__{column}_STD"] = otp.agg.stddev(column, biased=False)

        if self.schema.set_name in src.schema:  # fit only on train set, if splitting is used
            data_agg, _ = src[src[self.schema.set_name] == "TRAIN"]
        else:
            data_agg = src

        data_agg = data_agg.agg(agg_dict,
                                bucket_time='start')

        for column in columns:
            data_agg[f"__{column}_LOWER"] = data_agg[f"__{column}_MEAN"] - \
                self.std_num * data_agg[f"__{column}_STD"]
            data_agg[f"__{column}_HIGHER"] = data_agg[f"__{column}_MEAN"] + \
                self.std_num * data_agg[f"__{column}_STD"]
            # data_agg.drop(columns=[f"__{column}_MEAN", f"__{column}_STD"], inplace=True)

        src = otp.join_by_time([src, data_agg], policy="LATEST_TICKS")

        for column in columns:
            src[f"{column}{self.suffix}"] = src[column].apply(
                lambda x:
                    data_agg[f"__{column}_LOWER"] if x < data_agg[f"__{column}_LOWER"] else
                    data_agg[f"__{column}_HIGHER"] if x > data_agg[f"__{column}_HIGHER"] else
                    x
            )
        return src


class MinMaxScaler(SKLearnPreprocessor):
    """
    Data preprocessing class that scales data to a given range.

    Parameters
    ----------
    columns : list, optional
        List of columns to which preprocessing will be applied.
        By default None.
    transformed_range: tuple
        Desired range of transformed data.
        Default: (0, 1)
    """

    refit_on_predict = False
    deprocessable = True

    def __init__(self,
                 columns: Optional[list] = None,
                 transformed_range: tuple = (0, 1),
                 suffix: str = "",
                 group_by_column: bool = False):
        self.transformed_range = (0, 1) if transformed_range is None else transformed_range
        super().__init__(
            preprocessor_class=preprocessing.MinMaxScaler,
            fit_on_train=True,
            columns=columns,
            suffix=suffix,
            group_by_column=group_by_column,
            params=dict(feature_range=self.transformed_range)
        )
        # hack to save init params specifically for current class, not for the parent class
        self.save_init_params(dict(
            transformed_range=self.transformed_range,
            columns=columns
        ))

    def fit_ot(self, src: otp.Source):
        columns = self.column_names(src)
        agg_dict = {}
        for column in columns:
            agg_dict[f"__{column}_MIN"] = otp.agg.min(column)
            agg_dict[f"__{column}_MAX"] = otp.agg.max(column)
            # src[new_column] = src[column].apply(lambda x: )

        if self.schema.set_name in src.schema:  # fit only on train set, if splitting is used
            data_agg, _ = src[src[self.schema.set_name] == "TRAIN"]
        else:
            data_agg = src

        data_agg = data_agg.agg(agg_dict, bucket_time='start')
        src = otp.join_by_time([src, data_agg], policy="LATEST_TICKS")

        # save min and max values in a separate column for inverse transform
        self.schema.utility_columns += [f"__{column}_MIN" for column in columns]
        self.schema.utility_columns += [f"__{column}_MAX" for column in columns]

        return src

    def transform_ot(self, src: otp.Source):
        a = self.transformed_range[0]
        b = self.transformed_range[1]
        for new_column, column in self.iter_column_pairs(src):
            src[new_column] = src[column]
            src[new_column] = src[column].apply(
                lambda x:
                a + (x - src[f"__{column}_MIN"]) * (b - a) / (src[f"__{column}_MAX"] - src[f"__{column}_MIN"]))
        return src

    def inverse_transform(self, prediction_df: pd.DataFrame = None) -> pd.DataFrame:
        a = self.transformed_range[0]
        b = self.transformed_range[1]
        _df = prediction_df.copy()
        for new_column, column in self._columns_map.items():
            if new_column not in _df.columns:
                continue
            cmin = f"__{column}_MIN"
            cmax = f"__{column}_MAX"
            if cmin in _df.columns and cmax in _df.columns:
                _df[column] = (_df[cmax] - _df[cmin]) / (b - a) * (_df[new_column] - a) + _df[cmin]
            else:
                _df[column] = self.preprocessors[column].inverse_transform(_df[new_column].values.reshape(-1, 1))
        return _df


class LaggedDifferences(BasePreprocess):
    """
    Data preprocessing class that calculates the difference between the current and lag values (of time series).

    Note: this preprocessing will make first `lag` rows contain NaN values. Filter them out before training.

    Parameters
    ----------
    columns : list
        List of columns to which preprocessing will be applied.
    lag: int
        Value of the lag to be used.
        This value is equals, how many rows will be removed from the beggining of the dataframe.
        Default: 39
    """

    refit_on_predict = True
    deprocessable = True

    def __init__(self,
                 lag: int = 39,
                 columns: Union[list, str] = "__all__",
                 suffix: str = "",):
        super().__init__(columns=columns, lag=lag, suffix=suffix)
        self.lag = lag
        self.suffix = suffix
        self._shift_df = None

    def transform_ot(self, src: otp.Source):
        """
        Calculates lagged differences for the given columns.

        Parameters
        ----------
        src: otp.Source
            Source to calulate lagged differences for.
        columns : list, optional
            List of columns to calculate lagged differences for.

        Returns
        -------
        pd.DataFrame
            Source with calculated lagged differences in a new columns.
        """
        for new_col, col in self.iter_column_pairs(src):
            # calculate lagged values in a separate columns
            # to use it in inverse_transform() method
            utility_column = f'__LAGGED_{self.lag}_{col}'
            src[utility_column] = src[col][-self.lag]
            src[new_col] = src[col] - src[utility_column]
            self.schema.utility_columns += [utility_column]
        return src

    def fit_pandas(self, df: pd.DataFrame):
        """Preprocess data by calculating the difference between the current and lag values (of time series).

        Parameters
        ----------
        df : pd.DataFrame
            Dataframe to be preprocessed.
        """
        self._shift_df = df.loc[:, self.column_names(df)].shift(self.lag).copy(deep=True)
        return df

    def transform_pandas(self, df: pd.DataFrame):
        if self._shift_df is None:
            raise RuntimeError("LaggedDifferences cannot be applied without being fitted on data first")

        columns = self.column_names(df)
        suffix_columns = [f"{column}{self.suffix}" for column in columns]
        df[suffix_columns] = df[columns].subtract(self._shift_df[columns])

        return df

    def inverse_transform(self, prediction_df: pd.DataFrame = None) -> pd.DataFrame:
        """Reverse process data by adding the lagged values to the corresponding prediction values.

        Parameters
        ----------
        prediction_df : pd.DataFrame
            Dataframe to be deprocessed.

        Returns
        -------
        pd.DataFrame
            Deprocessed dataframe with the same shape as the input dataframe or None if prediction_df is None.
        """
        if prediction_df is None:
            return None

        if self._shift_df is None:
            # if operator transformed data in onetick context
            # then we should extract _shift_df from utility_columns
            self._shift_df = pd.DataFrame()
            for column, original_column in self._columns_map.items():
                utility_column = f'__LAGGED_{self.lag}_{original_column}'
                if utility_column in prediction_df.columns:
                    self._shift_df[original_column] = prediction_df[utility_column]
                else:
                    raise RuntimeError("LaggedDifferences cannot deprocess without being fitted on all columns first")

        _df = prediction_df.copy(deep=True)
        # todo: which is less: _df.columns or self._columns_map?
        columns = [column for column in _df.columns if column in self._columns_map]
        original_columns = [self._columns_map[column] for column in columns]
        _df[original_columns] = _df[columns].values + self._shift_df.loc[_df.index, original_columns].values

        return _df


class IntradayAveraging(BasePreprocess):
    """
    Data preprocessing class that calculates the difference between the current value and the average value of the same
    intraday interval over the past N days.

    Note: this preprocessing will remove the first `bins`*`window_days` rows from the dataframe.

    Parameters
    ----------
    columns : list
        List of columns to which preprocessing will be applied.
    window_days: int
        Number of days for averaging.
        Default: 5
    bins: int
        Number of intraday intervals.
        If None is specified, then it is determined automatically by the number of unique hh:mm buckets.
        Default: None
    datetime_column: str
        Name of the column that stores the datetime values.
        Default: 'Time'
    suffix: str
        Suffix to be added to the column names.
        Default: ''
    """

    refit_on_predict = True
    deprocessable = True

    def __init__(self,
                 columns: Union[list, str] = "__all__",
                 window_days=5,
                 bins=None,
                 datetime_column='Time',
                 suffix: str = ""):
        super().__init__(columns=columns,
                         window_days=window_days,
                         bins=bins,
                         datetime_column=datetime_column,
                         suffix=suffix)
        self.window_days = window_days
        self.bins = bins
        self.datetime_column = datetime_column
        self._agg_df = None
        self._columns = None

    def fit_pandas(self, df: pd.DataFrame):
        """Preprocess (timeseries) data by calculating the difference between the current value
        and the average value of the same intraday interval in the past period of time.

        Parameters
        ----------
        df : pd.DataFrame
            Dataframe to be preprocessed.
        """
        columns = self.column_names(df)
        df.loc[:, self.datetime_column] = pd.to_datetime(df[self.datetime_column])
        df.loc[:, '__hhmm'] = df[self.datetime_column].dt.time

        agg_df = df[columns + ['__hhmm']].copy(deep=True)
        agg_df = agg_df.groupby(by='__hhmm').rolling(self.window_days).mean()
        agg_df = agg_df.shift(1).reset_index(level=0).sort_index()

        # TODO Now solution below adjust for num-type index, it needs to find best option to cut incorrect shift vals
        if self.bins is None:
            self.bins = df['__hhmm'].nunique()
        min_agg_index = self.bins * self.window_days + min(df.index)
        agg_df.loc[agg_df.index < min_agg_index, :] = np.nan
        self._agg_df = agg_df
        self._columns = columns
        return df

    def transform_pandas(self, df: pd.DataFrame):
        assert self._agg_df is not None, "IntradayAveraging cannot be applied without being fitted on data first"
        new_columns, columns = zip(*self.iter_column_pairs(df))
        df[list(new_columns)] = df[list(columns)] - self._agg_df[list(columns)]
        return df

    def inverse_transform(self, prediction_df: pd.DataFrame = None) -> pd.DataFrame:
        """Reverse process data by adding the average value of the intraday intervals
        to the corresponding prediction values.

        Parameters
        ----------
        prediction_df : pd.DataFrame
            Dataframe to be deprocessed.

        Returns
        -------
        pd.DataFrame
            Reverse processed dataframe with the same shape as the input dataframe or None if prediction_df is None.
        """
        assert self._agg_df is not None, "IntradayAveraging cannot deprocess without being fitted on data first"

        if prediction_df is None:
            return None

        _df = prediction_df.copy(deep=True)
        # todo: which is less: _df.columns or self._columns_map?
        columns = [column for column in _df.columns if column in self._columns_map]
        original_columns = [self._columns_map[column] for column in columns]
        _df[original_columns] = _df[columns].values + self._agg_df.loc[_df.index][original_columns].values

        return _df


class GroupByColumn(BasePreprocess):
    def __init__(self,
                 preprocessor,
                 columns: Union[list, str] = "__all__",
                 ):
        super().__init__(
            preprocessor=preprocessor,
            columns=columns,
            suffix="")
        self.preprocessor = preprocessor
        self._groups = None
        self._processors = {}

    def get_init_params(self):
        """Override get_init_params in order to replace GroupByColumn with nested preprocessor
        and add group_by_column to its dict.
        """
        _params = copy.deepcopy(self.preprocessor.get_init_params())
        _params['group_by_column'] = True
        return _params

    def fit_pandas(self, df: pd.DataFrame):
        self._groups = df.groupby(self.schema.group_column_name)
        for group, group_df in self._groups:
            # clone preprocessor and save it for transform
            self._processors[group] = copy.deepcopy(self.preprocessor)
            self._processors[group].schema = self.schema

            # fit each preprocessor on a particular ticker
            # todo: do we need copy below?
            self._processors[group].fit(group_df.copy())
        return df

    def transform_pandas(self, df: pd.DataFrame):
        assert self._groups is not None, "GroupByColumn cannot be applied without being fitted on data first"

        # apply transform to each group
        for group, group_df in self._groups:
            transformed_group = self._processors[group].transform(group_df)

            # initialize new columns with NaNs
            new_columns = set(transformed_group.columns) - set(df.columns)
            for new_column in new_columns:
                if new_column not in df.columns:
                    df[new_column] = np.nan

            df.loc[group_df.index] = transformed_group
        return df

    def fit_ot(self, src: otp.Source):
        return self.preprocessor.fit_ot(src)

    def transform_ot(self, src: otp.Source):
        return self.preprocessor.transform_ot(src)

    def inverse_transform(self, prediction_df: pd.DataFrame = None) -> pd.DataFrame:
        if self.preprocessor.deprocessable is False:
            return prediction_df

        assert self._groups is not None, "GroupByColumn cannot be applied without being fitted on data first"

        if prediction_df is None:
            return None

        _df = prediction_df.copy(deep=True)
        _df = _df.groupby(self.schema.group_column_name, group_keys=False).apply(
            lambda x: self._processors[x.name].inverse_transform(x))

        # drop group column index
        # todo: check if it is correct???
        # _df.index = _df.index.droplevel(0)
        _df = _df.sort_index()
        return _df


class RefittableLabelEncoder(preprocessing.LabelEncoder):
    """
    Same as LabelEncoder, but be fitted on new data without losing the mapping of the old data.
    SKLearn LabelEncoder is not refittable because it throws an error
    if new classes are encountered during transform.
    """

    def fit(self, y, **kwargs):
        if hasattr(self, 'classes_'):
            self.classes_ = np.unique(np.concatenate((self.classes_, np.unique(y))))

        super().fit(y)
        return self

    def fit_transform(self, y, **kwargs):
        self.fit(y)
        return self.transform(y)