FateWithShellmiaoComment/Fate/python/federatedml/evaluation/metrics/classification_metric.py

import copy
import sys

import numpy as np
import pandas as pd
from scipy.stats import stats
from sklearn.metrics import accuracy_score
from sklearn.metrics import precision_score
from sklearn.metrics import recall_score
from sklearn.metrics import average_precision_score

ROUND_NUM = 6


def neg_pos_count(labels: np.ndarray, pos_label: int):
    pos_num = ((labels == pos_label) + 0).sum()
    neg_num = len(labels) - pos_num
    return pos_num, neg_num


def sort_score_and_label(labels: np.ndarray, pred_scores: np.ndarray):
    labels = np.array(labels)
    pred_scores = np.array(pred_scores)

    sort_idx = np.flip(pred_scores.argsort())
    sorted_labels = labels[sort_idx]
    sorted_scores = pred_scores[sort_idx]

    return sorted_labels, sorted_scores


class ConfusionMatrix(object):

    @staticmethod
    def compute(sorted_labels: list, sorted_pred_scores: list, score_thresholds: list, ret: list, pos_label=1):

        for ret_type in ret:
            assert ret_type in ['tp', 'tn', 'fp', 'fn']

        sorted_labels = np.array(sorted_labels)
        sorted_scores = np.array(sorted_pred_scores)
        sorted_labels[sorted_labels != pos_label] = 0
        sorted_labels[sorted_labels == pos_label] = 1
        score_thresholds = np.array([score_thresholds]).transpose()
        pred_labels = (sorted_scores > score_thresholds) + 0

        ret_dict = {}
        if 'tp' in ret or 'tn' in ret:
            match_arr = (pred_labels + sorted_labels)
            if 'tp' in ret:
                tp_num = (match_arr == 2).sum(axis=-1)
                ret_dict['tp'] = tp_num
            if 'tn' in ret:
                tn_num = (match_arr == 0).sum(axis=-1)
                ret_dict['tn'] = tn_num

        if 'fp' in ret or 'fn' in ret:
            match_arr = (sorted_labels - pred_labels)
            if 'fp' in ret:
                fp_num = (match_arr == -1).sum(axis=-1)
                ret_dict['fp'] = fp_num
            if 'fn' in ret:
                fn_num = (match_arr == 1).sum(axis=-1)
                ret_dict['fn'] = fn_num

        return ret_dict


class ThresholdCutter(object):

    @staticmethod
    def cut_by_step(sorted_scores, steps=0.01):
        assert isinstance(steps, float) and (0 < steps < 1)
        thresholds = list(set(sorted_scores))
        thresholds, cuts = ThresholdCutter.__filt_threshold(thresholds, 0.01)
        score_threshold = thresholds

        return score_threshold, cuts

    @staticmethod
    def fixed_interval_threshold(steps=0.01):
        intervals = np.array([i for i in range(0, 100)])
        intervals = intervals * steps
        return intervals

    @staticmethod
    def cut_by_index(sorted_scores):
        cuts = np.array([c / 100 for c in range(100)])
        data_size = len(sorted_scores)
        indexs = [int(data_size * cut) for cut in cuts]
        score_threshold = [sorted_scores[idx] for idx in indexs]
        return score_threshold, cuts

    @staticmethod
    def __filt_threshold(thresholds, step):
        cuts = list(map(float, np.arange(0, 1, step)))
        size = len(list(thresholds))
        thresholds.sort(reverse=True)
        index_list = [int(size * cut) for cut in cuts]
        new_thresholds = [thresholds[idx] for idx in index_list]

        return new_thresholds, cuts

    @staticmethod
    def cut_by_quantile(scores, quantile_list=None, interpolation='nearest', remove_duplicate=True):

        if quantile_list is None:  # default is 20 intervals
            quantile_list = [round(i * 0.05, 3) for i in range(20)] + [1.0]
        quantile_val = np.quantile(scores, quantile_list, interpolation=interpolation)
        if remove_duplicate:
            quantile_val = sorted(list(set(quantile_val)))
        else:
            quantile_val = sorted(list(quantile_val))

        if len(quantile_val) == 1:
            quantile_val = [np.min(scores), np.max(scores)]

        return quantile_val


class KS(object):

    @staticmethod
    def compute(labels, pred_scores, pos_label=1, fixed_interval_threshold=True):
        sorted_labels, sorted_scores = sort_score_and_label(labels, pred_scores)

        threshold, cuts = ThresholdCutter.cut_by_index(sorted_scores)
        confusion_mat = ConfusionMatrix.compute(sorted_labels, sorted_scores, threshold, ret=['tp', 'fp'],
                                                pos_label=pos_label)
        pos_num, neg_num = neg_pos_count(sorted_labels, pos_label=pos_label)

        assert pos_num > 0 and neg_num > 0, "error when computing KS metric, pos sample number and neg sample number" \
                                            "must be larger than 0"

        tpr_arr = confusion_mat['tp'] / pos_num
        fpr_arr = confusion_mat['fp'] / neg_num

        tpr = np.append(tpr_arr, np.array([1.0]))
        fpr = np.append(fpr_arr, np.array([1.0]))
        cuts = np.append(cuts, np.array([1.0]))

        ks_curve = tpr[:-1] - fpr[:-1]
        ks_val = np.max(ks_curve)

        return ks_val, fpr, tpr, threshold, cuts


class BiClassMetric(object):

    def __init__(self, cut_method='step', remove_duplicate=False, pos_label=1):
        assert cut_method in ['step', 'quantile']
        self.cut_method = cut_method
        self.remove_duplicate = remove_duplicate  # available when cut_method is quantile
        self.pos_label = pos_label

    def prepare_confusion_mat(self, labels, scores, add_to_end=True, ):
        sorted_labels, sorted_scores = sort_score_and_label(labels, scores)

        score_threshold, cuts = None, None

        if self.cut_method == 'step':
            score_threshold, cuts = ThresholdCutter.cut_by_step(sorted_scores, steps=0.01)
            if add_to_end:
                score_threshold.append(min(score_threshold) - 0.001)
                cuts.append(1)

        elif self.cut_method == 'quantile':
            score_threshold = ThresholdCutter.cut_by_quantile(sorted_scores, remove_duplicate=self.remove_duplicate)
            score_threshold = list(np.flip(score_threshold))

        confusion_mat = ConfusionMatrix.compute(sorted_labels, sorted_scores, score_threshold,
                                                ret=['tp', 'fp', 'fn', 'tn'], pos_label=self.pos_label)

        return confusion_mat, score_threshold, cuts

    def compute(self, labels, scores, ):
        confusion_mat, score_threshold, cuts = self.prepare_confusion_mat(labels, scores, )
        metric_scores = self.compute_metric_from_confusion_mat(confusion_mat)
        return list(metric_scores), score_threshold, cuts

    def compute_metric_from_confusion_mat(self, *args):
        raise NotImplementedError()


class Lift(BiClassMetric):
    """
    Compute lift
    """

    @staticmethod
    def _lift_helper(val):

        tp, fp, fn, tn, labels_num = val[0], val[1], val[2], val[3], val[4]

        lift_x_type, lift_y_type = [], []

        for label_type in ['1', '0']:

            if label_type == '0':
                tp, tn = tn, tp
                fp, fn = fn, fp

            if labels_num == 0:
                lift_x = 1
                denominator = 1
            else:
                lift_x = (tp + fp) / labels_num
                denominator = (tp + fn) / labels_num

            if tp + fp == 0:
                numerator = 1
            else:
                numerator = tp / (tp + fp)

            if denominator == 0:
                lift_y = sys.float_info.max
            else:
                lift_y = numerator / denominator

            lift_x_type.insert(0, lift_x)
            lift_y_type.insert(0, lift_y)

        return lift_x_type, lift_y_type

    def compute(self, labels, pred_scores, pos_label=1):

        confusion_mat, score_threshold, cuts = self.prepare_confusion_mat(labels, pred_scores, add_to_end=False, )

        lifts_y, lifts_x = self.compute_metric_from_confusion_mat(confusion_mat, len(labels), )

        return lifts_y, lifts_x, list(score_threshold)

    def compute_metric_from_confusion_mat(self, confusion_mat, labels_len, ):

        labels_nums = np.zeros(len(confusion_mat['tp'])) + labels_len

        rs = map(self._lift_helper, zip(confusion_mat['tp'], confusion_mat['fp'],
                                        confusion_mat['fn'], confusion_mat['tn'], labels_nums))

        rs = list(rs)

        lifts_x, lifts_y = [i[0] for i in rs], [i[1] for i in rs]

        return lifts_y, lifts_x


class Gain(BiClassMetric):
    """
    Compute Gain
    """

    @staticmethod
    def _gain_helper(val):

        tp, fp, fn, tn, num_label = val[0], val[1], val[2], val[3], val[4]

        gain_x_type, gain_y_type = [], []

        for pos_label in ['1', '0']:

            if pos_label == '0':
                tp, tn = tn, tp
                fp, fn = fn, fp

            if num_label == 0:
                gain_x = 1
            else:
                gain_x = float((tp + fp) / num_label)

            num_positives = tp + fn
            if num_positives == 0:
                gain_y = 1
            else:
                gain_y = float(tp / num_positives)

            gain_x_type.insert(0, gain_x)
            gain_y_type.insert(0, gain_y)

        return gain_x_type, gain_y_type

    def compute(self, labels, pred_scores, pos_label=1):

        confusion_mat, score_threshold, cuts = self.prepare_confusion_mat(labels, pred_scores, add_to_end=False, )

        gain_y, gain_x = self.compute_metric_from_confusion_mat(confusion_mat, len(labels))

        return gain_y, gain_x, list(score_threshold)

    def compute_metric_from_confusion_mat(self, confusion_mat, labels_len):

        labels_nums = np.zeros(len(confusion_mat['tp'])) + labels_len

        rs = map(self._gain_helper, zip(confusion_mat['tp'], confusion_mat['fp'],
                                        confusion_mat['fn'], confusion_mat['tn'], labels_nums))

        rs = list(rs)

        gain_x, gain_y = [i[0] for i in rs], [i[1] for i in rs]

        return gain_y, gain_x


class BiClassPrecision(BiClassMetric):
    """
    Compute binary classification precision
    """

    def compute_metric_from_confusion_mat(self, confusion_mat, formatted=True, impute_val=1.0):
        numerator = confusion_mat['tp']
        denominator = (confusion_mat['tp'] + confusion_mat['fp'])
        zero_indexes = (denominator == 0)
        denominator[zero_indexes] = 1
        precision_scores = numerator / denominator
        precision_scores[zero_indexes] = impute_val  # impute_val is for prettifying when drawing pr curves

        if formatted:
            score_formatted = [[0, i] for i in precision_scores]
            return score_formatted
        else:
            return precision_scores


class MultiClassPrecision(object):
    """
    Compute multi-classification precision
    """

    def compute(self, labels, pred_scores):
        all_labels = sorted(set(labels).union(set(pred_scores)))
        return precision_score(labels, pred_scores, average=None), all_labels


class BiClassRecall(BiClassMetric):
    """
    Compute binary classification recall
    """

    def compute_metric_from_confusion_mat(self, confusion_mat, formatted=True):
        recall_scores = confusion_mat['tp'] / (confusion_mat['tp'] + confusion_mat['fn'])

        if formatted:
            score_formatted = [[0, i] for i in recall_scores]
            return score_formatted
        else:
            return recall_scores


class MultiClassRecall(object):
    """
    Compute multi-classification recall
    """

    def compute(self, labels, pred_scores):
        all_labels = sorted(set(labels).union(set(pred_scores)))
        return recall_score(labels, pred_scores, average=None), all_labels


class BiClassAccuracy(BiClassMetric):
    """
    Compute binary classification accuracy
    """

    def compute(self, labels, scores, normalize=True):
        confusion_mat, score_threshold, cuts = self.prepare_confusion_mat(labels, scores)
        metric_scores = self.compute_metric_from_confusion_mat(confusion_mat, normalize=normalize)
        return list(metric_scores), score_threshold[: len(metric_scores)], cuts[: len(metric_scores)]

    def compute_metric_from_confusion_mat(self, confusion_mat, normalize=True):
        rs = (confusion_mat['tp'] + confusion_mat['tn']) / \
             (confusion_mat['tp'] + confusion_mat['tn'] + confusion_mat['fn'] + confusion_mat['fp']) if normalize \
            else (confusion_mat['tp'] + confusion_mat['tn'])
        return rs[:-1]


class MultiClassAccuracy(object):
    """
    Compute multi-classification accuracy
    """

    def compute(self, labels, pred_scores, normalize=True):
        return accuracy_score(labels, pred_scores, normalize=normalize)


class FScore(object):
    """
    Compute F score from bi-class confusion mat
    """

    @staticmethod
    def compute(labels, pred_scores, beta=1, pos_label=1):
        sorted_labels, sorted_scores = sort_score_and_label(labels, pred_scores)
        _, cuts = ThresholdCutter.cut_by_step(sorted_scores, steps=0.01)
        fixed_interval_threshold = ThresholdCutter.fixed_interval_threshold()
        confusion_mat = ConfusionMatrix.compute(sorted_labels, sorted_scores,
                                                fixed_interval_threshold,
                                                ret=['tp', 'fp', 'fn', 'tn'], pos_label=pos_label)

        precision_computer = BiClassPrecision()
        recall_computer = BiClassRecall()
        p_score = precision_computer.compute_metric_from_confusion_mat(confusion_mat, formatted=False)
        r_score = recall_computer.compute_metric_from_confusion_mat(confusion_mat, formatted=False)

        beta_2 = beta * beta
        denominator = (beta_2 * p_score + r_score)
        denominator[denominator == 0] = 1e-6  # in case denominator is 0
        numerator = (1 + beta_2) * (p_score * r_score)
        f_score = numerator / denominator

        return f_score, fixed_interval_threshold, cuts


class PSI(object):

    def compute(self, train_scores: list, validate_scores: list, train_labels=None, validate_labels=None,
                debug=False, str_intervals=False, round_num=3, pos_label=1):
        """
        train/validate scores: predicted scores on train/validate set
        train/validate labels: true labels
        debug: print debug message
        if train&validate labels are not None, count positive sample percentage in every interval
        pos_label: pos label
        round_num： round number
        str_intervals: return str intervals
        """

        train_scores = np.array(train_scores)
        validate_scores = np.array(validate_scores)
        quantile_points = ThresholdCutter().cut_by_quantile(train_scores)

        train_count = self.quantile_binning_and_count(train_scores, quantile_points)
        validate_count = self.quantile_binning_and_count(validate_scores, quantile_points)

        train_pos_perc, validate_pos_perc = None, None

        if train_labels is not None and validate_labels is not None:
            assert len(train_labels) == len(train_scores) and len(validate_labels) == len(validate_scores)
            train_labels, validate_labels = np.array(train_labels), np.array(validate_labels)
            train_pos_count = self.quantile_binning_and_count(train_scores[train_labels == pos_label], quantile_points)
            validate_pos_count = self.quantile_binning_and_count(validate_scores[validate_labels == pos_label],
                                                                 quantile_points)

            train_pos_perc = np.array(train_pos_count['count']) / np.array(train_count['count'])
            validate_pos_perc = np.array(validate_pos_count['count']) / np.array(validate_count['count'])

            # handle special cases
            train_pos_perc[train_pos_perc == np.inf] = -1
            validate_pos_perc[validate_pos_perc == np.inf] = -1
            train_pos_perc[np.isnan(train_pos_perc)] = 0
            validate_pos_perc[np.isnan(validate_pos_perc)] = 0

        if debug:
            print(train_count)
            print(validate_count)

        assert (train_count['interval'] == validate_count['interval']), 'train count interval is not equal to ' \
                                                                        'validate count interval'

        expected_interval = np.array(train_count['count'])
        actual_interval = np.array(validate_count['count'])

        expected_interval = expected_interval.astype(np.float)
        actual_interval = actual_interval.astype(np.float)

        psi_scores, total_psi, expected_interval, actual_interval, expected_percentage, actual_percentage \
            = self.psi_score(expected_interval, actual_interval, len(train_scores), len(validate_scores))

        intervals = train_count['interval'] if not str_intervals else PSI.intervals_to_str(train_count['interval'],
                                                                                           round_num=round_num)

        if train_labels is None and validate_labels is None:
            return psi_scores, total_psi, expected_interval, expected_percentage, actual_interval, actual_percentage, \
                intervals
        else:
            return psi_scores, total_psi, expected_interval, expected_percentage, actual_interval, actual_percentage, \
                train_pos_perc, validate_pos_perc, intervals

    @staticmethod
    def quantile_binning_and_count(scores, quantile_points):
        """
        left edge and right edge of last interval are closed
        """

        assert len(quantile_points) >= 2

        left_bounds = copy.deepcopy(quantile_points[:-1])
        right_bounds = copy.deepcopy(quantile_points[1:])

        last_interval_left = left_bounds.pop()
        last_interval_right = right_bounds.pop()

        bin_result_1, bin_result_2 = None, None

        if len(left_bounds) != 0 and len(right_bounds) != 0:
            bin_result_1 = pd.cut(scores, pd.IntervalIndex.from_arrays(left_bounds, right_bounds, closed='left'))

        bin_result_2 = pd.cut(scores, pd.IntervalIndex.from_arrays([last_interval_left], [last_interval_right],
                                                                   closed='both'))

        count1 = None if bin_result_1 is None else bin_result_1.value_counts().reset_index()
        count2 = bin_result_2.value_counts().reset_index()

        # if predict scores are the same, count1 will be None, only one interval exists
        final_interval = list(count1['index']) + list(count2['index']) if count1 is not None else list(count2['index'])
        final_count = list(count1[0]) + list(count2[0]) if count1 is not None else list(count2[0])
        rs = {'interval': final_interval, 'count': final_count}

        return rs

    @staticmethod
    def interval_psi_score(val):
        expected, actual = val[0], val[1]
        return (actual - expected) * np.log(actual / expected)

    @staticmethod
    def intervals_to_str(intervals, round_num=3):
        str_intervals = []
        for interval in intervals:
            left_bound, right_bound = '[', ']'
            if interval.closed == 'left':
                right_bound = ')'
            elif interval.closed == 'right':
                left_bound = '('
            str_intervals.append("{}{}, {}{}".format(left_bound, round(interval.left, round_num),
                                                     round(interval.right, round_num), right_bound))

        return str_intervals

    @staticmethod
    def psi_score(expected_interval: np.ndarray, actual_interval: np.ndarray, expect_total_num, actual_total_num,
                  debug=False):

        expected_interval[expected_interval == 0] = 1e-6  # in case no overlap samples

        actual_interval[actual_interval == 0] = 1e-6  # in case no overlap samples

        expected_percentage = expected_interval / expect_total_num
        actual_percentage = actual_interval / actual_total_num

        if debug:
            print(expected_interval)
            print(actual_interval)
            print(expected_percentage)
            print(actual_percentage)

        psi_scores = list(map(PSI.interval_psi_score, zip(expected_percentage, actual_percentage)))
        psi_scores = np.array(psi_scores)
        total_psi = psi_scores.sum()
        return psi_scores, total_psi, expected_interval, actual_interval, expected_percentage, actual_percentage


class KSTest(object):

    @staticmethod
    def compute(train_scores, validate_scores):
        """
        train/validate scores: predicted scores on train/validate set
        """
        return stats.ks_2samp(train_scores, validate_scores).pvalue


class AveragePrecisionScore(object):

    @staticmethod
    def compute(train_scores, validate_scores, train_labels, validate_labels):
        """
            train/validate scores: predicted scores on train/validate set
            train/validate labels: true labels
        """
        train_mAP = average_precision_score(train_labels, train_scores)
        validate_mAP = average_precision_score(validate_labels, validate_scores)
        return abs(train_mAP - validate_mAP)


class Distribution(object):

    @staticmethod
    def compute(train_scores: list, validate_scores: list):
        """
        train/validate scores: predicted scores on train/validate set
        """
        train_scores = np.array(train_scores)
        validate_scores = np.array(validate_scores)
        validate_scores = dict(validate_scores)
        count = 0
        for key, value in train_scores:
            if key in validate_scores.keys() and value != validate_scores.get(key):
                count += 1
        return count / len(train_scores)