"""逾期数据预估模块.
基于特征分箱统计（feature_bin_stats）计算分箱逾期率，
并利用加权逾期率对无标签样本进行逾期风险预估。
核心功能:
- 支持从原始数据（含target或逾期天数）自动分箱并提取逾期率
- 支持从现成分箱表（feature_bin_stats输出或人工提供）直接提取逾期率
- 将分箱逾期率应用于无标签样本，计算加权逾期率
- 提供系数设置功能，调整预测偏差
- 输出包含逾期率预估结果的报告数据（DataFrame）
设计原则:
1. 遵循sklearn的fit/transform API风格
2. 与feature_bin_stats保持一致的数据格式
3. 支持多逾期标签（overdue+dpds）场景
"""
import numpy as np
import pandas as pd
import warnings
from typing import Union, List, Dict, Optional, Tuple, Any
from copy import deepcopy
from sklearn.base import BaseEstimator, TransformerMixin
from ..utils.overdue import overdue_label, validate_overdue_operator
from ..utils.input_utils import normalize_dpd_values
from .feature_analyzer import feature_bin_stats
from .mining.base import _mining_workload
from ..utils.parallel import ParallelizableMixin, resolve_n_jobs, validate_parallel_config
def _overdue_target_transform(task):
"""计算单个逾期标签的基础及调整后逾期率。"""
predictor, values, target_name, bin_labels = task
rates = predictor.bin_rates_.get(target_name, {})
_, base_rates = predictor._assign_bins_and_rates(values, rates)
coefficients = predictor._resolve_coefficients(rates, target_name)
adjusted = predictor._apply_coefficients(bin_labels, base_rates, coefficients)
return target_name, base_rates, adjusted
[文档]
class OverduePredictor(ParallelizableMixin, BaseEstimator, TransformerMixin):
"""逾期率预测器.
基于特征分箱对应的逾期率，对无标签样本进行加权逾期率预测。
支持从原始数据或现成分箱表两种方式获取分箱逾期率。
**两种拟合模式**
模式一：从原始数据拟合
- 传入含target或逾期天数的DataFrame，自动分箱并计算各箱逾期率
- 支持overdue+dpds多标签场景
模式二：从分箱表拟合
- 传入已有的分箱统计表（feature_bin_stats输出或人工构建的DataFrame）
- 直接从分箱表中提取各箱逾期率
**参数**
:param feature: 特征名称，用于分箱和预估
:param target: 目标变量名称，默认为'target'
:param overdue: 逾期天数字段名称或列表，如 'MOB1' 或 ['MOB1', 'MOB3']
:param dpds: 逾期定义天数或列表，如 7 或 [0, 7, 30]
- 默认逾期天数 > dpds 为坏样本(1)，可通过 overdue_operator 修改比较符
:param method: 分箱方法，默认'mdlp'
:param max_n_bins: 最大分箱数，默认5
:param min_bin_size: 每箱最小样本占比，默认0.05
:param missing_separate: 是否将缺失值单独分箱，默认True
:param coefficients: 逾期率调整系数，支持以下格式：
- None: 不调整（默认）
- float: 对所有分箱统一乘以该系数
- dict: 按分箱标签指定系数，如 {'(-inf, 300]': 1.2, '(300, 500]': 0.9}
- 'auto': 自动基于整体逾期率偏差校正
:param bad_rate_col: 分箱表中逾期率列名，默认为'坏样本率'
- 单标签时直接使用该列名
- 多标签时需包含目标名称，函数会自动匹配
:param bin_label_col: 分箱表中分箱标签列名，默认为'分箱标签'
:param rules: 自定义分箱切分点列表，如 [300, 500, 700]
:param desc: 特征描述，用于报告展示
:param bin_params: 传递给feature_bin_stats的额外参数
:param overdue_operator: 比较符（>、>=、<、<=）；None 时沿用 bin_params 配置，否则默认 >。
显式参数优先；> 的灰样本为 (0, dpd]，>= 为 (0, dpd)，< 和 <= 不剔灰。
:param del_grey: 是否按 overdue_operator 对应区间剔除灰样本；``<``、``<=`` 下不剔除。
兼容读取 ``bin_params['del_grey']``，默认 False
:param binner: 已拟合分箱器或原始数据拟合模板；结构化预测复用真实切点及类别归属
:param bin_specs: 分箱表输入时使用的 {稳定分箱编号或标签: BinSpec}；编号映射要求表有“分箱”列
:param table_mode: legacy 允许只含展示标签的旧表并明确警告；strict要求binner或bin_specs
**属性**
- bin_table_: 拟合后的分箱统计表
- bin_rates_: 各分箱的逾期率字典 {分箱标签: 逾期率}
- splits_: 分箱切分点
- feature_names_in_: 输入特征名称
- target_names_: 目标标签名称列表
- coefficients_: 实际使用的调整系数
**参考样例**
>>> import numpy as np
>>> import pandas as pd
>>> from hscredit.report.overdue_predictor import OverduePredictor
>>>
>>> # 准备有标签的训练数据
>>> train_df = pd.DataFrame({
... 'score': np.random.randn(1000) * 100 + 500,
... 'target': np.random.randint(0, 2, 1000)
... })
>>>
>>> # 方式一：从原始数据拟合（自动分箱计算各箱逾期率）
>>> predictor = OverduePredictor(feature='score', target='target', max_n_bins=5)
>>> predictor.fit(train_df)
>>>
>>> # 对无标签数据预测（根据样本所在分箱加权计算逾期率）
>>> test_df = pd.DataFrame({'score': np.random.randn(200) * 100 + 500})
>>> result = predictor.transform(test_df)
>>> print(result.head())
>>>
>>> # 设置调整系数（对逾期率进行整体缩放校正）
>>> predictor.set_coefficients(1.1)
>>> result_adjusted = predictor.transform(test_df)
>>>
>>> # 方式二：从分箱表拟合（直接使用现成分箱逾期率，无需原始数据）
>>> bin_table = pd.DataFrame({
... '分箱标签': ['(-inf, 400]', '(400, 500]', '(500, 600]', '(600, +inf)'],
... '坏样本率': [0.15, 0.08, 0.04, 0.02]
... })
>>> predictor2 = OverduePredictor(feature='score')
>>> predictor2.fit(bin_table)
>>> result2 = predictor2.transform(test_df)
"""
def __init__(
self,
feature: str,
target: Optional[str] = None,
overdue: Optional[Union[str, List[str]]] = None,
dpds: Optional[Union[int, List[int]]] = None,
method: str = 'mdlp',
max_n_bins: int = 5,
min_bin_size: float = 0.05,
missing_separate: bool = True,
coefficients: Optional[Union[float, Dict[str, float], str]] = None,
bad_rate_col: str = '坏样本率',
bin_label_col: str = '分箱标签',
rules: Optional[List] = None,
desc: Optional[str] = None,
bin_params: Optional[Dict] = None,
n_jobs: Union[int, float] = -1,
parallel_backend: Optional[str] = None,
parallel_config: Optional[Dict[str, Any]] = None,
del_grey: Optional[bool] = None,
*,
overdue_operator: Optional[str] = None,
binner=None,
bin_specs=None,
table_mode: str = "legacy",
):
self.feature = feature
self.target = target
self.overdue = overdue
self.dpds = dpds
self.method = method
self.max_n_bins = max_n_bins
self.min_bin_size = min_bin_size
self.missing_separate = missing_separate
self.coefficients = coefficients
self.bad_rate_col = bad_rate_col
self.bin_label_col = bin_label_col
self.rules = rules
self.desc = desc
self.bin_params = bin_params
self.n_jobs = n_jobs
self.parallel_backend = parallel_backend
self.parallel_config = parallel_config
self.del_grey = del_grey
if overdue_operator is not None:
validate_overdue_operator(overdue_operator)
self.overdue_operator = overdue_operator
self.binner = binner
self.bin_specs = bin_specs
self.table_mode = table_mode
[文档]
def fit(self, X: Union[pd.DataFrame, pd.Series], y=None) -> "OverduePredictor":
"""拟合预估器.
支持两种输入模式:
1. 传入DataFrame（含target或逾期天数列）: 自动分箱并计算逾期率
2. 传入分箱表DataFrame（含分箱标签和逾期率列）: 直接提取逾期率
:param X: 训练数据DataFrame或分箱表DataFrame
:param y: sklearn兼容参数，此处不使用（目标列从X中提取）
:return: self
"""
if not isinstance(X, pd.DataFrame):
raise TypeError("X 必须是 pd.DataFrame，支持原始数据或分箱表")
validate_parallel_config(self.parallel_backend, self.parallel_config)
resolve_n_jobs(self.n_jobs, task_count=1)
if self.table_mode not in {"legacy", "strict"}:
raise ValueError("table_mode 必须为 legacy 或 strict")
if self.binner is not None and self.bin_specs is not None:
raise ValueError("binner 与 bin_specs 只能指定一个，避免分箱语义冲突")
learned_attrs = (
'feature_names_in_', 'target_names_', 'bin_table_', 'bin_rates_',
'splits_', 'coefficients_', 'bin_weights_', 'overall_rates_',
'binner_', 'bin_specs_', 'bin_labels_', 'prediction_mode_',
)
candidate = deepcopy(self)
for name in learned_attrs:
if hasattr(candidate, name):
delattr(candidate, name)
candidate.feature_names_in_ = [candidate.feature]
if candidate._is_bin_table(X):
candidate._fit_from_bin_table(X)
else:
candidate._fit_from_raw_data(X)
for name in learned_attrs:
if hasattr(self, name):
delattr(self, name)
for name in learned_attrs:
if hasattr(candidate, name):
setattr(self, name, getattr(candidate, name))
return self
def _is_bin_table(self, df: pd.DataFrame) -> bool:
"""判断输入是否为分箱表.
分箱表的特征：
- 包含 bin_label_col 指定的分箱标签列
- 包含逾期率相关的列（坏样本率 或 多级表头下包含坏样本率）
:param df: 输入DataFrame
:return: 是否为分箱表
"""
if df.empty:
return False
cols = df.columns.tolist()
# 单层表头检查
if self.bin_label_col in cols:
if self.bad_rate_col in cols:
return True
# 检查是否有含"坏样本率"的列（多标签时）
for c in cols:
if isinstance(c, str) and '坏样本率' in c:
return True
return False # 有分箱标签列但无坏样本率列，不是有效分箱表
# 多级表头检查
if isinstance(df.columns, pd.MultiIndex):
for c in df.columns:
col_name = c[-1] if isinstance(c, (tuple, list)) else c
if col_name == self.bin_label_col:
# 多级表头时也检查是否有坏样本率
for c2 in df.columns:
col_name2 = c2[-1] if isinstance(c2, (tuple, list)) else c2
if isinstance(col_name2, str) and '坏样本率' in col_name2:
return True
return False
return False
def _fit_from_raw_data(self, df: pd.DataFrame) -> None:
"""从原始数据拟合，自动分箱并计算逾期率.
:param df: 含target或逾期天数的DataFrame
"""
if self.bin_specs is not None:
raise ValueError("原始数据拟合请使用 binner 或 rules；bin_specs 用于已有分箱统计表")
if self.overdue is not None:
# 检查逾期天数字段是否存在
overdue_cols = [self.overdue] if isinstance(self.overdue, str) else self.overdue
missing_cols = [c for c in overdue_cols if c not in df.columns]
if missing_cols:
raise ValueError(f"数据集缺少逾期天数字段: {missing_cols}")
elif self.target is not None:
if self.target not in df.columns:
raise ValueError(f"数据集缺少目标变量字段: {self.target}")
else:
raise ValueError("必须传入 target 或 overdue+dpds 参数")
# 构建目标变量名称列表
self.target_names_ = self._build_target_names()
# 调用 feature_bin_stats 生成分箱表
bin_kwargs = {
'method': self.method,
'max_n_bins': self.max_n_bins,
'min_bin_size': self.min_bin_size,
'missing_separate': self.missing_separate,
'margins': True,
}
if self.rules is not None:
bin_kwargs['rules'] = self.rules
if self.bin_params:
bin_kwargs.update(self.bin_params)
if self.binner is not None:
bin_kwargs['binner'] = self.binner
bin_kwargs.pop('return_rules', None)
bin_kwargs['return_binner'] = True
if self.del_grey is not None:
bin_kwargs['del_grey'] = bool(self.del_grey)
bin_kwargs['overdue_operator'] = self._resolved_overdue_operator()
result = feature_bin_stats(
df,
feature=self.feature,
target=self.target,
overdue=self.overdue,
dpds=self.dpds,
desc=self.desc or self.feature,
n_jobs=self.n_jobs,
parallel_backend=self.parallel_backend,
parallel_config=self.parallel_config,
**bin_kwargs,
)
table, binners = result
self.bin_table_ = table
self._extract_bin_rates_from_table(table)
self._attach_bin_specs(table, binners[self.feature])
# 提取分箱切分点（尝试从bin_table中获取）
self._extract_splits_from_raw_data(df)
def _fit_from_bin_table(self, df: pd.DataFrame) -> None:
"""从现成分箱表拟合，直接提取逾期率.
:param df: 分箱统计表DataFrame
"""
self.bin_table_ = df.copy()
# 解析目标名称（从多级表头或列名推断）
self.target_names_ = self._infer_target_names(df)
self._extract_bin_rates_from_table(df)
self._attach_bin_specs(df, self.binner)
self._extract_splits_from_bin_table(df)
def _attach_bin_specs(self, table, binner=None):
"""以稳定箱号关联风险表与执行语义，显示标签仅用于输出和系数映射。"""
from ..core.binning.spec import BinSpec
label_col = next(c for c in table.columns if (c[-1] if isinstance(c, tuple) else c) == self.bin_label_col)
rows = table.loc[table[label_col].astype(str).ne('合计')]
labels = rows[label_col].astype(str)
if labels.duplicated().any():
raise ValueError("分箱标签重复，无法唯一关联风险率；请保留稳定箱号并提供唯一展示标签")
id_col = next((c for c in table.columns if (c[-1] if isinstance(c, tuple) else c) == '分箱'), None)
supplied = self.bin_specs
if supplied is None and binner is None:
if self.table_mode == 'strict':
raise ValueError("strict分箱表预测必须提供 binner 或 bin_specs，不能从展示字符串推断精确边界")
self.prediction_mode_ = 'legacy_labels'
warnings.warn("当前使用legacy分箱展示标签解析；舍入边界及含逗号类别可能不精确，请提供binner/bin_specs", FutureWarning, stacklevel=3)
return
specs = BinSpec.from_binner(binner, self.feature) if supplied is None else dict(supplied)
if not specs or not all(isinstance(spec, BinSpec) and spec.feature == self.feature for spec in specs.values()):
raise ValueError("bin_specs 必须包含同一特征的 BinSpec")
keyed_by_id = all(isinstance(key, (int, np.integer)) for key in specs)
if keyed_by_id:
if any(int(key) != spec.bin_id for key, spec in specs.items()):
raise ValueError("bin_specs键必须与BinSpec.bin_id一致")
if id_col is None:
raise ValueError("编号型bin_specs/binner需要统计表保留稳定的“分箱”列；请使用feature_bin_stats(return_binner=True)")
ids = pd.to_numeric(rows[id_col], errors='coerce')
if ids.isna().any() or ids.duplicated().any() or (ids % 1 != 0).any():
raise ValueError("统计表中的分箱编号必须为唯一整数")
self.bin_labels_ = dict(zip(ids.astype(int), labels))
if not set(self.bin_labels_).issubset(specs):
raise ValueError("bin_specs未覆盖统计表中的全部分箱编号")
self.bin_specs_ = dict(specs)
else:
if not all(isinstance(key, str) for key in specs) or not set(labels).issubset(specs):
raise ValueError("标签型bin_specs必须覆盖统计表中的全部标签")
if len({spec.bin_id for spec in specs.values()}) != len(specs):
raise ValueError("bin_specs中的bin_id必须唯一")
self.bin_specs_ = {spec.bin_id: spec for spec in specs.values()}
self.bin_labels_ = {spec.bin_id: label for label, spec in specs.items()}
if binner is not None:
self.binner_ = binner
self.prediction_mode_ = 'structured'
def _resolved_overdue_operator(self) -> str:
"""显式比较符优先，否则沿用分箱配置。"""
value = self.overdue_operator
if value is None:
value = (self.bin_params or {}).get("overdue_operator", ">")
return validate_overdue_operator(value)
def __setstate__(self, state):
"""兼容恢复新增比较符参数之前保存的预测器。"""
state = dict(state)
state.setdefault("overdue_operator", None)
state.setdefault("binner", None)
state.setdefault("bin_specs", None)
state.setdefault("table_mode", "legacy")
state.setdefault("prediction_mode_", "legacy_labels")
super().__setstate__(state)
if hasattr(self, 'bin_table_') and not hasattr(self, 'bin_weights_'):
self._capture_reference_weights(self.bin_table_)
def _build_target_names(self) -> List[str]:
"""构建目标变量名称列表.
:return: 目标名称列表
"""
if self.overdue is not None and self.dpds is not None:
overdue_list = [self.overdue] if isinstance(self.overdue, str) else self.overdue
dpd_list = normalize_dpd_values(self.dpds)
names = []
for mob in overdue_list:
for d in dpd_list:
names.append(overdue_label(mob, d, self._resolved_overdue_operator()))
return names
elif self.target is not None:
return [self.target]
return []
def _infer_target_names(self, df: pd.DataFrame) -> List[str]:
"""从分箱表推断目标名称.
:param df: 分箱表
:return: 目标名称列表
"""
if isinstance(df.columns, pd.MultiIndex):
# 多级表头：收集所有非"分箱详情"的顶级列名
names = []
for col in df.columns:
top_level = col[0] if isinstance(col, tuple) else col
if top_level != '分箱详情' and top_level not in names:
names.append(top_level)
return names if names else ['target']
else:
return ['target']
def _extract_bin_rates_from_table(self, table: pd.DataFrame) -> None:
"""从分箱表中提取各分箱逾期率.
:param table: 分箱统计表
"""
self.bin_rates_ = {}
if isinstance(table.columns, pd.MultiIndex):
# 多级表头：每个目标标签对应一组逾期率
for target_name in self.target_names_:
rates = self._extract_single_target_rates(table, target_name)
self.bin_rates_[target_name] = rates
# 单标签时提供便捷访问
if len(self.target_names_) == 1:
self.bin_rates_['_default'] = self.bin_rates_[self.target_names_[0]]
else:
# 单层表头
rates = self._extract_single_target_rates_flat(table)
self.bin_rates_['_default'] = rates
if self.target_names_:
for name in self.target_names_:
self.bin_rates_[name] = rates
self._capture_reference_weights(table)
def _capture_reference_weights(self, table: pd.DataFrame) -> None:
"""保存每标签的有效箱权重和合计率，禁止从箱数猜测样本占比。"""
self.bin_weights_ = {}
self.overall_rates_ = {}
def find_column(name, target_name):
if not isinstance(table.columns, pd.MultiIndex):
return name if name in table.columns else None
for group in (target_name, '分箱详情'):
for column in table.columns:
if column[0] == group and column[-1] == name:
return column
return None
label_col = next((c for c in table.columns if (c[-1] if isinstance(c, tuple) else c) == self.bin_label_col), None)
if label_col is None:
return
labels = table[label_col].astype(str)
for name, rates in self.bin_rates_.items():
target = self.target_names_[0] if name == '_default' and self.target_names_ else name
weight_col = find_column('样本总数', target)
if weight_col is None:
weight_col = find_column('样本占比', target)
weights = {}
if weight_col is not None:
values = pd.to_numeric(table[weight_col], errors='coerce')
for label, value in zip(labels, values):
if label == '合计':
continue
if pd.isna(value) or not np.isfinite(value) or value < 0:
raise ValueError(f"分箱 '{label}' 的样本权重必须为非负有限数值")
weights[label] = float(value)
self.bin_weights_[name] = weights
rate_col = find_column(self.bad_rate_col, target)
if rate_col is not None and labels.eq('合计').any():
value = table.loc[labels.eq('合计'), rate_col].iloc[0]
if pd.notna(value):
self.overall_rates_[name] = float(value)
if name not in self.overall_rates_ and sum(weights.values()) > 0:
self.overall_rates_[name] = sum(weights.get(label, 0.0) * rate for label, rate in rates.items()) / sum(weights.values())
for rate in rates.values():
if not np.isfinite(rate) or not 0 <= rate <= 1:
raise ValueError("分箱坏样本率必须为0到1之间的有限数值")
if name in self.overall_rates_ and not 0 <= self.overall_rates_[name] <= 1:
raise ValueError("整体坏样本率必须为0到1之间的有限数值")
def _reference_key(self, rates):
return next((name for name, candidate in self.bin_rates_.items() if candidate is rates), None)
def _extract_single_target_rates(self, table: pd.DataFrame, target_name: str) -> Dict[str, float]:
"""从多级表头表中提取指定目标的逾期率.
:param table: 多级表头分箱表
:param target_name: 目标名称
:return: {分箱标签: 逾期率}
"""
rates = {}
# 查找分箱标签列
bin_label_col = None
for col in table.columns:
col_name = col[-1] if isinstance(col, tuple) else col
if col_name == self.bin_label_col:
bin_label_col = col
break
if bin_label_col is None:
raise ValueError(f"分箱表中未找到 '{self.bin_label_col}' 列")
# 查找该目标下的坏样本率列
bad_rate_col = None
for col in table.columns:
if isinstance(col, tuple):
if col[0] == target_name and col[1] == self.bad_rate_col:
bad_rate_col = col
break
elif col == self.bad_rate_col:
bad_rate_col = col
break
if bad_rate_col is None:
# 尝试模糊匹配：找目标下含"坏样本率"的列
for col in table.columns:
if isinstance(col, tuple) and col[0] == target_name:
col_sub = col[-1] if isinstance(col, (tuple, list)) else col
if '坏样本率' in str(col_sub):
bad_rate_col = col
break
if bad_rate_col is None:
raise ValueError(f"分箱表中未找到目标 '{target_name}' 对应的 '{self.bad_rate_col}' 列")
# 提取逾期率（排除合计行）
for idx, row in table.iterrows():
label = row[bin_label_col]
if label == '合计':
continue
rate = row[bad_rate_col]
rates[str(label)] = float(rate) if pd.notna(rate) else 0.0
return rates
def _extract_single_target_rates_flat(self, table: pd.DataFrame) -> Dict[str, float]:
"""从单层表头表中提取逾期率.
:param table: 单层表头分箱表
:return: {分箱标签: 逾期率}
"""
rates = {}
if self.bin_label_col not in table.columns:
raise ValueError(f"分箱表中未找到 '{self.bin_label_col}' 列")
if self.bad_rate_col not in table.columns:
# 尝试模糊匹配
rate_col = None
for c in table.columns:
if '坏样本率' in str(c):
rate_col = c
break
if rate_col is None:
raise ValueError(f"分箱表中未找到 '{self.bad_rate_col}' 列")
else:
rate_col = self.bad_rate_col
for idx, row in table.iterrows():
label = row[self.bin_label_col]
if label == '合计':
continue
rate = row[rate_col]
rates[str(label)] = float(rate) if pd.notna(rate) else 0.0
return rates
def _extract_splits_from_raw_data(self, df: pd.DataFrame) -> None:
"""从原始数据中提取分箱切分点.
:param df: 原始数据
"""
if self.rules is not None:
self.splits_ = np.array(self.rules)
else:
self.splits_ = None
def _extract_splits_from_bin_table(self, df: pd.DataFrame) -> None:
"""从分箱表中解析分箱切分点.
:param df: 分箱表
"""
if self.bin_label_col not in df.columns:
self.splits_ = None
return
labels = df[self.bin_label_col].tolist()
splits = self._parse_bin_labels_to_splits(labels)
self.splits_ = splits
def _parse_bin_labels_to_splits(self, labels: List[str]) -> Optional[np.ndarray]:
"""从分箱标签解析分箱切分点.
支持的标签格式:
- '(-inf, 300]', '(300, 500]', '(500, +inf)'
- 'missing', 'special'
:param labels: 分箱标签列表
:return: 切分点数组
"""
import re
splits = []
for label in labels:
label = str(label).strip()
if label in ('missing', 'special', '合计', ''):
continue
# 匹配区间格式: (left, right] 或 [left, right)
pattern = r'[\(\[]\s*([^\s,]+)\s*,\s*([^\s\)]+)\s*[\)\]]'
match = re.match(pattern, label)
if match:
right = match.group(2)
if right not in ('+inf', 'inf'):
try:
val = float(right)
if val not in splits:
splits.append(val)
except ValueError:
pass
if splits:
return np.array(sorted(splits))
return None
def _transform_single_target(self, X: pd.DataFrame) -> pd.DataFrame:
"""单标签逾期率预估.
:param X: 待预估数据
:return: 包含预估结果的数据
"""
rates = self.bin_rates_.get('_default', self.bin_rates_.get(self.target_names_[0] if self.target_names_ else 'target', {}))
# 为每个样本分配分箱和逾期率
bin_labels, base_rates = self._assign_bins_and_rates(X[self.feature], rates)
# 应用调整系数
coefficients_map = self._resolve_coefficients(rates)
adjusted_rates = self._apply_coefficients(bin_labels, base_rates, coefficients_map)
# 写入结果
X[f'{self.feature}_分箱'] = bin_labels
X[f'{self.feature}_基础逾期率'] = base_rates
X[f'{self.feature}_预测逾期率'] = adjusted_rates
return X
def _transform_multi_target(self, X: pd.DataFrame) -> pd.DataFrame:
"""多标签逾期率预测.
:param X: 待预测数据
:return: 包含多标签预测结果的数据
"""
# 分箱只需要做一次
first_target = self.target_names_[0]
first_rates = self.bin_rates_.get(first_target, {})
bin_labels, _ = self._assign_bins_and_rates(X[self.feature], first_rates)
X[f'{self.feature}_分箱'] = bin_labels
tasks = [(self, X[self.feature], target_name, bin_labels) for target_name in self.target_names_]
target_results = self._parallel_execute(
_overdue_target_transform,
tasks,
task_labels=self.target_names_,
default_backend="threading",
workload=_mining_workload(
X,
len(tasks),
operation="多标签逾期率预测",
cost_per_item=6.0,
),
)
for target_name, base_rates, adjusted_rates in target_results:
X[f'{self.feature}_{target_name}_基础逾期率'] = base_rates
X[f'{self.feature}_{target_name}_预测逾期率'] = adjusted_rates
return X
def _assign_bins_and_rates(
self,
values: pd.Series,
rates: Dict[str, float],
) -> Tuple[List[str], List[float]]:
"""将样本值映射到分箱，并分配对应的逾期率.
:param values: 特征值序列
:param rates: {分箱标签: 逾期率}
:return: (分箱标签列表, 逾期率列表)
"""
if getattr(self, 'prediction_mode_', None) == 'structured':
return self._assign_structured_bins(values, rates)
# 类别箱只匹配类别值，不能把无法解析的标签视为覆盖全域的数值区间。
bin_intervals = self._parse_rate_intervals(rates)
category_labels = [
label for label in rates
if label not in bin_intervals and label not in ('missing', 'special', 'unknown', '合计', '未知')
]
# 按箱矢量化匹配，避免每个样本重复解析类别和遍历Python区间。
values = values.reset_index(drop=True)
missing = values.isna().to_numpy()
fallback_label = 'unknown' if 'unknown' in rates else ('special' if 'special' in rates else '未知')
labels = np.full(len(values), fallback_label, dtype=object)
fallback_rate = rates.get(fallback_label, self._get_overall_rate(rates))
base = np.full(len(values), fallback_rate, dtype=float)
assigned = missing.copy()
labels[missing] = 'missing'
base[missing] = rates.get('missing', self._get_overall_rate(rates))
category_map = {str(label).strip(): label for label in category_labels}
for label in category_labels:
for part in str(label).split(','):
category_map.setdefault(part.strip(), label)
mapped = values.astype(str).str.strip().map(category_map)
selected = mapped.notna().to_numpy() & ~assigned
labels[selected] = mapped[selected].to_numpy()
base[selected] = mapped[selected].map(rates).to_numpy(dtype=float)
assigned |= selected
numeric = pd.to_numeric(values, errors='coerce').to_numpy(dtype=float, na_value=np.nan)
for label, (left, right, left_inc, right_inc) in bin_intervals.items():
if label in ('missing', 'special', 'unknown', '合计', '未知'):
continue
selected = ~assigned & ~np.isnan(numeric)
if left is not None:
selected &= numeric >= left if left_inc else numeric > left
if right is not None:
selected &= numeric <= right if right_inc else numeric < right
labels[selected] = label
base[selected] = rates[label]
assigned |= selected
return labels.tolist(), base.tolist()
def _assign_structured_bins(self, values, rates):
"""先用真实分箱器/语义取得箱号，再取风险率，完全不解析展示文本。"""
values = values.reset_index(drop=True)
if hasattr(self, 'binner_'):
codes = self.binner_.transform(values.to_frame(name=self.feature), metric='indices')[self.feature].to_numpy()
else:
codes = np.full(len(values), None, dtype=object)
assigned = np.zeros(len(values), dtype=bool)
for bin_id, spec in self.bin_specs_.items():
mask = spec.mask(values).to_numpy(dtype=bool)
if np.any(assigned & mask):
raise ValueError("bin_specs存在重叠分箱，同一样本不能匹配多个箱")
codes[mask] = bin_id
assigned |= mask
labels = [self.bin_labels_.get(int(code), 'missing' if int(code) == -1 else '未知') if code is not None and not pd.isna(code) else '未知' for code in codes]
baseline = self._get_overall_rate(rates)
return labels, [rates.get(label, baseline) for label in labels]
def _parse_rate_intervals(self, rates: Dict[str, float]) -> Dict[str, Tuple]:
"""解析分箱标签为区间元组.
:param rates: {分箱标签: 逾期率}
:return: {标签: (left, right, left_inclusive, right_inclusive)}
"""
import re
intervals = {}
for label in rates:
label_str = str(label).strip()
if label_str in ('missing', 'special', '合计', '未知'):
intervals[label_str] = (None, None, False, False)
continue
# 匹配 (left, right] 或 [left, right)
pattern = r'([\(\[])\s*([^\s,]+)\s*,\s*([^\s\)]+)\s*([\)\]])'
match = re.match(pattern, label_str)
if match:
left_bracket, left_str, right_str, right_bracket = match.groups()
left_inc = left_bracket == '['
right_inc = right_bracket == ']'
left = None if left_str in ('-inf', 'inf') else float(left_str)
right = None if right_str in ('+inf', 'inf', '-inf') else float(right_str)
intervals[label_str] = (left, right, left_inc, right_inc)
return intervals
def _resolve_coefficients(
self,
rates: Dict[str, float],
target_name: Optional[str] = None,
) -> Dict[str, float]:
"""解析并确定实际使用的调整系数.
:param rates: 各分箱逾期率
:param target_name: 目标名称（多标签时区分）
:return: {分箱标签: 系数}
"""
if self.coefficients is None:
return {label: 1.0 for label in rates}
if isinstance(self.coefficients, (int, float)):
return {label: float(self.coefficients) for label in rates}
if isinstance(self.coefficients, dict):
result = {label: 1.0 for label in rates}
for label, coef in self.coefficients.items():
if label in rates:
result[label] = float(coef)
return result
if self.coefficients == 'auto':
overall_rate = self._get_overall_rate(rates)
weights = np.array([self._estimate_weight(label, rates) for label in rates], dtype=float)
if not weights.sum() > 0:
raise ValueError("auto调整需要分箱表中的样本总数或样本占比，不能按箱数猜测权重")
weights = weights / weights.sum()
rate_values = np.array(list(rates.values()), dtype=float)
weighted_rate = float(weights @ rate_values)
if np.isclose(weighted_rate, overall_rate, rtol=1e-10, atol=1e-12):
factor = 1.0
elif overall_rate > weights[rate_values > 0].sum():
raise ValueError("整体坏样本率无法由现有非零风险箱校准得到")
else:
# 以参考箱占比匹配整体基准；考虑概率截断，避免截断后均值偏移。
positive = rate_values[rate_values > 0]
low, high = 0.0, max(1.0, float(1.0 / positive.min())) if len(positive) else 1.0
if not np.isfinite(high):
raise ValueError("分箱风险率过小，无法稳定计算auto调整系数")
for _ in range(64):
middle = (low + high) / 2
if float(weights @ np.minimum(rate_values * middle, 1.0)) < overall_rate:
low = middle
else:
high = middle
factor = (low + high) / 2
return {label: factor for label in rates}
return {label: 1.0 for label in rates}
def _estimate_weight(self, label: str, rates: Dict[str, float]) -> float:
"""估算分箱权重（用于auto系数模式）.
使用拟合阶段保存的样本总数或样本占比；缺失时返回0。
:param label: 分箱标签
:param rates: 逾期率字典
:return: 权重
"""
return getattr(self, 'bin_weights_', {}).get(self._reference_key(rates), {}).get(label, 0.0)
def _get_overall_rate(self, rates: Dict[str, float]) -> float:
"""获取整体逾期率（从分箱率加权计算或从合计行获取）.
:param rates: 逾期率字典
:return: 整体逾期率
"""
if '合计' in rates:
return rates['合计']
reference = getattr(self, 'overall_rates_', {}).get(self._reference_key(rates))
if reference is not None:
return reference
# 简单平均
valid_rates = [r for l, r in rates.items() if l not in ('合计', 'missing', 'special')]
return np.mean(valid_rates) if valid_rates else 0.0
def _apply_coefficients(
self,
bin_labels: List[str],
base_rates: List[float],
coefficients_map: Dict[str, float],
) -> List[float]:
"""应用调整系数到基础逾期率.
:param bin_labels: 分箱标签列表
:param base_rates: 基础逾期率列表
:param coefficients_map: 系数映射
:return: 调整后的逾期率列表
"""
adjusted = []
for label, rate in zip(bin_labels, base_rates):
coef = coefficients_map.get(label, 1.0)
adjusted_rate = rate * coef
adjusted.append(min(adjusted_rate, 1.0)) # 逾期率不超过1
return adjusted
[文档]
def set_coefficients(self, coefficients: Union[float, Dict[str, float], str]) -> "OverduePredictor":
"""设置或更新逾期率调整系数.
:param coefficients: 调整系数，支持：
- float: 统一系数
- dict: 按分箱标签指定系数
- 'auto': 自动校正
- None: 取消调整
:return: self
"""
self.coefficients = coefficients
self.coefficients_ = coefficients
return self
[文档]
def get_report(self, metric: str = 'count') -> pd.DataFrame:
"""获取逾期率预估报告.
输出各分箱的逾期率、样本分布及预估信息。
:param metric: 统计口径，'count'（订单口径）或 'amount'（金额口径）
:return: 报告DataFrame
**参考样例**
>>> predictor = OverduePredictor(feature='score', target='target') # 初始化逾期率预估器
>>> predictor.fit(train_df) # 拟合分箱表并计算各箱逾期率
>>> report = predictor.get_report() # 获取分箱统计报告（含逾期率、样本数等）
>>> print(report)
"""
if not hasattr(self, 'bin_table_'):
raise ValueError("预估器尚未拟合，请先调用fit方法")
table = self.bin_table_.copy()
# 添加系数信息
if self.coefficients is not None:
coefficients_map = self._resolve_coefficients(
self.bin_rates_.get('_default', list(self.bin_rates_.values())[0] if self.bin_rates_ else {})
)
if isinstance(table.columns, pd.MultiIndex):
# 多级表头：在第一个目标下添加系数列
first_target = self.target_names_[0] if self.target_names_ else table.columns[1][0]
coef_col = (first_target, '调整系数')
adjusted_col = (first_target, '调整后逾期率')
rates = self.bin_rates_.get(first_target, {})
bin_col = None
for c in table.columns:
c_name = c[-1] if isinstance(c, tuple) else c
if c_name == self.bin_label_col:
bin_col = c
break
if bin_col is not None:
coefs = []
adjusted = []
for _, row in table.iterrows():
label = str(row[bin_col])
coef = coefficients_map.get(label, 1.0)
coefs.append(coef)
rate = rates.get(label, 0.0)
adjusted.append(min(rate * coef, 1.0))
table[coef_col] = coefs
table[adjusted_col] = adjusted
else:
# 单层表头
rates = self.bin_rates_.get('_default', {})
coefs = []
adjusted = []
for _, row in table.iterrows():
label = str(row.get(self.bin_label_col, ''))
coef = coefficients_map.get(label, 1.0)
coefs.append(coef)
rate = rates.get(label, 0.0)
adjusted.append(min(rate * coef, 1.0))
table['调整系数'] = coefs
table['调整后逾期率'] = adjusted
return table
[文档]
def predict(self, X: pd.DataFrame) -> Union[pd.Series, Dict[str, pd.Series]]:
"""预测逾期率（简化接口）.
直接返回逾期率预测结果，不含分箱和基础逾期率列。
:param X: 待预测数据
:return: 逾期率Series（单标签）或 {目标名: 逾期率Series}（多标签）
**参考样例**
>>> predictor = OverduePredictor(feature='score', target='target')
>>> predictor.fit(train_df)
>>> predicted_rates = predictor.predict(test_df) # 直接返回逾期率Series（单标签）或字典（多标签）
"""
result = self.transform(X)
if len(self.target_names_) <= 1:
rate_col = f'{self.feature}_预测逾期率'
return result[rate_col]
else:
output = {}
for target_name in self.target_names_:
rate_col = f'{self.feature}_{target_name}_预测逾期率'
output[target_name] = result[rate_col]
return output
[文档]
def overdue_prediction_report(
data: pd.DataFrame,
feature: str,
target: Optional[str] = None,
overdue: Optional[Union[str, List[str]]] = None,
dpds: Optional[Union[int, List[int]]] = None,
predict_data: Optional[pd.DataFrame] = None,
coefficients: Optional[Union[float, Dict[str, float], str]] = None,
method: str = 'mdlp',
max_n_bins: int = 5,
min_bin_size: float = 0.05,
missing_separate: bool = True,
bin_table: Optional[pd.DataFrame] = None,
rules: Optional[List] = None,
desc: Optional[str] = None,
excel_writer=None,
sheet: str = "逾期率预估报告",
n_jobs: Union[int, float] = -1,
parallel_backend: Optional[str] = None,
parallel_config: Optional[Dict[str, Any]] = None,
*,
overdue_operator: Optional[str] = None,
**kwargs,
) -> pd.DataFrame:
"""逾期率预估报告便捷函数.
统一的入口函数，支持从原始数据或分箱表进行逾期率预估，
并可输出包含预估结果的报告。
:param data: 有标签数据集（含target或逾期天数），或分箱表
:param feature: 特征名称
:param target: 目标变量名称
:param overdue: 逾期天数字段名称或列表
:param dpds: 逾期定义天数或列表
:param predict_data: 待预估的无标签数据（可选），
传入后会计算各样本的预估逾期率
:param coefficients: 逾期率调整系数
:param method: 分箱方法，默认'mdlp'
:param max_n_bins: 最大分箱数，默认5
:param min_bin_size: 每箱最小样本占比，默认0.05
:param missing_separate: 是否将缺失值单独分箱，默认True
:param bin_table: 现成分箱表（可选），传入后直接使用而不从data计算
:param rules: 自定义分箱切分点列表
:param desc: 特征描述
:param excel_writer: Excel文件路径或ExcelWriter对象（可选），用于输出报告
:param sheet: Excel工作表名称
:return: 包含预估结果的DataFrame
**参考样例**
>>> from hscredit.report.overdue_predictor import overdue_prediction_report
>>>
>>> # 方式一：从原始数据生成报告（自动拟合+预估）
>>> report = overdue_prediction_report(
... train_df, feature='score', target='target',
... predict_data=test_df, coefficients=1.1
... )
>>>
>>> # 方式二：从分箱表生成报告（复用现成分箱逾期率）
>>> report = overdue_estimation_report(
... bin_table, feature='score',
... predict_data=test_df
... )
>>>
>>> # 方式三：多标签场景（同时预估MOB1/MOB3等多个时间窗口的逾期率）
>>> report = overdue_estimation_report(
... train_df, feature='score',
... overdue='MOB1', dpds=[7, 15, 30],
... predict_data=test_df,
... excel_writer='overdue_report.xlsx'
... )
:param overdue_operator: 逾期标签比较符，支持 ``>``、``>=``、``<``、``<=``，默认 ``>``。
满足比较条件记为坏样本(1)，否则为好样本(0)。
"""
# 确定拟合数据来源
fit_data = bin_table if bin_table is not None else data
predictor = OverduePredictor(
feature=feature,
target=target,
overdue=overdue,
dpds=dpds,
overdue_operator=overdue_operator,
method=method,
max_n_bins=max_n_bins,
min_bin_size=min_bin_size,
missing_separate=missing_separate,
coefficients=coefficients,
rules=rules,
desc=desc,
bin_params=kwargs or None,
n_jobs=n_jobs,
parallel_backend=parallel_backend,
parallel_config=parallel_config,
)
predictor.fit(fit_data)
# 获取分箱报告
report = predictor.get_report()
# 对无标签数据预测
if predict_data is not None and isinstance(predict_data, pd.DataFrame):
predict_result = predictor.transform(predict_data)
# 将预测结果附加到报告
predict_summary = _build_prediction_summary(predict_result, predictor)
report = pd.concat([report, pd.DataFrame([{'_section': '分隔'}]), predict_summary], ignore_index=True)
# 输出到Excel
if excel_writer is not None:
from ..excel import ExcelWriter, dataframe2excel
from ..utils import init_setting
init_setting()
owns_writer = isinstance(excel_writer, str)
writer = ExcelWriter() if owns_writer else excel_writer
worksheet = writer.get_sheet_by_name(sheet)
dataframe2excel(
report, writer, worksheet,
sheet_name=sheet if owns_writer else None,
percent_cols=['坏样本率', '样本占比', '好样本占比', '坏样本占比',
'调整后逾期率', '预测逾期率'],
condition_cols=['坏样本率', '调整后逾期率'],
start_row=2,
)
if owns_writer:
writer.save(excel_writer)
return report
def _build_prediction_summary(predict_result: pd.DataFrame, predictor: OverduePredictor) -> pd.DataFrame:
"""构建预测汇总统计.
:param predict_result: transform输出结果
:param predictor: 预测器实例
:return: 汇总统计DataFrame
"""
rows = []
feature = predictor.feature
if len(predictor.target_names_) <= 1:
rate_col = f'{feature}_预测逾期率'
bin_col = f'{feature}_分箱'
if rate_col in predict_result.columns:
rows.append({
'指标名称': feature,
'指标含义': '预测汇总',
'分箱标签': '预测样本总数',
'预测逾期率': len(predict_result),
})
rows.append({
'指标名称': feature,
'指标含义': '预测汇总',
'分箱标签': '平均预测逾期率',
'预测逾期率': predict_result[rate_col].mean(),
})
if bin_col in predict_result.columns:
bin_dist = predict_result[bin_col].value_counts()
for bin_label, count in bin_dist.items():
subset = predict_result[predict_result[bin_col] == bin_label]
avg_rate = subset[rate_col].mean()
rows.append({
'指标名称': feature,
'指标含义': '分箱预测',
'分箱标签': str(bin_label),
'样本总数': count,
'预测逾期率': avg_rate,
})
else:
for target_name in predictor.target_names_:
rate_col = f'{feature}_{target_name}_预测逾期率'
if rate_col in predict_result.columns:
rows.append({
'指标名称': feature,
'指标含义': f'{target_name} 预测汇总',
'分箱标签': '平均预测逾期率',
'预测逾期率': predict_result[rate_col].mean(),
})
return pd.DataFrame(rows)