"""评分卡风格特征粗筛器.
参考 scorecardpipeline.FeatureSelection 的常用粗筛流程，
按缺失率、IV、相关性、单一值占比顺序筛选特征，
但完整复用 hscredit 现有筛选器体系与双风格 API。
**参考样例**
>>> from hscredit.core.selectors import ScorecardFeatureSelection
>>> import pandas as pd
>>> import numpy as np
>>> np.random.seed(42)
>>> X = pd.DataFrame(np.random.randn(1000, 10), columns=[f'f{i}' for i in range(10)]) # 10个特征
>>> y = pd.Series(np.random.randint(0, 2, 1000)) # 目标变量
>>> selector = ScorecardFeatureSelection(
... null_threshold=0.95, # 移除缺失率>95%的特征
... iv_threshold=0.02, # 移除IV<0.02的特征
... corr_threshold=0.7, # 移除相关性>0.7的特征
... mode_threshold=0.95, # 移除众数占比>95%的特征
... )
>>> selector.fit(X, y)
>>> print(selector.selected_features_)
"""
from typing import Union, List, Optional, Dict, Any
import copy
import numpy as np
import pandas as pd
from .base import BaseFeatureSelector
from .null_selector import NullSelector
from .iv_selector import IVSelector
from .corr_selector import CorrSelector
from .mode_selector import ModeSelector
[文档]
class ScorecardFeatureSelection(BaseFeatureSelector):
"""评分卡风格特征粗筛器.
按如下顺序执行特征筛选：
1. 缺失率筛选（NullSelector）
2. IV值筛选（IVSelector）
3. 相关性筛选（CorrSelector）
4. 单一值筛选（ModeSelector）
与 scorecardpipeline.FeatureSelection 的思路一致，
但参数命名和基础行为遵循 hscredit 风格：
- 缺失率阈值使用 ``null_threshold``
- IV阈值使用 ``iv_threshold``
- 相关性阈值使用 ``corr_threshold``
- 单一值占比阈值使用 ``mode_threshold``
- 强制保留使用 ``include``
- 强制剔除使用 ``exclude`` / ``force_drop``
- 支持 sklearn 风格 ``fit(X, y)``
- 支持 scorecardpipeline 风格 ``fit(df)``
**参数**
:param null_threshold: 缺失率阈值，默认为0.95；设为None可关闭该阶段
:param iv_threshold: IV阈值，默认为0.02；设为None可关闭该阶段
:param corr_threshold: 相关性阈值，默认为0.7；设为None可关闭该阶段
:param mode_threshold: 单一值占比阈值，默认为0.95；设为None可关闭该阶段
:param corr_method: 相关系数计算方法，默认为'pearson'
- 'pearson': 皮尔逊相关系数
- 'spearman': 斯皮尔曼等级相关系数
- 'kendall': 肯德尔相关系数
:param corr_metric: 相关性筛选的保留指标，默认为'iv'
- 'iv': 信息值（需要目标变量y）
- 'ks': KS统计量
- 'lift': LIFT值
- 'bad_rate': 坏样本率
指标通过分箱后的bin_tables_计算得到。
:param corr_weights: 自定义相关性筛选权重，优先级高于corr_metric
:param corr_binning_params: 透传给CorrSelector的分箱参数
:param iv_regularization: IV计算正则项，默认为1.0
:param mode_dropna: 计算单一值占比时是否排除缺失值，默认为True
:param target: 目标变量列名，默认为'target'
:param include: 强制保留特征列表
:param exclude: 强制剔除特征列表
:param force_drop: 强制剔除特征列表，最终会合并到exclude
:param target_rm: 默认False保留已有目标列；仅显式True时移除目标
:param n_jobs: 并行任务数
**参考样例**
>>> from hscredit.core.selectors import ScorecardFeatureSelection
>>> import pandas as pd
>>> import numpy as np
>>> np.random.seed(42)
>>> X = pd.DataFrame(np.random.randn(1000, 10), columns=[f'f{i}' for i in range(10)])
>>> y = pd.Series(np.random.randint(0, 2, 1000))
>>> selector = ScorecardFeatureSelection(
... null_threshold=0.95,
... iv_threshold=0.02,
... corr_threshold=0.7,
... mode_threshold=0.95,
... )
>>> selector.fit(X, y)
>>> print(selector.selected_features_)
>>> print(selector.stage_report_df_)
"""
method_name = "评分卡特征粗筛"
def __init__(
self,
null_threshold: Optional[float] = 0.95,
iv_threshold: Optional[float] = 0.02,
corr_threshold: Optional[float] = 0.7,
mode_threshold: Optional[float] = 0.95,
corr_method: str = "pearson",
corr_metric: str = "iv",
corr_weights: Optional[Union[pd.Series, Dict[str, float], List[float]]] = None,
corr_binning_params: Optional[Dict[str, Any]] = None,
iv_regularization: float = 1.0,
mode_dropna: bool = True,
target: str = "target",
include: Optional[List[str]] = None,
exclude: Optional[List[str]] = None,
force_drop: Optional[List[str]] = None,
target_rm: bool = False,
n_jobs: Optional[Union[int, float]] = -1,
binner: Optional[Any] = None,
binning_params: Optional[Dict[str, Any]] = None,
parallel_backend: Optional[str] = None,
parallel_config: Optional[Dict[str, Any]] = None,
):
super().__init__(
target=target,
include=include,
exclude=exclude,
force_drop=force_drop,
threshold="multi-stage",
n_jobs=n_jobs,
binner=binner,
binning_params=binning_params,
parallel_backend=parallel_backend,
parallel_config=parallel_config,
target_rm=target_rm,
)
self.null_threshold = null_threshold
self.iv_threshold = iv_threshold
self.corr_threshold = corr_threshold
self.mode_threshold = mode_threshold
self.corr_method = corr_method
self.corr_metric = corr_metric
self.corr_weights = corr_weights
self.corr_binning_params = corr_binning_params
self.iv_regularization = iv_regularization
self.mode_dropna = mode_dropna
self.target_rm = target_rm
def _check_input(self, X, y=None):
X, y = super()._check_input(X, y)
self._validate_configuration(y, check_target=False)
if not isinstance(self.target_rm, (bool, np.bool_)):
raise ValueError("target_rm 必须为布尔值")
self.planned_stages_ = [
{"key": key, "name": name, "selector": selector, "enabled": self._is_stage_enabled(threshold)}
for key, name, selector, threshold in [
("empty", "缺失率筛选", "NullSelector", self.null_threshold),
("iv", "IV值筛选", "IVSelector", self.iv_threshold),
("corr", "相关性筛选", "CorrSelector", self.corr_threshold),
("identical", "单一值筛选", "ModeSelector", self.mode_threshold),
]
]
self.executed_stages_ = []
self.skipped_stages_ = {
i: "配置关闭该阶段" for i, stage in enumerate(self.planned_stages_) if not stage["enabled"]
}
return X, y
def _initialize_empty_selection_result(self):
super()._initialize_empty_selection_result()
self.stage_selectors_ = {}
self.stage_reports_ = []
self.stage_report_df_ = pd.DataFrame()
self.skipped_stages_.update(
{i: "父级强制操作后无待筛选特征" for i, stage in enumerate(self.planned_stages_) if stage["enabled"]}
)
[文档]
def fit(
self,
X: Union[pd.DataFrame, np.ndarray],
y: Optional[Union[pd.Series, np.ndarray]] = None,
) -> "ScorecardFeatureSelection":
"""拟合评分卡风格筛选器并同步兼容属性。"""
return super().fit(X, y)
def _finalize_fit(self) -> None:
"""在隔离候选对象上整理最终选择结果。"""
self._finalize_selection_result()
def _fit_impl(
self,
X: pd.DataFrame,
y: Optional[Union[pd.Series, np.ndarray]],
) -> None:
"""按评分卡粗筛顺序执行筛选。"""
self._validate_configuration(y)
self._get_feature_names(X)
# 所有阶段只读输入；首阶段不复制完整高维 DataFrame。
current_X = X
self.stage_selectors_ = {}
self.stage_reports_ = []
self.stage_report_df_ = pd.DataFrame()
self.corr_iv_scores_ = None
all_dropped = []
if self._is_stage_enabled(self.null_threshold) and len(current_X.columns) > 0:
current_X = self._run_stage(
stage_key="empty",
stage_name="缺失率筛选",
selector=NullSelector(
threshold=self.null_threshold,
target=self.target,
n_jobs=self.n_jobs,
parallel_backend=self.parallel_backend,
parallel_config=self.parallel_config,
),
current_X=current_X,
y=y,
all_dropped=all_dropped,
)
iv_scores = None
if self._is_stage_enabled(self.iv_threshold) and len(current_X.columns) > 0:
iv_selector = IVSelector(
threshold=self.iv_threshold,
target=self.target,
regularization=self.iv_regularization,
n_jobs=self.n_jobs,
parallel_backend=self.parallel_backend,
parallel_config=self.parallel_config,
)
current_X = self._run_stage(
stage_key="iv",
stage_name="IV值筛选",
selector=iv_selector,
current_X=current_X,
y=y,
all_dropped=all_dropped,
)
iv_scores = getattr(iv_selector, "scores_", None)
if self._is_stage_enabled(self.corr_threshold) and len(current_X.columns) > 0:
corr_weights = self._resolve_corr_weights(current_X, y, iv_scores)
corr_binning_kwargs = {}
if self.corr_binning_params is not None:
corr_binning_kwargs["binning_params"] = self.corr_binning_params
elif getattr(self, "_binner_instance", None) is not None or corr_weights is not None:
# 外层已分箱或已有明确权重时，关闭内部 CorrSelector 的构造默认分箱。
corr_binning_kwargs["binning_params"] = None
corr_selector = CorrSelector(
threshold=self.corr_threshold,
method=self.corr_method,
metric=self.corr_metric,
weights=corr_weights,
target=self.target,
n_jobs=self.n_jobs,
parallel_backend=self.parallel_backend,
parallel_config=self.parallel_config,
**corr_binning_kwargs,
)
current_X = self._run_stage(
stage_key="corr",
stage_name="相关性筛选",
selector=corr_selector,
current_X=current_X,
y=y,
all_dropped=all_dropped,
)
if self._is_stage_enabled(self.mode_threshold) and len(current_X.columns) > 0:
current_X = self._run_stage(
stage_key="identical",
stage_name="单一值筛选",
selector=ModeSelector(
threshold=self.mode_threshold,
dropna=self.mode_dropna,
target=self.target,
n_jobs=self.n_jobs,
parallel_backend=self.parallel_backend,
parallel_config=self.parallel_config,
),
current_X=current_X,
y=y,
all_dropped=all_dropped,
)
self.selected_features_ = current_X.columns.tolist()
self.scores_ = None
if len(all_dropped) > 0:
self.dropped_ = pd.concat(all_dropped, ignore_index=True)
else:
self.dropped_ = pd.DataFrame(columns=["特征", "剔除原因", "筛选阶段", "筛选阶段名称", "筛选器"])
self.removed_features_ = self.dropped_["特征"].tolist() if len(self.dropped_) > 0 else []
if self.stage_reports_:
self.stage_report_df_ = pd.DataFrame(self.stage_reports_)
self.skipped_stages_.update(
{
i: "上游无剩余特征"
for i, stage in enumerate(self.planned_stages_)
if stage["enabled"] and i not in self.executed_stages_
}
)
def _run_stage(
self,
stage_key: str,
stage_name: str,
selector: BaseFeatureSelector,
current_X: pd.DataFrame,
y: Optional[Union[pd.Series, np.ndarray]],
all_dropped: List[pd.DataFrame],
) -> pd.DataFrame:
"""执行单个筛选阶段并记录结果。"""
input_count = current_X.shape[1]
selector.fit(current_X, y)
self.executed_stages_.append(
next(i for i, stage in enumerate(self.planned_stages_) if stage["key"] == stage_key)
)
self.stage_selectors_[stage_key] = selector
self.stage_reports_.append(
{
"阶段键": stage_key,
"阶段名称": stage_name,
"筛选器": selector.__class__.__name__,
"输入特征数": input_count,
"选中特征数": len(selector.selected_features_),
"剔除特征数": len(getattr(selector, "removed_features_", [])),
"阈值": getattr(selector, "threshold", None),
}
)
if hasattr(selector, "dropped_") and selector.dropped_ is not None and len(selector.dropped_) > 0:
dropped = selector.dropped_.copy(deep=False)
dropped["筛选阶段"] = stage_key
dropped["筛选阶段名称"] = stage_name
dropped["筛选器"] = selector.__class__.__name__
all_dropped.append(dropped)
selected = selector.selected_features_
if len(selected) == 0:
return current_X.iloc[:, 0:0]
if selected == list(current_X.columns):
return current_X
return current_X.loc[:, selected]
def _resolve_corr_weights(
self,
X: pd.DataFrame,
y: Optional[Union[pd.Series, np.ndarray]],
iv_scores: Optional[pd.Series],
) -> Optional[Union[pd.Series, Dict[str, float], List[float]]]:
"""为相关性筛选解析保留权重。"""
if self.corr_weights is not None:
return self.corr_weights
if iv_scores is not None and str(self.corr_metric).lower() == "iv":
return iv_scores.reindex(X.columns).fillna(0.0)
if str(self.corr_metric).lower() == "iv":
if y is None:
raise ValueError("启用 corr_threshold 且使用 IV 作为保留指标时，需要传入 y 或 target 列")
iv_selector = IVSelector(
threshold=float("-inf"),
target=self.target,
regularization=self.iv_regularization,
n_jobs=self.n_jobs,
parallel_backend=self.parallel_backend,
parallel_config=self.parallel_config,
)
iv_selector.fit(X, y)
self.corr_iv_scores_ = iv_selector.scores_.reindex(X.columns).fillna(0.0)
return self.corr_iv_scores_
return None
def _validate_configuration(
self,
y: Optional[Union[pd.Series, np.ndarray]],
check_target: bool = True,
) -> None:
"""校验配置合法性。"""
self._validate_ratio_threshold(self.null_threshold, "null_threshold")
self._validate_ratio_threshold(self.corr_threshold, "corr_threshold")
self._validate_ratio_threshold(self.mode_threshold, "mode_threshold")
if self._is_stage_enabled(self.iv_threshold) and (
not isinstance(self.iv_threshold, (int, float, np.number))
or not np.isfinite(self.iv_threshold)
or self.iv_threshold < 0
):
raise ValueError("iv_threshold 必须是非负有限数值")
if (
not isinstance(self.iv_regularization, (int, float, np.number))
or not np.isfinite(self.iv_regularization)
or self.iv_regularization <= 0
):
raise ValueError("iv_regularization 必须是正有限数值")
requires_target = self._is_stage_enabled(self.iv_threshold)
requires_target = requires_target or (
self._is_stage_enabled(self.corr_threshold)
and self.corr_weights is None
and str(self.corr_metric).lower() == "iv"
)
requires_target = requires_target or (
self._is_stage_enabled(self.corr_threshold)
and self.corr_weights is None
and str(self.corr_metric).lower() != "iv"
)
if check_target and requires_target and y is None:
raise ValueError("当前筛选配置需要目标变量，请使用 fit(X, y) 或 fit(df) 且 df 包含 target 列")
@staticmethod
def _is_stage_enabled(threshold: Optional[float]) -> bool:
"""判断筛选阶段是否启用。"""
return threshold is not None and threshold is not False
@staticmethod
def _validate_ratio_threshold(threshold: Optional[float], name: str) -> None:
"""校验 0-1 比例阈值。"""
if threshold is None or threshold is False:
return
if not isinstance(threshold, (int, float, np.number)) or not np.isfinite(threshold) or not 0 <= threshold <= 1:
raise ValueError(f"{name} 必须在 [0, 1] 范围内")
def _finalize_selection_result(self) -> None:
"""整理最终选择结果并补齐兼容属性。"""
selected_set = set(self.selected_features_)
self.selected_features_ = [c for c in self._feature_names if c in selected_set]
self.n_features_ = len(self.selected_features_)
if hasattr(self, "dropped_") and self.dropped_ is not None and len(self.dropped_) > 0:
dropped_df = self.dropped_.copy()
if "筛选阶段" not in dropped_df.columns:
dropped_df["筛选阶段"] = None
if "筛选阶段名称" not in dropped_df.columns:
dropped_df["筛选阶段名称"] = None
if "筛选器" not in dropped_df.columns:
dropped_df["筛选器"] = None
force_drop_mask = dropped_df["剔除原因"].astype(str).str.contains("强制剔除", na=False)
dropped_df.loc[force_drop_mask, "筛选阶段"] = "force_drop"
dropped_df.loc[force_drop_mask, "筛选阶段名称"] = "强制剔除"
dropped_df = dropped_df.loc[~dropped_df["特征"].isin(self.selected_features_)].copy()
dropped_df = dropped_df.drop_duplicates(subset=["特征"], keep="first").reset_index(drop=True)
self.dropped_ = dropped_df
self.removed_features_ = dropped_df["特征"].tolist()
else:
self.dropped_ = pd.DataFrame(columns=["特征", "剔除原因", "筛选阶段", "筛选阶段名称", "筛选器"])
self.removed_features_ = []
self.select_columns = list(self.selected_features_)
if (
not self.target_rm
and getattr(self, "target_present_at_fit_", False)
and self.target not in self.select_columns
):
self.select_columns.append(self.target)
self.dropped = pd.DataFrame(
{
"variable": self.dropped_["特征"] if len(self.dropped_) > 0 else pd.Series(dtype=object),
"rm_reason": self.dropped_["筛选阶段"] if len(self.dropped_) > 0 else pd.Series(dtype=object),
}
)
[文档]
def get_selection_report(self) -> Dict[str, Any]:
"""获取包含阶段明细的筛选报告。"""
if hasattr(self, "_selection_report_dict_"):
return copy.deepcopy(self._selection_report_dict_)
report = super().get_selection_report()
if report.get("状态") == "未拟合":
return report
report["阈值"] = {
"null_threshold": self.null_threshold,
"iv_threshold": self.iv_threshold,
"corr_threshold": self.corr_threshold,
"mode_threshold": self.mode_threshold,
}
if hasattr(self, "stage_reports_"):
report["阶段明细"] = copy.deepcopy(self.stage_reports_)
return report