easy-ml/feature/strategy_iv.py

# -*- coding:utf-8 -*-
"""
@author: yq
@time: 2024/1/2
@desc: iv值及单调性筛选类
"""
import json
from itertools import combinations_with_replacement
from typing import List, Dict, Tuple

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import scorecardpy as sc
import seaborn as sns
from pandas.core.dtypes.common import is_numeric_dtype
from tqdm import tqdm

from commom import f_display_images_by_side, NumpyEncoder
from entitys import DataSplitEntity, CandidateFeatureEntity, DataPreparedEntity, DataFeatureEntity, MetricFucEntity
from .feature_utils import f_judge_monto, f_get_corr, f_get_ivf, f_format_bin, f_monto_contrast
from .filter_strategy_base import FilterStrategyBase


class StrategyIv(FilterStrategyBase):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def _f_get_iv_by_bins(self, bins) -> pd.DataFrame:
        iv = {key_: [round(value_['total_iv'].max(), 4)] for key_, value_ in bins.items()}
        iv = pd.DataFrame.from_dict(iv, orient='index', columns=['IV']).reset_index()
        iv = iv.sort_values('IV', ascending=False).reset_index(drop=True)
        iv.columns = ['变量', 'IV']
        return iv

    def _f_get_var_corr_image(self, train_woe):
        if len(train_woe.columns.to_list()) <= 1:
            return None
        train_corr = f_get_corr(train_woe)
        plt.figure(figsize=(12, 12))
        sns.heatmap(train_corr, vmax=1, square=True, cmap='RdBu', annot=True)
        plt.title('Variables Correlation', fontsize=15)
        plt.yticks(rotation=0)
        plt.xticks(rotation=90)
        path = self.data_process_config.f_get_save_path(f"var_corr.png")
        plt.savefig(path)
        return path

    def _f_save_var_trend(self, bins, x_columns_candidate, prefix):
        image_path_list = []
        for k in x_columns_candidate:
            bin_df = bins[k]
            # bin_df["bin"] = bin_df["bin"].apply(lambda x: re.sub(r"(\d+\.\d+)",
            #                                                      lambda m: "{:.2f}".format(float(m.group(0))), x))
            sc.woebin_plot(bin_df)
            path = self.data_process_config.f_get_save_path(f"{prefix}_{k}.png")
            plt.savefig(path)
            image_path_list.append(path)
        return image_path_list

    def _f_get_bins_by_breaks(self, data: pd.DataFrame, candidate_dict: Dict[str, CandidateFeatureEntity],
                              y_column=None):
        y_column = self.data_process_config.y_column if y_column is None else y_column
        special_values = self.data_process_config.special_values
        x_columns_candidate = list(candidate_dict.keys())
        breaks_list = {}
        for column, candidate in candidate_dict.items():
            breaks_list[column] = candidate.breaks_list
        bins = sc.woebin(data[x_columns_candidate + [y_column]], y=y_column, breaks_list=breaks_list,
                         special_values=special_values, print_info=False)
        return bins

    def _f_corr_filter(self, data: DataSplitEntity, candidate_dict: Dict[str, CandidateFeatureEntity]) -> List[str]:
        # 相关性剔除变量
        corr_threshold = self.data_process_config.corr_threshold
        breaks_list = self.data_process_config.breaks_list
        train_data = data.train_data
        x_columns_candidate = list(candidate_dict.keys())

        bins = self._f_get_bins_by_breaks(train_data, candidate_dict)
        train_woe = sc.woebin_ply(train_data[x_columns_candidate], bins, print_info=False)
        corr_df = f_get_corr(train_woe)
        corr_dict = corr_df.to_dict()
        for column, corr in corr_dict.items():
            column = column.replace("_woe", "")
            if column not in x_columns_candidate:
                continue
            for challenger_column, challenger_corr in corr.items():
                challenger_column = challenger_column.replace("_woe", "")
                if challenger_corr < corr_threshold or column == challenger_column \
                        or challenger_column not in x_columns_candidate:
                    continue
                iv_max = candidate_dict[column].iv_max
                challenger_iv_max = candidate_dict[challenger_column].iv_max
                if iv_max > challenger_iv_max:
                    if challenger_column not in breaks_list.keys():
                        x_columns_candidate.remove(challenger_column)
                else:
                    if column not in breaks_list.keys():
                        x_columns_candidate.remove(column)
                    break
        return x_columns_candidate

    def _f_wide_filter(self, data: DataSplitEntity) -> Dict:
        # 粗筛变量
        train_data = data.train_data
        test_data = data.test_data
        special_values = self.data_process_config.special_values
        breaks_list = self.data_process_config.breaks_list.copy()
        y_column = self.data_process_config.y_column
        iv_threshold_wide = self.data_process_config.iv_threshold_wide
        x_columns_candidate = self.data_process_config.x_columns_candidate
        if x_columns_candidate is None or len(x_columns_candidate) == 0:
            x_columns_candidate = train_data.columns.tolist()
        if y_column in x_columns_candidate:
            x_columns_candidate.remove(y_column)

        bins_train = sc.woebin(train_data[x_columns_candidate + [y_column]], y=y_column, bin_num_limit=5,
                               special_values=special_values, breaks_list=breaks_list, print_info=False)

        for column, bin in bins_train.items():
            breaks_list[column] = list(bin['breaks'])
        bins_test = None
        if test_data is not None and len(test_data) != 0:
            bins_test = sc.woebin(test_data[x_columns_candidate + [y_column]], y=y_column,
                                  special_values=special_values, breaks_list=breaks_list, print_info=False)
        bins_iv_dict = {}
        for column, bin_train in bins_train.items():
            train_iv = bin_train['total_iv'][0]
            test_iv = 0
            if bins_test is not None:
                bin_test = bins_test[column]
                test_iv = bin_test['total_iv'][0]
            iv_max = train_iv + test_iv
            if train_iv < iv_threshold_wide:
                continue
            bins_iv_dict[column] = {"iv_max": iv_max, "breaks_list": breaks_list[column]}
        return bins_iv_dict

    def _f_get_best_bins_numeric(self, data: DataSplitEntity, x_column: str):
        # 贪婪搜索【训练集】及【测试集】加起来【iv】值最高的且【单调】的分箱
        interval = self.data_process_config.bin_search_interval
        iv_threshold = self.data_process_config.iv_threshold
        special_values = self.data_process_config.get_special_values(x_column)
        breaks_list = self.data_process_config.get_breaks_list(x_column)
        y_column = self.data_process_config.y_column
        sample_rate = self.data_process_config.sample_rate
        format_bin = self.data_process_config.format_bin
        pos_neg_cnt = self.data_process_config.pos_neg_cnt
        monto_contrast_change_cnt = self.data_process_config.monto_contrast_change_cnt

        def _n0(x):
            return sum(x == 0)

        def _n1(x):
            return sum(x == 1)

        def _f_distribute_balls(balls, boxes):
            # 计算在 balls - 1 个空位中放入 boxes - 1 个隔板的方法数
            total_ways = combinations_with_replacement(range(balls + boxes - 1), boxes - 1)
            distribute_list = []
            # 遍历所有可能的隔板位置
            for combo in total_ways:
                # 根据隔板位置分配球
                distribution = [0] * boxes
                start = 0
                for i, divider in enumerate(combo):
                    distribution[i] = divider - start + 1
                    start = divider + 1
                distribution[-1] = balls - start  # 最后一个箱子的球数
                # 确保每个箱子至少有一个球
                if all(x > 0 for x in distribution):
                    distribute_list.append(distribution)
            return distribute_list

        def _get_sv_bins(df, x_column, y_column, special_values):
            # special_values_bins
            sv_bin_list = []
            for special in special_values:
                dtm = df[df[x_column] == special]
                if len(dtm) != 0:
                    dtm['bin'] = [str(special)] * len(dtm)
                    binning = dtm.groupby(['bin'], group_keys=False)[y_column].agg(
                        [_n0, _n1]).reset_index().rename(columns={'_n0': 'good', '_n1': 'bad'})
                    binning['is_special_values'] = [True] * len(binning)
                    sv_bin_list.append(binning)
            return sv_bin_list

        def _get_bin_left_value(bin: str):
            if "," not in bin:
                return float(bin)
            left = bin.split(",")[0]
            return float(left[1:])

        def _get_bins(df, x_column, y_column, breaks_list):
            dtm = pd.DataFrame({'y': df[y_column], 'value': df[x_column]})
            bstbrks = [-np.inf] + breaks_list + [np.inf]
            labels = ['[{},{})'.format(bstbrks[i], bstbrks[i + 1]) for i in range(len(bstbrks) - 1)]
            dtm.loc[:, 'bin'] = pd.cut(dtm['value'], bstbrks, right=False, labels=labels)
            dtm['bin'] = dtm['bin'].astype(str)
            bins = dtm.groupby(['bin'], group_keys=False)['y'].agg([_n0, _n1]) \
                .reset_index().rename(columns={'_n0': 'good', '_n1': 'bad'})
            bins['is_special_values'] = [False] * len(bins)
            bins["ordered"] = bins['bin'].apply(_get_bin_left_value)
            bins = bins.sort_values(by=["ordered"], ascending=[True])
            return bins

        def _get_badprob(bins):
            bins['count'] = bins['good'] + bins['bad']
            bins['badprob'] = bins['bad'] / bins['count']
            bad_prob = bins[bins['is_special_values'] == False]['badprob'].values.tolist()
            return bad_prob

        def _calculation_iv(bins, judge_monto=True, pos_neg_cnt=1):
            # 单调性判断
            bad_prob = _get_badprob(bins)
            if judge_monto and not f_judge_monto(bad_prob, pos_neg_cnt):
                return -1
            # 计算iv
            infovalue = pd.DataFrame({'good': bins['good'], 'bad': bins['bad']}) \
                .replace(0, 0.9) \
                .assign(
                DistrBad=lambda x: x.bad / sum(x.bad),
                DistrGood=lambda x: x.good / sum(x.good)
            ) \
                .assign(iv=lambda x: (x.DistrBad - x.DistrGood) * np.log(x.DistrBad / x.DistrGood)) \
                .iv
            bins['bin_iv'] = infovalue
            bins['total_iv'] = bins['bin_iv'].sum()
            iv = bins['total_iv'].values[0]
            return iv

        def _f_sampling(distribute_list: list, sample_rate: float):
            # 采样，完全贪婪搜索耗时太长
            sampled_list = distribute_list[::int(1 / sample_rate)]
            return sampled_list

        train_data = data.train_data
        train_data_filter = train_data[~train_data[x_column].isin(special_values)]
        train_data_filter = train_data_filter.sort_values(by=x_column, ascending=True)
        train_data_x = train_data_filter[x_column]
        train_data_x_describe = train_data_x.describe(percentiles=[0.1, 0.9])
        train_data_x_max = train_data_x.max()

        test_data = data.test_data
        test_data_filter = None
        if test_data is not None and len(test_data) != 0:
            test_data_filter = test_data[~test_data[x_column].isin(special_values)]
            test_data_filter = test_data_filter.sort_values(by=x_column, ascending=True)

        # 构造数据切分点
        # 计算 2 - 5 箱的情况
        distribute_list = []
        points_list = []
        for bin_num in list(range(2, 6)):
            distribute_list_cache = _f_distribute_balls(int(1 / interval), bin_num)
            # 4箱及以上得采样，不然耗时太久
            sample_num = 1000 * sample_rate
            if sample_rate <= 0.15:
                sample_num *= 2
            if bin_num == 4 and len(distribute_list_cache) >= sample_num:
                distribute_list_cache = _f_sampling(distribute_list_cache, sample_num / len(distribute_list_cache))
            sample_num = 4000 * sample_rate
            if bin_num == 5 and len(distribute_list_cache) >= sample_num:
                distribute_list_cache = _f_sampling(distribute_list_cache, sample_num / len(distribute_list_cache))
            distribute_list.extend(distribute_list_cache)
        for distribute in distribute_list:
            point_list_cache = []
            point_percentile_list = [sum(distribute[0:idx + 1]) * interval for idx, _ in enumerate(distribute[0:-1])]
            for point_percentile in point_percentile_list:
                point = train_data_x.iloc[int(len(train_data_x) * point_percentile)]
                point = float(point)
                if format_bin:
                    point = f_format_bin(train_data_x_describe, point)
                point = round(point, 2)
                if point == 0:
                    continue
                if point not in point_list_cache and point < train_data_x_max:
                    point_list_cache.append(point)
            if point_list_cache not in points_list and len(point_list_cache) != 0:
                points_list.append(point_list_cache)
        # IV与单调性过滤
        # 获取2 - 5 箱的情况下最佳分箱
        bins_enum = {}
        iv_max = 0
        breaks_list_target = None
        judge_monto = True
        if len(breaks_list) != 0:
            points_list = [breaks_list]
            judge_monto = False
        train_sv_bin_list = _get_sv_bins(train_data, x_column, y_column, special_values)
        test_sv_bin_list = None
        if test_data_filter is not None:
            test_sv_bin_list = _get_sv_bins(test_data, x_column, y_column, special_values)
        for point_list in points_list:
            is_discard = 0
            discard_reason = ""
            is_monto = 1
            is_monto_contrast = 1
            train_bins = _get_bins(train_data_filter, x_column, y_column, point_list)
            # 与special_values合并计算iv
            for sv_bin in train_sv_bin_list:
                train_bins = pd.concat((train_bins, sv_bin))
            # _calculation_iv包含了单调性判断，并排除了特殊值
            train_iv = _calculation_iv(train_bins, judge_monto, pos_neg_cnt)
            # 只限制训练集的单调性与iv值大小
            if train_iv < iv_threshold:
                discard_reason = f"训练集iv小于阈值{iv_threshold}"
                is_discard = 1
                is_monto = 0

            test_iv = 0
            if test_data_filter is not None:
                test_bins = _get_bins(test_data_filter, x_column, y_column, point_list)
                for sv_bin in test_sv_bin_list:
                    test_bins = pd.concat((test_bins, sv_bin))
                test_iv = _calculation_iv(test_bins, judge_monto, pos_neg_cnt)
                # 趋势一致性判断
                train_bad_prob = _get_badprob(train_bins)
                test_bad_prob = _get_badprob(test_bins)
                if not f_monto_contrast(train_bad_prob, test_bad_prob, monto_contrast_change_cnt) \
                        and len(breaks_list) == 0:
                    discard_reason = f"变量趋势一致性不够"
                    is_discard = 1
                    is_monto_contrast = 0

            iv = train_iv + test_iv

            if len(breaks_list) == 0:
                bin_num = len(point_list) + 1
                if bin_num not in bins_enum.keys():
                    bins_enum[bin_num] = []
                bins_enum[bin_num].append({
                    "is_discard": is_discard,
                    "is_monto": is_monto,
                    "is_monto_contrast": is_monto_contrast,
                    "discard_reason": discard_reason,
                    "point_list": point_list,
                    "iv": iv,
                })

            if iv > iv_max and not is_discard:
                iv_max = iv
                breaks_list_target = point_list

        # 各个分箱数下的最佳分箱点
        bins_enum_best_point = []
        for k, v in bins_enum.items():
            df_bin_enum = pd.DataFrame(data=v)
            df_bin_enum.sort_values(by=["is_discard", "is_monto", "is_monto_contrast", "iv"],
                                    ascending=[True, False, False, False], inplace=True)
            bins_enum_best_point.append(df_bin_enum.iloc[0]["point_list"])

        return iv_max, breaks_list_target, bins_enum_best_point

    def filter(self, data: DataSplitEntity, *args, **kwargs) -> Tuple[
        Dict[str, CandidateFeatureEntity], Dict[str, List[CandidateFeatureEntity]]]:
        # 粗筛
        bins_iv_dict = self._f_wide_filter(data)
        x_columns_candidate = list(bins_iv_dict.keys())
        candidate_num = self.data_process_config.candidate_num
        candidate_dict: Dict[str, CandidateFeatureEntity] = {}
        numeric_candidate_dict_all: Dict[str, List[CandidateFeatureEntity]] = {}
        for x_column in tqdm(x_columns_candidate):
            if is_numeric_dtype(data.train_data[x_column]):
                iv_max, breaks_list, bins_enum_best_point = self._f_get_best_bins_numeric(data, x_column)
                if len(bins_enum_best_point) != 0:
                    numeric_candidate_dict_all[x_column] = []
                    for point in bins_enum_best_point:
                        numeric_candidate_dict_all[x_column].append(CandidateFeatureEntity(x_column, point, 0))
                if breaks_list is None:
                    continue
                candidate_dict[x_column] = CandidateFeatureEntity(x_column, breaks_list, iv_max)
            else:
                # 字符型暂时用scorecardpy来处理
                candidate_dict[x_column] = CandidateFeatureEntity(x_column, bins_iv_dict[x_column]["breaks_list"],
                                                                  bins_iv_dict[x_column]["iv_max"])

        # 相关性进一步剔除变量
        x_columns_candidate = self._f_corr_filter(data, candidate_dict)
        candidate_list: List[CandidateFeatureEntity] = []
        for x_column, v in candidate_dict.items():
            if x_column in x_columns_candidate:
                candidate_list.append(v)

        candidate_list.sort(key=lambda x: x.iv_max, reverse=True)
        candidate_list = candidate_list[0:candidate_num]
        candidate_dict = {}
        for candidate in candidate_list:
            candidate_dict[candidate.x_column] = candidate
        return candidate_dict, numeric_candidate_dict_all

    def feature_generate(self, data: DataSplitEntity, candidate_dict: Dict[str, CandidateFeatureEntity], *args,
                         **kwargs) -> DataPreparedEntity:
        train_data = data.train_data
        val_data = data.val_data
        test_data = data.test_data
        y_column = self.data_process_config.y_column
        x_columns_candidate = list(candidate_dict.keys())
        bins = self._f_get_bins_by_breaks(train_data, candidate_dict)

        train_woe = sc.woebin_ply(train_data[x_columns_candidate], bins, print_info=False)
        train_data_feature = DataFeatureEntity(pd.concat((train_woe, train_data[y_column]), axis=1),
                                               train_woe.columns.tolist(), y_column)

        val_data_feature = None
        if val_data is not None and len(val_data) != 0:
            val_woe = sc.woebin_ply(val_data[x_columns_candidate], bins, print_info=False)
            val_data_feature = DataFeatureEntity(pd.concat((val_woe, val_data[y_column]), axis=1),
                                                 train_woe.columns.tolist(), y_column)

        test_data_feature = None
        if test_data is not None and len(test_data) != 0:
            test_woe = sc.woebin_ply(test_data[x_columns_candidate], bins, print_info=False)
            test_data_feature = DataFeatureEntity(pd.concat((test_woe, test_data[y_column]), axis=1),
                                                  train_woe.columns.tolist(), y_column)
        return DataPreparedEntity(train_data_feature, val_data_feature, test_data_feature, bins=bins,
                                  data_split_original=data)

    def feature_report(self, data: DataSplitEntity, candidate_dict: Dict[str, CandidateFeatureEntity],
                       numeric_candidate_dict_all: Dict[str, List[CandidateFeatureEntity]],
                       *args, **kwargs) -> Dict[str, MetricFucEntity]:
        y_column = self.data_process_config.y_column
        jupyter = self.data_process_config.jupyter
        x_columns_candidate = list(candidate_dict.keys())
        train_data = data.train_data
        test_data = data.test_data

        metric_value_dict = {}
        # 样本分布
        metric_value_dict["样本分布"] = MetricFucEntity(table=data.get_distribution(y_column), table_font_size=10,
                                                    table_cell_width=3)
        # 变量iv及psi
        train_bins = self._f_get_bins_by_breaks(train_data, candidate_dict)
        train_iv = self._f_get_iv_by_bins(train_bins)

        if test_data is not None and len(test_data) != 0:
            # 计算psi仅需把y改成识别各自训练集测试集即可
            psi_df = pd.concat((train_data, test_data))
            psi_df["#target#"] = [1] * len(train_data) + [0] * len(test_data)
            psi = self._f_get_bins_by_breaks(psi_df, candidate_dict, y_column="#target#")
            psi = self._f_get_iv_by_bins(psi)
            psi.columns = ['变量', 'psi']
            train_iv = pd.merge(train_iv, psi, on="变量", how="left")

            # 变量趋势-测试集
            test_bins = self._f_get_bins_by_breaks(test_data, candidate_dict)
            image_path_list = self._f_save_var_trend(test_bins, x_columns_candidate, "test")
            metric_value_dict["变量趋势-测试集"] = MetricFucEntity(image_path=image_path_list, image_size=4)

        metric_value_dict["变量iv"] = MetricFucEntity(table=train_iv, table_font_size=10, table_cell_width=3)
        # 变量趋势-训练集
        image_path_list = self._f_save_var_trend(train_bins, x_columns_candidate, "train")
        metric_value_dict["变量趋势-训练集"] = MetricFucEntity(image_path=image_path_list, image_size=4)
        # 变量有效性
        train_woe = sc.woebin_ply(train_data[x_columns_candidate], train_bins, print_info=False)
        var_corr_image_path = self._f_get_var_corr_image(train_woe)
        # vif
        vif_df = f_get_ivf(train_woe)
        metric_value_dict["变量有效性"] = MetricFucEntity(image_path=var_corr_image_path, table=vif_df)

        if jupyter:
            from IPython import display

            display.display(metric_value_dict["样本分布"].table)
            # 打印变量iv
            display.display(metric_value_dict["变量iv"].table)
            # 打印vif
            display.display(metric_value_dict["变量有效性"].table)
            # 打印变量相关性
            f_display_images_by_side(metric_value_dict["变量有效性"].image_path, display, width=800)

            # 打印变量趋势
            var_trend_train = metric_value_dict["变量趋势-训练集"].image_path
            var_trend_test = None
            metric_test = metric_value_dict.get("变量趋势-测试集")
            if metric_test is not None:
                var_trend_test = metric_test.image_path
            f_display_images_by_side(var_trend_train, display, title="变量趋势训练集", image_path_list2=var_trend_test,
                                     title2="变量趋势测试集")
            # 打印breaks_list
            breaks_list = {}
            for x_column, feature in candidate_dict.items():
                breaks_list[x_column] = feature.breaks_list
            print("变量切分点:")
            print(json.dumps(breaks_list, ensure_ascii=False, indent=2, cls=NumpyEncoder))

            # 打印所有变量的推荐切分点
            print("-----不同分箱数下变量的推荐切分点-----")
            for x_column, features in numeric_candidate_dict_all.items():
                print(f"-----【{x_column}】-----")
                print(f"切分点:")
                var_trend_images_train = []
                var_trend_images_test = []
                for feature in features:
                    print(json.dumps(feature.breaks_list, ensure_ascii=False, cls=NumpyEncoder))
                    var_breaks_list = [str(i) for i in feature.breaks_list]
                    var_trend_bins_train = self._f_get_bins_by_breaks(train_data, {x_column: feature})
                    image_path = self._f_save_var_trend(var_trend_bins_train, [x_column],
                                                        f"train_{x_column}_{'_'.join(var_breaks_list)}")
                    var_trend_images_train.append(image_path[0])
                    if metric_test is not None:
                        var_trend_bins_test = self._f_get_bins_by_breaks(test_data, {x_column: feature})
                        image_path = self._f_save_var_trend(var_trend_bins_test, [x_column],
                                                            f"test_{x_column}_{'_'.join(var_breaks_list)}")
                        var_trend_images_test.append(image_path[0])

                f_display_images_by_side(var_trend_images_train, display, title=f"训练集",
                                         image_path_list2=var_trend_images_test,
                                         title2="测试集")

        return metric_value_dict