"""IV值筛选器.
使用信息价值（IV）进行特征筛选，是金融风控场景的核心筛选方法。
**参考样例**
>>> from hscredit.core.selectors import IVSelector
>>> import pandas as pd
>>> import numpy as np
>>> np.random.seed(42)
>>> X = pd.DataFrame(np.random.randn(1000, 5), columns=[f'f{i}' for i in range(5)]) # 5个特征
>>> y = pd.Series(np.random.randint(0, 2, 1000)) # 目标变量（0=好，1=坏）
>>> selector = IVSelector(threshold=0.02) # 筛选IV>0.02的特征
>>> selector.fit(X, y)
>>> print(selector.selected_features_)
"""
from typing import Union, List, Optional, Dict, Any
import numpy as np
import pandas as pd
from .base import BaseFeatureSelector
from ...utils.parallel import ParallelWorkload
def _compute_iv_single(x: np.ndarray, y: np.ndarray, regularization: float = 1.0) -> float:
"""计算单个特征的IV值。
:param x: 特征值数组
:param y: 目标变量数组
:param regularization: 正则化参数，避免除零
:return: IV值
"""
if regularization <= 0:
raise ValueError("regularization 必须大于 0")
x = np.asarray(x)
y = np.asarray(y)
if x.shape[0] != y.shape[0]:
raise ValueError("特征与目标变量长度不一致")
# 处理缺失值 - 兼容category和object类型
# 先转换为object类型，然后用isnull判断
if isinstance(x, pd.Series):
has_missing = x.isnull().values
else:
# 如果是numpy数组，尝试转换为Series以使用isnull
try:
has_missing = pd.Series(x).isnull().values
except Exception:
# 如果转换失败，使用pd.isnull直接判断
has_missing = pd.isnull(x)
valid = ~has_missing
x_valid = x[valid]
y_valid = y[valid]
if len(x_valid) == 0:
return 0.0
# 获取唯一值
uniques = np.unique(x_valid)
n_cats = len(uniques)
if n_cats <= 1:
return 0.0
labels = set(np.unique(y_valid).tolist())
if not labels.issubset({0, 1}):
raise ValueError("IV 计算要求目标变量只包含 0 和 1")
if labels != {0, 1}:
return 0.0
# 统计好坏样本
event_mask = y_valid == 1
nonevent_mask = y_valid == 0
event_tot = np.count_nonzero(event_mask) + n_cats * regularization
nonevent_tot = np.count_nonzero(nonevent_mask) + n_cats * regularization
event_rates = np.zeros(n_cats, dtype=np.float64)
nonevent_rates = np.zeros(n_cats, dtype=np.float64)
for i, cat in enumerate(uniques):
mask = x_valid == cat
event_rates[i] = np.count_nonzero(mask & event_mask) + regularization
nonevent_rates[i] = np.count_nonzero(mask & nonevent_mask) + regularization
event_rates /= event_tot
nonevent_rates /= nonevent_tot
# 计算IV
ivs = (event_rates - nonevent_rates) * np.log(np.maximum(event_rates, 1e-10) / np.maximum(nonevent_rates, 1e-10))
return np.sum(ivs).item()
def _compute_iv_feature(task):
"""编码并计算单个特征 IV。"""
feature, series, y, regularization = task
if series.dtype.name in ["object", "category"]:
values = pd.factorize(series)[0].astype(float)
values[pd.isna(series).to_numpy()] = np.nan
else:
values = series.values
return feature, _compute_iv_single(values, y, regularization)
[文档]
class IVSelector(BaseFeatureSelector):
"""IV值筛选器.
使用信息价值（Information Value）筛选特征。
IV是金融风控中衡量特征预测能力的核心指标。
IV值解释:
- < 0.02: 无预测能力
- 0.02 - 0.1: 弱预测能力
- 0.1 - 0.3: 中等预测能力
- 0.3 - 0.5: 强预测能力
- > 0.5: 极强预测能力（可能过拟合）
**支持的数据类型:**
- 数值型特征（int, float）
- 类别型特征（object, category）
**参数**
:param threshold: IV阈值，默认为0.02
- 0.02: 仅保留IV值大于0.02的特征
:param target: 目标变量列名，默认为'target'
:param regularization: 正则化参数，默认为1.0
:param n_jobs: 并行计算的任务数
**参考样例**
::
>>> from hscredit.core.selectors import IVSelector
>>> import pandas as pd
>>> import numpy as np
>>> np.random.seed(42)
>>> X = pd.DataFrame(np.random.randn(1000, 5), columns=[f'f{i}' for i in range(5)])
>>> y = pd.Series(np.random.randint(0, 2, 1000))
>>> selector = IVSelector(threshold=0.02)
>>> selector.fit(X, y)
>>> print(selector.selected_features_)
>>> print(selector.scores_) # 查看IV值
>>> # 仅传分箱参数：内部创建并训练 OptimalBinning
>>> selector = IVSelector(
... threshold=0.02,
... binning_params={'method': 'best_iv', 'max_n_bins': 5},
... )
>>> selector.fit(X, y)
>>> # 传入配置好的未训练实例：筛选器自动训练
>>> from hscredit.core.binning import OptimalBinning
>>> binner = OptimalBinning(method='best_iv', max_n_bins=5)
>>> selector = IVSelector(threshold=0.02, binner=binner).fit(X, y)
>>> # 传入已训练实例：直接复用规则，不重新训练
>>> trained_binner = OptimalBinning(method='best_iv').fit(X, y)
>>> selector = IVSelector(
... threshold=0.02,
... binner=trained_binner,
... binning_params={'method': 'uniform'}, # binner 优先，本参数被忽略
... ).fit(X, y)
**参数**
除继承自 :class:`~hscredit.core.selectors.base.BaseFeatureSelector` 的通用参数
（``target`` / ``include`` / ``exclude`` / ``force_drop`` / ``n_jobs``）外：
:param threshold: IV 保留阈值，``IV >= threshold`` 的特征被保留，默认为 ``0.02``
:param regularization: 计算 WOE/IV 时的加性平滑系数，避免某类别好/坏样本数为 0
导致取对数发散，默认为 ``1.0``
:param binner: 可选的已配置分箱器实例。未训练实例会自动训练，已训练实例直接复用
:param binning_params: 可选的 ``OptimalBinning`` 构造参数字典。未传 ``binner`` 时，
内部创建分箱器并将原始数据转换为分箱 index 后计算 IV
.. note::
本筛选器按特征的**唯一取值**直接计算 IV（类别型先 ``factorize``），适合已分箱
或基数较低的特征；连续特征建议先用 :class:`~hscredit.core.binning.OptimalBinning`
分箱后再筛选，或传入 ``binner`` 参数。
**引用**
Information Value 用于变量筛选见 Siddiqi, N. (2006). *Credit Risk Scorecards.*
Wiley；阈值经验区间（0.02/0.1/0.3/0.5）为业界通行标准。
"""
method_name = "IV值筛选"
def __init__(
self,
threshold: float = 0.02,
target: str = "target",
regularization: float = 1.0,
include: Optional[List[str]] = None,
exclude: Optional[List[str]] = None,
force_drop: Optional[List[str]] = None,
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,
threshold=threshold,
include=include,
exclude=exclude,
force_drop=force_drop,
n_jobs=n_jobs,
binner=binner,
binning_params=binning_params,
parallel_backend=parallel_backend,
parallel_config=parallel_config,
)
self.regularization = regularization
def _fit_impl(
self,
X: pd.DataFrame,
y: Optional[Union[pd.Series, np.ndarray]],
) -> None:
"""拟合IV值筛选器。
:param X: 输入特征DataFrame
:param y: 目标变量
"""
self._get_feature_names(X)
if y is None:
raise ValueError("IVSelector 需要目标变量 y")
y = np.asarray(y)
results = self._parallel_execute(
_compute_iv_feature,
((col, X[col], y, self.regularization) for col in X.columns),
task_labels=X.columns,
default_backend="threading",
workload=ParallelWorkload(
task_count=X.shape[1],
rows=X.shape[0],
columns=X.shape[1],
data_bytes=int(X.memory_usage(deep=True).sum()),
cost_per_item=5.0,
capability="thread_safe",
releases_gil=True,
operation="IV字段计算",
),
)
iv_values = np.array([score for _, score in results])
self.scores_ = pd.Series(iv_values, index=X.columns)
# 选择IV值大于等于阈值的特征
selected_mask = iv_values >= self.threshold
self.selected_features_ = X.columns[selected_mask].tolist()
# 构建详细的dropped_记录，包含IV值
dropped_cols = X.columns[~selected_mask].tolist()
if len(dropped_cols) > 0:
self.dropped_ = pd.DataFrame(
{
"特征": dropped_cols,
"剔除原因": [f"IV值({self.scores_[col]:.4f}) <= 阈值({self.threshold})" for col in dropped_cols],
"IV值": [self.scores_[col] for col in dropped_cols],
"阈值": [self.threshold] * len(dropped_cols),
}
)
else:
self.dropped_ = pd.DataFrame(columns=["特征", "剔除原因", "IV值", "阈值"])
[文档]
def get_iv_interpretation(self) -> pd.DataFrame:
"""获取IV值的中文解释。
:return: 包含IV值及解释的DataFrame
"""
if not hasattr(self, "scores_"):
return pd.DataFrame()
def interpret_iv(iv):
if iv < 0.02:
return "无预测能力"
elif iv < 0.1:
return "弱预测能力"
elif iv < 0.3:
return "中等预测能力"
elif iv < 0.5:
return "强预测能力"
else:
return "极强预测能力（可能过拟合）"
df = pd.DataFrame({"特征": self.scores_.index, "IV值": self.scores_.values, "预测能力": [interpret_iv(iv) for iv in self.scores_.values]})
return df.sort_values("IV值", ascending=False)