# -*- coding: utf-8 -*-
"""
@author: yq
@time: 2025/4/3
@desc: 值标准化，类似于分箱
"""
import os
from typing import Dict, List
import joblib
import pandas as pd
import xgboost as xgb
from pandas.core.dtypes.common import is_numeric_dtype
from commom import GeneralException, f_display_title
from data import DataExplore
from entitys import DataSplitEntity, MetricFucResultEntity
from enums import ResultCodesEnum, ContextEnum, FileEnum
from feature.feature_strategy_base import FeatureStrategyBase
from init import context
from .utils import f_format_value, OneHot, f_format_bin
class StrategyNorm(FeatureStrategyBase):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.x_columns = None
self.one_hot_encoder_dict: Dict[str, OneHot] = {}
self.points_dict: Dict[str, List[float]] = {}
def _f_fast_filter(self, data: DataSplitEntity) -> List[str]:
y_column = self.ml_config.y_column
x_columns = self.ml_config.x_columns
columns_exclude = self.ml_config.columns_exclude
format_bin = self.ml_config.format_bin
params_xgb = self.ml_config.params_xgb
max_feature_num = self.ml_config.max_feature_num
columns_anns = self.ml_config.columns_anns
train_data = data.train_data.copy()
test_data = data.test_data.copy()
# 特征列配置
if len(x_columns) == 0:
x_columns = train_data.columns.tolist()
if y_column in x_columns:
x_columns.remove(y_column)
for column in columns_exclude:
if column in x_columns:
x_columns.remove(column)
# 简单校验数据类型一致性
check_msg = DataExplore.check_type(data.data[x_columns])
if check_msg != "":
print(f"数据类型分析:\n{check_msg}\n同一变量请保持数据类型一致")
raise GeneralException(ResultCodesEnum.ILLEGAL_PARAMS, message=f"数据类型错误.")
# 数据处理
model_columns = []
num_columns = []
str_columns = []
for x_column in x_columns:
if is_numeric_dtype(train_data[x_column]):
num_columns.append(x_column)
# 粗分箱
if format_bin:
data_x_describe = train_data[x_column].describe(percentiles=[0.1, 0.9])
points = f_format_bin(data_x_describe)
if points is not None:
self.points_dict[x_column] = points
train_data[x_column] = train_data[x_column].apply(lambda x: f_format_value(points, x))
test_data[x_column] = test_data[x_column].apply(lambda x: f_format_value(points, x))
else:
str_columns.append(x_column)
one_hot_encoder = OneHot()
one_hot_encoder.fit(data.data, x_column)
one_hot_encoder.encoder(train_data)
one_hot_encoder.encoder(test_data)
model_columns.extend(one_hot_encoder.columns_onehot)
self.one_hot_encoder_dict[x_column] = one_hot_encoder
model_columns.extend(num_columns)
# 重要性剔除弱变量
model = xgb.XGBClassifier(objective=params_xgb.get("objective"),
n_estimators=params_xgb.get("num_boost_round"),
max_depth=params_xgb.get("max_depth"),
learning_rate=params_xgb.get("learning_rate"),
random_state=params_xgb.get("random_state"),
reg_alpha=params_xgb.get("alpha"),
subsample=params_xgb.get("subsample"),
colsample_bytree=params_xgb.get("colsample_bytree"),
importance_type='weight'
)
model.fit(X=train_data[model_columns], y=train_data[y_column],
eval_set=[(train_data[model_columns], train_data[y_column]),
(test_data[model_columns], test_data[y_column])],
eval_metric=params_xgb.get("eval_metric"),
early_stopping_rounds=params_xgb.get("early_stopping_rounds"),
verbose=False,
)
# 重要合并，字符型变量重要性为各one-hot子变量求和
importance = model.feature_importances_
feature = []
importance_weight = []
for x_column in num_columns:
for i, j in zip(model_columns, importance):
if i == x_column:
feature.append(x_column)
importance_weight.append(j)
break
for x_column in str_columns:
feature_cache = 0
for i, j in zip(model_columns, importance):
if i.startswith(f"{x_column}("):
feature_cache += j
feature.append(x_column)
importance_weight.append(feature_cache)
anns = [columns_anns.get(column, "-") for column in feature]
df_importance = pd.DataFrame({'feature': feature, f'importance_weight': importance_weight, "释义": anns})
df_importance.sort_values(by=["importance_weight"], ascending=[False], inplace=True)
df_importance.reset_index(drop=True, inplace=True)
df_importance_rank = df_importance[df_importance["importance_weight"] > 0]
df_importance_rank.reset_index(drop=True, inplace=True)
x_columns_filter = list(df_importance_rank["feature"])[0:max_feature_num]
context.set_filter_info(ContextEnum.FILTER_FAST,
f"筛选前变量数量:{len(x_columns)}\n{x_columns}\n"
f"快速筛选剔除变量数量:{len(x_columns) - len(x_columns_filter)}", detail=df_importance)
context.set(ContextEnum.XGB_COLUMNS_NUM, num_columns)
context.set(ContextEnum.XGB_POINTS, self.points_dict)
return x_columns_filter
def feature_search(self, data: DataSplitEntity, *args, **kwargs):
x_columns = self._f_fast_filter(data)
# 排个序，防止因为顺序原因导致的可能的bug
x_columns.sort()
self.x_columns = x_columns
context.set(ContextEnum.XGB_COLUMNS_SELECTED, x_columns)
def variable_analyse(self, *args, **kwargs):
pass
def feature_generate(self, data: pd.DataFrame, *args, **kwargs) -> pd.DataFrame:
df = data.copy()
model_columns = []
for x_column in self.x_columns:
if x_column in self.points_dict.keys():
points = self.points_dict[x_column]
df[x_column] = df[x_column].apply(lambda x: f_format_value(points, x))
model_columns.append(x_column)
elif x_column in self.one_hot_encoder_dict.keys():
one_hot_encoder = self.one_hot_encoder_dict[x_column]
one_hot_encoder.encoder(df)
model_columns.extend(one_hot_encoder.columns_onehot)
else:
model_columns.append(x_column)
return df[model_columns]
def feature_save(self, *args, **kwargs):
if self.x_columns is None:
GeneralException(ResultCodesEnum.NOT_FOUND, message=f"feature不存在")
path = self.ml_config.f_get_save_path(FileEnum.FEATURE_PKL.value)
feature_info = {
"x_columns": self.x_columns,
"one_hot_encoder_dict": self.one_hot_encoder_dict,
"points_dict": self.points_dict,
}
joblib.dump(feature_info, path)
print(f"feature save to【{path}】success. ")
def feature_load(self, path: str, *args, **kwargs):
if os.path.isdir(path):
path = os.path.join(path, FileEnum.FEATURE_PKL.value)
if not os.path.isfile(path) or FileEnum.FEATURE_PKL.value not in path:
raise GeneralException(ResultCodesEnum.NOT_FOUND, message=f"特征信息【{FileEnum.FEATURE_PKL.value}】不存在")
feature_info = joblib.load(path)
self.x_columns = feature_info["x_columns"]
self.one_hot_encoder_dict = feature_info["one_hot_encoder_dict"]
self.points_dict = feature_info["points_dict"]
print(f"feature load from【{path}】success.")
def feature_report(self, data: DataSplitEntity, *args, **kwargs) -> Dict[str, MetricFucResultEntity]:
y_column = self.ml_config.y_column
metric_value_dict = {}
# 样本分布
metric_value_dict["样本分布"] = MetricFucResultEntity(table=data.get_distribution(y_column), table_font_size=10,
table_cell_width=3)
self.jupyter_print(metric_value_dict)
return metric_value_dict
def jupyter_print(self, metric_value_dict, *args, **kwargs):
from IPython import display
max_feature_num = self.ml_config.max_feature_num
filter_fast = context.get(ContextEnum.FILTER_FAST)
f_display_title(display, "样本分布")
display.display(metric_value_dict["样本分布"].table)
df_importance = filter_fast["detail"]
df_importance = df_importance[df_importance["feature"].isin(self.x_columns)]
f_display_title(display, "入模变量")
display.display(df_importance)
f_display_title(display, "快速筛选过程")
print(f"剔除变量重要性排名{max_feature_num}以后的变量")
print(filter_fast.get("overview"))
display.display(filter_fast["detail"])