"""LIFT筛选器.
使用LIFT@ratio值进行特征筛选，支持自定义覆盖率和方向。
**参考样例**
>>> from hscredit.core.selectors import LiftSelector
>>> 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)) # 目标变量
>>> selector = LiftSelector(threshold=0.5, ratio=0.10) # 筛选LIFT>0.5且覆盖率10%的特征
>>> selector.fit(X, y)
>>> print(selector.selected_features_)
"""
from typing import Union, List, Optional, Literal, Tuple, Dict, Any
import numpy as np
import pandas as pd
from .base import BaseFeatureSelector
from ...utils.parallel import ParallelWorkload
def _compute_lift_single(
x: np.ndarray,
y: np.ndarray,
ratio: float = 0.10,
ascending: bool = False,
) -> float:
"""计算单个特征在指定排序方向下的LIFT@ratio值.
将样本按特征值排序后，取头部 ratio 比例的样本，
计算该子群的坏样本率与整体坏样本率的比值。
:param x: 特征值数组
:param y: 目标变量数组
:param ratio: 覆盖率，默认0.10（LIFT@10%）
:param ascending: 排序方向，默认False（降序，取最大值头部）
:return: LIFT值
"""
n = len(x)
if n == 0:
return 1.0
# 特征无变异，无区分能力
if len(np.unique(x)) <= 1:
return 1.0
base_bad_rate = np.mean(y)
if base_bad_rate == 0 or base_bad_rate == 1:
return 1.0
# 头部样本数量（至少1个）
k = max(1, int(np.ceil(n * ratio)))
# 按特征值排序
if ascending:
order = np.argsort(x, kind="stable") # 升序：最小值在前
else:
order = np.argsort(x, kind="stable")[::-1] # 降序：最大值在前
# 取头部 k 个样本
top_idx = order[:k]
top_bad_rate = np.mean(y[top_idx])
lift = top_bad_rate / base_bad_rate
return float(lift)
def _compute_lift_with_direction(
x: np.ndarray,
y: np.ndarray,
ratio: float = 0.10,
direction: str = "auto",
) -> Tuple[float, float, float, str]:
"""计算单个特征的LIFT得分（支持方向判断）.
:param x: 特征值数组
:param y: 目标变量数组
:param ratio: 覆盖率
:param direction: 方向模式
- 'auto': 同时计算两个方向，只比较各自目标方向上的改善
- 'bad': 仅计算找坏人的LIFT（降序取头部，LIFT越高越好）
- 'good': 仅计算找好人的LIFT（升序取头部，LIFT越低越好）
:return: (score, lift_bad, lift_good, best_direction)
- score: 指定方向相对 LIFT=1 的有效改善，越大区分力越强
- lift_bad: 降序LIFT值（找坏人方向）
- lift_good: 升序LIFT值（找好人方向）
- best_direction: 最优方向 'bad' 或 'good'
"""
if direction == "bad":
lift_bad = _compute_lift_single(x, y, ratio, ascending=False)
score = max(lift_bad - 1.0, 0.0)
return score, lift_bad, np.nan, "bad"
if direction == "good":
lift_good = _compute_lift_single(x, y, ratio, ascending=True)
score = max(1.0 - lift_good, 0.0)
return score, np.nan, lift_good, "good"
# auto: 同时计算两个方向，只奖励方向正确的改善。
lift_bad = _compute_lift_single(x, y, ratio, ascending=False)
lift_good = _compute_lift_single(x, y, ratio, ascending=True)
dist_bad = max(lift_bad - 1.0, 0.0)
dist_good = max(1.0 - lift_good, 0.0)
if dist_bad >= dist_good:
return dist_bad, lift_bad, lift_good, "bad"
else:
return dist_good, lift_bad, lift_good, "good"
def _compute_lift_feature(task):
"""计算单个特征的 LIFT 详情。"""
feature, values, y, ratio, direction = task
return (feature,) + _compute_lift_with_direction(values, y, ratio, direction)
[文档]
class LiftSelector(BaseFeatureSelector):
"""LIFT筛选器.
使用LIFT@ratio值筛选特征，支持找坏人、找好人、自动三种方向模式。
LIFT衡量特征在头部覆盖率下对目标群体的提升程度。
**LIFT@ratio% 计算方式**
1. 将样本按特征值排序
2. 取头部 ratio 比例的样本
3. LIFT = 该子群坏样本率 / 整体坏样本率
**方向模式**
| direction | 含义 | 评分方式 |
|-----------|------|----------|
| auto | 自动选择最优方向（默认） | score = max(LIFT_bad-1, 1-LIFT_good, 0) |
| bad | 仅评估找坏人能力 | score = max(LIFT_bad - 1, 0) |
| good | 仅评估找好人能力 | score = max(1 - LIFT_good, 0) |
**评分含义**
score 只度量目标方向相对基准 LIFT=1 的改善，反向偏离按 0 计:
- score = 0: 无区分能力（LIFT = 1）
- score = 4.0: 强找坏人能力（LIFT_bad=5.0）；找好人得分上限为 1（LIFT_good=0）
- 内部经验: score >= 0.5 通常认为有一定区分力
**参数**
:param threshold: 目标方向的改善得分阈值，默认0.5
- 仅保留 score >= threshold 的特征
- threshold=0.5 等价于旧版 LIFT >= 1.5（找坏人方向）
- 内部经验: 风控场景常用 0.5~1.0
:param ratio: LIFT计算的覆盖率，默认0.10（即LIFT@10%）
- 内部经验: 风控场景常用 lift@5% 或 lift@10%
:param direction: 方向模式，默认'auto'
- 'auto': 同时计算两个方向，取最优（推荐）
- 'bad': 仅评估找坏人能力（降序取头部，LIFT > 1）
- 'good': 仅评估找好人能力（升序取头部，LIFT < 1）
:param target: 目标变量列名，默认为'target'
:param include: 强制保留的特征列表
:param exclude: 强制剔除的特征列表
:param n_jobs: 并行计算的任务数
**属性**
- scores\_: 各特征在目标方向上的改善得分，pd.Series
- lift_detail\_: 各特征的LIFT详情表，pd.DataFrame
包含列: LIFT_bad, LIFT_good, best_direction, score
**参考样例**
>>> from hscredit.core.selectors import LiftSelector
>>> 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 = LiftSelector(threshold=0.5, ratio=0.10)
>>> selector.fit(X, y)
>>> print(selector.lift_detail_) # 查看各特征两个方向的LIFT
>>>
>>> # 仅评估找坏人能力
>>> selector = LiftSelector(direction='bad', threshold=0.5)
>>> selector.fit(X, y)
>>>
>>> # 仅评估找好人能力
>>> selector = LiftSelector(direction='good', threshold=0.5)
>>> selector.fit(X, y)
**引用**
LIFT@k%（头部覆盖率下的提升度）是响应/风险模型的标准评估口径，参见 lift chart
https://en.wikipedia.org/wiki/Lift_(data_mining) 及 Siddiqi, N. (2006).
*Credit Risk Scorecards.* Wiley。
"""
method_name = "LIFT筛选"
def __init__(
self,
threshold: float = 0.5,
ratio: float = 0.10,
direction: Literal["auto", "bad", "good"] = "auto",
target: str = "target",
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.ratio = ratio
self.direction = direction
def _fit_impl(
self,
X: pd.DataFrame,
y: Optional[Union[pd.Series, np.ndarray]],
) -> None:
"""拟合LIFT筛选器。
:param X: 输入特征DataFrame
:param y: 目标变量
"""
self._get_feature_names(X)
if y is None:
raise ValueError("LiftSelector 需要目标变量 y")
if not 0 < float(self.ratio) <= 1:
raise ValueError("ratio 必须在 (0, 1] 范围内")
if self.direction not in {"auto", "bad", "good"}:
raise ValueError("direction 必须是 'auto'、'bad' 或 'good'")
y = np.asarray(y)
results = self._parallel_execute(
_compute_lift_feature,
((col, X[col].values, y, self.ratio, self.direction) 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=8.0,
capability="thread_safe",
releases_gil=True,
operation="LIFT字段排序",
),
)
# 解包结果
scores = np.array([r[1] for r in results])
lift_bad = np.array([r[2] for r in results])
lift_good = np.array([r[3] for r in results])
best_dirs = [r[4] for r in results]
# 评分只奖励目标方向上的改善
self.scores_ = pd.Series(scores, index=X.columns)
# LIFT详情表
self.lift_detail_ = pd.DataFrame(
{
"LIFT_bad": lift_bad,
"LIFT_good": lift_good,
"best_direction": best_dirs,
"score": scores,
},
index=X.columns,
)
# 选择 score >= threshold 的特征
selected_mask = scores >= self.threshold
self.selected_features_ = X.columns[selected_mask].tolist()
# 生成剔除原因
dir_label = {"auto": "自动", "bad": "找坏人", "good": "找好人"}
self._drop_reason = f"LIFT@{self.ratio:.0%} 方向改善得分 < {self.threshold}" f"（方向: {dir_label.get(self.direction, self.direction)}）"