"""WOE (Weight of Evidence) 编码器.
提供直接计算WOE的编码功能，不依赖分箱模块。
"""
from typing import Optional, List, Dict, Union, Any
import numpy as np
import pandas as pd
from .base import BaseEncoder
from ._category_protocol import CategoryToken, MISSING, UNKNOWN
from ...exceptions import NotFittedError, FeatureNotFoundError
[文档]
class WOEEncoder(BaseEncoder):
"""WOE (证据权重) 编码器.
直接对类别特征计算WOE值，不依赖分箱功能。
WOE计算公式：
WOE = ln(P(坏样本|类别) / P(好样本|类别)) = ln(坏样本占比/好样本占比)
**参数**
:param cols: 需要编码的列名列表。如果为None，则自动识别所有类别型列
:param regularization: 正则化参数，防止除零，默认为1.0
:param woe_clip: WOE值截断阈值，默认为5.0
当某个分箱无坏样本或无好样本时，WOE可能变得极大（如±10以上），
这会导致评分卡中对应分箱的分数异常。
设置此参数可将WOE限制在[-woe_clip, woe_clip]范围内。
设置为None则不进行截断。
:param handle_unknown: transform 时遇到 fit 未见过的类别的处理方式，默认为 ``'value'``：
- ``'value'``：编码为 0.0（中性 WOE）
- ``'return_nan'``：编码为 NaN
- ``'error'``：抛出异常
:param handle_missing: 缺失值（NaN）的处理方式，默认为 ``'value'``：
- ``'value'``：编码为 0.0（中性 WOE）
- ``'return_nan'``：编码为 NaN
- ``'error'``：fit/transform 遇缺失即抛出异常
:param drop_invariant: 是否删除方差为0的列，默认为False
:param return_df: 是否返回DataFrame，默认为True
:param target: scorecardpipeline 风格的目标列名，提供后 fit 时从 X 中提取该列作为 y
:param training_mode: 默认 in_sample；oof 时 fit_transform 生成折外训练编码，
transform 使用全训练集映射。时间切分未覆盖的起始行保留 NaN，见 oof_coverage_。
:param cv: 折外编码的折数或 sklearn 切分器；fit_transform 的 groups 参数按位置传入
:param random_state: 整数 cv 随机分折的种子，默认 42
**属性**
- mapping_: WOE编码映射字典，格式为 {col: {category: woe_value}}
- iv_: 各特征的IV值，格式为 {col: iv_value}
**参考样例**
>>> from hscredit.core.encoders import WOEEncoder
>>> encoder = WOEEncoder(cols=['category', 'score'])
>>> X_encoded = encoder.fit_transform(X, y)
>>> print(encoder.iv_)
>>>
>>> # 获取IV摘要
>>> summary = encoder.summary()
>>> print(summary)
导出/加载（与 toad、scorecardpipeline 规则格式互通）::
>>> rules = encoder.export(to_json='woe_rules.json')
>>> WOEEncoder().load('woe_rules.json')
**注意**
与 :class:`~hscredit.core.binning.BaseBinning` 的 ``metric='woe'`` 不同，本编码器
直接对原始类别取值计算 WOE、不做数值分箱，适合基数适中的类别特征。``regularization``
采用加性平滑以避免某类别好/坏样本数为 0 时 WOE 取 ±∞。
**引用**
WOE（证据权重）与 IV（信息价值）出自信息论，系统应用于信用评分见
Siddiqi, N. (2006). *Credit Risk Scorecards.* Wiley；
公式与直观解释参考
https://www.listendata.com/2015/03/weight-of-evidence-woe-and-information.html
"""
# iv_ 为各特征信息价值，随映射一并序列化以便 import_mapping 后仍可查
_EXTRA_STATE_ATTRS = ["iv_", "_legacy_string_keys_"]
_TARGET_TYPE = "binary"
_OUTPUT_ATTRS = {"hscredit_encoding": "woe", "hscredit_source": "WOEEncoder"}
def __init__(
self,
cols: Optional[List[str]] = None,
regularization: float = 1.0,
woe_clip: Optional[float] = 5.0,
handle_unknown: str = 'value',
handle_missing: str = 'value',
drop_invariant: bool = False,
return_df: bool = True,
target: Optional[str] = None,
n_jobs: Optional[Union[int, float]] = -1,
parallel_backend: Optional[str] = None,
parallel_config: Optional[Dict[str, Any]] = None,
training_mode: str = "in_sample",
cv: Any = 5,
random_state: Optional[int] = 42,
passthrough_target: bool = False,
):
"""初始化WOE编码器。
:param cols: 需要编码的列名列表
:param regularization: 正则化参数，防止除零，默认为1.0
:param woe_clip: WOE值截断阈值，默认为5.0
当某个分箱的WOE绝对值超过此阈值时会被截断。
这可以防止因分箱中无坏样本或无好样本导致的极端WOE值，
避免评分卡中对应分箱的分数过大。
设置为None则不进行截断。
:param handle_unknown: 处理未知类别的方式，默认为'value'
:param handle_missing: 处理缺失值的方式，默认为'value'
:param drop_invariant: 是否删除方差为0的列，默认为False
:param return_df: 是否返回DataFrame，默认为True
:param target: scorecardpipeline风格的目标列名。如果提供，fit时从X中提取该列作为y
"""
super().__init__(
cols=cols,
drop_invariant=drop_invariant,
return_df=return_df,
handle_unknown=handle_unknown,
handle_missing=handle_missing,
target=target,
n_jobs=n_jobs,
parallel_backend=parallel_backend,
parallel_config=parallel_config,
passthrough_target=passthrough_target,
)
self.regularization = regularization
self.woe_clip = woe_clip
self.training_mode = training_mode
self.cv = cv
self.random_state = random_state
self.iv_: Dict[str, float] = {}
self._legacy_string_keys_ = False
def _get_category_cols(self, X: pd.DataFrame) -> List[str]:
"""自动识别需要编码的列。
WOEEncoder支持数值型和类别型列，因此返回所有列（除了目标列）。
:param X: 输入数据
:return: 列名列表
"""
return X.columns.tolist()
def _fit(self, X: pd.DataFrame, y: Optional[pd.Series] = None):
"""拟合WOE编码器。
:param X: 输入数据，shape (n_samples, n_features)
:param y: 目标变量，二分类 (0/1)
:raises ValueError: 当y为空或目标变量不是二元时抛出
"""
if y is None:
raise ValueError("WOEEncoder是有监督编码器，必须提供目标变量y")
y = pd.Series(y).astype(int)
unique = y.unique()
if len(unique) != 2:
raise ValueError(f"目标变量必须是二元的，当前有{len(unique)}个唯一值")
if not set(unique).issubset({0, 1}):
raise ValueError("目标变量必须是0和1")
self._fit_columns(X, y, state_attrs=("mapping_", "iv_"))
def _fit_column(self, column, values, y=None):
total_good = (y == 0).sum()
total_bad = (y == 1).sum()
woe_map, iv = self._fit_categorical(values, y, total_good, total_bad)
return {"mapping_": woe_map, "iv_": iv}
def _fit_categorical(
self, x: pd.Series, y: pd.Series, total_good: int, total_bad: int
) -> tuple:
"""拟合类别特征的WOE。
:param x: 特征列
:param y: 目标变量
:param total_good: 好样本总数
:param total_bad: 坏样本总数
:return: WOE映射和IV值的元组 (woe_map, iv)
"""
woe_map = {}
codes, categories = pd.factorize(x, sort=False)
valid = codes >= 0
good_counts = np.bincount(codes[valid], weights=(np.asarray(y)[valid] == 0), minlength=len(categories))
bad_counts = np.bincount(codes[valid], weights=(np.asarray(y)[valid] == 1), minlength=len(categories))
iv = 0.0
for category, good_count, bad_count in zip(categories, good_counts, bad_counts):
woe = self._compute_woe(good_count, bad_count, total_good, total_bad)
woe_map[category] = woe
good_dist = (good_count + self.regularization) / (total_good + 2 * self.regularization)
bad_dist = (bad_count + self.regularization) / (total_bad + 2 * self.regularization)
iv += (bad_dist - good_dist) * np.log(bad_dist / good_dist)
if self.handle_missing == 'value':
woe_map[MISSING] = 0.0
elif self.handle_missing == 'return_nan':
woe_map[MISSING] = np.nan
if self.handle_unknown == 'value':
woe_map[UNKNOWN] = 0.0
elif self.handle_unknown == 'return_nan':
woe_map[UNKNOWN] = np.nan
return woe_map, iv
def _compute_woe(
self, good_count: int, bad_count: int, total_good: int, total_bad: int
) -> float:
"""计算WOE值（带正则化和截断）。
WOE = ln(坏样本占比 / 好样本占比)
与 toad、scorecardpipeline 及 hscredit 分箱模块保持一致。
坏样本集中的箱 WOE > 0，好样本集中的箱 WOE < 0，
LR 系数为正，便于理解。
当某个分箱无坏样本或无好样本时，WOE值可能变得极大，
此时会根据 woe_clip 参数进行截断，防止评分卡分数异常。
:param good_count: 好样本数量
:param bad_count: 坏样本数量
:param total_good: 好样本总数
:param total_bad: 坏样本总数
:return: WOE值
"""
good_rate = (good_count + self.regularization) / (total_good + 2 * self.regularization)
bad_rate = (bad_count + self.regularization) / (total_bad + 2 * self.regularization)
woe = np.log(bad_rate / good_rate)
# 截断极端WOE值，防止评分卡分数异常
if self.woe_clip is not None:
woe = np.clip(woe, -self.woe_clip, self.woe_clip)
return woe
def _compute_iv_categorical(
self, x: pd.Series, y: pd.Series, total_good: int, total_bad: int
) -> float:
"""计算类别特征的IV。
:param x: 特征列
:param y: 目标变量
:param total_good: 好样本总数
:param total_bad: 坏样本总数
:return: IV值
"""
iv = 0.0
for category in x.dropna().unique():
mask = x == category
good_count = (y[mask] == 0).sum()
bad_count = (y[mask] == 1).sum()
good_dist = (good_count + self.regularization) / (total_good + 2 * self.regularization)
bad_dist = (bad_count + self.regularization) / (total_bad + 2 * self.regularization)
iv += (bad_dist - good_dist) * np.log(bad_dist / good_dist)
return iv
def _transform(self, X: pd.DataFrame, y: Optional[pd.Series] = None) -> pd.DataFrame:
"""转换数据为WOE编码。
:param X: 输入数据，shape (n_samples, n_features)
:param y: 目标变量（可选）
:return: 编码后的数据
"""
output = self._transform_columns(X, y)
output.attrs['hscredit_encoding'] = 'woe'
output.attrs['hscredit_source'] = 'WOEEncoder'
return output
def _transform_column(self, column, values, y=None, context=None):
woe_map = self.mapping_[column]
mapped = self._map_values(values, woe_map)
# 仅旧格式允许数字字符串回退，维持传统export→load互操作。
# fit及v2类型映射按真实类别类型匹配，不将未见字符串悄悄当数值类别。
unmapped = mapped.isna() & values.notna()
if unmapped.any() and getattr(self, "_legacy_string_keys_", False):
str_map = {
str(k): v
for k, v in woe_map.items()
if not isinstance(k, CategoryToken) and not (isinstance(k, float) and pd.isna(k))
}
if str_map:
mapped.loc[unmapped] = values.loc[unmapped].astype(str).map(str_map)
# 旧JSON的整数键是"1"；Excel/pandas遇缺失可能把输入提升为1.0。
# 仅在显式旧格式且普通字符串回退仍未命中时，兼容这类整数浮点值。
remaining = mapped.isna() & values.notna()
if remaining.any():
def legacy_key(value):
if isinstance(value, (float, np.floating)) and np.isfinite(value) and float(value).is_integer():
return str(int(value))
return str(value)
mapped.loc[remaining] = values.loc[remaining].map(legacy_key).map(str_map)
return self._apply_missing_unknown(column, values, mapped, 0.0)
[文档]
def get_iv(self) -> Dict[str, float]:
"""获取各特征的IV值。
:return: 特征名到IV值的映射字典
"""
return self.iv_
[文档]
def get_mapping(self, col: Optional[str] = None) -> Union[Dict, Dict[str, Dict]]:
"""获取WOE编码映射。
:param col: 列名。如果提供，返回该列的映射；
如果为None，返回所有列的映射
:return: WOE映射字典。当 col 指定时返回 {category: woe_value}，
col 为 None 时返回 {col: {category: woe_value}}
:raises NotFittedError: 当编码器尚未拟合时抛出
:raises FeatureNotFoundError: 当指定的 col 不在编码器中时抛出
"""
if not hasattr(self, 'mapping_') or not self.mapping_:
raise NotFittedError("WOEEncoder 尚未拟合，请先调用 fit 方法")
if col is None:
return self.mapping_
if col not in self.mapping_:
raise FeatureNotFoundError(f"列 '{col}' 不在编码器中，请检查列名是否正确")
return self.mapping_[col]
[文档]
def summary(self) -> pd.DataFrame:
"""获取 WOE 编码摘要表（按 IV 降序）。
对每个已编码特征给出 IV 值及对应的预测能力评级，评级阈值为：
- IV < 0.02：无预测力
- 0.02 ≤ IV < 0.1：弱预测力
- 0.1 ≤ IV < 0.3：中等预测力
- 0.3 ≤ IV < 0.5：强预测力
- IV ≥ 0.5：超强预测力（需检查是否标签泄漏）
:return: 含 ``特征`` / ``IV值`` / ``预测能力`` 三列的 DataFrame，按 ``IV值`` 降序；
未拟合或无特征时返回空 DataFrame
**参考样例**
>>> encoder.fit(X, y)
>>> encoder.summary()
"""
if not self.iv_:
return pd.DataFrame()
summary = []
for col, iv in self.iv_.items():
if iv < 0.02:
power = '无预测力'
elif iv < 0.1:
power = '弱预测力'
elif iv < 0.3:
power = '中等预测力'
elif iv < 0.5:
power = '强预测力'
else:
power = '超强预测力(需检查)'
summary.append({
'特征': col,
'IV值': round(iv, 4),
'预测能力': power,
})
return pd.DataFrame(summary).sort_values('IV值', ascending=False)
[文档]
def export(self, to_json: Optional[str] = None) -> Dict[str, Dict]:
"""导出WOE编码规则，兼容 toad/scorecardpipeline 格式.
导出格式与 toad.WOETransformer.export() 和 scorecardpipeline.WOETransformer.export() 保持一致。
:param to_json: 可选，JSON 文件保存路径。如果提供，将规则保存到该文件
:return: WOE编码规则字典，格式为 {feature: {value: woe_value, ...}, ...}
**参考样例**
>>> from hscredit.core.encoders import WOEEncoder
>>> encoder = WOEEncoder(cols=['category', 'city'])
>>> encoder.fit(X, y)
>>>
>>> # 导出为字典
>>> rules = encoder.export()
>>>
>>> # 导出并保存到 JSON 文件
>>> rules = encoder.export(to_json='woe_rules.json')
**与 toad/scorecardpipeline 的兼容性**
导出的规则可以直接被 toad 和 scorecardpipeline 加载:
>>> import toad
>>> transformer = toad.transform.WOETransformer()
>>> transformer.load(rules)
>>>
>>> from scorecardpipeline import WOETransformer
>>> transformer = WOETransformer()
>>> transformer.load(rules)
"""
import json
if not hasattr(self, 'mapping_') or not self.mapping_:
raise ValueError("WOEEncoder 尚未拟合，请先调用 fit 方法")
# 构建与 toad 兼容的格式: {feature: {value: woe_value}}
rules = {}
for col, woe_map in self.mapping_.items():
# 将 WOE 映射转换为可 JSON 序列化的格式
col_rules = {}
for value, woe in woe_map.items():
# 处理特殊值
if value == UNKNOWN:
continue # toad 不保存 __UNKNOWN__
if value == MISSING or pd.isna(value):
if 'nan' in col_rules:
raise ValueError("传统 WOE JSON 缺失键与业务类别冲突，请使用 export_mapping()")
col_rules['nan'] = woe # toad 使用字符串 'nan'
else:
key = str(value)
if key in col_rules:
raise ValueError("传统 WOE JSON 字符串键存在类型碰撞，请使用 export_mapping() 的类型化协议")
col_rules[key] = float(woe)
rules[col] = col_rules
if to_json is not None:
# 确保目录存在
import os
dir_path = os.path.dirname(to_json)
if dir_path and not os.path.exists(dir_path):
os.makedirs(dir_path, exist_ok=True)
with open(to_json, 'w', encoding='utf-8') as f:
json.dump(rules, f, ensure_ascii=False, indent=2)
return rules
[文档]
def load(self, from_json: Union[str, Dict], update: bool = False) -> 'WOEEncoder':
"""加载WOE编码规则，兼容 toad/scorecardpipeline 格式.
从字典或 JSON 文件加载WOE编码规则，支持 toad 和 scorecardpipeline 导出的格式。
:param from_json: WOE规则字典或 JSON 文件路径
- 字典: {'category': {'A': 0.5, 'B': -0.3}}
- 文件路径: 'woe_rules.json'
:param update: 是否更新现有规则（而非替换），默认为 False
:return: self，支持链式调用
**参考样例**
>>> from hscredit.core.encoders import WOEEncoder
>>> encoder = WOEEncoder()
>>>
>>> # 从字典加载
>>> rules = {'category': {'A': 0.5, 'B': -0.3}}
>>> encoder.load(rules)
>>>
>>> # 从 JSON 文件加载
>>> encoder.load('woe_rules.json')
>>>
>>> # 更新现有规则
>>> encoder.load({'new_feature': {'X': 0.2}}, update=True)
**与 toad/scorecardpipeline 的兼容性**
可以直接加载 toad 和 scorecardpipeline 导出的规则:
>>> import toad
>>> toad_transformer = toad.transform.WOETransformer()
>>> toad_transformer.fit(df, y)
>>> rules = toad_transformer.export()
>>>
>>> encoder = WOEEncoder()
>>> encoder.load(rules)
"""
import json
self._legacy_string_keys_ = True
if isinstance(from_json, str):
# 从文件加载
with open(from_json, 'r', encoding='utf-8') as f:
rules = json.load(f)
else:
# 直接使用字典
rules = from_json
if not update:
self.mapping_ = {}
self.cols_ = []
# 规范化容器：update=True 时若在未 fit 的新实例上调用，
# cols_/mapping_ 仍为基类初始的 None，下方成员判断会崩溃，这里统一兜底为空容器
if self.cols_ is None:
self.cols_ = []
if self.mapping_ is None:
self.mapping_ = {}
# 加载规则
for col, col_rules in rules.items():
if col not in self.mapping_:
self.mapping_[col] = {}
if col not in self.cols_:
self.cols_.append(col)
for value, woe in col_rules.items():
# toad/scorecardpipeline 用字符串 'nan' 表示缺失键
if value == 'nan':
self.mapping_[col][MISSING] = woe
else:
# 保持原始（字符串）键，不做 int/float 强转：
# 否则数字型字符串类别（如城市码/商品码 '100'）会被转成 int，
# 与 transform 时的字符串输入不匹配，导致 WOE 全部落到未知值（静默错误）。
# 数值型输入的兼容由 _transform 的字符串回退映射保证。
self.mapping_[col][value] = woe
# 添加未知值处理
if self.handle_unknown == 'value':
self.mapping_[col][UNKNOWN] = 0.0
elif self.handle_unknown == 'return_nan':
self.mapping_[col][UNKNOWN] = np.nan
self._is_fitted = True
return self