imputer.py

import copy
import functools

import numpy as np

from federatedml.feature.fate_element_type import NoneType
from federatedml.feature.instance import Instance
from federatedml.statistic import data_overview
from federatedml.statistic.data_overview import get_header
from federatedml.statistic.statics import MultivariateStatisticalSummary
from federatedml.util import LOGGER
from federatedml.util import consts


class Imputer(object):
    """
    This class provides basic strategies for values replacement. It can be used as missing filled or outlier replace.
    You can use the statistics such as mean, median or max of each column to fill the missing value or replace outlier.
    """

    def __init__(self, missing_value_list=None):
        """
        Parameters
        ----------
        missing_value_list: list, the value to be replaced. Default None, if is None, it will be set to list of blank, none, null and na,
                            which regarded as missing filled. If not, it can be outlier replace, and missing_value_list includes the outlier values
        """
        if missing_value_list is None:
            self.missing_value_list = ['', 'none', 'null', 'na', 'None', np.nan]
        else:
            self.missing_value_list = missing_value_list

        self.abnormal_value_list = copy.deepcopy(self.missing_value_list)
        for i, v in enumerate(self.missing_value_list):
            if v != v:
                self.missing_value_list[i] = np.nan
                self.abnormal_value_list[i] = NoneType()
        self.abnormal_value_set = set(self.abnormal_value_list)
        self.support_replace_method = ['min', 'max', 'mean', 'median', 'designated']
        self.support_output_format = {
            'str': str,
            'float': float,
            'int': int,
            'origin': None
        }

        self.support_replace_area = {
            'min': 'col',
            'max': 'col',
            'mean': 'col',
            'median': 'col',
            'designated': 'col'
        }

        self.cols_fit_impute_rate = []
        self.cols_transform_impute_rate = []
        self.cols_replace_method = []
        self.skip_cols = []

    def get_missing_value_list(self):
        return self.missing_value_list

    def get_cols_replace_method(self):
        return self.cols_replace_method

    def get_skip_cols(self):
        return self.skip_cols

    def get_impute_rate(self, mode="fit"):
        if mode == "fit":
            return list(self.cols_fit_impute_rate)
        elif mode == "transform":
            return list(self.cols_transform_impute_rate)
        else:
            raise ValueError("Unknown mode of {}".format(mode))

    @staticmethod
    def replace_missing_value_with_cols_transform_value_format(data, transform_list, missing_value_list,
                                                               output_format, skip_cols):
        _data = copy.deepcopy(data)
        replace_cols_index_list = []
        if isinstance(_data, Instance):
            for i, v in enumerate(_data.features):
                if v in missing_value_list and i not in skip_cols:
                    _data.features[i] = output_format(transform_list[i])
                    replace_cols_index_list.append(i)
                else:
                    _data[i] = output_format(v)
        else:
            for i, v in enumerate(_data):
                if str(v) in missing_value_list and i not in skip_cols:
                    _data[i] = output_format(transform_list[i])
                    replace_cols_index_list.append(i)
                else:
                    _data[i] = output_format(v)

        return _data, replace_cols_index_list

    @staticmethod
    def replace_missing_value_with_cols_transform_value(data, transform_list, missing_value_list, skip_cols):
        _data = copy.deepcopy(data)
        replace_cols_index_list = []
        if isinstance(_data, Instance):
            new_features = []
            for i, v in enumerate(_data.features):
                if v in missing_value_list and i not in skip_cols:
                    # _data.features[i] = transform_list[i]
                    new_features.append(transform_list[i])
                    replace_cols_index_list.append(i)
                else:
                    new_features.append(v)
            if replace_cols_index_list:
                # new features array will have lowest compatible dtype
                _data.features = np.array(new_features)
        else:
            for i, v in enumerate(_data):
                if str(v) in missing_value_list and i not in skip_cols:
                    _data[i] = str(transform_list[i])
                    replace_cols_index_list.append(i)

        return _data, replace_cols_index_list

    @staticmethod
    def replace_missing_value_with_replace_value_format(data, replace_value, missing_value_list, output_format):
        _data = copy.deepcopy(data)
        replace_cols_index_list = []
        if isinstance(_data, Instance):
            for i, v in enumerate(_data.features):
                if v in missing_value_list:
                    _data.features[i] = replace_value
                    replace_cols_index_list.append(i)
                else:
                    _data[i] = output_format(_data[i])
        else:
            for i, v in enumerate(_data):
                if str(v) in missing_value_list:
                    _data[i] = output_format(replace_value)
                    replace_cols_index_list.append(i)
                else:
                    _data[i] = output_format(_data[i])

        return _data, replace_cols_index_list

    @staticmethod
    def replace_missing_value_with_replace_value(data, replace_value, missing_value_list):
        _data = copy.deepcopy(data)
        replace_cols_index_list = []
        if isinstance(_data, Instance):
            new_features = []
            for i, v in enumerate(_data.features):
                if v in missing_value_list:
                    # _data.features[i] = replace_value
                    new_features.append(replace_value)
                    replace_cols_index_list.append(i)
                else:
                    new_features.append(v)
            if replace_cols_index_list:
                # make sure new features array has lowest compatible dtype
                _data.features = np.array(new_features)
        else:
            for i, v in enumerate(_data):
                if str(v) in missing_value_list:
                    _data[i] = str(replace_value)
                    replace_cols_index_list.append(i)

        return _data, replace_cols_index_list

    @staticmethod
    def __get_cols_transform_method(data, replace_method, col_replace_method):
        header = get_header(data)
        if col_replace_method:
            replace_method_per_col = {col_name: col_replace_method.get(col_name, replace_method) for col_name in header}
        else:
            replace_method_per_col = {col_name: replace_method for col_name in header}
        skip_cols = [v for v in header if replace_method_per_col[v] is None]

        return replace_method_per_col, skip_cols

    def __get_cols_transform_value(self, data, replace_method, replace_value=None):
        """

        Parameters
        ----------
        data: input data
        replace_method: dictionary of (column name, replace_method_name) pairs

        Returns
        -------
        list of transform value for each column, length equal to feature count of input data

        """
        summary_obj = MultivariateStatisticalSummary(data, -1, abnormal_list=self.abnormal_value_list)
        header = get_header(data)
        cols_transform_value = {}
        if isinstance(replace_value, list):
            if len(replace_value) != len(header):
                raise ValueError(
                    f"replace value {replace_value} length does not match with header {header}, please check.")
        for i, feature in enumerate(header):
            if replace_method[feature] is None:
                transform_value = 0
            elif replace_method[feature] == consts.MIN:
                transform_value = summary_obj.get_min()[feature]
            elif replace_method[feature] == consts.MAX:
                transform_value = summary_obj.get_max()[feature]
            elif replace_method[feature] == consts.MEAN:
                transform_value = summary_obj.get_mean()[feature]
            elif replace_method[feature] == consts.MEDIAN:
                transform_value = summary_obj.get_median()[feature]
            elif replace_method[feature] == consts.DESIGNATED:
                if isinstance(replace_value, list):
                    transform_value = replace_value[i]
                else:
                    transform_value = replace_value
                LOGGER.debug(f"replace value for feature {feature} is: {transform_value}")
            else:
                raise ValueError("Unknown replace method:{}".format(replace_method))
            if (data.schema.get("meta",{}).get("exclusive_data_type",{}).get(feature) is not None):
                data_type = data.schema["meta"]["exclusive_data_type"][feature]
                if ("str" == data_type):
                    cols_transform_value[feature] = str(transform_value)
                elif ("int" == data_type):
                    cols_transform_value[feature] = int(transform_value)
                else:
                    cols_transform_value[feature] = float(transform_value)
            else:
                cols_transform_value[feature] = transform_value

        LOGGER.debug(f"cols_transform value is: {cols_transform_value}")
        cols_transform_value = [cols_transform_value[key] for key in header]
        # cols_transform_value = {i: round(cols_transform_value[key], 6) for i, key in enumerate(header)}
        LOGGER.debug(f"cols_transform value is: {cols_transform_value}")
        return cols_transform_value

    @staticmethod
    def _transform_nan(instance):
        feature_shape = instance.features.shape[0]
        new_features = []

        for i in range(feature_shape):
            if instance.features[i] != instance.features[i]:
                new_features.append(NoneType())
            else:
                new_features.append(instance.features[i])
        new_instance = copy.deepcopy(instance)
        new_instance.features = np.array(new_features)
        return new_instance

    def __fit_replace(self, data, replace_method, replace_value=None, output_format=None,
                      col_replace_method=None):
        replace_method_per_col, skip_cols = self.__get_cols_transform_method(data, replace_method, col_replace_method)

        schema = data.schema
        if isinstance(data.first()[1], Instance):
            data = data.mapValues(lambda v: Imputer._transform_nan(v))
            data.schema = schema
        cols_transform_value = self.__get_cols_transform_value(data, replace_method_per_col,
                                                               replace_value=replace_value)
        self.skip_cols = skip_cols
        skip_cols = [get_header(data).index(v) for v in skip_cols]
        if output_format is not None:
            f = functools.partial(Imputer.replace_missing_value_with_cols_transform_value_format,
                                  transform_list=cols_transform_value, missing_value_list=self.abnormal_value_set,
                                  output_format=output_format, skip_cols=set(skip_cols))
        else:
            f = functools.partial(Imputer.replace_missing_value_with_cols_transform_value,
                                  transform_list=cols_transform_value, missing_value_list=self.abnormal_value_set,
                                  skip_cols=set(skip_cols))

        transform_data = data.mapValues(f)
        self.cols_replace_method = replace_method_per_col
        LOGGER.info(
            "finish replace missing value with cols transform value, replace method is {}".format(replace_method))
        return transform_data, cols_transform_value

    def __transform_replace(self, data, transform_value, replace_area, output_format, skip_cols):
        skip_cols = [get_header(data).index(v) for v in skip_cols]

        schema = data.schema
        if isinstance(data.first()[1], Instance):
            data = data.mapValues(lambda v: Imputer._transform_nan(v))
            data.schema = schema

        if replace_area == 'all':
            if output_format is not None:
                f = functools.partial(Imputer.replace_missing_value_with_replace_value_format,
                                      replace_value=transform_value, missing_value_list=self.abnormal_value_set,
                                      output_format=output_format)
            else:
                f = functools.partial(Imputer.replace_missing_value_with_replace_value,
                                      replace_value=transform_value, missing_value_list=self.abnormal_value_set)
        elif replace_area == 'col':
            if output_format is not None:
                f = functools.partial(Imputer.replace_missing_value_with_cols_transform_value_format,
                                      transform_list=transform_value, missing_value_list=self.abnormal_value_set,
                                      output_format=output_format,
                                      skip_cols=set(skip_cols))
            else:
                f = functools.partial(Imputer.replace_missing_value_with_cols_transform_value,
                                      transform_list=transform_value, missing_value_list=self.abnormal_value_set,
                                      skip_cols=set(skip_cols))
        else:
            raise ValueError("Unknown replace area {} in Imputer".format(replace_area))

        return data.mapValues(f)

    @staticmethod
    def __get_impute_number(some_data):
        impute_num_list = None
        data_size = None

        for line in some_data:
            processed_data = line[1][0]
            index_list = line[1][1]
            if not data_size:
                if isinstance(processed_data, Instance):
                    data_size = data_overview.get_instance_shape(processed_data)
                else:
                    data_size = len(processed_data)
                # data_size + 1, the last element of impute_num_list used to count the number of "some_data"
                impute_num_list = [0 for _ in range(data_size + 1)]

            impute_num_list[data_size] += 1
            for index in index_list:
                impute_num_list[index] += 1

        return np.array(impute_num_list)

    def __get_impute_rate_from_replace_data(self, data):
        impute_number_statics = data.applyPartitions(self.__get_impute_number).reduce(lambda x, y: x + y)
        cols_impute_rate = impute_number_statics[:-1] / impute_number_statics[-1]

        return cols_impute_rate

    def fit(self, data, replace_method=None, replace_value=None, output_format=consts.ORIGIN,
            col_replace_method=None):
        """
        Apply imputer for input data
        Parameters
        ----------
        data: Table, each data's value should be list
        replace_method: str, the strategy of imputer, like min, max, mean or designated and so on. Default None
        replace_value: str, if replace_method is designated, you should assign the replace_value which will be used to replace the value in imputer_value_list
        output_format: str, the output data format. The output data can be 'str', 'int', 'float'. Default origin, the original format as input data
        col_replace_method: dict of (col_name, replace_method), any col_name not included will take replace_method

        Returns
        ----------
        fit_data:data_instance, data after imputer
        cols_transform_value: list, the replace value in each column
        """
        if output_format not in self.support_output_format:
            raise ValueError("Unsupport output_format:{}".format(output_format))

        output_format = self.support_output_format[output_format]

        if isinstance(replace_method, str):
            replace_method = replace_method.lower()
            if replace_method not in self.support_replace_method:
                raise ValueError("Unknown replace method:{}".format(replace_method))
        elif replace_method is None and col_replace_method is None:
            if isinstance(data.first()[1], Instance):
                replace_value = 0
            else:
                replace_value = '0'
        elif replace_method is None and col_replace_method is not None:
            LOGGER.debug(f"perform computation on selected cols only: {col_replace_method}")
        else:
            raise ValueError("parameter replace_method should be str or None only")
        if isinstance(col_replace_method, dict):
            for col_name, method in col_replace_method.items():
                method = method.lower()
                if method not in self.support_replace_method:
                    raise ValueError("Unknown replace method:{}".format(method))
                col_replace_method[col_name] = method

        process_data, cols_transform_value = self.__fit_replace(data, replace_method, replace_value, output_format,
                                                                col_replace_method=col_replace_method)

        self.cols_fit_impute_rate = self.__get_impute_rate_from_replace_data(process_data)
        process_data = process_data.mapValues(lambda v: v[0])
        process_data.schema = data.schema

        return process_data, cols_transform_value

    def transform(self, data, transform_value, output_format=consts.ORIGIN, skip_cols=None):
        """
        Transform input data using Imputer with fit results
        Parameters
        ----------
        data: Table, each data's value should be list
        transform_value:
        output_format: str, the output data format. The output data can be 'str', 'int', 'float'. Default origin, the original format as input data

        Returns
        ----------
        transform_data:data_instance, data after transform
        """
        if output_format not in self.support_output_format:
            raise ValueError("Unsupport output_format:{}".format(output_format))

        output_format = self.support_output_format[output_format]
        skip_cols = [] if skip_cols is None else skip_cols

        # Now all of replace_method is "col", remain replace_area temporarily
        # replace_area = self.support_replace_area[replace_method]
        replace_area = "col"
        process_data = self.__transform_replace(data, transform_value, replace_area, output_format, skip_cols)
        self.cols_transform_impute_rate = self.__get_impute_rate_from_replace_data(process_data)
        process_data = process_data.mapValues(lambda v: v[0])
        process_data.schema = data.schema

        return process_data