"""统一分箱接口 - 整合所有分箱方法.
提供统一的 fit/transform 接口，支持全部分箱方法（共 17 种，
见 ``OptimalBinning.VALID_METHODS``）：
- 无监督方法: uniform, quantile, kmeans, kernel_density
- 有监督方法: tree, cart, chi, mdlp, best_ks, best_iv, best_lift,
target_bad_rate, monotonic, genetic, smooth
- 运筹规划: or_tools, cp_sat
支持指定切割点 (user_splits) 和单调性约束。
使用 core.metrics 中的指标计算方法。
"""
import logging
import inspect
from copy import deepcopy
from typing import Union, List, Dict, Optional, Any, Callable, Iterable
import numpy as np
import pandas as pd
import warnings
from ...exceptions import NotFittedError, ParallelExecutionError
from ...utils.parallel import ParallelWorkload, parallel_execute
from .base import BaseBinning
from ._contracts import (
HandleUnknown,
UNKNOWN_BIN,
UserSplitsFixed,
is_missing_marker,
validate_handle_unknown,
)
from ._categorical import (
CategoryOrder,
assign_category_groups,
restore_category_groups,
)
from .uniform_binning import UniformBinning
from .quantile_binning import QuantileBinning
from .tree_binning import TreeBinning
from .chi_merge_binning import ChiMergeBinning
from .best_ks_binning import BestKSBinning
from .best_iv_binning import BestIVBinning
from .mdlp_binning import MDLPBinning
from .or_binning import ORBinning, ORTOOLS_AVAILABLE
from .cp_sat_binning import CPSATBinning
from .cart_binning import CartBinning
from .kmeans_binning import KMeansBinning
from .genetic_binning import GeneticBinning
from .smooth_binning import SmoothBinning
from .kernel_density_binning import KernelDensityBinning
from .best_lift_binning import BestLiftBinning
from .target_bad_rate_binning import TargetBadRateBinning
from .monotonic_binning import MonotonicBinning
# 从 metrics 导入指标计算方法
from ..metrics._binning import (
iv_for_splits,
ks_for_splits,
compare_splits_iv,
compare_splits_ks,
)
logger = logging.getLogger(__name__)
def _auto_method_score_worker(task):
"""拟合一个自动分箱候选并返回稳定的候选得分。"""
method, X, y, feature, criterion, parallel_backend, parallel_config = task
try:
binner = OptimalBinning(
method=method,
verbose=False,
n_jobs=1,
parallel_backend=parallel_backend,
parallel_config=parallel_config,
)
binner.fit(X[[feature]], y)
bin_table = binner.bin_tables_.get(feature)
if bin_table is None or len(bin_table) == 0:
return method, None, None
if criterion == "ks":
score = float(bin_table["分档KS值"].max()) if "分档KS值" in bin_table.columns else 0.0
elif "指标IV值" in bin_table.columns:
score = float(bin_table["指标IV值"].iloc[0])
elif "分档IV值" in bin_table.columns:
score = float(bin_table["分档IV值"].sum())
else:
score = 0.0
return method, score, None
except Exception as exc:
return method, None, f"{type(exc).__name__}: {exc}"
[文档]
class OptimalBinning(BaseBinning):
"""统一分箱接口 - 整合所有分箱方法.
提供统一的 fit/transform 接口，支持所有分箱方法。
融合 MonotonicBinning 的单调性约束功能。
支持指定切割点 (user_splits) 和预分箱。
**架构设计原则**
1. **OptimalBinning 作为统一入口**：集成所有分箱方法，支持预分箱功能
2. **独立分箱模块保持简单**：各个具体分箱类（如BestIVBinning、MDLPBinning等）
只执行一次分箱，不包含预分箱逻辑
3. **预分箱在 OptimalBinning 层面实现**：通过 prebinning 参数在统一接口层实现
预分箱+二次分箱的两阶段分箱流程
:param target: 目标变量列名，默认为'target'
:param method: 分箱方法，默认为 ``'mdlp'``。可取以下枚举值（按类别）：
*无监督（不使用标签决定切分）*
- ``'uniform'``：等距分箱，按数值范围等宽切分，快但对偏态敏感
- ``'quantile'``：等频分箱，各箱样本量均衡，对异常值稳健，常作预分箱
- ``'kmeans'``：K-Means 聚类分箱，按数值聚类结构切分，适合自然分组
- ``'kernel_density'``：核密度分箱，以分布谷值为边界，适合多峰分布
*有监督（结合标签寻优）*
- ``'tree'``：决策树分箱，取决策树分裂点，贴合目标
- ``'cart'``：CART 分箱（对齐 optbinning 预分箱），支持 p-value 检验/回归目标
- ``'chi'``：卡方分箱（ChiMerge），合并分布无显著差异的相邻箱
- ``'mdlp'``：MDLP 信息论分箱，自动确定分箱数（默认）
- ``'best_ks'``：最大化 KS 统计量
- ``'best_iv'``：最大化 IV（信息价值），评分卡常用
- ``'best_lift'``：最大化头部箱提升度 Lift，偏策略拒绝场景
- ``'target_bad_rate'``：按目标坏样本率梯度切分
- ``'monotonic'``：单调最优分箱，强约束坏样本率/WOE 单调（含 U/倒U）
- ``'genetic'``：遗传算法全局寻优，约束复杂时使用
- ``'smooth'``：平滑分箱，对箱内坏样本率做平滑收缩，适合小样本
*运筹规划（数学规划全局最优）*
- ``'or_tools'``：基于 Google OR-Tools 的整数规划分箱
- ``'cp_sat'``：基于 CP-SAT 求解器的约束规划分箱
:param max_n_bins: 最大分箱数，默认为5
:param min_n_bins: 最小分箱数，默认为2
:param min_bin_size: 每箱最小样本数或占比，默认为0.01
:param max_bin_size: 每箱最大样本数或占比，默认为None（不限制）
:param min_bad_rate: 每箱最小坏样本率，默认为0.0。坏样本率低于该值的分箱将与相邻
分箱合并；无论该参数取值，坏样本率为 0/1 的退化分箱都会被合并以避免 WOE 异常
（等宽 uniform / 等频 quantile 为保持切分结构精确，不参与该约束）
:param monotonic: 坏样本率单调性约束，默认为False
- False: 不要求单调性
- True 或 'auto': 自动检测最佳趋势（允许单增、单减、正U、倒U）
- 'auto_asc_desc': 自动检测，但只允许单增或单减（不允许U型）
- 'auto_heuristic': 使用启发式方法自动检测
- 'ascending': 强制坏样本率递增
- 'descending': 强制坏样本率递减
- 'peak': 允许单峰形态(先升后降，倒U型)
- 'valley': 允许单谷形态(先降后升，正U型)
- 'peak_heuristic': 使用启发式方法检测峰值
- 'valley_heuristic': 使用启发式方法检测谷值
:param user_splits: 用户指定的切分点，支持:
- Dict[str, List]: 每个特征的切分点，如 {'age': [25, 35, 45]}
- Callable: 函数返回切分点
:param user_splits_fixed: 用户切分点固定配置。支持全局布尔值，或按字段配置布尔值/布尔列表
:param prebinning: 预分箱方法，支持:
- str: 预分箱方法名（所有VALID_METHODS中的方法都可作为预分箱方法）
- BaseBinning: 预分箱器实例
- Dict: 预分箱配置，如 {'method': 'cart', 'max_n_bins': 20}
默认为None（不进行预分箱）
:param prebinning_params: 预分箱参数，当prebinning为str时使用
:param special_codes: 特殊值列表，默认为None
:param cat_cutoff: 类别型变量处理阈值，默认为None
:param random_state: 随机种子，默认为None
:param verbose: 是否输出详细信息，默认为False
:param decimal: 数值型切分点保留的小数位数，默认为4
:param woe_clip: WOE值截断阈值，默认为None
当某个分箱无坏样本或无好样本时，WOE可能变得极大（如±10以上），
这会导致评分卡中对应分箱的分数异常。
设置此参数可将WOE限制在[-woe_clip, woe_clip]范围内。
:param kwargs: 其他分箱方法特定参数
**参考样例**
>>> from hscredit.core.binning import OptimalBinning
>>> # 使用MDLP分箱（默认，直接分箱）
>>> binner = OptimalBinning(method='mdlp', max_n_bins=5)
>>> binner.fit(X, y)
>>> # 使用最优IV分箱（直接分箱）
>>> binner = OptimalBinning(method='best_iv', max_n_bins=5)
>>> binner.fit(X, y)
>>> # 使用CART分箱（直接分箱）
>>> binner = OptimalBinning(method='cart', max_n_bins=5)
>>> binner.fit(X, y)
>>> # 使用预分箱（MDLP先进行CART预分箱成20箱，再优化为5箱）
>>> binner = OptimalBinning(method='mdlp', prebinning='cart', prebinning_params={'max_n_bins': 20})
>>> binner.fit(X, y)
>>> # 使用预分箱器实例
>>> pre_binner = OptimalBinning(method='cart', max_n_bins=20)
>>> binner = OptimalBinning(method='best_iv', prebinning=pre_binner)
>>> binner.fit(X, y)
>>> # 使用quantile预分箱（先将数据分成20等份，再进行MDLP分箱）
>>> binner = OptimalBinning(method='mdlp', prebinning='quantile', prebinning_params={'max_n_bins': 20})
>>> binner.fit(X, y)
>>> # 自动为某特征选择最优分箱方法
>>> best = OptimalBinning.auto_select_method(X, y, 'age')
>>> binner = OptimalBinning(method=best).fit(X, y)
**引用**
最优分箱（optimal binning）的两阶段（预分箱 + 二次合并）框架与单调/规划求解参考
optbinning：Navas-Palencia, G. (2020). *Optimal binning: mathematical programming
formulation.* arXiv:2001.08025. https://arxiv.org/abs/2001.08025 ；
各子方法的具体出处见对应分箱类文档。
"""
# 所有支持的分箱方法
VALID_METHODS = [
"uniform",
"quantile",
"tree",
"chi",
"best_ks",
"best_iv",
"mdlp",
"or_tools",
"cp_sat",
"cart",
"kmeans",
"monotonic",
"genetic",
"smooth",
"kernel_density",
"best_lift",
"target_bad_rate",
]
[文档]
@classmethod
def validate_method(cls, method: str) -> str:
"""校验并归一化分箱方法名.
统一所有入口的 method 校验逻辑：类型检查、大小写与首尾空白归一化、
枚举值校验。不支持别名（如 ``'iv'``、``'chi2'`` 会被拒绝）。
:param method: 分箱方法名
:return: 归一化后的方法名（小写、去除首尾空白）
:raises ValueError: method 不是字符串或不在 ``VALID_METHODS`` 中时抛出
"""
if not isinstance(method, str):
raise ValueError(f"method 必须是字符串，当前类型为 {type(method).__name__}: {method!r}，" f"可选: {cls.VALID_METHODS}")
normalized = method.strip().lower()
if normalized not in cls.VALID_METHODS:
raise ValueError(f"不支持的method: {method}，可选: {cls.VALID_METHODS}")
return normalized
def __init__(
self,
target: str = "target",
method: str = "mdlp",
max_n_bins: int = 5,
min_n_bins: int = 2,
min_bin_size: Union[float, int] = 0.01,
max_bin_size: Optional[Union[float, int]] = None,
min_bad_rate: float = 0.0,
monotonic: Union[bool, str] = False,
missing_separate: bool = True,
user_splits: Optional[Union[Dict[str, List], Callable]] = None,
user_splits_fixed: UserSplitsFixed = None,
prebinning: Optional[Union[str, "BaseBinning", Dict]] = None,
prebinning_params: Optional[Dict] = None,
special_codes: Optional[List] = None,
cat_cutoff: Optional[Union[float, int]] = None,
category_order: CategoryOrder = None,
handle_unknown: HandleUnknown = UNKNOWN_BIN,
random_state: Optional[int] = None,
verbose: Union[bool, int] = False,
decimal: int = 4,
woe_clip: Optional[float] = None,
n_jobs: Union[int, float] = -1,
parallel_backend: Optional[str] = None,
parallel_config: Optional[Dict[str, Any]] = None,
**kwargs,
):
super().__init__(
target=target,
missing_separate=missing_separate,
max_n_bins=max_n_bins,
min_n_bins=min_n_bins,
min_bin_size=min_bin_size,
max_bin_size=max_bin_size,
min_bad_rate=min_bad_rate,
monotonic=monotonic,
special_codes=special_codes,
cat_cutoff=cat_cutoff,
user_splits=user_splits,
user_splits_fixed=user_splits_fixed,
category_order=category_order,
handle_unknown=handle_unknown,
random_state=random_state,
verbose=verbose,
decimal=decimal,
woe_clip=woe_clip,
n_jobs=n_jobs,
parallel_backend=parallel_backend,
parallel_config=parallel_config,
)
# 仅校验不归一化：sklearn clone 要求构造参数按对象原样保存，
# 大小写/空白归一化推迟到 fit 时统一进行
self.validate_method(method)
self.method = method
self.user_splits = user_splits
self.user_splits_fixed = user_splits_fixed
self.prebinning = prebinning
self.prebinning_params = prebinning_params
control_option_names = {
"lift_refine",
"lift_focus_weight",
"sample_stability_weight",
"lift_refine_max_bins",
"monotonic_bonus_weight",
}
invalid_kwargs = sorted(set(kwargs) - self._method_parameter_names() - control_option_names)
if invalid_kwargs:
if invalid_kwargs[0] == "n_bins":
raise ValueError("n_bins 参数已移除，请使用 max_n_bins")
raise ValueError(f"参数 {invalid_kwargs[0]!r} 对 OptimalBinning 无效")
self._fit_control_options = {name: kwargs[name] for name in control_option_names if name in kwargs}
# 清理kwargs，移除不应该传递给底层分箱器的参数
self.kwargs = self._clean_kwargs(kwargs)
self._defer_categorical_adapter = True
self._binner = None
self._prebinner = None
self._ordinary_binner_ = None
self._ordinary_features_ = ()
self.monotonic_trend_ = {}
def _clean_kwargs(self, kwargs: Dict) -> Dict:
"""清理kwargs，移除不应该传递给底层分箱器的参数.
这些参数是OptimalBinning特有的，底层分箱器不需要。
"""
# 需要过滤的参数列表
invalid_keys = [
"prebinning",
"prebinning_params",
"user_splits",
"user_splits_fixed",
"method",
"lift_refine",
"lift_focus_weight",
"sample_stability_weight",
"lift_refine_max_bins",
"monotonic_bonus_weight",
]
return {k: v for k, v in kwargs.items() if k not in invalid_keys}
@classmethod
def _method_parameter_names(cls) -> set:
"""返回所有直接分箱器可接收的方法专属参数名。"""
names = set()
for binner_class in cls._method_classes().values():
for name, parameter in inspect.signature(binner_class.__init__).parameters.items():
if name != "self" and parameter.kind not in (parameter.VAR_POSITIONAL, parameter.VAR_KEYWORD):
names.add(name)
return names
@classmethod
def _method_classes(cls) -> Dict[str, type]:
return {
"uniform": UniformBinning,
"quantile": QuantileBinning,
"tree": TreeBinning,
"chi": ChiMergeBinning,
"best_ks": BestKSBinning,
"best_iv": BestIVBinning,
"mdlp": MDLPBinning,
"or_tools": ORBinning,
"cp_sat": CPSATBinning,
"cart": CartBinning,
"kmeans": KMeansBinning,
"monotonic": MonotonicBinning,
"genetic": GeneticBinning,
"smooth": SmoothBinning,
"kernel_density": KernelDensityBinning,
"best_lift": BestLiftBinning,
"target_bad_rate": TargetBadRateBinning,
}
[文档]
def get_params(self, deep: bool = True) -> Dict[str, Any]:
"""在不改变 ``**kwargs`` 入口的前提下暴露方法专属参数。"""
params = super().get_params(deep=deep)
params.update(self.kwargs)
return params
[文档]
def set_params(self, **params) -> "OptimalBinning":
"""支持 sklearn 为显式参数和方法专属 ``**kwargs`` 参数赋值。"""
if not params:
return self
explicit = set(inspect.signature(self.__init__).parameters) - {"self", "kwargs"}
method_params = self._method_parameter_names()
control_params = {
"lift_refine",
"lift_focus_weight",
"sample_stability_weight",
"lift_refine_max_bins",
"monotonic_bonus_weight",
}
invalid = [name for name in params if name.split("__", 1)[0] not in explicit | method_params | control_params]
if invalid:
valid = sorted(explicit | method_params | control_params | set(self.kwargs))
raise ValueError(f"参数 {invalid[0]!r} 对 OptimalBinning 无效。可用参数为: {valid}")
nested = {name: value for name, value in params.items() if "__" in name}
direct = {name: value for name, value in params.items() if "__" not in name}
if "handle_unknown" in direct:
direct["handle_unknown"] = validate_handle_unknown(direct["handle_unknown"])
for name, value in direct.items():
if name in explicit:
setattr(self, name, value)
else:
self.kwargs[name] = value
if name in control_params:
self._fit_control_options[name] = value
if nested:
super().set_params(**nested)
return self
[文档]
def fit(
self, X: Union[pd.DataFrame, np.ndarray], y: Optional[Union[pd.Series, np.ndarray]] = None, **kwargs
) -> "OptimalBinning":
"""拟合分箱.
:param X: 训练数据
:param y: 目标变量
:param kwargs: 其他参数
:return: 拟合后的分箱器
"""
# 防御 set_params 绕过 __init__ 校验的场景，统一归一化 method
self.method = self.validate_method(self.method)
X, y = self._check_input(X, y)
imported_features = [
feature
for feature in getattr(self, "_imported_rule_features_", ())
if getattr(self, "_rules_imported_", False) and feature in X.columns
]
excluded_features = set()
if isinstance(self.user_splits, dict):
excluded_features.update(self.user_splits)
elif callable(self.user_splits):
excluded_features.update(X.columns)
excluded_features.update(imported_features)
self._record_category_orders(X, y, excluded_features=excluded_features)
if imported_features:
self._fit_imported_and_ordinary_features(X, y, imported_features)
self._finalize_reserved_bins(X, y)
self._is_fitted = True
return self
# 如果已经拟合过（例如通过import_rules），只计算统计信息
if self._is_fitted:
self._update_bin_stats(X, y)
self._finalize_reserved_bins(X, y)
return self
# 如果指定了 user_splits，优先使用
if self.user_splits is not None:
self._fit_with_user_splits(X, y)
self._finalize_reserved_bins(X, y)
self._is_fitted = True
return self
elif self.prebinning is not None:
# 使用预分箱
self._fit_with_prebinning(X, y)
# 统一后处理：围绕头尾Lift与样本稳定性微调切分点
# 默认开启，可通过 lift_refine=False 关闭
if self._fit_control_options.get("lift_refine", True) and self.method != "uniform":
self._refine_splits_for_lift_stability(X, y)
# quantile 方法需保持分位数切分点精确，跳过所有后处理
if self.method == "quantile":
self._restrict_to_prebin_candidates(
X,
y,
{feature: self._prebinner.splits_.get(feature, np.array([])) for feature in X.columns},
)
self._finalize_reserved_bins(X, y)
self._is_fitted = True
return self
# 统一收口约束：确保不同方法都遵守单调性/最小箱/最大箱限制
self._apply_post_fit_constraints(
X,
y,
enforce_monotonic=self.method != "monotonic",
enforce_bad_rate=self.method not in ("uniform", "quantile"),
)
self._restrict_to_prebin_candidates(
X,
y,
{feature: self._prebinner.splits_.get(feature, np.array([])) for feature in X.columns},
)
else:
# 使用指定方法
self._fit_with_method(X, y)
# quantile 方法需保持分位数切分点精确，跳过所有后处理
if self.method == "quantile":
if not self._reserved_bins_are_finalized(X.columns):
self._finalize_reserved_bins(X, y)
self._is_fitted = True
return self
# 统一后处理：围绕头尾Lift与样本稳定性微调切分点
# 默认开启，可通过 lift_refine=False 关闭
if self._fit_control_options.get("lift_refine", True) and self.method != "uniform":
self._refine_splits_for_lift_stability(X, y)
# 统一收口约束：确保不同方法都遵守单调性/最小箱/最大箱限制
self._apply_post_fit_constraints(
X,
y,
enforce_monotonic=self.method != "monotonic",
enforce_bad_rate=self.method not in ("uniform", "quantile"),
)
self._finalize_reserved_bins(X, y)
self._is_fitted = True
return self
def _fit_imported_and_ordinary_features(self, X: pd.DataFrame, y: pd.Series, imported_features: List[str]) -> None:
"""按特征分派导入规则与普通算法，并按输入列顺序合并最终状态。"""
imported_set = set(imported_features)
ordinary_features = [feature for feature in X.columns if feature not in imported_set]
# 导入规则是不可再优化的固定输入，只补充本轮样本统计。
self._update_bin_stats(X[imported_features], y)
ordinary_binner = None
if ordinary_features:
ordinary_X = X[ordinary_features]
# 使用同一估计器配置创建干净事务候选；ordinary 子集沿用完整并行预算，
# 并完整经过现有工厂、预分箱、类别适配和后处理链路。
ordinary_binner = self._make_fit_transaction_candidate((ordinary_X, y), {})
ordinary_binner.fit(ordinary_X, y)
for state_name in self._FEATURE_DICT_STATE:
imported_state = getattr(self, state_name, {})
ordinary_state = getattr(ordinary_binner, state_name, {}) if ordinary_binner is not None else {}
merged = {}
for feature in X.columns:
source = imported_state if feature in imported_set else ordinary_state
if feature in source:
merged[feature] = deepcopy(source[feature])
has_state = hasattr(self, state_name) or (
ordinary_binner is not None and hasattr(ordinary_binner, state_name)
)
if has_state or merged:
setattr(self, state_name, merged)
for state_name in self._FEATURE_SET_STATE:
imported_state = getattr(self, state_name, set())
ordinary_state = getattr(ordinary_binner, state_name, set()) if ordinary_binner is not None else set()
setattr(
self,
state_name,
{
feature
for feature in X.columns
if feature in (imported_state if feature in imported_set else ordinary_state)
},
)
imported_woe = getattr(self, "_woe_maps_", {})
ordinary_woe = getattr(ordinary_binner, "_woe_maps_", {}) if ordinary_binner is not None else {}
self._woe_maps_ = {
feature: deepcopy((imported_woe if feature in imported_set else ordinary_woe)[feature])
for feature in X.columns
if feature in (imported_woe if feature in imported_set else ordinary_woe)
}
# 只保留本轮 ordinary 子模型及其 delegate；transform 按特征分区调用，
# imported 特征不会进入这个只覆盖 ordinary 子集的估计器。
self._ordinary_binner_ = ordinary_binner
self._ordinary_features_ = tuple(ordinary_features)
self._binner = getattr(ordinary_binner, "_binner", None)
self._prebinner = getattr(ordinary_binner, "_prebinner", None)
def _update_bin_stats(self, X: pd.DataFrame, y: pd.Series):
"""更新分箱统计信息（用于已导入规则的情况）.
:param X: 特征数据
:param y: 目标变量
"""
for feature in self.splits_.keys():
if feature not in X.columns:
continue
# 获取特征类型
feature_type = self.feature_types_.get(feature, "numerical")
# 获取切分点
splits = self.splits_[feature]
# 对于类别型变量，优先使用_cat_bins_
if feature_type == "categorical" and feature in self._cat_bins_:
bins = self._apply_bins(X[feature], self._cat_bins_[feature], feature_type, feature)
else:
bins = self._apply_bins(X[feature], splits, feature_type, feature)
# 计算分箱统计
self.bin_tables_[feature] = self._compute_bin_stats(feature, X[feature], y, bins)
if feature_type == "categorical":
self._validate_categorical_constraints(feature, y)
def _fit_with_user_splits(self, X: pd.DataFrame, y: pd.Series):
"""分派用户规则字段与普通字段，并按输入顺序合并状态。"""
if callable(self.user_splits):
user_features = list(X.columns)
elif isinstance(self.user_splits, dict):
user_features = [feature for feature in X.columns if self._has_explicit_user_rule(feature)]
else:
raise ValueError("user_splits 必须是字段规则字典、可调用对象或 None")
user_set = set(user_features)
ordinary_features = [feature for feature in X.columns if feature not in user_set]
if user_features:
user_X = X[user_features]
try:
self._fit_features(user_X, y, "_fit_user_split_feature")
except ParallelExecutionError as exc:
# user_splits 历史上直接暴露规则校验异常；并行化不能把
# ValueError/KeyError 等公共异常永久包装为并行内部异常。
if exc.__cause__ is not None:
raise exc.__cause__
raise
if self.method != "quantile":
if self._fit_control_options.get("lift_refine", True) and self.method != "uniform":
self._refine_splits_for_lift_stability(user_X, y)
self._apply_post_fit_constraints(
user_X,
y,
enforce_monotonic=self.method != "monotonic",
enforce_bad_rate=self.method not in ("uniform", "quantile"),
)
ordinary_binner = None
if ordinary_features:
ordinary_X = X[ordinary_features]
ordinary_binner = self._make_fit_transaction_candidate((ordinary_X, y), {})
# 关键边界：普通字段不得再次进入 user_splits 分支，而应完整使用
# 当前配置的 method/prebinning 及其逐特征并行实现。
ordinary_binner.user_splits = None
ordinary_binner.user_splits_fixed = None
ordinary_binner.fit(ordinary_X, y)
self._merge_user_and_ordinary_feature_states(
X,
user_features=user_features,
ordinary_features=ordinary_features,
ordinary_binner=ordinary_binner,
)
def _fit_user_split_feature(self, feature: str, x: pd.Series, y: pd.Series) -> None:
"""在隔离 worker 中直接应用单个字段的用户规则并计算统计。"""
self._fit_common_user_split_feature(feature, x, y)
def _merge_user_and_ordinary_feature_states(
self,
X: pd.DataFrame,
*,
user_features: List[str],
ordinary_features: List[str],
ordinary_binner: Optional["OptimalBinning"],
) -> None:
"""确定性合并规则 worker 与普通 method 子模型的逐特征状态。"""
user_set = set(user_features)
for state_name in self._FEATURE_DICT_STATE:
user_state = getattr(self, state_name, {})
ordinary_state = getattr(ordinary_binner, state_name, {}) if ordinary_binner is not None else {}
merged = {}
for feature in X.columns:
source = user_state if feature in user_set else ordinary_state
if feature in source:
# worker/普通子模型已完成事务隔离；提交阶段与直接 method
# 拟合一样共享只读逐特征产物，避免高维场景复制大量分箱表。
merged[feature] = source[feature]
if hasattr(self, state_name) or ordinary_binner is not None or merged:
setattr(self, state_name, merged)
for state_name in self._FEATURE_SET_STATE:
user_state = getattr(self, state_name, set())
ordinary_state = getattr(ordinary_binner, state_name, set()) if ordinary_binner is not None else set()
setattr(
self,
state_name,
{
feature
for feature in X.columns
if feature in (user_state if feature in user_set else ordinary_state)
},
)
user_woe = getattr(self, "_woe_maps_", {})
ordinary_woe = getattr(ordinary_binner, "_woe_maps_", {}) if ordinary_binner is not None else {}
self._woe_maps_ = {
feature: (user_woe if feature in user_set else ordinary_woe)[feature]
for feature in X.columns
if feature in (user_woe if feature in user_set else ordinary_woe)
}
self._ordinary_binner_ = ordinary_binner
self._ordinary_features_ = tuple(ordinary_features)
self._binner = getattr(ordinary_binner, "_binner", None)
self._prebinner = getattr(ordinary_binner, "_prebinner", None)
def _merge_user_category_groups_with_method(
self,
feature: str,
x: pd.Series,
y: pd.Series,
groups: List[List],
) -> List[List]:
"""把用户类别箱作为原子单位，使用当前方法决定整箱合并边界。"""
group_codes = assign_category_groups(
feature,
x,
groups,
special_codes=self.special_codes,
missing_separate=self.missing_separate,
handle_unknown=UNKNOWN_BIN,
)
encoded = pd.Series(group_codes, index=x.index, name=feature, dtype=float)
encoded.loc[encoded < 0] = np.nan
method_binner = OptimalBinning(
target=self.target,
method=self.method,
max_n_bins=min(self.max_n_bins, len(groups)),
min_n_bins=min(self.min_n_bins, len(groups)),
min_bin_size=self.min_bin_size,
max_bin_size=self.max_bin_size,
min_bad_rate=self.min_bad_rate,
monotonic=self.monotonic,
missing_separate=False,
special_codes=None,
cat_cutoff=None,
random_state=self.random_state,
verbose=False,
decimal=self.decimal,
woe_clip=self.woe_clip,
handle_unknown=UNKNOWN_BIN,
n_jobs=self.n_jobs,
parallel_backend=self.parallel_backend,
parallel_config=self.parallel_config,
**self.kwargs,
).fit(encoded.to_frame(), y)
numeric_splits = method_binner.splits_.get(feature, np.array([]))
atomic_groups = restore_category_groups(list(range(len(groups))), numeric_splits)
return [
[value for group_index in atomic_group for value in groups[group_index]] for atomic_group in atomic_groups
]
def _merge_common_user_category_groups_with_method(
self,
feature: str,
x: pd.Series,
y: pd.Series,
groups: List[List],
) -> List[List]:
"""OptimalBinning 使用所选主方法合并可变的用户类别原子组。"""
return self._merge_user_category_groups_with_method(feature, x, y, groups)
def _get_prebinning_params(self, override_dict: Optional[Dict] = None) -> Dict:
"""获取预分箱参数.
:param override_dict: 覆盖默认参数的字典
:return: 预分箱器参数字典
"""
base_params = {
"target": self.target,
"max_n_bins": 20, # 默认预分箱为20箱
"min_n_bins": 2,
"min_bin_size": self.min_bin_size,
"max_bin_size": self.max_bin_size,
"min_bad_rate": self.min_bad_rate,
"monotonic": self.monotonic,
"missing_separate": self.missing_separate,
"special_codes": self.special_codes,
"cat_cutoff": self.cat_cutoff,
"random_state": self.random_state,
"verbose": False, # 预分箱默认不输出详细信息
"decimal": self.decimal,
"woe_clip": self.woe_clip,
"category_order": self.category_order,
"handle_unknown": self.handle_unknown,
"n_jobs": self.n_jobs,
"parallel_backend": self.parallel_backend,
"parallel_config": self.parallel_config,
}
if self.prebinning_params is not None:
if not isinstance(self.prebinning_params, dict):
raise ValueError("prebinning_params 必须是参数字典或 None")
base_params.update({key: value for key, value in self.prebinning_params.items() if key != "method"})
if override_dict:
base_params.update({key: value for key, value in override_dict.items() if key != "method"})
return base_params
def _fit_with_prebinning(self, X: pd.DataFrame, y: pd.Series):
"""使用预分箱进行分箱.
先使用预分箱方法生成初始切分点，再使用主方法进行优化。
参考optbinning的实现，如MDLP分箱前先用CART预分箱。
支持的预分箱方法：所有 VALID_METHODS 中的方法都可以作为预分箱方法。
"""
# 1. 创建预分箱器
if isinstance(self.prebinning, BaseBinning):
# 传入的是分箱器实例
self._prebinner = self.prebinning
elif isinstance(self.prebinning, str):
# 传入的是方法名
pre_params = self._get_prebinning_params()
self._prebinner = OptimalBinning(method=self.prebinning, **pre_params)
elif isinstance(self.prebinning, dict):
# 传入的是配置字典
pre_method = self.prebinning.get("method", "cart")
pre_params = self._get_prebinning_params(self.prebinning)
self._prebinner = OptimalBinning(method=pre_method, **pre_params)
else:
raise ValueError(f"不支持的prebinning类型: {type(self.prebinning)}")
# 2. 执行预分箱
if self.verbose:
logger.info(f"执行预分箱: {getattr(self._prebinner, 'method', self._prebinner.__class__.__name__)}")
self._prebinner.fit(X, y)
# 3. 获取预分箱的切分点作为初始切分点
pre_splits = {}
for feature in X.columns:
if feature in self._prebinner.splits_:
pre_splits[feature] = self._prebinner.splits_[feature]
# 4. 所有主方法都真实拟合，使方法专属参数完整生效；随后把结果约束到
# 预分箱候选边界。预分箱只定义搜索空间，不能替代主方法结果。
self._fit_with_method(X, y)
self._restrict_to_prebin_candidates(X, y, pre_splits)
def _fit_with_method_and_prebins(self, X: pd.DataFrame, y: pd.Series, pre_splits: Dict[str, np.ndarray]):
"""使用预分箱切分点进行优化分箱."""
for feature in X.columns:
feature_type = self._detect_feature_type(X[feature])
self.feature_types_[feature] = feature_type
if feature_type == "categorical":
# 预分类别组是不可拆分原子单元，仍由主方法决定是否合并。
if feature in self._prebinner.splits_:
groups = self._prebinner._cat_bins_.get(feature, self._prebinner.splits_[feature])
groups = self._merge_user_category_groups_with_method(feature, X[feature], y, groups)
self._cat_bins_[feature] = groups
self.splits_[feature] = groups
self.n_bins_[feature] = len(groups)
self._category_orders_[feature] = [value for group in groups for value in group if pd.notna(value)]
self._category_code_maps_[feature] = [
(value, index) for index, value in enumerate(self._category_orders_[feature])
]
self._categorical_numeric_splits_[feature] = (
np.cumsum([len(group) for group in groups], dtype=float)[:-1] - 0.5
)
bins = self._assign_categorical_bins(feature, X[feature])
self.bin_tables_[feature] = self._compute_bin_stats(feature, X[feature], y, bins)
continue
# 数值型特征：在预分箱切分点基础上优化
initial_splits = pre_splits.get(feature, np.array([]))
# 兼容异常格式（如 tuple/list 嵌套），统一清洗为有序数值切分点
if len(initial_splits) > 0:
cleaned = []
for v in list(initial_splits):
if isinstance(v, (list, tuple, np.ndarray)):
for t in v:
if pd.notna(t):
try:
tv = float(t)
if np.isfinite(tv):
cleaned.append(tv)
except Exception:
pass
else:
if pd.notna(v):
try:
vv = float(v)
if np.isfinite(vv):
cleaned.append(vv)
except Exception:
pass
initial_splits = np.unique(np.sort(np.array(cleaned, dtype=float))) if cleaned else np.array([])
if len(initial_splits) == 0:
# 没有预分箱切分点，使用默认方法
self._fit_single_feature(X[[feature]], y, feature)
continue
# 使用预分箱切分点生成初始分箱
x_clean = X[feature].dropna()
y_clean = y[x_clean.index]
# 根据主方法进行优化
if self.method == "best_iv":
optimized_splits = self._optimize_iv_splits(x_clean, y_clean, initial_splits)
elif self.method == "best_ks":
optimized_splits = self._optimize_ks_splits(x_clean, y_clean, initial_splits)
elif self.method == "chi":
optimized_splits = self._optimize_chi_merge_splits(x_clean, y_clean, initial_splits)
elif self.method == "mdlp":
# MDLP: 如果预分箱超出限制，使用IV优化进行合并
optimized_splits = self._optimize_mdlp_splits(x_clean, y_clean, initial_splits)
else:
optimized_splits = initial_splits
self.splits_[feature] = self._round_splits(optimized_splits)
self.n_bins_[feature] = len(self.splits_[feature]) + 1
# 计算最终分箱统计
final_bins = self._apply_bins(X[feature], self.splits_[feature], feature_type)
self.bin_tables_[feature] = self._compute_bin_stats(feature, X[feature], y, final_bins)
def _restrict_to_prebin_candidates(
self,
X: pd.DataFrame,
y: pd.Series,
pre_splits: Dict[str, np.ndarray],
) -> None:
"""把主方法结果约束到预分箱候选边界，避免预分箱退化为结果复制。"""
for feature in X.columns:
if self.feature_types_.get(feature) == "categorical":
groups = self._prebinner._cat_bins_.get(feature)
if groups:
groups = self._merge_user_category_groups_with_method(feature, X[feature], y, groups)
self._cat_bins_[feature] = groups
self.splits_[feature] = groups
self.n_bins_[feature] = len(groups)
self._category_orders_[feature] = [value for group in groups for value in group if pd.notna(value)]
self._category_code_maps_[feature] = [
(value, index) for index, value in enumerate(self._category_orders_[feature])
]
self._categorical_numeric_splits_[feature] = (
np.cumsum([len(group) for group in groups], dtype=float)[:-1] - 0.5
)
bins = self._assign_categorical_bins(feature, X[feature])
self.bin_tables_[feature] = self._compute_bin_stats(feature, X[feature], y, bins)
continue
candidates = np.asarray(pre_splits.get(feature, np.array([])), dtype=float)
candidates = np.unique(np.sort(candidates[np.isfinite(candidates)]))
if len(candidates) == 0:
continue
proposed = np.asarray(self.splits_.get(feature, np.array([])), dtype=float)
proposed = proposed[np.isfinite(proposed)]
projected = (
np.unique([candidates[int(np.argmin(np.abs(candidates - split)))] for split in proposed])
if len(proposed)
else np.array([], dtype=float)
)
required = min(max(0, self.min_n_bins - 1), len(candidates), max(0, self.max_n_bins - 1))
if len(projected) < required:
remaining = [
value for value in candidates if not np.any(np.isclose(projected, value, rtol=0, atol=1e-12))
]
for value in remaining:
projected = np.unique(np.append(projected, value))
if len(projected) >= required:
break
if len(projected) > self.max_n_bins - 1:
projected = self._optimize_iv_splits(X[feature].dropna(), y.loc[X[feature].dropna().index], projected)
self.splits_[feature] = self._round_splits(projected)
self.n_bins_[feature] = len(projected) + 1
bins = self._apply_bins(X[feature], self.splits_[feature], "numerical", feature)
self.bin_tables_[feature] = self._compute_bin_stats(feature, X[feature], y, bins)
def _fit_single_feature(self, X: pd.DataFrame, y: pd.Series, feature: str):
"""对单个特征使用主方法分箱."""
temp_binner = OptimalBinning(
method=self.method,
max_n_bins=self.max_n_bins,
min_bin_size=self.min_bin_size,
random_state=self.random_state,
verbose=False,
decimal=self.decimal,
n_jobs=self.n_jobs,
parallel_backend=self.parallel_backend,
parallel_config=self.parallel_config,
)
temp_binner.fit(X, y)
self.splits_[feature] = temp_binner.splits_[feature]
self.n_bins_[feature] = temp_binner.n_bins_[feature]
self.bin_tables_[feature] = temp_binner.bin_tables_[feature]
def _copy_categorical_state_from(self, binner: BaseBinning) -> None:
"""复制预分箱器或底层分箱器的完整类别状态。"""
for attribute in (
"_cat_bins_",
"_category_orders_",
"_category_code_maps_",
"_categorical_numeric_splits_",
):
if hasattr(binner, attribute):
setattr(self, attribute, getattr(binner, attribute).copy())
def _copy_finalized_bin_state_from(self, binner: BaseBinning) -> None:
"""复制底层分箱器已经完成统一收口的逐特征状态。"""
self._copy_categorical_state_from(binner)
for attribute in (
"_user_splits_fixed_masks_",
"_user_missing_bin_targets_",
"_missing_bin_targets_",
"_woe_maps_",
"_recorded_bins_",
):
setattr(self, attribute, deepcopy(getattr(binner, attribute, {})))
self._reserved_bins_finalized_ = set(getattr(binner, "_reserved_bins_finalized_", set()))
def _reserved_bins_are_finalized(self, features: Iterable[Any]) -> bool:
"""判断当前字段是否都已有可直接复用的最终分箱产物。"""
finalized = getattr(self, "_reserved_bins_finalized_", set())
return all(
feature in finalized
and feature in self.bin_tables_
and feature in self._woe_maps_
and feature in self._recorded_bins_
for feature in features
)
def _optimize_iv_splits(self, x: pd.Series, y: pd.Series, initial_splits: np.ndarray) -> np.ndarray:
"""基于预分箱切分点优化IV."""
if len(initial_splits) <= self.max_n_bins - 1:
return initial_splits
# 计算每个预分箱的IV贡献
bins = np.digitize(x, initial_splits)
bin_stats = []
for bin_idx in range(len(initial_splits) + 1):
mask = bins == bin_idx
if mask.sum() == 0:
continue
y_bin = y[mask]
bad_rate = y_bin.mean()
bin_stats.append(
{
"bin": bin_idx,
"count": mask.sum(),
"bad_rate": bad_rate,
"split": initial_splits[bin_idx] if bin_idx < len(initial_splits) else None,
}
)
# 基于IV贡献合并相邻箱，直到满足max_n_bins
current_splits = list(initial_splits)
while len(current_splits) >= self.max_n_bins:
# 找到IV损失最小的合并方案
min_iv_loss = float("inf")
merge_idx = -1
for i in range(len(current_splits)):
temp_splits = current_splits[:i] + current_splits[i + 1 :]
iv_loss = self._calculate_iv_loss(x, y, current_splits, temp_splits)
if iv_loss < min_iv_loss:
min_iv_loss = iv_loss
merge_idx = i
if merge_idx >= 0:
current_splits.pop(merge_idx)
else:
break
return np.array(current_splits)
def _optimize_ks_splits(self, x: pd.Series, y: pd.Series, initial_splits: np.ndarray) -> np.ndarray:
"""基于预分箱切分点优化KS."""
if len(initial_splits) <= self.max_n_bins - 1:
return initial_splits
# 类似IV优化，但基于KS统计量
current_splits = list(initial_splits)
while len(current_splits) >= self.max_n_bins:
# 找到KS损失最小的合并方案
min_ks_loss = float("inf")
merge_idx = -1
for i in range(len(current_splits)):
temp_splits = current_splits[:i] + current_splits[i + 1 :]
ks_loss = self._calculate_ks_loss(x, y, current_splits, temp_splits)
if ks_loss < min_ks_loss:
min_ks_loss = ks_loss
merge_idx = i
if merge_idx >= 0:
current_splits.pop(merge_idx)
else:
break
return np.array(current_splits)
def _optimize_chi_merge_splits(self, x: pd.Series, y: pd.Series, initial_splits: np.ndarray) -> np.ndarray:
"""基于预分箱切分点进行卡方合并."""
if len(initial_splits) <= self.max_n_bins - 1:
return initial_splits
current_splits = list(initial_splits)
while len(current_splits) >= self.max_n_bins:
min_chi2 = float("inf")
merge_idx = -1
for i in range(len(current_splits)):
temp_splits = current_splits[:i] + current_splits[i + 1 :]
chi2 = self._calculate_chi2_for_splits(x, y, temp_splits)
if chi2 < min_chi2:
min_chi2 = chi2
merge_idx = i
if merge_idx >= 0 and min_chi2 < 3.841: # 卡方阈值
current_splits.pop(merge_idx)
else:
break
return np.array(current_splits)
def _optimize_mdlp_splits(self, x: pd.Series, y: pd.Series, initial_splits: np.ndarray) -> np.ndarray:
"""基于预分箱切分点优化MDLP分箱.
MDLP算法本身会根据信息增益自动确定分箱数。
当使用预分箱时，如果预分箱的分箱数超过max_n_bins，
使用坏样本率差异最小的策略合并相邻分箱。
:param x: 特征数据
:param y: 目标变量
:param initial_splits: 预分箱切分点
:return: 优化后的切分点
"""
if len(initial_splits) <= self.max_n_bins - 1:
return initial_splits
# 使用坏样本率差异作为合并标准
current_splits = list(initial_splits)
while len(current_splits) >= self.max_n_bins:
# 计算每个分箱的统计信息
bins = np.digitize(x, current_splits)
bin_stats = []
for bin_idx in range(len(current_splits) + 1):
mask = bins == bin_idx
if mask.sum() > 0:
y_bin = y[mask]
bin_stats.append({"bin": bin_idx, "count": mask.sum(), "bad_rate": y_bin.mean()})
# 找到坏样本率差异最小的相邻分箱
min_diff = float("inf")
merge_idx = -1
for i in range(len(bin_stats) - 1):
diff = abs(bin_stats[i]["bad_rate"] - bin_stats[i + 1]["bad_rate"])
if diff < min_diff:
min_diff = diff
merge_idx = i
# 合并选中的切分点
if merge_idx >= 0:
current_splits.pop(merge_idx)
else:
break
return np.array(current_splits)
def _calculate_iv_loss(self, x: pd.Series, y: pd.Series, splits_before: List, splits_after: List) -> float:
"""计算合并前后的IV损失.
IV损失 = 合并前IV - 合并后IV
值越小表示合并带来的信息损失越小。
使用 metrics.binning_metrics.compare_splits_iv 方法。
"""
result = compare_splits_iv(
x.values if isinstance(x, pd.Series) else x,
y.values if isinstance(y, pd.Series) else y,
np.array(splits_before),
np.array(splits_after),
)
if isinstance(result, (tuple, list)):
return float(result[0])
return float(result)
def _calculate_ks_loss(self, x: pd.Series, y: pd.Series, splits_before: List, splits_after: List) -> float:
"""计算合并前后的KS损失.
KS损失 = 合并前KS - 合并后KS
值越小表示合并带来的区分度损失越小。
使用 metrics.binning_metrics.compare_splits_ks 方法。
"""
result = compare_splits_ks(
x.values if isinstance(x, pd.Series) else x,
y.values if isinstance(y, pd.Series) else y,
np.array(splits_before),
np.array(splits_after),
)
if isinstance(result, (tuple, list)):
return float(result[0])
return float(result)
def _calculate_total_iv(self, x: pd.Series, y: pd.Series, splits: List) -> float:
"""计算给定切分点的总IV值.
使用 metrics.binning_metrics.iv_for_splits 方法。
"""
if len(splits) == 0:
return 0.0
return iv_for_splits(
x.values if isinstance(x, pd.Series) else x, y.values if isinstance(y, pd.Series) else y, np.array(splits)
)
def _calculate_max_ks(self, x: pd.Series, y: pd.Series, splits: List) -> float:
"""计算给定切分点的最大KS值.
使用 metrics.binning_metrics.ks_for_splits 方法。
"""
if len(splits) == 0:
return 0.0
return ks_for_splits(
x.values if isinstance(x, pd.Series) else x, y.values if isinstance(y, pd.Series) else y, np.array(splits)
)
def _calculate_chi2_for_splits(self, x: pd.Series, y: pd.Series, splits: List) -> float:
"""计算给定切分点的卡方统计量."""
if len(splits) == 0:
return 0.0
bins = np.digitize(x, splits)
contingency = pd.crosstab(bins, y).values
if contingency.shape[0] < 2:
return 0.0
row_sum = contingency.sum(axis=1)
col_sum = contingency.sum(axis=0)
total = contingency.sum()
if total == 0:
return 0.0
expected = np.outer(row_sum, col_sum) / total
expected = np.maximum(expected, 1e-10)
chi2 = ((contingency - expected) ** 2 / expected).sum()
return chi2
def _evaluate_lift_stability_score(
self,
x: pd.Series,
y: pd.Series,
splits: np.ndarray,
min_samples: int,
focus_weight: float,
sample_weight: float,
) -> float:
"""评估切分点综合分数（越大越好）：头尾Lift + 稳定性 + 单调性倾向."""
if len(splits) == 0:
return -np.inf
bins = np.digitize(x, splits)
n_bins = len(splits) + 1
total = len(y)
total_bad = float(np.sum(y))
if total == 0 or total_bad <= 0 or total_bad >= total:
return -np.inf
counts = np.bincount(bins, minlength=n_bins).astype(float)
bad_counts = np.bincount(bins, weights=y, minlength=n_bins).astype(float)
if np.any(counts < min_samples):
return -np.inf
bad_rates = bad_counts / np.maximum(counts, 1.0)
overall_bad_rate = total_bad / total
lifts = bad_rates / np.maximum(overall_bad_rate, 1e-10)
# 1) 头尾Lift：优先看两端箱体（按数值顺序）
edge_lift_left = float(lifts[0])
edge_lift_right = float(lifts[-1])
edge_strength = abs(edge_lift_right - edge_lift_left)
edge_extreme = max(
max(0.0, edge_lift_left - 1.0),
max(0.0, 1.0 - edge_lift_left),
max(0.0, edge_lift_right - 1.0),
max(0.0, 1.0 - edge_lift_right),
)
# 2) 全局头尾（最大/最小Lift）
max_lift = float(np.max(lifts))
min_lift = float(np.min(lifts))
global_tail_strength = max(0.0, max_lift - 1.0) + max(0.0, 1.0 - min_lift)
tail_strength = 0.65 * edge_strength + 0.2 * edge_extreme + 0.15 * global_tail_strength
# 3) 样本稳定性（极端箱不能太小）
top_idx = int(np.argmax(lifts))
bottom_idx = int(np.argmin(lifts))
top_ratio = counts[top_idx] / total
bottom_ratio = counts[bottom_idx] / total
min_ratio = float(np.min(counts) / total)
stability = np.log1p(top_ratio * 100.0) + np.log1p(bottom_ratio * 100.0) + np.log1p(min_ratio * 100.0)
# 4) 单调性倾向：减少拐点，尤其在auto_asc_desc下
diffs = np.diff(bad_rates)
signs = np.sign(diffs)
non_zero = signs[signs != 0]
sign_changes = 0 if len(non_zero) <= 1 else int(np.sum(non_zero[1:] * non_zero[:-1] < 0))
monotonic_bonus_weight = float(self._fit_control_options.get("monotonic_bonus_weight", 0.4))
monotonic_bonus = -float(sign_changes)
if self.monotonic in ["ascending", "descending", "auto_asc_desc"]:
monotonic_bonus *= 1.5
iv_value = iv_for_splits(
x.values if isinstance(x, pd.Series) else x, y.values if isinstance(y, pd.Series) else y, np.array(splits)
)
n_bins_score = np.log1p(max(0, len(splits)))
return (
focus_weight * tail_strength
+ sample_weight * stability
+ monotonic_bonus_weight * monotonic_bonus
+ 0.08 * iv_value
+ 0.25 * n_bins_score
)
def _refine_splits_for_lift_stability(self, X: pd.DataFrame, y: pd.Series):
"""对已有分箱结果做局部搜索：先删点，再补点，兼顾头尾Lift与单调性."""
methods_to_refine = {
"uniform",
"quantile",
"tree",
"chi",
"best_ks",
"best_iv",
"mdlp",
"cart",
"kmeans",
"genetic",
"smooth",
"kernel_density",
"best_lift",
"target_bad_rate",
}
if self.method not in methods_to_refine:
return
min_samples_abs = self._get_min_samples(len(y))
focus_weight = float(self._fit_control_options.get("lift_focus_weight", 3.0))
sample_weight = float(self._fit_control_options.get("sample_stability_weight", 0.2))
max_search_bins = int(self._fit_control_options.get("lift_refine_max_bins", self.max_n_bins))
strict_mono = self.monotonic in ["ascending", "descending", "auto_asc_desc"]
def _is_ok_monotonic(xv: pd.Series, yv: pd.Series, sp: np.ndarray) -> bool:
if not strict_mono:
return True
b = np.digitize(xv, sp)
cnt = np.bincount(b, minlength=len(sp) + 1).astype(float)
bad = np.bincount(b, weights=yv, minlength=len(sp) + 1).astype(float)
br = bad / np.maximum(cnt, 1.0)
d = np.diff(br)
return bool(np.all(d >= -1e-10) or np.all(d <= 1e-10))
for feature, splits in list(self.splits_.items()):
if self.feature_types_.get(feature) != "numerical":
continue
splits_arr = np.array(splits, dtype=float) if len(splits) > 0 else np.array([])
if len(splits_arr) == 0:
continue
x = pd.to_numeric(X[feature], errors="coerce")
valid_mask = x.notna()
if self.special_codes:
for code in self.special_codes:
valid_mask &= x != code
x_valid = x[valid_mask]
y_valid = y[valid_mask]
if len(x_valid) < max(min_samples_abs * self.min_n_bins, 50):
continue
current = np.unique(np.sort(splits_arr))
best = current.copy()
best_score = self._evaluate_lift_stability_score(
x_valid, y_valid, best, min_samples_abs, focus_weight, sample_weight
)
# Step1: 删点搜索（提升鲁棒性）
improved = True
while improved and len(best) > 1 and len(best) >= self.min_n_bins:
improved = False
candidate_best = None
candidate_score = best_score
for i in range(len(best)):
cand = np.delete(best, i)
if len(cand) < max(1, self.min_n_bins - 1):
continue
if not _is_ok_monotonic(x_valid, y_valid, cand):
continue
score = self._evaluate_lift_stability_score(
x_valid, y_valid, cand, min_samples_abs, focus_weight, sample_weight
)
if score > candidate_score:
candidate_score = score
candidate_best = cand
if candidate_best is not None:
best = candidate_best
best_score = candidate_score
improved = True
# Step2: 补点搜索（避免过度合并，强化头尾区分）
max_splits_allowed = max(1, max_search_bins - 1)
pool = np.unique(np.quantile(x_valid, np.linspace(0.05, 0.95, 19)))
pool = np.array([v for v in pool if np.isfinite(v)], dtype=float)
improved = True
while improved and len(best) < max_splits_allowed:
improved = False
candidate_best = None
candidate_score = best_score
for c in pool:
if np.any(np.isclose(best, c, atol=1e-10, rtol=0)):
continue
cand = np.unique(np.sort(np.append(best, c)))
if len(cand) > max_splits_allowed:
continue
if not _is_ok_monotonic(x_valid, y_valid, cand):
continue
score = self._evaluate_lift_stability_score(
x_valid, y_valid, cand, min_samples_abs, focus_weight, sample_weight
)
if score > candidate_score:
candidate_score = score
candidate_best = cand
if candidate_best is not None:
best = candidate_best
best_score = candidate_score
improved = True
if len(best) > 0 and not np.array_equal(best, current):
self.splits_[feature] = self._round_splits(best)
self.n_bins_[feature] = len(self.splits_[feature]) + 1
bins = self._apply_bins(X[feature], self.splits_[feature], "numerical", feature)
self.bin_tables_[feature] = self._compute_bin_stats(feature, X[feature], y, bins)
def _fit_with_method(self, X: pd.DataFrame, y: pd.Series):
"""使用指定方法进行分箱.
注意：各个独立的分箱模块（如BestIVBinning、MDLPBinning等）
保持简单，只执行一次分箱。预分箱功能是OptimalBinning特有的，
通过prebinning参数在OptimalBinning层面实现。
"""
# 基础参数（适用于大多数方法）- 过滤掉不相关的参数
base_params = {
"max_n_bins": self.max_n_bins,
"min_n_bins": self.min_n_bins,
"min_bin_size": self.min_bin_size,
"max_bin_size": self.max_bin_size,
"min_bad_rate": self.min_bad_rate,
"monotonic": self.monotonic,
"missing_separate": self.missing_separate,
"special_codes": self.special_codes,
"cat_cutoff": self.cat_cutoff,
"random_state": self.random_state,
"verbose": self.verbose,
"decimal": self.decimal,
"category_order": self.category_order,
"handle_unknown": self.handle_unknown,
"n_jobs": self.n_jobs,
"parallel_backend": self.parallel_backend,
"parallel_config": self.parallel_config,
}
base_params.update(self.kwargs)
# 需要 target 参数的方法
target_params = {"target": self.target, **base_params}
# 需要 cat_cutoff 参数的方法
full_params = {
"target": self.target,
"max_n_bins": self.max_n_bins,
"min_n_bins": self.min_n_bins,
"min_bin_size": self.min_bin_size,
"max_bin_size": self.max_bin_size,
"min_bad_rate": self.min_bad_rate,
"monotonic": self.monotonic,
"missing_separate": self.missing_separate,
"special_codes": self.special_codes,
"cat_cutoff": self.cat_cutoff,
"random_state": self.random_state,
"verbose": self.verbose,
"decimal": self.decimal,
"woe_clip": self.woe_clip,
"category_order": self.category_order,
"handle_unknown": self.handle_unknown,
"n_jobs": self.n_jobs,
"parallel_backend": self.parallel_backend,
"parallel_config": self.parallel_config,
}
full_params.update(self.kwargs)
if self.method == "uniform":
uniform_params = target_params.copy()
uniform_params.setdefault("force_numerical", False)
uniform_params["category_order"] = self.category_order
uniform_params["handle_unknown"] = self.handle_unknown
self._binner = UniformBinning(**uniform_params)
elif self.method == "quantile":
quantile_params = target_params.copy()
quantile_params.setdefault("force_numerical", False)
self._binner = QuantileBinning(**quantile_params)
elif self.method == "tree":
self._binner = TreeBinning(**target_params)
elif self.method == "chi":
self._binner = ChiMergeBinning(**target_params)
elif self.method == "best_ks":
self._binner = BestKSBinning(**full_params)
elif self.method == "best_iv":
self._binner = BestIVBinning(**full_params)
elif self.method == "mdlp":
self._binner = MDLPBinning(**target_params)
elif self.method == "or_tools":
if not ORTOOLS_AVAILABLE:
raise ImportError("OR-Tools 未安装，无法使用 or_tools 方法。" "请使用 pip install ortools 安装。")
or_params = full_params.copy()
or_params.setdefault("objective", "iv")
or_params.setdefault("time_limit", 30)
self._binner = ORBinning(**or_params)
elif self.method == "cp_sat":
if not ORTOOLS_AVAILABLE:
raise ImportError("OR-Tools 未安装，无法使用 cp_sat 方法。" "请使用 pip install ortools 安装。")
cp_sat_params = full_params.copy()
cp_sat_params.setdefault("objective", "iv")
cp_sat_params.setdefault("time_limit", 30)
cp_sat_params.setdefault("n_prebins", 50)
self._binner = CPSATBinning(**cp_sat_params)
elif self.method == "cart":
self._binner = CartBinning(**full_params)
elif self.method == "kmeans":
kmeans_params = {
"max_n_bins": self.max_n_bins,
"min_n_bins": self.min_n_bins,
"min_bin_size": self.min_bin_size,
"max_bin_size": self.max_bin_size,
"min_bad_rate": self.min_bad_rate,
"monotonic": self.monotonic,
"missing_separate": self.missing_separate,
"special_codes": self.special_codes,
"cat_cutoff": self.cat_cutoff,
"random_state": self.random_state,
"verbose": self.verbose,
"decimal": self.decimal,
"force_numerical": False,
"category_order": self.category_order,
"handle_unknown": self.handle_unknown,
"n_jobs": self.n_jobs,
"parallel_backend": self.parallel_backend,
"parallel_config": self.parallel_config,
}
kmeans_params.update(self.kwargs)
self._binner = KMeansBinning(**kmeans_params)
elif self.method == "monotonic":
mono_params = {
"monotonic": self.monotonic if self.monotonic else "auto",
"max_n_bins": self.max_n_bins,
"min_n_bins": self.min_n_bins,
"min_bin_size": self.min_bin_size,
"max_bin_size": self.max_bin_size,
"min_bad_rate": self.min_bad_rate,
"special_codes": self.special_codes,
"missing_separate": self.missing_separate,
"cat_cutoff": self.cat_cutoff,
"random_state": self.random_state,
"verbose": self.verbose,
"decimal": self.decimal,
"category_order": self.category_order,
"handle_unknown": self.handle_unknown,
"n_jobs": self.n_jobs,
"parallel_backend": self.parallel_backend,
"parallel_config": self.parallel_config,
}
mono_params.update(self.kwargs)
self._binner = MonotonicBinning(**mono_params)
elif self.method == "genetic":
self._binner = GeneticBinning(**base_params)
elif self.method == "smooth":
self._binner = SmoothBinning(**base_params)
elif self.method == "kernel_density":
kernel_params = {
"target": self.target,
"max_n_bins": self.max_n_bins,
"min_n_bins": self.min_n_bins,
"min_bin_size": self.min_bin_size,
"max_bin_size": self.max_bin_size,
"min_bad_rate": self.min_bad_rate,
"monotonic": self.monotonic,
"special_codes": self.special_codes,
"missing_separate": self.missing_separate,
"cat_cutoff": self.cat_cutoff,
"random_state": self.random_state,
"verbose": self.verbose,
"decimal": self.decimal,
"category_order": self.category_order,
"handle_unknown": self.handle_unknown,
"n_jobs": self.n_jobs,
"parallel_backend": self.parallel_backend,
"parallel_config": self.parallel_config,
}
kernel_params.update(self.kwargs)
self._binner = KernelDensityBinning(**kernel_params)
elif self.method == "best_lift":
self._binner = BestLiftBinning(**target_params)
elif self.method == "target_bad_rate":
self._binner = TargetBadRateBinning(**base_params)
else:
# fit 入口已统一校验，此处为兜底防御（如子类改写校验逻辑）
raise ValueError(f"不支持的method: {self.method}，可选: {self.VALID_METHODS}")
# woe_clip 属于统一统计契约；部分直接分箱器未把它重复暴露在构造签名中。
self._binner.woe_clip = self.woe_clip
self._binner.fit(X, y)
# 复制属性
self.splits_ = self._binner.splits_
self.n_bins_ = self._binner.n_bins_
self.bin_tables_ = self._binner.bin_tables_
self.feature_types_ = self._binner.feature_types_
if self.method == "quantile":
self._copy_finalized_bin_state_from(self._binner)
else:
self._copy_categorical_state_from(self._binner)
if hasattr(self._binner, "ks_stats_"):
self.ks_stats_ = self._binner.ks_stats_
if hasattr(self._binner, "iv_stats_"):
self.iv_stats_ = self._binner.iv_stats_
if hasattr(self._binner, "monotonic_trend_"):
self.monotonic_trend_ = self._binner.monotonic_trend_
def _regroup_categorical_features(self, X: pd.DataFrame, y: pd.Series) -> None:
"""对所有类别型特征重新按坏样本率分组（List[List]），统一受 max_n_bins/min_bin_size 约束。
各底层分箱方法对类别型特征的处理并不一致（如 best_iv/best_ks 退化为数值编码标签、
mdlp 不合并类别），此处在 OptimalBinning 层统一重算，保证：
每个类别归入某一组、组数不超过 ``max_n_bins``、小样本/退化类别并入相邻组、
缺失值按 ``missing_separate`` 单独成箱。
"""
for feature in list(self.feature_types_.keys()):
if self.feature_types_.get(feature) != "categorical":
continue
if feature not in X.columns:
continue
groups = self._group_categories_by_badrate(X[feature], y)
if not groups:
continue
self._cat_bins_[feature] = groups
self.splits_[feature] = groups
self.n_bins_[feature] = len(groups)
bins = self._apply_bins(X[feature], groups, "categorical", feature)
self.bin_tables_[feature] = self._compute_bin_stats(feature, X[feature], y, bins)
def _group_categories_by_badrate(self, x: pd.Series, y: pd.Series) -> List[List]:
"""将类别按坏样本率排序后合并为不超过 max_n_bins 组（List[List]）。
- 按各类别坏样本率升序排列，相邻（坏样本率最接近）类别优先合并；
- 样本量低于 min_bin_size 的类别、以及退化组（坏样本率 0/1）并入相邻组；
- 在不低于 min_n_bins 的前提下迭代合并，直至满足 max_n_bins 与样本量约束；
- 缺失值/特殊值不参与分组（由 missing_separate / special_codes 单独成箱）。
:return: List[List]，每个子列表为一组类别名（字符串），按组坏样本率升序排列
"""
mask = x.notna()
if self.special_codes:
for code in self.special_codes:
mask = mask & (x != code)
xv = x[mask]
yv = y[mask]
if len(xv) == 0:
return []
cat_df = pd.DataFrame({"c": xv.astype(str).values, "y": np.asarray(yv, dtype=float)})
grp = cat_df.groupby("c")["y"].agg(["sum", "count"])
cats = list(grp.index)
order = sorted(
range(len(cats)),
key=lambda i: (grp["sum"].iloc[i] / grp["count"].iloc[i]) if grp["count"].iloc[i] > 0 else 0.0,
)
groups = [[cats[i]] for i in order]
gcount = [float(grp["count"].iloc[i]) for i in order]
gbad = [float(grp["sum"].iloc[i]) for i in order]
min_samples = self._get_min_samples(len(xv))
max_bins = max(1, self.max_n_bins)
min_bins = max(1, self.min_n_bins)
def brate(i: int) -> float:
return gbad[i] / gcount[i] if gcount[i] > 0 else 0.0
def merge(lo: int, hi: int) -> None:
groups[lo] = groups[lo] + groups[hi]
gcount[lo] += gcount[hi]
gbad[lo] += gbad[hi]
del groups[hi], gcount[hi], gbad[hi]
for _ in range(1000):
n = len(groups)
if n <= min_bins:
break
small = [i for i in range(n) if gcount[i] < min_samples]
degenerate = [i for i in range(n) if gbad[i] <= 0 or gbad[i] >= gcount[i]]
if not (small or degenerate or n > max_bins):
break
if small:
target = min(small, key=lambda k: gcount[k])
elif degenerate:
target = degenerate[0]
else:
# 仅需满足 max_n_bins：合并坏样本率差异最小的相邻组
diffs = [abs(brate(k + 1) - brate(k)) for k in range(n - 1)]
k = int(np.argmin(diffs))
merge(k, k + 1)
continue
if target == 0:
neighbor = 1
elif target == n - 1:
neighbor = n - 2
else:
neighbor = (
target - 1
if abs(brate(target) - brate(target - 1)) <= abs(brate(target) - brate(target + 1))
else target + 1
)
merge(min(target, neighbor), max(target, neighbor))
# 合并后重新按坏样本率排序，保持相邻性语义
idx = sorted(range(len(groups)), key=brate)
groups = [groups[i] for i in idx]
gcount = [gcount[i] for i in idx]
gbad = [gbad[i] for i in idx]
return groups
def _get_default_splits(self, x: pd.Series, y: pd.Series, feature_type: str):
"""获取默认切分点."""
if feature_type == "categorical":
return x.dropna().unique().tolist()
else:
# 使用等频分箱
x_clean = x.dropna()
if len(x_clean) > 0:
quantiles = np.linspace(0, 1, self.max_n_bins + 1)
return np.percentile(x_clean, quantiles[1:-1] * 100)
return np.array([])
def _apply_bins(
self, x: pd.Series, splits: Union[np.ndarray, List], feature_type: str, feature: Optional[str] = None
) -> np.ndarray:
"""应用分箱.
类别规则只支持严格 ``List[List]``，并优先使用已规范化的 ``_cat_bins_``。
"""
if feature_type == "categorical" and feature in self._cat_bins_:
return self._assign_categorical_bins(feature, x)
if isinstance(splits, list):
if not splits or not all(isinstance(group, list) for group in splits):
raise ValueError("类别分箱规则必须是非空 List[List]")
bins = np.zeros(len(x), dtype=int)
# 类别型比较时，将 Series 转为字符串以避免类型不匹配导致的静默失败
x_str = x.astype(str).where(x.notna(), other=np.nan)
has_nan_group = False
for i, group in enumerate(splits):
for value in group:
if is_missing_marker(value):
bins[x.isna()] = i
has_nan_group = True
else:
bins[x_str == str(value)] = i
if not has_nan_group and self.missing_separate:
bins[x.isna()] = -1
if self.special_codes:
for code in self.special_codes:
bins[(x == code) | (x_str == str(code))] = -2
return bins
else:
bins = np.zeros(len(x), dtype=int)
if self.missing_separate:
bins[x.isna()] = -1
if self.special_codes:
for code in self.special_codes:
bins[x == code] = -2
mask = x.notna()
if self.special_codes:
for code in self.special_codes:
mask = mask & (x != code)
if len(splits) > 0:
bins[mask] = np.digitize(x[mask], splits)
else:
bins[mask] = 0
return bins
def _apply_monotonic_adjustment(self, X: pd.DataFrame, y: pd.Series):
"""基于当前 method 切分点执行单调性收口。"""
super()._apply_monotonic_adjustment(X, y)
def _is_peak_pattern(self, bad_rates: np.ndarray) -> bool:
"""检查是否为峰值模式（倒U型）.
:param bad_rates: 坏样本率数组
:return: 是否为峰值模式
"""
if len(bad_rates) < 3:
return False
t = np.argmax(bad_rates)
if t == 0 or t == len(bad_rates) - 1:
return False
# 检查前半部分递增，后半部分递减
left_asc = np.all(bad_rates[1 : t + 1] - bad_rates[:t] >= -1e-10)
right_desc = np.all(bad_rates[t + 1 :] - bad_rates[t:-1] <= 1e-10)
return left_asc and right_desc
def _is_valley_pattern(self, bad_rates: np.ndarray) -> bool:
"""检查是否为谷值模式（U型）.
:param bad_rates: 坏样本率数组
:return: 是否为谷值模式
"""
if len(bad_rates) < 3:
return False
t = np.argmin(bad_rates)
if t == 0 or t == len(bad_rates) - 1:
return False
# 检查前半部分递减，后半部分递增
left_desc = np.all(bad_rates[1 : t + 1] - bad_rates[:t] <= 1e-10)
right_asc = np.all(bad_rates[t + 1 :] - bad_rates[t:-1] >= -1e-10)
return left_desc and right_asc
def _transform_with_wrapper_state(self, X: pd.DataFrame, metric: str, **kwargs) -> pd.DataFrame:
"""使用 Optimal 已合并的规则状态转换 imported 或 wrapper 特征。"""
return self._transform_binning_features(
X,
metric,
lambda feature: self._apply_bins(
X[feature],
self._cat_bins_[feature]
if self.feature_types_[feature] == "categorical" and feature in self._cat_bins_
else self.splits_[feature],
self.feature_types_[feature],
feature,
),
extra_metric=self._transform_lift_metric if self.method == "best_lift" else None,
)
[文档]
def get_stats(self, feature: Optional[str] = None) -> Dict[str, Any]:
"""获取分箱统计信息.
返回各特征的IV值、KS值、分箱表等统计指标。
:param feature: 特征名，如果为None则返回所有特征的统计，默认为None
:return: 统计信息字典
- 如果指定了feature，返回 {'n_bins_': int, 'bin_table': DataFrame, 'ks': float, 'iv': float, 'monotonic_trend': str}
- 如果未指定feature，返回 {feature_name: stats, ...} 格式的字典
:raises ValueError: 如果分箱器尚未拟合
:raises KeyError: 如果指定特征不存在
**参考样例**
>>> binner = OptimalBinning(method='best_iv')
>>> binner.fit(X, y)
>>>
>>> # 获取单个特征的统计
>>> stats = binner.get_stats('age')
>>> print(stats['n_bins_']) # 分箱数
>>> print(stats['iv']) # IV值
>>> print(stats['ks']) # KS值
>>> print(stats['bin_table']) # 分箱统计表
>>>
>>> # 获取所有特征的统计
>>> all_stats = binner.get_stats()
>>> for feat, s in all_stats.items():
... print(f"{feat}: IV={s.get('iv', 'N/A')}")
"""
if not self._is_fitted:
raise NotFittedError("分箱器尚未拟合，请先调用fit方法")
if feature is not None:
if feature not in self.bin_tables_:
raise KeyError(f"特征 '{feature}' 未找到")
stats = {
"n_bins_": self.n_bins_[feature],
"bin_table": self.bin_tables_[feature],
}
if hasattr(self, "ks_stats_") and feature in self.ks_stats_:
stats["ks"] = self.ks_stats_[feature]
if hasattr(self, "iv_stats_") and feature in self.iv_stats_:
stats["iv"] = self.iv_stats_[feature]
if feature in self.monotonic_trend_:
stats["monotonic_trend"] = self.monotonic_trend_[feature]
return stats
else:
return {f: self.get_stats(f) for f in self.splits_.keys()}
[文档]
@staticmethod
def auto_select_method(
X: pd.DataFrame,
y: pd.Series,
feature: str,
methods: Optional[List[str]] = None,
criterion: str = "iv",
n_jobs: Union[int, float] = -1,
parallel_backend: Optional[str] = None,
parallel_config: Optional[Dict[str, Any]] = None,
) -> str:
"""自动选择最优分箱方法.
对指定特征遍历多种分箱方法，选择最优的一个。
适用于在不了解特征分布时自动选择最佳分箱策略。
:param X: 特征数据 DataFrame
:param y: 目标变量 Series，二分类 (0/1)
:param feature: 待评估的特征名（必须是X中的列）
:param methods: 待评估的方法列表，默认为 ['uniform', 'quantile', 'tree', 'chi',
'best_ks', 'best_iv', 'mdlp', 'cart', 'kmeans']
:param criterion: 选择标准，默认为 'iv'
- 'iv': 选择IV值最大的方法
- 'ks': 选择KS值最大的方法
:return: 最优方法名（字符串）
:raises Warning: 某个方法执行失败时输出警告但继续评估其他方法
**参考样例**
>>> binner = OptimalBinning()
>>> best_method = OptimalBinning.auto_select_method(X, y, 'age')
>>> print(f"最优方法: {best_method}")
>>>
>>> # 自定义方法列表
>>> best = OptimalBinning.auto_select_method(
... X, y, 'income',
... methods=['best_iv', 'cart', 'mdlp'],
... criterion='ks'
... )
>>>
>>> # 使用最优方法进行分箱
>>> binner = OptimalBinning(method=best_method)
>>> binner.fit(X[[feature]], y)
"""
if methods is None:
methods = ["uniform", "quantile", "tree", "chi", "best_ks", "best_iv", "mdlp", "cart", "kmeans"]
else:
if not methods:
raise ValueError("methods 不能为空列表")
# 统一校验并归一化，避免非法方法被逐个 warn 后返回无效结果
methods = [OptimalBinning.validate_method(m) for m in methods]
tasks = [(method, X, y, feature, criterion, parallel_backend, parallel_config) for method in methods]
data_bytes = int(X[[feature]].memory_usage(deep=True).sum())
if hasattr(y, "memory_usage"):
data_bytes += int(y.memory_usage(deep=True))
results = parallel_execute(
_auto_method_score_worker,
tasks,
n_jobs=n_jobs,
parallel_backend=parallel_backend,
parallel_config=parallel_config,
task_labels=methods,
default_backend="loky",
has_parallel_children=True,
workload=ParallelWorkload(
task_count=len(tasks),
rows=len(X),
columns=len(tasks),
data_bytes=data_bytes,
cost_per_item=20.0,
capability="process_safe",
has_parallel_children=True,
operation="自动分箱方法评估",
),
)
best_method = None
best_score = -np.inf
for method, score, error in results:
if error is not None:
warnings.warn(f"方法 {method} 在特征 {feature} 上失败: {error}")
continue
if score is not None and score > best_score:
best_score = score
best_method = method
if best_method is None:
raise ValueError(f"特征 '{feature}' 的所有候选分箱方法均执行失败或未产生有效分箱")
return best_method
if __name__ == "__main__":
# 测试代码
print("=" * 70)
print("OptimalBinning - 统一分箱接口测试")
print("=" * 70)
# 生成测试数据
np.random.seed(42)
n = 1000
X = pd.DataFrame(
{
"feature1": np.random.normal(0, 1, n),
"feature2": np.random.uniform(0, 100, n),
}
)
y = (X["feature1"] + X["feature2"] / 100 > 0.5).astype(int)
# 测试各种方法
methods_to_test = ["uniform", "quantile", "tree", "chi", "best_iv", "cart", "kmeans", "mdlp"]
for method in methods_to_test:
print(f"\n测试方法: {method}")
print("-" * 50)
try:
binner = OptimalBinning(method=method, max_n_bins=5, verbose=False)
binner.fit(X, y)
table = binner.get_bin_table("feature1")
print(f" 分箱数: {len(table)}")
print(f" 前3行:\n{table[['分箱', '样本总数', '坏样本率']].head(3)}")
except Exception as e:
print(f" 错误: {e}")
print("\n" + "=" * 70)
print("测试完成!")
print("=" * 70)