# -*- coding: utf-8 -*-
"""
@author: yq
@time: 2024/11/1
@desc: 数据处理配置类
"""
import json
import os
from typing import List, Union
from commom import GeneralException, f_get_datetime
from config import BaseConfig
from enums import ResultCodesEnum
from init import warning_ignore
class MlConfigEntity():
def __init__(self,
y_column: str,
project_name: str = None,
x_columns: List[str] = [],
columns_exclude: List[str] = [],
columns_include: List[str] = [],
columns_anns: dict = {},
bin_search_interval: float = 0.05,
bin_sample_rate: float = 0.1,
iv_threshold: float = 0.01,
corr_threshold: float = 0.4,
psi_threshold: float = 0.2,
vif_threshold: float = 10,
monto_shift_threshold=1,
trend_shift_threshold=0,
max_feature_num: int = 10,
special_values: Union[dict, list, str] = None,
breaks_list: dict = None,
format_bin: str = False,
jupyter_print=False,
bin_detail_print=True,
stress_test=False,
stress_sample_times=100,
stress_bad_rate_list: List[float] = [],
model_type="lr",
feature_strategy="woe",
rules=[],
fill_method: str = None,
fill_value=None,
*args, **kwargs):
self._model_type = model_type
self._feature_strategy = feature_strategy
self._psi_threshold = psi_threshold
self._vif_threshold = vif_threshold
# 排除的x列
self._columns_exclude = columns_exclude
# 强制保留的x列
self._columns_include = columns_include
# 变量注释
self._columns_anns = columns_anns
# 是否开启下输出内容
self._stress_test = stress_test
# jupyter下输出内容
self._stress_sample_times = stress_sample_times
# jupyter下输出内容
self._stress_bad_rate_list = stress_bad_rate_list
# jupyter下输出内容
self._jupyter_print = jupyter_print
# jupyter下输出内容
self._bin_detail_print = bin_detail_print
# 单调性允许变化次数
self._monto_shift_threshold = monto_shift_threshold
# 变量趋势一致性允许变化次数
self._trend_shift_threshold = trend_shift_threshold
# 是否启用粗分箱
self._format_bin = format_bin
# 项目名称，和缓存路径有关
self._project_name = project_name
# 定义y变量
self._y_column = y_column
# 候选x变量
self._x_columns = x_columns
# 缺失值填充方法
self._fill_method = fill_method
# 缺失值填充值
self._fill_value = fill_value
# 使用iv筛变量时的阈值
self._iv_threshold = iv_threshold
# 贪婪搜索分箱时数据粒度大小，应该在0.01-0.1之间
self._bin_search_interval = bin_search_interval
# 最终保留多少x变量
self._max_feature_num = max_feature_num
self._special_values = special_values
self._breaks_list = breaks_list
# 变量相关性阈值
self._corr_threshold = corr_threshold
# 贪婪搜索采样比例,只针对4箱5箱时有效
self._bin_sample_rate = bin_sample_rate
# 加减分规则
self._rules = rules
if self._project_name is None or len(self._project_name) == 0:
self._base_dir = os.path.join(BaseConfig.train_path, f"{f_get_datetime()}")
else:
self._base_dir = os.path.join(BaseConfig.train_path, self._project_name)
self._include = columns_include + list(self.breaks_list.keys())
os.makedirs(self._base_dir, exist_ok=True)
print(f"项目路径:【{self._base_dir}】")
if self._jupyter_print:
warning_ignore()
@property
def model_type(self):
return self._model_type
@property
def feature_strategy(self):
return self._feature_strategy
@property
def psi_threshold(self):
return self._psi_threshold
@property
def vif_threshold(self):
return self._vif_threshold
@property
def stress_test(self):
return self._stress_test
@property
def stress_sample_times(self):
return self._stress_sample_times
@property
def stress_bad_rate_list(self):
return self._stress_bad_rate_list
@property
def jupyter_print(self):
return self._jupyter_print
@property
def bin_detail_print(self):
return self._bin_detail_print
@property
def base_dir(self):
return self._base_dir
@property
def monto_shift_threshold(self):
return self._monto_shift_threshold
@property
def trend_shift_threshold(self):
return self._trend_shift_threshold
@property
def format_bin(self):
return self._format_bin
@property
def project_name(self):
return self._project_name
@property
def bin_sample_rate(self):
return self._bin_sample_rate
@property
def rules(self):
return self._rules
@property
def corr_threshold(self):
return self._corr_threshold
@property
def max_feature_num(self):
return self._max_feature_num
@property
def y_column(self):
return self._y_column
@property
def x_columns(self):
return self._x_columns
@property
def columns_exclude(self):
return self._columns_exclude
@property
def columns_include(self):
return self._columns_include
@property
def columns_anns(self):
return self._columns_anns
@property
def fill_value(self):
return self._fill_value
@property
def fill_method(self):
return self._fill_method
@property
def iv_threshold(self):
return self._iv_threshold
@property
def bin_search_interval(self):
return self._bin_search_interval
@property
def special_values(self):
if self._special_values is None or len(self._special_values) == 0:
return None
if isinstance(self._special_values, str):
return [self._special_values]
if isinstance(self._special_values, (dict, list)):
return self._special_values
return None
def get_special_values(self, column: str = None):
if self._special_values is None or len(self._special_values) == 0:
return []
if isinstance(self._special_values, str):
return [self._special_values]
if isinstance(self._special_values, list):
return self._special_values
if isinstance(self._special_values, dict) and column is not None:
return self._special_values.get(column, [])
return []
@property
def breaks_list(self):
if self._breaks_list is None:
return {}
if isinstance(self._breaks_list, dict):
return self._breaks_list
return {}
def get_breaks_list(self, column: str = None):
if self._breaks_list is None or len(self._breaks_list) == 0:
return []
if isinstance(self._breaks_list, dict) and column is not None:
return self._breaks_list.get(column, [])
return []
def is_include(self, column: str) -> bool:
return column in self._include
def f_get_save_path(self, file_name: str) -> str:
path = os.path.join(self._base_dir, file_name)
return path
@staticmethod
def from_config(config_path: str):
"""
从配置文件生成实体类
"""
if os.path.isdir(config_path):
config_path = os.path.join(config_path, "mlcfg.json")
if os.path.exists(config_path):
with open(config_path, mode="r", encoding="utf-8") as f:
j = json.loads(f.read())
else:
raise GeneralException(ResultCodesEnum.NOT_FOUND, message=f"指配置文件【{config_path}】不存在")
print(f"mlcfg load from【{config_path}】success. ")
return MlConfigEntity(**j)
def config_save(self):
path = self.f_get_save_path("mlcfg.json")
with open(path, mode="w", encoding="utf-8") as f:
j = {k.lstrip("_"): v for k, v in self.__dict__.items()}
j = json.dumps(j, ensure_ascii=False)
f.write(j)
print(f"mlcfg save to【{path}】success. ")
if __name__ == "__main__":
pass