报告 hscredit.report
报告生成与规则挖掘:特征分析、规则分析、Swap 分析、逾期预测、模型报告、模型对比, 以及格式化 Excel 输出工具。
报告模块.
提供专业的模型报告生成功能。
子模块: - feature_analyzer: 特征分箱统计与自动分析 - rule_analysis: 规则集与多标签规则分析 - swap_analysis: 规则置换风险分析 - overdue_estimator: 逾期数据预估
- hscredit.report.feature_bin_stats(data, feature, target=None, overdue=None, dpds=None, rules=None, method='mdlp', desc=None, binner=None, max_n_bins=5, min_bin_size=0.05, missing_separate=True, prebinning=None, prebinning_params=None, return_cols=None, return_rules=False, del_grey=False, margins=False, amount=None, verbose=0, monotonic=None, long_format=False, n_jobs=-1, parallel_backend=None, parallel_config=None, **kwargs)[源代码]
特征分箱统计表,汇总统计特征每个分箱的各项指标信息.
支持单特征或多特征,支持单目标或多逾期标签+逾期天数组合分析。 当传入 overdue 和 dpds 时,会生成多级表头展示不同标签组合下的分箱统计。
- 参数:
data (DataFrame) -- 数据集
feature (str | List[str]) -- 特征名称或特征名称列表
target (str | None) -- 目标变量名称,默认 None
overdue (str | List[str] | None) -- 逾期天数字段名称或列表,如 'MOB1' 或 ['MOB1', 'MOB3']
dpds (int | List[int] | None) -- 逾期定义天数或列表,如 7 或 [0, 7, 30] - 逾期天数 > dpds 为坏样本(1),其他为好样本(0)
rules (List | Dict[str, List] | None) -- 自定义分箱规则,支持 list(所有特征统一规则)或 dict(按特征名映射规则)。 对 rules 中未包含的特征,按 method 参数重新训练分箱器。 优先级: binner > rules > method
method (str) --
分箱方法,可选(与 OptimalBinning.VALID_METHODS 一致,共17种): - 无监督方法: 'uniform'(等宽), 'quantile'(等频), 'kmeans'(K-Means聚类),
'kernel_density'(核密度)
- 有监督方法: 'tree'(决策树), 'cart'(CART), 'chi'(卡方), 'mdlp'(信息论),
'best_ks'(最优KS), 'best_iv'(最优IV), 'best_lift'(Best Lift), 'target_bad_rate'(目标坏样本率), 'monotonic'(单调性), 'genetic'(遗传算法), 'smooth'(平滑)
- 运筹规划方法: 'or_tools'(OR-Tools整数规划,需安装 ortools),
'cp_sat'(CP-SAT约束规划,需安装 ortools)
默认: 'mdlp'
desc (str | Dict[str, str] | None) -- 特征描述,支持 str(单个特征)或 dict(多个特征)
binner (BaseBinning | Dict[str, BaseBinning] | None) -- 分箱器,支持以下三种传入方式: - BaseBinning(已训练): 对其中已包含的特征直接使用,未包含的特征按 method 参数重新训练 - BaseBinning(未训练): 作为模板,对每个特征 deepcopy 后 fit - Dict[str, BaseBinning]: 按特征名映射的已训练分箱器字典,未包含的特征按 method 参数重新训练 优先级: binner > rules > method
max_n_bins (int) -- 最大分箱数,默认 5
min_bin_size (float) -- 每箱最小样本占比,默认 0.05
missing_separate (bool) -- 是否将缺失值单独分箱,默认 True
prebinning (str | BaseBinning | Dict | None) -- 预分箱配置,参数格式与 OptimalBinning 保持一致,默认 'quantile'。 - None: 不使用预分箱 - str: 预分箱方法名(如 'quantile' / 'tree') - BaseBinning: 预分箱器实例 - Dict: 预分箱配置字典
prebinning_params (Dict[str, Any] | None) -- 预分箱参数(传给 OptimalBinning.prebinning_params)。 默认 None,此时会使用 {'max_n_bins': 100},即先等频100箱再合并。
return_cols (List[str] | None) -- 指定返回的列名列表,默认返回所有列
return_rules (bool) -- 是否返回分箱规则,默认 False
del_grey (bool) -- 是否删除逾期天数 (0, dpds] 的灰样本,仅 overdue 起作用时有用 - True: 剔除灰样本,不同目标下样本数不同,样本数相关列按目标单独显示 - False: 保留灰样本,不同目标下样本数相同,样本数相关列作为公共列
margins (bool) -- 是否在分箱表最后添加合计行,默认 False - True: 在最后一行显示合计,缺失值和特殊值放在正常分箱之后、合计之前
amount (str | None) -- 金额字段名称,用于金额口径分析。传入后会增加金额总数、金额占比等指标
verbose (int) -- 是否输出详细信息,默认 0
monotonic (str | bool | None) -- 单调性约束,控制分箱后坏样本率的单调方向,透传给 OptimalBinning。可选值: - None: 不强制单调性约束 - 'auto_asc': 自动判断并强制单调递增 - 'auto_desc': 自动判断并强制单调递减 - 'auto_asc_desc': 自动选择最优方向(递增/递减),默认选项 - 'peak': 先升后降,适用于评分类特征 - 'valley': 先降后升 - bool: True=强制升序,False=强制降序 注意:需配合 method 参数使用,部分 method 默认已包含单调约束(如 'monotonic' 方法)
long_format (bool) --
分箱表输出格式,默认 False - False: 沿用原样式。多目标时使用多级表头(
分箱详情+ 各逾期标签)按列展开 - True: 长格式输出。各逾期标签纵向堆叠,新增逾期标签列标识目标,列顺序为 指标名称/指标含义/逾期标签/分箱标签/样本总数/样本占比/好样本数/坏样本数/... 单目标时同样会输出
逾期标签列。margins=True时按各逾期标签分组分别追加合计行kwargs -- 其他分箱器参数(如 lift_refine、prebinning 等)
n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict[str, Any] | None)
- 返回:
pd.DataFrame: 特征分箱统计表
Tuple[pd.DataFrame, Dict]: 当 return_rules=True 时返回 (统计表, 分箱规则)
- 返回类型:
DataFrame | Tuple[DataFrame, Dict]
参考样例
>>> # 单特征单目标分析 >>> table = feature_bin_stats(data, 'score', target='target', method='mdlp') >>> >>> # 单特征多逾期标签分析 >>> table = feature_bin_stats(data, 'score', overdue=['MOB1', 'MOB3'], dpds=[0, 7]) >>> >>> # 多特征分析 >>> table = feature_bin_stats(data, ['score', 'age'], overdue='MOB1', dpds=7) >>> >>> # 使用自定义分箱规则 >>> table = feature_bin_stats(data, 'score', rules=[300, 500, 700]) >>> >>> # 使用单调性分箱 >>> table = feature_bin_stats(data, 'score', method='mdlp', monotonic='peak') >>> >>> # 使用单调性约束 + 强制升序 >>> table = feature_bin_stats(data, 'score', method='mdlp', monotonic='auto_asc') >>> >>> # 直接使用 monotonic 方法 >>> table = feature_bin_stats(data, 'score', method='monotonic', monotonic='peak') >>> >>> # 金额口径分析 >>> table = feature_bin_stats(data, 'score', target='target', amount='loan_amount') >>> >>> # 长格式输出:多逾期标签纵向堆叠,新增"逾期标签"列 >>> table = feature_bin_stats(data, 'score', overdue='MOB1', dpds=[15, 0], long_format=True)
- hscredit.report.feature_binning_summary(data, feature, methods='mdlp', bin_params=None, target=None, overdue=None, dpds=None, desc=None, max_n_bins=5, min_n_bins=2, min_bin_size=0.05, max_bin_size=None, min_bad_rate=0.0, missing_separate=True, prebinning=None, prebinning_params=None, special_codes=None, cat_cutoff=None, random_state=None, decimal=4, woe_clip=None, del_grey=False, margins=False, amount=None, verbose=0, monotonic=None, long_format=False, metrics=('分档KS值', 'LIFT值', '指标IV值', '坏样本数', '坏样本率'), n_jobs=-1, parallel_backend=None, parallel_config=None, **kwargs)[源代码]
对一个或多个字段执行多种分箱,并生成跨方法摘要。
bin_params支持两种格式:单层参数应用于所有方法;以 method 为 key、 参数字典为 value 的多层参数仅应用于对应方法。参数优先级为bin_params > 显式公共参数/kwargs。long_format控制binning_tables中每张分箱表的输出格式:默认False沿用原多级表头样式(多目标时按列展开);设为True时每张表按逾期标签列将各目标纵向堆叠输出(透传给feature_bin_stats())。摘要binning_summary的结构不受影响。metrics指定binning_summary中按目标汇总的指标及其展示顺序,默认['分档KS值', 'LIFT值', '指标IV值', '坏样本数', '坏样本率']。可选值为feature_bin_stats()输出的指标列,各指标的跨分箱聚合方式如下(合计行不参与):求和(sum):
样本总数/好样本数/坏样本数/分档IV值/样本占比/好样本占比/坏样本占比取最大值(max):
指标IV值/LIFT值/坏账改善/风险拒绝比/累积LIFT值/累积坏账改善/累计风险拒绝比/累积好样本数/累积坏样本数/分档KS值取绝对值最大值(max_abs):
分档WOE值总体坏样本率(bad_rate):
坏样本率按sum(坏样本数) / sum(样本总数)重新计算
传入不受支持的指标将抛出
ValueError。- 返回:
(binning_tables, binning_summary)。分箱表结构为{feature: {method: binning_table}},摘要使用两级列索引。- 参数:
data (DataFrame)
feature (str | List[str])
methods (str | List[str])
bin_params (Dict[str, Any] | None)
target (str | None)
overdue (str | List[str] | None)
dpds (int | List[int] | None)
desc (str | Dict[str, str] | None)
max_n_bins (int)
min_n_bins (int)
min_bin_size (float)
max_bin_size (float | int | None)
min_bad_rate (float)
missing_separate (bool)
prebinning (str | BaseBinning | Dict | None)
prebinning_params (Dict[str, Any] | None)
special_codes (List | None)
cat_cutoff (float | int | None)
random_state (int | None)
decimal (int)
woe_clip (float | None)
del_grey (bool)
margins (bool)
amount (str | None)
verbose (int)
monotonic (str | bool | None)
long_format (bool)
metrics (str | List[str])
n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict[str, Any] | None)
- 返回类型:
Tuple[Dict[str, Dict[str, DataFrame]], DataFrame]
参考样例
>>> tables, summary = feature_binning_summary( ... data, ['score', 'age'], methods=['quantile', 'mdlp'], ... overdue='MOB1', dpds=[3, 1, 0], max_n_bins=5, ... bin_params={'mdlp': {'min_bin_size': 0.1}}, ... )
- hscredit.report.feature_group_binning_summary(data, feature, methods='mdlp', date_col=None, freq='M', group_col=None, group_order='asc', dropna=True, bin_params=None, target=None, overdue=None, dpds=None, desc=None, max_n_bins=5, min_n_bins=2, min_bin_size=0.05, max_bin_size=None, min_bad_rate=0.0, missing_separate=True, prebinning=None, prebinning_params=None, special_codes=None, cat_cutoff=None, random_state=None, decimal=4, woe_clip=None, del_grey=False, margins=False, amount=None, verbose=0, monotonic=None, long_format=False, metrics=('分档KS值', 'LIFT值', '指标IV值', '坏样本数', '坏样本率'), n_jobs=-1, parallel_backend=None, parallel_config=None, **kwargs)[源代码]
统计日期周期或类别分组下的特征分箱效果。
分箱器在全量数据上拟合一次,各分组复用同一套分箱规则,因此不同日期周期或 类别分组下的坏样本率、LIFT、KS、IV 等指标可以直接横向比较。
- 参数:
data (DataFrame) -- 原始明细数据
feature (str | List[str]) -- 待分析特征名或特征名列表
methods (str | List[str]) -- 分箱方法或方法列表
date_col (str | None) -- 日期字段,与
freq配合生成时间分组;与group_col二选一freq (str) -- 日期频率,支持
D/W/M/Q,默认按月group_col (str | None) -- 类别分组字段;与
date_col二选一group_order (None | str | Callable[[Any], Any] | Sequence[Any]) -- 分组顺序,支持升序、降序、出现顺序、排序函数或显式列表
dropna (bool) -- 是否删除分组字段缺失样本;为 False 时归入“缺失”组
bin_params (Dict[str, Any] | None) -- 全局或按分箱方法配置的参数,规则与
feature_binning_summary()一致metrics (str | List[str]) -- summary 汇总指标,规则与
feature_binning_summary()一致target (str | None)
overdue (str | List[str] | None)
dpds (int | List[int] | None)
desc (str | Dict[str, str] | None)
max_n_bins (int)
min_n_bins (int)
min_bin_size (float)
max_bin_size (float | int | None)
min_bad_rate (float)
missing_separate (bool)
prebinning (str | BaseBinning | Dict | None)
prebinning_params (Dict[str, Any] | None)
special_codes (List | None)
cat_cutoff (float | int | None)
random_state (int | None)
decimal (int)
woe_clip (float | None)
del_grey (bool)
margins (bool)
amount (str | None)
verbose (int)
monotonic (str | bool | None)
long_format (bool)
n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict[str, Any] | None)
- 返回:
(binning_tables, binning_summary)。分箱表结构为{feature: {method: {group: binning_table}}};summary 使用两级列索引。- 返回类型:
Tuple[Dict[str, Dict[str, Dict[str, DataFrame]]], DataFrame]
参考样例
>>> tables, summary = feature_group_binning_summary( ... data, feature='score', methods=['quantile', 'mdlp'], ... date_col='申请日期', freq='M', overdue='MOB1', dpds=[3, 0], ... ) >>> category_tables, category_summary = feature_group_binning_summary( ... data, feature='score', group_col='商品类别', target='FPD', ... )
- hscredit.report.feature_efficiency_analysis(data, feature, manual_rules=None, target='target', overdue=None, dpd=0, auto_method='mdlp', desc=None, date_col=None, group_cols=None, date_freq='M', max_n_bins=5, min_bin_size=0.05, missing_separate=True, prebinning='quantile', prebinning_params=None, del_grey=False, margins=False, amount=None, figsize=(15, 10), trend_figsize=None, comparison_orientation='horizontal', auto_kwargs=None, trend_kwargs=None, output_dir=None, suffix='', quantiles=None, rule_decimals=4, save=None, n_jobs=-1, parallel_backend=None, parallel_config=None)[源代码]
特征效率分析:对比手工分箱与自动分箱效果,并输出趋势图。
适用于单个数值型指标或评分变量的快速效果评估。函数会: 1. 自动生成分位数分箱规则(默认使用 [0.01, 0.03, 0.05, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 0.95, 0.97, 0.99]) 2. 生成手工分箱与自动分箱两张分箱表 3. 输出 2×2 组合图:上排为手工/自动分箱图,下排为 KS/ROC 曲线 4. 当传入日期字段或分组字段时,额外输出手工分箱与自动分箱两张 bin_trend_plot 趋势图
- 参数:
data (DataFrame) -- 输入数据集
feature (str) -- 需要分析的特征名,建议为数值型指标/评分
manual_rules (List | Tuple | ndarray | Dict[str, List] | None) -- 手工分箱边界,支持 list 或 {feature: list}。默认 None,表示自动使用 quantiles 生成分箱边界
target (str) -- 目标变量列名,默认 target
overdue (str | List[str] | None) -- 逾期列名。传入后会基于 overdue > dpd 自动构造二分类目标
dpd (int) -- 逾期阈值,仅在 overdue 模式下使用,默认 0
auto_method (str) -- 自动分箱方法,默认 mdlp
desc (str | None) -- 特征中文描述,默认使用 feature
date_col (str | None) -- 日期列,传入后生成按时间分组的趋势图
group_cols (str | List[str] | None) -- 分组字段,支持单列或多列,传入后生成分组趋势图
date_freq (str) -- 日期聚合频率,默认 M
max_n_bins (int) -- 自动分箱最大箱数,默认 5
min_bin_size (float) -- 自动分箱最小箱占比,默认 0.05
missing_separate (bool) -- 缺失值是否单独分箱,默认 True
prebinning (str | BaseBinning | Dict | None) -- 预分箱配置,默认 quantile
prebinning_params (Dict[str, Any] | None) -- 预分箱参数,默认 None
del_grey (bool) -- overdue 模式下是否剔除灰样本,默认 False
margins (bool) -- 是否追加合计行,默认 False
amount (str | None) -- 金额字段,传入后输出金额口径分箱表
figsize (Tuple[float, float]) -- 2×2 组合图尺寸,默认 (15, 10)
trend_figsize (Tuple[float, float] | None) -- 趋势图尺寸,默认 None(由 bin_trend_plot 自动计算)
comparison_orientation (str) -- 两张分箱图的方向,默认 horizontal
auto_kwargs (Dict[str, Any] | None) -- 额外传给自动分箱 feature_bin_stats 的参数
trend_kwargs (Dict[str, Any] | None) -- 额外传给 bin_trend_plot 的参数
output_dir (str | None) -- 图片保存目录,默认 None(不落盘)
suffix (str) -- 保存文件名后缀,默认空字符串
quantiles (List[float] | None) -- 分位数列表,用于自动生成分箱边界。默认 [0.01, 0.03, 0.05, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 0.95, 0.97, 0.99]
rule_decimals (int) -- 分箱边界保留的小数位数,默认 4
save (str | None) -- 比较图片保存路径,如果提供则将生成的比较图保存至指定路径,默认为 None
n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict[str, Any] | None)
- 返回:
dict,包含分箱表、分箱规则、组合图与趋势图
- 返回类型:
Dict[str, Any]
Example:
>>> # 自动使用分位数生成手工分箱规则 >>> result = feature_efficiency_analysis( ... data=df, ... feature='score', ... target='target', ... auto_method='mdlp', ... date_col='apply_date' ... ) >>> result['manual_table'] >>> result['comparison_figure'] >>> # 手动指定分箱规则 >>> result = feature_efficiency_analysis( ... data=df, ... feature='score', ... manual_rules=[450, 520, 600, 680], ... target='target', ... auto_method='mdlp' ... )
- hscredit.report.auto_feature_analysis(data, features=None, target='target', overdue=None, dpds=None, date=None, data_summary_comment='', freq='M', excel_writer=None, sheet='分析报告', start_col=2, start_row=2, dropna=False, writer_params=None, bin_params=None, feature_map=None, corr=False, pictures=None, suffix='', output_dir='model_report', margins=False, amount=None, image_table_gap_rows=None, n_jobs=-1, parallel_backend=None, parallel_config=None, condition_color='F76E6C')[源代码]
自动特征分析.
用于三方数据评估或自有评分效果评估。生成包含数据集概况、特征分箱统计、 KS 曲线、分布图等内容的 Excel 分析结果。
- 参数:
data (DataFrame) -- 需要评估的数据集,需要包含目标变量
features -- 需要进行分析的特征名称,支持单个字符串或列表
target -- 目标变量名称
overdue -- 逾期天数字段名称,传入时会覆盖 target 参数
dpds -- 逾期定义方式,逾期天数 > DPD 为坏样本
date -- 日期列,用于时间维度分布分析
freq -- 日期统计粒度,默认按月 "M"
data_summary_comment -- 数据备注信息
excel_writer -- Excel 文件路径或 ExcelWriter 对象
sheet -- 工作表名称
start_col -- 起始列
start_row -- 起始行
dropna -- 是否剔除缺失值
writer_params -- Excel 写入器初始化参数
bin_params -- 分箱统计参数,支持 feature_bin_stats 的参数
feature_map -- 特征名称映射字典
corr -- 是否计算特征相关性
pictures -- 需要生成的图片列表,支持 ["ks", "hist", "bin"]
suffix -- 文件名后缀,避免同名文件被覆盖
output_dir -- 图片输出目录
margins -- 是否在每个特征分箱表末尾添加合计行,默认 False
amount -- 放款金额或余额字段名称。传入后同时生成订单口径和金额口径两张分箱表
image_table_gap_rows -- 图片区与分箱表之间的额外空行数
condition_color -- 条件格式颜色,默认使用副主题色
"F76E6C";支持颜色字符串、色阶列表或按列配置的字典
- 返回:
(end_row, end_col) 分析结束位置
参考样例
>>> from hscredit.report.feature_analyzer import auto_feature_analysis >>> auto_feature_analysis(data, features=['feature1'], target='target', excel_writer='分析结果.xlsx')
- hscredit.report.ruleset_analysis(datasets, rules, target='target', overdue=None, dpds=None, filter_cols=None, amount=None, n_jobs=-1, parallel_backend=None, parallel_config=None, **kwargs)[源代码]
用于D类调优时的规则集效果分析.
分析规则集在数据集上的应用效果,展示原始样本、每条规则命中效果、 各规则剩余样本以及所有规则合计命中效果。
- 参数:
datasets (DataFrame) -- 数据集
rules (List[Rule]) -- 规则列表
target (str) -- 目标变量名称
overdue (str | List[str] | None) -- 逾期天数字段名称(支持多标签,传入列表)
dpds (int | List[int] | None) -- 逾期定义方式(支持多标签,传入列表)
filter_cols (List[str] | None) -- 指定返回的字段列表
amount (str | None) -- 金额字段名称,用于金额口径分析
n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict | None)
- 返回:
规则集效果评估表。单标签时返回单层列结构,多标签时返回多层列结构(MultiIndex)
- 返回类型:
DataFrame
参考样例
>>> from hscredit.core.rules import Rule >>> from hscredit.report import ruleset_analysis >>> rules = [Rule("score < 600", name="低分"), Rule("多头 > 5", name="多头高")] >>> # 单标签 >>> ruleset_analysis(df, rules, target='FPD') >>> # 多逾期标签 + 金额口径 >>> ruleset_analysis(df, rules, overdue=['MOB1', 'MOB3'], dpds=[7, 0], amount='放款金额')
- hscredit.report.multi_label_rule_analysis(df, features, labels, miner_params=None, output_path='rule_analysis.xlsx', n_jobs=-1, parallel_backend=None, parallel_config=None)[源代码]
多标签规则分析(Excel 输出).
报告包含: - 规则汇总:各规则在每个标签下的覆盖率/坏率/LIFT/有效性分类 - 有效性矩阵:行=规则,列=标签,格=LIFT值 - 规则分类统计:按规则类型分组的汇总统计
- 参数:
df (DataFrame) -- 输入数据 DataFrame
features (List[str]) -- 参与挖掘的特征列表
labels (Dict[str, str]) -- 标签映射 {中文名: 列名}
miner_params (dict | None) -- 传递给 MultiLabelRuleMiner 的额外参数(如 min_support、min_lift)
output_path (str) -- 输出 Excel 文件路径,默认
'rule_analysis.xlsx'n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict | None)
- 返回:
输出文件路径(即
output_path)- 返回类型:
str
参考样例
>>> from hscredit.report import multi_label_rule_analysis >>> multi_label_rule_analysis( ... df, ... features=['score', '近六个月非银多头机构数', '青云24'], ... labels={'首逾7+': 'fpd7', '首逾0+': 'fpd0'}, ... output_path='多标签规则分析.xlsx', ... )
- hscredit.report.rule_swap_analysis(data, score, rules_in=None, rules_out=None, rules_base=None, reference_data=None, bin_table=None, target=None, overdue=None, dpds=None, score_weights=None, out_in_uplift=2.0, amount=None, sample_survival_rate=1.0, reverse_order=False, out_in_amount_fill=None, out_in_amount_col=None, bin_method='quantile', max_n_bins=10, min_bin_size=0.05, missing_separate=True, bin_params=None, rule_analysis_mode='independent', n_jobs=-1, parallel_backend=None, parallel_config=None)[源代码]
规则置入置出(Swap)分析。
整合自 scorecardpipeline 的
swapin_report和ruleset_analysis``(即 swapout_report), 只输出 ``swap_pipeline和swap_result,支持金额和订单口径。四象限定义
- 参数:
data (DataFrame) -- 全量样本集(包含 score 列 + rules_in/rules_out/rules_base 用到的所有特征列)
score (str | Dict[str, str]) -- 评分字段名(str)或多评分映射(Dict)
rules_in (List[Rule] | None) -- 置入规则集(List[Rule]),对应 out_in 象限
rules_out (List[Rule] | None) -- 置出规则集(可选),对应 in_out 象限
rules_base (List[Rule] | None) -- 基准拒绝规则集(可选),对应 out_out 象限
reference_data (DataFrame | None) -- 历史有表现参考数据集(包含 target 或 overdue+dpds)
bin_table (DataFrame | Dict[str, DataFrame] | None) -- 现成分箱表,支持: - pd.DataFrame:单评分分箱表 - Dict[str, pd.DataFrame]:多评分分箱表
{评分名: 分箱表}- None:自动从 reference_data 计算target (str | None) -- 目标变量名(与 bin_table 二选一)
overdue (str | List[str] | None) -- 逾期天数字段(多标签场景)
dpds (int | List[int] | None) -- 逾期天数阈值
score_weights (Dict[str, float] | None) -- 多模型权重(可选)
out_in_uplift (float) -- 置入风险上浮系数,默认 2.0
amount (str | None) -- 金额字段(可选),传入后同时输出金额口径报告
sample_survival_rate (float) -- 样本集幸存比例,默认 1.0
reverse_order (bool) -- 是否逆序展示(True: 从置入效果开始展示)
out_in_amount_fill (float | None) -- out_in 置入样本额度填充定值(可选)
out_in_amount_col (str | None) -- out_in 置入样本额度填充字段名(可选)
bin_method (str) -- 分箱方法,默认 'quantile'(仅 reference_data 模式生效)
max_n_bins (int) -- 最大分箱数,默认 10(仅 reference_data 模式生效)
min_bin_size (float) -- 每箱最小样本占比,默认 0.05(仅 reference_data 模式生效)
missing_separate (bool) -- 是否将缺失值单独分箱,默认 True
bin_params (dict | None) -- 额外分箱参数 dict,会透传给
feature_bin_statsrule_analysis_mode (str) -- 规则分析模式,默认 'independent'。 - 'independent':每条规则独立应用到全量 data,分别统计命中好坏分布。 - 'sequential':漏斗模式,每条规则在前一条拒绝后的剩余样本上分析。
n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict | None)
- 返回:
包含两张表的字典
swap_pipeline:分步骤通过率与逾期率变化(可逆序),支持订单/金额双口径swap_result:置换前后对比与业务增益
- 返回类型:
Dict[str, DataFrame]
参考样例
>>> from hscredit.core.rules import Rule >>> from hscredit.report.rule_analysis import rule_swap_analysis >>> >>> # 置入规则分析(传入历史参考数据,自动计算分箱表) >>> result = rule_swap_analysis( ... data=swap_data, ... score='score_a', ... rules_in=[rule_in], ... rules_base=[rule_base], ... reference_data=hist_data, ... target='target', ... amount='放款金额', ... ) >>> >>> print(result['swap_pipeline']) # 分步骤报告 >>> print(result['swap_result']) # 置换前后对比
>>> # 多逾期标签分析 >>> result = rule_swap_analysis( ... data=swap_data, ... score='score_a', ... rules_in=[rule_in], ... reference_data=hist_data, ... overdue='MOB1', ... dpds=[0, 7, 30], ... amount='放款金额', ... )
- hscredit.report.rule_report_table(report, rule_name=None, target_names=None, metrics=None, target_name='target', n_jobs=-1, parallel_backend=None, parallel_config=None)[源代码]
生成按逾期指标横向展开的规则详情表.
- 参数:
report (DataFrame) --
Rule.report返回的 DataFramerule_name (str | None) -- 展示用规则名称,默认使用报告中的指标名称
target_names (Mapping[str, str] | None) -- 逾期指标名称映射,如
{'MOB1 1+': 'fpd1'}metrics (Sequence[str] | None) -- 每个逾期指标需要展示的字段,默认使用内置
_DEFAULT_METRICStarget_name (str) -- 单标签报告的逾期指标名称,默认
"target"n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict[str, Any] | None)
- 返回:
两层列头的规则详情表
- 返回类型:
DataFrame
参考样例
>>> from hscredit.core.rules import Rule >>> from hscredit.report import rule_report_table >>> rep = Rule("score < 600").report(data, target='FPD') >>> rule_report_table(rep, rule_name='低分拒绝')
- hscredit.report.rule_target_analysis(report, current_pass_rate=1.0, rule_name=None, target_names=None, target_name='target', n_jobs=-1, parallel_backend=None, parallel_config=None)[源代码]
生成拒绝规则目标分析表.
绝对逾期改善为合计坏样本率减未命中坏样本率,绝对通过率为规则样本内的 未命中占比;相对逾期改善以合计坏样本率为分母,相对通过率则在现有策略 通过率基础上计算,即
current_pass_rate * 绝对通过率。- 参数:
report (DataFrame) --
Rule.report返回的 DataFramecurrent_pass_rate (float | None) -- 规则执行前的当前通过率,取值范围为[0, 1]
rule_name (str | None) -- 展示用规则名称
target_names (Mapping[str, str] | None) -- 逾期指标名称映射
target_name (str) -- 单标签报告的逾期指标名称,默认
"target"n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict[str, Any] | None)
- 返回:
两层列头的目标分析表
- 返回类型:
DataFrame
参考样例
>>> from hscredit.core.rules import Rule >>> from hscredit.report import rule_target_analysis >>> rep = Rule("score < 600").report(data, target='FPD') >>> # 当前通过率 0.8 时,评估该拒绝规则带来的逾期改善与通过率变化 >>> rule_target_analysis(rep, current_pass_rate=0.8, rule_name='低分拒绝')
- hscredit.report.rule_target_table(report, rule_name=None, target_names=None, metrics=None, target_name='target', n_jobs=-1, parallel_backend=None, parallel_config=None)[源代码]
生成规则、逾期指标和命中情况组成的纵向明细表.
与
rule_report_table()的横向展开不同,本函数按规则 × 逾期指标 × 命中情况逐行纵向罗列各项指标,便于直接落库或透视。- 参数:
report (DataFrame) --
Rule.report返回的 DataFramerule_name (str | None) -- 展示用规则名称,默认使用报告中的指标名称
target_names (Mapping[str, str] | None) -- 逾期指标名称映射,如
{'MOB1 1+': 'fpd1'}metrics (Sequence[str] | None) -- 需要展示的字段列表,默认使用内置
_DEFAULT_TARGET_METRICStarget_name (str) -- 单标签报告的逾期指标名称,默认
"target"n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict[str, Any] | None)
- 返回:
含
规则详情/逾期指标/命中情况及各指标列的纵向明细表- 返回类型:
DataFrame
参考样例
>>> from hscredit.core.rules import Rule >>> from hscredit.report import rule_target_table >>> rep = Rule("score < 600").report(data, target='FPD') >>> rule_target_table(rep, rule_name='低分拒绝')
- hscredit.report.rule_group_hit_table(group_reports, rule_name=None, target_names=None, metrics=None, target_name='target', n_jobs=-1, parallel_backend=None, parallel_config=None)[源代码]
生成多个样本分组下的规则命中效果对比表.
- 参数:
group_reports (Mapping[str, DataFrame]) -- 分组名称到
Rule.report结果的映射rule_name (str | None) -- 展示用规则名称
target_names (Mapping[str, str] | None) -- 逾期指标名称映射
metrics (Mapping[str, str] | None) -- 顶层展示名称到
Rule.report字段名的映射target_name (str) -- 单标签报告的逾期指标名称,默认
"target"n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict[str, Any] | None)
- 返回:
两层列头的分组命中对比表,不包含合计行
- 返回类型:
DataFrame
参考样例
>>> from hscredit.core.rules import Rule >>> from hscredit.report import rule_group_hit_table >>> r = Rule("score < 600") >>> # 对比训练集 / 测试集 / OOT 三个分组下同一规则的命中效果 >>> reports = { ... '训练集': r.report(train, target='FPD'), ... '测试集': r.report(test, target='FPD'), ... 'OOT': r.report(oot, target='FPD'), ... } >>> rule_group_hit_table(reports, rule_name='低分拒绝')
- hscredit.report.rule_group_compare(data, rule, date_col=None, freq='M', group_col=None, target='target', overdue=None, dpds=None, rule_name=None, target_names=None, metrics=None, prior_rules=None, amount=None, del_grey=False, dropna=True, group_order='asc', n_jobs=-1, parallel_backend=None, parallel_config=None, **kwargs)[源代码]
直接从原始数据生成分组下的规则命中效果对比表.
相比
rule_group_hit_table()需要在函数外手工切分样本并逐组调用Rule.report,本函数接收原始明细数据,按日期列 + 频率或分组字段自动切分样本,对每个分组调用同一规则的Rule.report``(支持 ``target单标签或overdue + dpds多标签口径),再汇总为分组对比表。- 参数:
data (DataFrame) -- 原始明细数据 DataFrame,需包含规则所需字段、目标/逾期字段及分组依据列
date_col (str | None) -- 日期列名,与
freq配合按时间周期分组(与group_col二选一)freq (str) -- 时间频率,
'D'日 /'W'周 /'M'月 /'Q'季度,默认'M'group_col (str | None) -- 分组字段列名,按其取值分组(与
date_col二选一)target (str) -- 目标变量列名,默认
"target",0=好样本,1=坏样本overdue (str | List[str] | None) -- 逾期天数字段名(可选,传入时以逾期天数>DPD定义坏样本,支持多标签)
dpds (int | List[int] | None) -- 逾期定义方式,逾期天数 > DPD 为坏样本,可传入列表支持多DPD联合分析
rule_name (str | None) -- 展示用规则名称,默认使用规则自身名称或报告中的指标名称
target_names (Mapping[str, str] | None) -- 逾期指标名称映射,如
{'MOB1 1+': 'fpd1'}metrics (Mapping[str, str] | None) -- 顶层展示名称到
Rule.report字段名的映射,默认_DEFAULT_GROUP_METRICSprior_rules (Rule | None) -- 先验规则(可选),每个分组内先排除命中先验规则的样本再评估
amount (str | None) -- 金额字段名(可选),传入时以金额口径而非样本数口径统计
del_grey (bool) -- 是否删除逾期天数在(0, DPD]区间内的灰度样本,默认为False
dropna (bool) -- 是否丢弃分组依据缺失的样本,默认为True;为False时缺失样本归入“缺失”分组
group_order (None | str | Callable[[Any], Any] | Sequence[Any]) --
分组排列方式,默认
"asc"升序。支持:"asc"/"desc"— 按分组标签升序 / 降序"appearance"— 按分组在数据中首次出现的顺序可调用对象 — 作为
sorted的key排序键分组名称序列 — 按给定顺序排列,未列出的分组按出现顺序追加在末尾
kwargs (Any) -- 透传给
Rule.report的其他参数(如desc、filter_cols、margins等)n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict[str, Any] | None)
- 返回:
两层列头的分组命中对比表,不包含合计行;列头第二层为各分组名称
- 抛出:
ValueError --
date_col与group_col未二选一,或所需列缺失时- 返回类型:
DataFrame
参考样例
>>> from hscredit.report import rule_group_compare >>> # 按放款月份对比同一拒绝规则在各月样本上的命中效果(多标签口径) >>> rule_group_compare( ... data, "score < 600", date_col='放款时间', freq='M', ... overdue=['MOB1'], dpds=[7, 0], rule_name='低分拒绝', ... ) >>> # 按商品类别分组、金额口径,并自定义分组展示顺序 >>> rule_group_compare( ... data, "score < 600", group_col='商品类别', target='FPD', ... amount='放款金额', group_order=['手机通讯', '电脑数码', '家用电器'], ... )
- hscredit.report.swap_out_report(data, rules, background=None, summary=None, describe=None, rule_summary=None, impact=None, target='target', overdue=None, dpds=None, save=None, verbose=False, methods='quantile', bin_params=None, features=None, amount=None, date_col=None, freq='M', group_col=None, current_pass_rate=1.0, prior_rules=None, del_grey=False, target_names=None, theme_color='2639E9', sheet_name='策略迭代', n_jobs=-1, parallel_backend=None, parallel_config=None, **kwargs)[源代码]
生成拒绝规则置换(策略迭代)分析报告,输出 hscredit 美化后的 Excel 文件.
参考「策略迭代参考模板」组织内容,输出两个工作表:
策略迭代主表:迭代背景、策略迭代总结、样本描述(样本情况 + 相关系数)、 规则变量效果(分箱详情)、业务影响情况分析(订单/金额口径并排)、规则效果分析 (命中/未命中明细,订单/金额口径并排)、规则稳定性分析(按时间或分组对比)。变量分箱明细表:各分箱方法的完整分箱表横向并排展示。
其中「规则变量效果」与「变量分箱」均源自
feature_binning_summary()的一次计算结果(binning_summary与binning_tables)。当rules仅传入单条 规则时,仅展示整体效果,不再拆分子规则。参数
- 参数:
data (DataFrame) -- 原始明细数据 DataFrame,需包含规则字段、目标/逾期字段及所需分析列
rules (str | Rule | Sequence[str | Rule]) -- 拒绝规则,单条
Rule/表达式字符串,或其列表; 多条规则时整体为各规则按「或」组合的并集,并逐条展示子规则效果background (str | List[str] | None) -- 迭代背景文本,
str或 ``list[str]``(列表自动加「序号、内容」)summary (str | List[str] | None) -- 策略迭代总结文本,
str或list[str]describe (str | List[str] | None) -- 样本描述文本,
str或list[str]rule_summary (str | List[str] | None) -- 规则表里效果说明文本,
str或list[str]impact (str | List[str] | None) -- 业务影响情况说明文本,
str或list[str]target (str) -- 目标变量列名,默认
"target",0=好样本,1=坏样本overdue (str | List[str] | None) -- 逾期天数字段名(可选,传入时以逾期天数>DPD定义坏样本,支持多标签)
dpds (int | List[int] | None) -- 逾期定义方式,逾期天数 > DPD 为坏样本,可传入列表支持多DPD联合分析
save (str | None) -- 报告保存路径(
.xlsx);为 None 时不落盘,仅返回 ExcelWriterverbose (bool) -- 是否打印计算进度,默认 False
methods (str | List[str]) -- 分箱详情所用分箱方法,
str或list[str],默认"quantile"bin_params (Dict[str, Any] | Dict[str, Dict[str, Any]] | None) -- 分箱参数,``dict``(所有方法统一)或 ``dict[method: dict]``(按方法名映射)
features (List[str] | None) -- 参与相关系数与分箱详情的指标列表,默认取所有规则引用到的字段
amount (str | None) -- 金额字段名(可选),传入时额外输出金额口径的业务影响与规则效果
date_col (str | None) -- 日期列名,与
freq配合做规则稳定性分析(与group_col二选一)freq (str) -- 稳定性分析的时间频率,
'D'/'W'/'M'/'Q',默认'M'group_col (str | None) -- 分组字段列名,做规则稳定性分析(与
date_col二选一)current_pass_rate (float) -- 规则执行前的当前通过率,取值 [0, 1],默认 1.0
prior_rules (Rule | None) -- 先验规则(可选),评估前先排除命中先验规则的样本
del_grey (bool) -- 是否删除逾期天数在 (0, DPD] 区间内的灰度样本,默认 False
target_names (Mapping[str, str] | None) -- 逾期指标名称映射,如
{'MOB1 7+': 'fpd7'}theme_color (str) -- Excel 主题色(不含 #),默认
"2639E9"sheet_name (str) -- 报告工作表名称,默认
"策略迭代"kwargs (Any) -- 透传给
Rule.report的其他参数(如desc、margins等)n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict[str, Any] | None)
- 返回:
ExcelWriter实例(已写入全部内容)
参考样例
>>> from hscredit.report import swap_out_report >>> swap_out_report( ... data, ... rules=["衡枢鉴真分老客版 < 0.05", "近六个月非银多头机构数 > 30"], ... background="为压降逾期,筛选近期放款老客样本进行策略迭代验证", ... summary=["低分拒绝规则性价比最高", "多头机构数规则作为补充收紧尾部"], ... overdue=["MOB1"], dpds=[7, 0], amount="放款金额", ... date_col="放款时间", freq="M", methods=["quantile", "mdlp"], ... save="策略迭代报告.xlsx", ... )
- class hscredit.report.ReferenceDataProvider(score_col='score', target_cols='target', amount_col=None, method='quantile', max_n_bins=10, custom_bins=None, n_jobs=-1, parallel_backend=None, parallel_config=None)[源代码]
基类:
BaseEstimator参考数据提供者.
从有标签的参考数据计算评分区间逾期率,用于swap分析中的风险预估。
- 参数:
score_col (str) -- 评分字段名
target_cols (str | List[str]) -- 目标变量字段名或列表
amount_col (str | None) -- 金额字段名(可选)
method (str) -- 分箱方法,默认'quantile'
max_n_bins (int) -- 最大分箱数,默认10
custom_bins (List[float] | None) -- 自定义分箱边界(可选)
n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict[str, Any] | None)
- class hscredit.report.SwapAnalyzer(config=None, ref_provider=None, n_jobs=-1, parallel_backend=None, parallel_config=None)[源代码]
基类:
BaseEstimatorSwap规则置换分析器.
对swap数据进行风险预估和指标计算。
- 参数:
config (SwapRiskConfig | None) -- Swap风险分析配置
ref_provider (ReferenceDataProvider | None) -- 参考数据提供者(可选)
n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict[str, Any] | None)
- analyze(swap_df, ref_provider=None)[源代码]
执行swap分析.
- 参数:
swap_df (DataFrame) -- swap数据集
ref_provider (ReferenceDataProvider | None) -- 参考数据提供者(可选)
- 返回:
SwapAnalysisResult分析结果
- 返回类型:
- class hscredit.report.SwapAnalysisResult(count_stats, amount_stats, count_combined, amount_combined, config, targets, pass_rate_analysis, risk_rejection_metrics, risk_rejection_metrics_dict, total_samples)[源代码]
基类:
objectSwap分析结果.
- 参数:
count_stats (Dict[SwapType, Dict]) -- 订单口径统计
amount_stats (Dict[SwapType, Dict] | None) -- 金额口径统计(可选)
count_combined (Dict[str, Dict]) -- 订单口径组合统计
amount_combined (Dict[str, Dict] | None) -- 金额口径组合统计(可选)
config (SwapRiskConfig) -- Swap风险分析配置
targets (List[str]) -- 目标变量列表
pass_rate_analysis (PassRateAnalysis) -- 通过率分析结果
risk_rejection_metrics (RiskRejectionMetrics) -- 风险拒绝率指标
risk_rejection_metrics_dict (Dict[str, RiskRejectionMetrics]) -- 各标签的风险拒绝率指标
total_samples (int) -- 总样本数
- get_detail_report(metric='count', target=None)[源代码]
生成详细报告(四象限).
- 参数:
metric (str) -- 'count'订单口径或'amount'金额口径
target (str) -- 指定标签,默认使用第一个
- 返回:
详细报告DataFrame
- 返回类型:
DataFrame
- get_risk_rejection_report_by_target(target=None)[源代码]
获取指定标签的风险拒绝率报告.
- 参数:
target (str) -- 目标变量名,默认使用第一个
- 返回:
风险拒绝率报告DataFrame
- 返回类型:
DataFrame
- get_summary_report(metric='count', target=None)[源代码]
生成汇总报告.
- 参数:
metric (str) -- 'count'订单口径或'amount'金额口径
target (str) -- 指定标签,默认使用第一个
- 返回:
汇总报告DataFrame
- 返回类型:
DataFrame
- property pass_rate_report: DataFrame
生成通过率分析报告.
- property risk_rejection_report: DataFrame
生成风险拒绝率报告.
- property summary_report_amount: DataFrame | None
金额口径汇总报告.
- property summary_report_count: DataFrame
订单口径汇总报告.
- class hscredit.report.SwapRiskConfig(score_col='score', swap_type_col='swap_type', amount_col=None, out_in_uplift=2.0, bin_method='quantile', max_n_bins=10, custom_bins=None, original_pass_rate=None, targets=None, target_aliases=None)[源代码]
基类:
objectSwap风险分析配置.
- 参数:
score_col (str) -- 评分字段名,用于风险预估
swap_type_col (str) -- swap类型字段名
amount_col (str | None) -- 金额字段名(可选)
out_in_uplift (float) -- out-in样本风险上浮因子,默认2.0
bin_method (str) -- 分箱方法,默认'quantile'
max_n_bins (int) -- 最大分箱数,默认10
custom_bins (List[float] | None) -- 自定义分箱边界(可选)
original_pass_rate (float | None) -- 原策略通过率(可选,用于无out-out数据场景)
targets (List[str] | None) -- 目标变量列表(可选)
target_aliases (Dict[str, str] | None) -- 目标变量别名(可选),用于报告展示
- amount_col: str | None = None
- bin_method: str = 'quantile'
- custom_bins: List[float] | None = None
- max_n_bins: int = 10
- original_pass_rate: float | None = None
- out_in_uplift: float = 2.0
- score_col: str = 'score'
- swap_type_col: str = 'swap_type'
- target_aliases: Dict[str, str] | None = None
- targets: List[str] | None = None
- hscredit.report.create_swap_dataset(df, original_rule_col, new_rule_col, score_col, swap_type_col='swap_type', amount_col=None, rule_type='reject', n_jobs=-1, parallel_backend=None, parallel_config=None)[源代码]
创建swap数据集.
- 参数:
df (DataFrame) -- 输入数据
original_rule_col (str) -- 原策略规则字段
new_rule_col (str) -- 新策略规则字段
score_col (str) -- 评分字段
swap_type_col (str) -- swap类型输出字段名
amount_col (str | None) -- 金额字段(可选)
rule_type (str) -- 规则类型,"reject"表示拒绝规则(1=拒绝,0=通过), "pass"表示通过规则(1=通过,0=拒绝)
n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict[str, Any] | None)
- 返回:
包含swap_type的数据集
- 返回类型:
DataFrame
- hscredit.report.create_swap_dataset_from_rules(df, original_rule, new_rule, score_col, swap_type_col='swap_type', amount_col=None, rule_type='reject', original_rule_name='original_reject', new_rule_name='new_reject', n_jobs=-1, parallel_backend=None, parallel_config=None)[源代码]
从Rule对象创建swap数据集.
- 参数:
df (DataFrame) -- 输入数据
original_rule -- 原策略Rule对象
new_rule -- 新策略Rule对象
score_col (str) -- 评分字段
swap_type_col (str) -- swap类型输出字段名
amount_col (str | None) -- 金额字段(可选)
rule_type (str) -- 规则类型,"reject"表示拒绝规则,"pass"表示通过规则
original_rule_name (str) -- 原策略规则结果临时字段名
new_rule_name (str) -- 新策略规则结果临时字段名
n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict[str, Any] | None)
- 返回:
包含swap_type的数据集
- 返回类型:
DataFrame
- hscredit.report.swap_analysis(swap_df, reference_df, score_col='score', target=None, overdue=None, dpds=None, swap_type_col='swap_type', amount_col=None, out_in_uplift=2.0, original_pass_rate=None, target_aliases=None, rules_out_out=None, rules_in_out=None, rules_in_in=None, rules_out_in=None, rule_execution_mode='parallel', n_jobs=-1, parallel_backend=None, parallel_config=None, **kwargs)[源代码]
统一的Swap分析入口函数.
传入数据集和参数配置,直接得到完整的swap分析结果。
- 参数:
swap_df (DataFrame) -- swap数据集(含swap_type和score)
reference_df (DataFrame) -- 参考数据集(含score和target/overdue)
score_col (str) -- 评分字段名,用于风险预估
target (str | None) -- 目标变量字段名(可选),与overdue+dpds二选一
overdue (str | List[str] | None) -- 逾期天数字段名或列表(可选),如'MOB1'或['MOB1','MOB3']
dpds (int | List[int] | None) -- 逾期定义天数或列表(可选),如15或[15,30] - 逾期天数>dpds为坏样本(1),其他为好样本(0)
swap_type_col (str) -- swap类型字段名
amount_col (str | None) -- 金额字段名(可选),用于金额口径分析
out_in_uplift (float) -- out-in风险上浮因子,默认2.0
original_pass_rate (float | None) -- 原策略通过率(可选),用于无out-out数据场景
target_aliases (Dict[str, str] | None) -- 目标变量别名(可选),如{'target_dpd15': 'DPD15+'}
rules_out_out (Rule | List[Rule] | None) -- out-out象限规则集,支持Rule或List[Rule](可选)
rules_in_out (Rule | List[Rule] | None) -- in-out象限规则集,支持Rule或List[Rule](可选)
rules_in_in (Rule | List[Rule] | None) -- in-in象限规则集,支持Rule或List[Rule](可选)
rules_out_in (Rule | List[Rule] | None) -- out-in象限规则集,支持Rule或List[Rule](可选)
rule_execution_mode (str) -- 模块内规则执行方式,'parallel'表示命中任意规则,'serial'表示按规则顺序命中第一个
kwargs -- 其他配置参数,如bin_method, max_n_bins, custom_bins等
n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict[str, Any] | None)
- 返回:
SwapAnalysisResult分析结果对象
- 返回类型:
参考样例
>>> # swap_df 需先通过 create_swap_dataset(或 create_swap_dataset_from_rules) >>> # 构造出 swap_type 列,再传入 swap_analysis >>> swap_df = create_swap_dataset( ... raw_df, original_rule_col='old_reject', new_rule_col='new_reject', ... score_col='score', ... ) >>> >>> # 单标签分析 >>> result = swap_analysis( ... swap_df, reference_df, ... score_col='score', ... target='target_dpd15' ... ) >>> >>> # 多标签分析(使用overdue+dpds) >>> result = swap_analysis( ... swap_df, reference_df, ... score_col='score', ... overdue='MOB1', ... dpds=[15, 30], ... target_aliases={'MOB1_15+': 'DPD15+', 'MOB1_30+': 'DPD30+'} ... ) >>> >>> # 获取结果 >>> result.summary_report_count >>> result.summary_report_amount >>> result.pass_rate_report >>> result.risk_rejection_report
- class hscredit.report.SwapType(value)[源代码]
基类:
EnumSwap四象限类型.
- IN_IN = 'in-in'
- IN_OUT = 'in-out'
- OUT_IN = 'out-in'
- OUT_OUT = 'out-out'
- class hscredit.report.OverduePredictor(feature, target=None, overdue=None, dpds=None, method='mdlp', max_n_bins=5, min_bin_size=0.05, missing_separate=True, coefficients=None, bad_rate_col='坏样本率', bin_label_col='分箱标签', rules=None, desc=None, bin_params=None, n_jobs=-1, parallel_backend=None, parallel_config=None)[源代码]
基类:
ParallelizableMixin,BaseEstimator,TransformerMixin逾期率预测器.
基于特征分箱对应的逾期率,对无标签样本进行加权逾期率预测。 支持从原始数据或现成分箱表两种方式获取分箱逾期率。
两种拟合模式
模式一:从原始数据拟合 - 传入含target或逾期天数的DataFrame,自动分箱并计算各箱逾期率 - 支持overdue+dpds多标签场景
模式二:从分箱表拟合 - 传入已有的分箱统计表(feature_bin_stats输出或人工构建的DataFrame) - 直接从分箱表中提取各箱逾期率
参数
- 参数:
feature (str) -- 特征名称,用于分箱和预估
target (str | None) -- 目标变量名称,默认为'target'
overdue (str | List[str] | None) -- 逾期天数字段名称或列表,如 'MOB1' 或 ['MOB1', 'MOB3']
dpds (int | List[int] | None) -- 逾期定义天数或列表,如 7 或 [0, 7, 30] - 逾期天数 > dpds 为坏样本(1),其他为好样本(0)
method (str) -- 分箱方法,默认'mdlp'
max_n_bins (int) -- 最大分箱数,默认5
min_bin_size (float) -- 每箱最小样本占比,默认0.05
missing_separate (bool) -- 是否将缺失值单独分箱,默认True
coefficients (float | Dict[str, float] | str | None) -- 逾期率调整系数,支持以下格式: - None: 不调整(默认) - float: 对所有分箱统一乘以该系数 - dict: 按分箱标签指定系数,如 {'(-inf, 300]': 1.2, '(300, 500]': 0.9} - 'auto': 自动基于整体逾期率偏差校正
bad_rate_col (str) -- 分箱表中逾期率列名,默认为'坏样本率' - 单标签时直接使用该列名 - 多标签时需包含目标名称,函数会自动匹配
bin_label_col (str) -- 分箱表中分箱标签列名,默认为'分箱标签'
rules (List | None) -- 自定义分箱切分点列表,如 [300, 500, 700]
desc (str | None) -- 特征描述,用于报告展示
bin_params (Dict | None) -- 传递给feature_bin_stats的额外参数
n_jobs (int | float | None)
parallel_backend (str | None)
parallel_config (Mapping[str, Any] | None)
属性
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)
- fit(X, y=None)[源代码]
拟合预估器.
支持两种输入模式: 1. 传入DataFrame(含target或逾期天数列): 自动分箱并计算逾期率 2. 传入分箱表DataFrame(含分箱标签和逾期率列): 直接提取逾期率
- 参数:
X (DataFrame | Series) -- 训练数据DataFrame或分箱表DataFrame
y -- sklearn兼容参数,此处不使用(目标列从X中提取)
- 返回:
self
- 返回类型:
- get_report(metric='count')[源代码]
获取逾期率预估报告.
输出各分箱的逾期率、样本分布及预估信息。
- 参数:
metric (str) -- 统计口径,'count'(订单口径)或 'amount'(金额口径)
- 返回:
报告DataFrame
- 返回类型:
DataFrame
参考样例
>>> predictor = OverduePredictor(feature='score', target='target') # 初始化逾期率预估器 >>> predictor.fit(train_df) # 拟合分箱表并计算各箱逾期率 >>> report = predictor.get_report() # 获取分箱统计报告(含逾期率、样本数等) >>> print(report)
- predict(X)[源代码]
预测逾期率(简化接口).
直接返回逾期率预测结果,不含分箱和基础逾期率列。
- 参数:
X (DataFrame) -- 待预测数据
- 返回:
逾期率Series(单标签)或 {目标名: 逾期率Series}(多标签)
- 返回类型:
Series | Dict[str, Series]
参考样例
>>> predictor = OverduePredictor(feature='score', target='target') >>> predictor.fit(train_df) >>> predicted_rates = predictor.predict(test_df) # 直接返回逾期率Series(单标签)或字典(多标签)
- hscredit.report.overdue_prediction_report(data, feature, target=None, overdue=None, dpds=None, predict_data=None, coefficients=None, method='mdlp', max_n_bins=5, min_bin_size=0.05, missing_separate=True, bin_table=None, rules=None, desc=None, excel_writer=None, sheet='逾期率预估报告', n_jobs=-1, parallel_backend=None, parallel_config=None, **kwargs)[源代码]
逾期率预估报告便捷函数.
统一的入口函数,支持从原始数据或分箱表进行逾期率预估, 并可输出包含预估结果的报告。
- 参数:
data (DataFrame) -- 有标签数据集(含target或逾期天数),或分箱表
feature (str) -- 特征名称
target (str | None) -- 目标变量名称
overdue (str | List[str] | None) -- 逾期天数字段名称或列表
dpds (int | List[int] | None) -- 逾期定义天数或列表
predict_data (DataFrame | None) -- 待预估的无标签数据(可选), 传入后会计算各样本的预估逾期率
coefficients (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 (DataFrame | None) -- 现成分箱表(可选),传入后直接使用而不从data计算
rules (List | None) -- 自定义分箱切分点列表
desc (str | None) -- 特征描述
excel_writer -- Excel文件路径或ExcelWriter对象(可选),用于输出报告
sheet (str) -- Excel工作表名称
n_jobs (int | float)
parallel_backend (str | None)
parallel_config (Dict[str, Any] | None)
- 返回:
包含预估结果的DataFrame
- 返回类型:
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' ... )
- class hscredit.report.ModelReport(model, X_train=None, y_train=None, X_test=None, y_test=None, X_oot=None, y_oot=None, feature_names=None, target=None, datasets=None, overdue=None, dpds=None, method='predict_proba', method_kwargs=None, explain_config=None, n_jobs=-1, parallel_backend=None, parallel_config=None, **kwargs)[源代码]
基类:
object面向报表输出的快速模型报告封装.
参考风控建模标准报告模板,对已训练模型一站式生成多 Sheet 结构的 Excel / HTML 报告,并提供各分项结果的获取方法(指标、分箱表、特征重要性、 描述统计、相关性等)。支持任意多个数据集(训练/测试/OOT…)的横向对比, 以及
overdue+dpds的多逾期标签构建。参数
(完整说明见
__init__())- 参数:
model -- 已训练好的模型,需实现
predict/predict_probadatasets (Optional[Union[List, Dict]]) -- 数据集字典或列表(推荐),如
{'train': df, 'test': df}; DataFrame 需含目标列或配合overdue/dpds构建标签X_train/y_train/X_test/y_test -- 兼容 sklearn 风格的数据传入方式
target (Optional[Union[str, Dict]]) -- 目标列名(sklearn/scorecardpipeline 风格)
overdue (Optional[Union[str, List[str]]]) -- 逾期天数列名或列表,配合
dpds自动构建 0/1 标签dpds (Optional[Union[int, float, List[Union[int, float]]]]) -- 逾期定义天数或列表(逾期天数 > dpds 记为坏样本)
feature_names (Optional[List[str]]) -- 特征名称列表,可选;None 时自动从模型
feature_names_/feature_names_in_获取method (Union[str, Callable])
method_kwargs (Optional[Dict[str, Any]])
explain_config (Optional[Dict[str, Any]])
parallel_backend (Optional[str])
parallel_config (Optional[Dict[str, Any]])
属性
model: 传入的已训练模型
feature_names: 最终使用的特征名称列表
_datasets: 解析后的各数据集(key ->
ReportDataset)
参考样例
>>> from hscredit.report import ModelReport >>> report = ModelReport(model, datasets={'train': train_df, 'test': test_df}) >>> report.get_metrics() # 各数据集指标对比 >>> report.get_feature_importance(top_n=20) >>> report.to_excel('模型报告.xlsx') # 导出多 Sheet 报告
- add_dataset(key, label, X, y=None, feature_names=None)[源代码]
添加额外数据集(如 OOT)用于报告.
- 参数:
key (str) -- 数据集标识
label (str) -- 数据集标签
X -- DataFrame(含目标列时 y 可为 None,自动构建标签)
y -- 标签列,None 时从 X 中通过 target / overdue+dpds 自动构建
feature_names (List[str] | None) -- 特征名列表
- get_bin_table(dataset='train', method='quantile', max_n_bins=10, amount_col=None, margins=True, label=None, labels=None)[源代码]
返回评分分箱表,并缓存同一报告调用中的确定性结果。
- 参数:
dataset (str)
method (str)
max_n_bins (int)
amount_col (str | None)
margins (bool)
label (str | None)
labels (List[str] | None)
- 返回类型:
DataFrame
- get_feature_bin_table(feature, dataset='train', max_n_bins=10, method='quantile', margins=True, amount_col=None, label=None, labels=None)[源代码]
返回特征分箱表,并缓存同一报告调用中的确定性结果。
- 参数:
feature (str)
dataset (str)
max_n_bins (int)
method (str)
margins (bool)
amount_col (str | None)
label (str | None)
labels (List[str] | None)
- 返回类型:
DataFrame
- get_metrics(label=None)[源代码]
KS / AUC / PSI 等核心指标.
- 参数:
label (str | None) -- 多标签模式下指定标签名,None 时使用 combined y
- 返回类型:
DataFrame
- summary()[源代码]
模型核心指标摘要表(多层列:统计指标 × 数据集;行:逾期指标).
参考
hscredit.report.rule_strategy中拒绝规则策略表的展示方式: 列为「统计指标 × 数据集」两层表头,便于同一指标在各数据集上横向对比; 行为不同逾期指标。overdue+dpds多标签模式下每个逾期标签独占一行, 单标签模式下仅一行(行名为目标列名)。- 返回:
行索引为
逾期指标、列为(统计指标, 数据集)两层表头的 DataFrame- 返回类型:
DataFrame
参考样例
>>> report = ModelReport(model, datasets={'train': tr, 'test': te}, ... overdue=['MOB1'], dpds=[7, 3, 0]) >>> report.summary() # 行: MOB1@7 / MOB1@3 / MOB1@0;列: (KS, 训练集) ...
- to_excel(filepath, *, n_bins=10, bin_method='quantile', amount_col=None, date_col=None, date_freq=None, group_col=None, with_plots=True, model_name=None, project_desc=None, feature_map=None, feature_info=None, show_lift=True, show_importance=True, feature_contribution_label_max_features=10, data_source=None, loc_cols=None)[源代码]
事务性生成 Excel;失败时恢复进入调用前的全部派生缓存。
- 参数:
filepath (str)
n_bins (int)
bin_method (str)
amount_col (str | None)
date_col (str | None)
date_freq (str | None)
group_col (str | None)
with_plots (bool)
model_name (str | None)
project_desc (str | None)
feature_map (Dict[str, str] | None)
feature_info (DataFrame | None)
show_lift (bool)
show_importance (bool)
feature_contribution_label_max_features (int | None)
data_source (str | None)
loc_cols (str | List[str] | None)
- 返回类型:
str
- hscredit.report.QuickModelReport
ModelReport的别名
- hscredit.report.auto_model_report(model, datasets=None, X_train=None, y_train=None, X_test=None, y_test=None, X_oot=None, y_oot=None, feature_names=None, target=None, overdue=None, dpds=None, excel_path=None, verbose=True, n_bins=10, bin_method='quantile', amount_col=None, date_col=None, date_freq=None, group_col=None, with_plots=True, model_name=None, project_desc=None, feature_map=None, feature_info=None, show_lift=True, show_importance=True, feature_contribution_label_max_features=10, data_source=None, loc_cols=None, method='predict_proba', method_kwargs=None, n_jobs=-1, parallel_backend=None, parallel_config=None, **kwargs)[源代码]
一键生成模型报告.
数据集传入支持三种方式,内部统一规整为
{数据集名称: (X, y)}结构:datasets 为 dict:直接以 key 作为数据集名称, 如
{'建模集': df, 'OOT': df},DataFrame 需包含目标列,或通过 overdue/dpds 自动构建标签datasets 为 list:依次命名为 数据集1、数据集2、...、数据集N
X_train/X_test/X_oot 参数:依次命名为 训练集、测试集、跨时间验证集
标签解析遵循 hscredit 统一传参风格:显式传入 y 优先(sklearn 风格); 否则通过 target 列名从 X 中提取(scorecardpipeline 风格); 传入 overdue+dpds 组合时直接忽略 target。
overdue/dpds 用法(自动从 X 构建二分类标签):
# 单阈值:MOB 任一期间 DPD > 5 则 y=1 auto_model_report(model, X_train=df, overdue='dpds', dpds=5) # 多阈值:MOB1 DPD>15 或 MOB3 DPD>7 任一触发则 y=1 auto_model_report( model, X_train=df, overdue=['dpds_m1', 'dpds_m3'], dpds=[15, 7, 0], )
示例:
# 方式1: datasets dict(DataFrame 直接传入,X 中含目标列) auto_model_report(model, datasets={'train': train_df, 'test': test_df}, excel_path='report.xlsx') # 方式1: datasets list(自动命名) auto_model_report(model, datasets=[train_df, test_df], excel_path='report.xlsx') # 方式1: overdue/dpds 自动构建标签 auto_model_report( model, datasets={'train': df}, overdue='dpds', dpds=[15, 7, 0], excel_path='report.xlsx', ) # 方式2: 兼容 sklearn API(分离 X/y) auto_model_report( model, X_train=X, y_train=y, X_test=X_val, y_test=y_val, excel_path='report.xlsx', )
- 参数:
model -- 训练好的模型(ScoreCard / XGBoost / LightGBM / sklearn 等)
datasets (List | Dict | None) -- 数据集字典/列表,字典键直接作为数据集名称(推荐)
X_train -- 训练集特征(命名为 训练集)
y_train -- 训练集标签,None 时从 X_train 中自动构建
X_test -- 测试集特征(命名为 测试集)
y_test -- 测试集标签,None 时从 X_test 中自动构建
X_oot -- 跨时间验证集特征(命名为 跨时间验证集)
y_oot -- 跨时间验证集标签,None 时从 X_oot 中自动构建
feature_names (List[str] | None) -- 特征名称列表,可选;None 时自动从模型
feature_names_/feature_names_in_获取target (str | Dict | None) -- 目标列配置,str 为列名,dict 为 {'overdue': col, 'dpds': threshold}
overdue (str | List[str] | None) -- 逾期列名(str)或多个列名(List[str]),与 dpds 配合自动构建标签
dpds (int | float | List[int | float] | None) -- 逾期天数阈值(int/float)或多个阈值(List),与 overdue 配合使用
excel_path (str | None) -- Excel 报告输出路径
verbose (bool) -- 是否打印控制台报告
n_bins (int) -- 分箱数
bin_method (str) -- 分箱方法
amount_col (str | None) -- 金额字段(用于金额口径分析)
date_col (str | None) -- 日期字段(用于分月分析)
date_freq (str | None) -- 日期频率,支持 'D', 'W', 'M', 'Q' 等(默认自动推断)
group_col (str | None) -- 分组字段(用于分组坏样本率分析)
with_plots (bool) -- 是否生成图表
model_name (str | None) -- 模型名称
project_desc (str | None) -- 项目描述
feature_map (Dict[str, str] | None) -- 特征名称到含义的映射
feature_info (DataFrame | None) -- 特征部署信息表
show_lift (bool) -- 是否在报告中显示 LIFT 曲线
show_importance (bool) -- 是否在报告中显示特征重要性
feature_contribution_label_max_features (int | None) -- 贡献图显示数据标签的最大特征数;默认 10,
None始终显示,0 始终隐藏data_source (str | None) -- 数据源描述
loc_cols (str | List[str] | None) -- 定位字段(订单号等),支持 str 或 List[str],用于生产测试用例列
method (str | Callable) -- 数据集唯一预测方法,默认
predict_proba,也支持 callablemethod_kwargs (Dict[str, Any] | None) -- callable 同名参数的显式覆盖字典
n_jobs -- 并行工作数;-1 自动保留 CPU,None 使用兼容串行模式
parallel_backend (str | None) -- joblib 后端,如
threading或lokyparallel_config (Dict[str, Any] | None) -- joblib 其他并行配置
kwargs -- 透传给 callable 的额外同名参数
- 返回:
ModelReport 实例
- 返回类型:
- hscredit.report.compare_models(models, X_train, y_train, X_test=None, y_test=None, excel_path=None, n_jobs=-1, parallel_backend=None, parallel_config=None)[源代码]
横向对比多个模型的评估指标.
对每个模型分别构建
ModelReport并取其summary(), 按输入映射顺序纵向拼接为一张对比表;任一模型失败时立即抛出并保留原始异常链。- 参数:
models (Dict[str, object]) -- 模型名称到模型对象的映射,如
{'XGB': xgb_model, 'LR': lr_model}X_train -- 训练集特征
y_train -- 训练集标签(0/1)
X_test -- 测试集特征,可选
y_test -- 测试集标签,可选
excel_path (str | None) -- 可选,若提供则将对比表导出到该 Excel 路径
n_jobs -- 模型外层并行工作数;-1 自动保留 CPU
parallel_backend (str | None) -- joblib 后端,如
threading或lokyparallel_config (Dict[str, Any] | None) -- joblib 其他并行配置
- 返回:
含
模型名称列的指标对比DataFrame- 返回类型:
DataFrame
参考样例
>>> from hscredit.report import compare_models >>> result = compare_models( ... {'XGBoost': xgb_model, '逻辑回归': lr_model}, ... X_train, y_train, X_test, y_test, ... excel_path='模型对比.xlsx', ... ) >>> print(result)
- hscredit.report.population_drift(expected, actual, features, target_col=None, date_col=None, score_col=None, n_bins=10, output='客群偏移监控报告.xlsx', *, target=None, n_jobs=-1, parallel_backend=None, parallel_config=None)[源代码]
生成客群偏移综合监控结果(Excel).
报告包含以下Sheet: - 总览: 各特征PSI汇总及稳定性等级 - 特征分布对比: 各特征在基准/实际数据集的分箱分布对比 - 逾期率对比 (可选): 各特征分箱下的逾期率差异 - 评分分布 (可选): 模型评分在两个数据集的分布对比
- 参数:
expected (DataFrame) -- 基准数据集(如训练集)
actual (DataFrame) -- 实际/监控数据集(如生产数据)
features (List[str]) -- 监控特征列表
target_col (str | None) -- 目标变量列名(如有,则生成逾期率对比Sheet)
date_col (str | None) -- 时间列名(如有,用于按时间拆分实际数据)
score_col (str | None) -- 评分列名(如有,则生成评分分布Sheet)
n_bins (int) -- 分箱数
output (str) -- 输出文件路径
target (str | None) -- target_col 的别名
- 返回:
输出文件路径(即
output)- 返回类型:
str
参考样例
>>> from hscredit.report import population_drift >>> # 基础:仅特征分布 PSI 对比 >>> population_drift(train_df, prod_df, features=['score', 'age', 'income']) >>> >>> # 含逾期率与评分分布对比 >>> population_drift( ... train_df, prod_df, ... features=['score', 'age'], ... target_col='FPD', ... score_col='model_score', ... output='客群偏移监控报告.xlsx', ... )
备注
ExcelWriter / dataframe2excel 见 Excel hscredit.excel;规则挖掘器在下方
hscredit.report.mining 小节统一说明。
规则挖掘 hscredit.report.mining
单特征 / 多特征交叉 / 多标签规则挖掘器与树规则提取器。
规则挖掘模块.
提供从数据中自动挖掘规则的功能,包括: - 单特征规则挖掘 - 多特征交叉规则挖掘 - 决策树规则提取(支持DT、RF、GBDT、XGBoost、孤立森林) - 规则评估指标
代码风格参考hscredit的binning模块和Rule模块,fit方法兼容scorecardpipeline风格。
- 示例:
>>> from hscredit.report.mining import SingleFeatureRuleMiner >>> miner = SingleFeatureRuleMiner(target='target') >>> miner.fit(df) >>> rules = miner.get_top_rules(top_n=10)
>>> from hscredit.report.mining import TreeRuleExtractor >>> extractor = TreeRuleExtractor(algorithm='rf', max_depth=5) >>> extractor.fit(X, y) >>> rules = extractor.extract_rules()
>>> from hscredit.report.mining import DecisionTreeAnalyzer >>> analyzer = DecisionTreeAnalyzer(target='target', features=['age', 'income']) >>> analyzer.fit(df_train) >>> print(analyzer.evaluate([('测试', df_test)], metric_type='ks'))
- class hscredit.report.mining.SingleFeatureRuleMiner(target='target', exclude_cols=None, method='mdlp', max_n_bins=10, min_n_bins=2, min_bin_size=0.01, max_bin_size=None, monotonic=False, chi2_threshold=3.841, significance_level=0.05, min_lift=1.5, min_samples=10, special_codes=None, cat_cutoff=None, random_state=None, verbose=False, n_jobs=-1, parallel_backend=None, parallel_config=None, features=None, feature_map=None, **binning_kwargs)[源代码]
基类:
BaseRuleMiner单特征规则挖掘器.
对数据各特征的不同阈值进行效度分布分析,挖掘高区分度的单特征规则。
代码风格参考hscredit的binning模块,fit方法兼容scorecardpipeline风格。 支持hscredit中所有分箱方法。
- 参数:
target (str) -- 目标变量列名,默认为'target'
exclude_cols (List[str] | None) -- 需要排除的列名列表
features (str | List[str] | None) -- 参与挖掘的字段名或字段名列表,默认None(使用全部候选字段)
feature_map (Dict[str, str] | None) -- 字段名到字段含义的映射,用于规则报告展示
method (str) -- 分箱方法,取值与 OptimalBinning.VALID_METHODS 完全一致,不支持别名。 默认为'mdlp'
max_n_bins (int) -- 最大分箱数,默认20。同binning模块的max_n_bins
min_n_bins (int) -- 最小分箱数,默认2。同binning模块的min_n_bins
min_bin_size (float | int) -- 每箱最小样本数或占比,默认0.05 - 如果 < 1, 表示占比 (如 0.05 表示 5%) - 如果 >= 1, 表示绝对数量
max_bin_size (float | int | None) -- 每箱最大样本数或占比,默认None
monotonic (bool | str) -- 坏样本率单调性约束,默认False - False: 不要求单调性 - True 或 'auto': 自动检测并应用最佳单调方向 - 'ascending': 强制坏样本率递增 - 'descending': 强制坏样本率递减 - 'peak': 允许单峰形态(先升后降) - 'valley': 允许单谷形态(先降后升)
chi2_threshold (float) -- 卡方分箱合并阈值,默认3.841 (p=0.05, df=1)
significance_level (float) -- 卡方显著性水平,默认0.05
min_lift (float) -- 最小lift阈值,用于筛选规则,默认1.1
min_samples (int) -- 最小样本数,默认10。同binning模块的约束
special_codes (List | None) -- 特殊值列表,这些值会被单独处理,默认None
cat_cutoff (float | int | None) -- 类别型变量处理阈值,默认None
random_state (int | None) -- 随机种子,默认None
verbose (bool | int) -- 是否输出详细信息,默认False
binning_kwargs -- 分箱方法的其他参数,通过**kwargs传入
n_jobs (int | float | None)
parallel_backend (str | None)
parallel_config (Mapping[str, Any] | None)
参考样例
>>> miner = SingleFeatureRuleMiner(target='ISBAD', method='quantile', max_n_bins=20) # 等频分箱:每箱样本数大致相等 >>> miner.fit(df) >>> rules = miner.get_top_rules(top_n=10, metric='lift') # 获取TOP10规则,按LIFT降序 >>> miner = SingleFeatureRuleMiner(target='ISBAD', method='chi', max_n_bins=10, chi2_threshold=3.841) # 卡方分箱:相近坏率的箱自动合并 >>> miner.fit(df) >>> miner = SingleFeatureRuleMiner(target='ISBAD', method='best_iv', max_n_bins=5, monotonic=True) # 最优IV分箱:自动选择IV最大且坏率单调的分箱方案 >>> miner.fit(df) >>> feature_rules = miner.analyze_feature('age', max_n_bins=10) # 分析单个特征的规则分布
- 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']
- analyze_feature(feature, max_n_bins=None, **kwargs)[源代码]
分析单个特征.
- 参数:
feature (str) -- 特征名
max_n_bins (int | None) -- 分箱数量(None则使用初始化参数)
kwargs -- 其他覆盖参数
- 返回:
各阈值指标DataFrame
- 返回类型:
DataFrame
- fit(X, y=None, feature_names=None, **kwargs)[源代码]
拟合挖掘器.
- 参数:
X (DataFrame | ndarray) -- 训练数据,DataFrame或numpy数组
y (ndarray | Series | None) -- 目标变量(可选)
feature_names (str | List[str] | None) -- 本次拟合使用的字段名或字段名列表,优先于构造参数
featureskwargs -- 额外参数,可覆盖初始化参数
- 返回:
self
- 返回类型:
- get_binning_table(feature)[源代码]
获取特征的分箱统计表(如果使用了hscredit分箱方法).
- 参数:
feature (str) -- 特征名
- 返回:
分箱统计表或None
- 返回类型:
DataFrame | None
- get_rule_objects(min_lift=None, min_samples=None, target=None, datasets=None, sort_by='lift', ascending=False, top_n=10000)[源代码]
获取Rule对象列表(与get_rules保持一致).
- 参数:
min_lift (float | None)
min_samples (int | None)
target (str | None)
datasets (DataFrame | None)
sort_by (str)
ascending (bool)
top_n (int)
- 返回类型:
List[Rule]
- get_rules(min_lift=None, min_samples=None, target=None, datasets=None, sort_by='lift', ascending=False, top_n=10000)[源代码]
获取挖掘规则(只使用Rule.expr,并基于Rule.report命中结果).
- 参数:
min_lift (float | None)
min_samples (int | None)
target (str | None)
datasets (DataFrame | None)
sort_by (str)
ascending (bool)
top_n (int)
- 返回类型:
List[Rule]
- get_top_rules(top_n=10, metric='lift', feature=None, min_lift=None, min_samples=None, ascending=False, datasets=None, target=None)[源代码]
获取Top规则(使用Rule.expr与Rule.report命中结果,支持缺失值规则).
- 参数:
top_n (int)
metric (str)
feature (str | None)
min_lift (float | None)
min_samples (int | None)
ascending (bool)
datasets (DataFrame | None)
target (str | None)
- 返回类型:
DataFrame
- class hscredit.report.mining.MultiFeatureRuleMiner(target='target', exclude_cols=None, method='quantile', max_n_bins=5, min_n_bins=2, min_bin_size=0.01, max_bin_size=None, monotonic=False, cat_cutoff=None, min_samples=10, min_lift=1.1, special_codes=None, random_state=None, verbose=False, n_jobs=-1, parallel_backend=None, parallel_config=None, features=None, feature_map=None, **binning_kwargs)[源代码]
基类:
BaseRuleMiner多特征交叉规则挖掘器.
生成双特征交叉分析结果,支持hscredit所有分箱方法。
代码风格参考hscredit的binning模块,fit方法兼容scorecardpipeline风格。
- 参数:
target (str) -- 目标变量列名,默认为'target'
exclude_cols (List[str] | None) -- 需要排除的列名列表
features (str | List[str] | None) -- 参与挖掘的字段名或字段名列表,默认None(使用全部候选字段)
feature_map (Dict[str, str] | None) -- 字段名到字段含义的映射,用于规则报告展示
method (str) -- 分箱方法,取值与 OptimalBinning.VALID_METHODS 完全一致,不支持别名。 默认为'quantile'
max_n_bins (int) -- 最大分箱数,默认5。超过此值的数值型特征将分箱
min_n_bins (int) -- 最小分箱数,默认2
min_bin_size (float | int) -- 每箱最小样本数或占比,默认0.01
max_bin_size (float | int | None) -- 每箱最大样本数或占比,默认None
monotonic (bool | str) -- 坏样本率单调性约束,默认False
cat_cutoff (float | int | None) -- 类别型变量处理阈值,默认None
min_samples (int) -- 最小样本数,默认10
min_lift (float) -- 最小lift阈值,默认1.1
special_codes (List | None) -- 特殊值列表,默认None
random_state (int | None) -- 随机种子,默认None
verbose (bool | int) -- 是否输出详细信息,默认False
binning_kwargs -- 分箱方法的其他参数,通过**kwargs传入
n_jobs (int | float | None)
parallel_backend (str | None)
parallel_config (Mapping[str, Any] | None)
参考样例
>>> miner = MultiFeatureRuleMiner(target='ISBAD', method='quantile', max_n_bins=5) # 等频分箱:双特征交叉规则挖掘 >>> miner.fit(df) >>> cross_matrix = miner.generate_cross_matrix('age', 'income') # 生成年龄×收入的交叉分箱矩阵 >>> rules = miner.get_cross_rules('age', 'income', top_n=10) # 获取TOP10交叉规则,按LIFT排序 >>> miner = MultiFeatureRuleMiner(target='ISBAD', method='chi', max_n_bins=4) # 卡方分箱:自动合并坏率相近的交叉箱 >>> miner.fit(df)
- 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']
- fit(X, y=None, feature_names=None, **kwargs)[源代码]
拟合挖掘器.
- 参数:
X (DataFrame | ndarray) -- 训练数据
y (ndarray | Series | None) -- 目标变量(可选)
feature_names (str | List[str] | None) -- 本次拟合使用的字段名或字段名列表,优先于构造参数
featureskwargs -- 额外参数,可覆盖初始化参数
- 返回:
self
- 返回类型:
- generate_cross_matrix(feature1, feature2, custom_splits1=None, custom_splits2=None, **kwargs)[源代码]
生成双特征交叉矩阵.
- 参数:
feature1 (str) -- 第一个特征
feature2 (str) -- 第二个特征
custom_splits1 (List[float] | None) -- 特征1的自定义分箱切分点
custom_splits2 (List[float] | None) -- 特征2的自定义分箱切分点
kwargs -- 其他参数
- 返回:
交叉矩阵DataFrame
- 返回类型:
DataFrame
- get_all_cross_rules(top_n=10, metric='lift', max_feature_pairs=50, min_samples=None, min_lift=None)[源代码]
获取所有特征对的交叉规则.
- 参数:
top_n (int) -- 每对特征的返回规则数
metric (str) -- 排序指标
max_feature_pairs (int) -- 最大特征对数量
min_samples (int | None) -- 最小样本数
min_lift (float | None) -- 最小lift
- 返回:
所有交叉规则DataFrame
- 返回类型:
DataFrame
- get_binning_table(feature)[源代码]
获取特征的分箱统计表(如果使用了hscredit分箱方法).
- 参数:
feature (str) -- 特征名
- 返回:
分箱统计表或None
- 返回类型:
DataFrame | None
- get_cross_rules(feature1, feature2, top_n=10, metric='lift', min_samples=None, min_lift=None, datasets=None, target=None)[源代码]
获取双特征交叉的top规则(使用rule_expr+Rule.report命中结果).
- 参数:
feature1 (str)
feature2 (str)
top_n (int)
metric (str)
min_samples (int | None)
min_lift (float | None)
datasets (DataFrame | None)
target (str | None)
- 返回类型:
DataFrame
- get_rule_objects(top_n=10, metric='lift', target=None, datasets=None, min_samples=None, min_lift=None)[源代码]
获取Rule对象列表(与get_rules保持一致).
- 参数:
top_n (int)
metric (str)
target (str | None)
datasets (DataFrame | None)
min_samples (int | None)
min_lift (float | None)
- 返回类型:
List[Rule]
- get_rules(top_n=10, metric='lift', target=None, datasets=None, min_samples=None, min_lift=None)[源代码]
获取挖掘的交叉规则(只使用Rule.expr,并基于Rule.report命中结果).
- 参数:
top_n (int)
metric (str)
target (str | None)
datasets (DataFrame | None)
min_samples (int | None)
min_lift (float | None)
- 返回类型:
List[Rule]
- plot_2d_binning(feature1, feature2, metric='lift', figsize=(14, 10), title=None, save_path=None, cmap=None)[源代码]
绘制二维分箱图(主图+上下文分箱轴).
中心主图:二维网格指标热力图
顶部:feature2 分箱轴(与主图共享x轴)
右侧:feature1 分箱轴(与主图共享y轴)
- 参数:
feature1 (str)
feature2 (str)
metric (str)
figsize (Tuple[int, int])
title (str | None)
save_path (str | None)
cmap (str | None)
- plot_cross_heatmap(feature1, feature2, metric='lift', figsize=(12, 10), title=None, save_path=None, cmap=None)[源代码]
绘制交叉热力图.
- 参数:
feature1 (str) -- 特征1
feature2 (str) -- 特征2
metric (str) -- 指标
figsize (Tuple[int, int]) -- 图大小
title (str | None) -- 自定义标题
save_path (str | None) -- 保存路径
cmap (str | None) -- 颜色映射,默认使用RdYlGn
- 返回:
matplotlib Figure对象
- class hscredit.report.mining.MultiLabelRuleMiner(labels=None, label_names=None, min_support=0.02, min_lift=1.5, max_rules=10, n_bins=10, exclude_cols=None, n_jobs=-1, parallel_backend=None, parallel_config=None)[源代码]
基类:
BaseRuleMiner多标签规则挖掘器.
支持同时针对多个标签挖掘规则,并分析规则在不同标签下的有效性差异。
典型应用场景: - 长短期标签都有效的强规则(稳定拒绝规则) - 仅短期标签有效(可能是偶发风险,谨慎使用) - 仅长期标签有效(长期风险,可做预警规则) - 两标签均无效(噪声规则,丢弃)
- 参数:
labels (List[str] | None) -- 标签列名列表,如 ['mob3_30', 'mob6_30']
label_names (List[str] | None) -- 标签中文名列表,如 ['短期标签(MOB3@30)', '长期标签(MOB6@30)']
min_support (float) -- 最小支持度(规则覆盖率),默认 0.02
min_lift (float) -- 最小 LIFT 值,默认 1.5
max_rules (int) -- 每个特征最大规则数,默认 10
n_bins (int) -- 数值特征分箱数,默认 10
exclude_cols (List[str] | None) -- 需要排除的列名列表
n_jobs (int | float | None)
parallel_backend (str | None)
parallel_config (Mapping[str, Any] | None)
参考样例
>>> miner = MultiLabelRuleMiner( ... labels=['mob3_30', 'mob6_30'], # 同时分析两个标签 ... label_names=['短期标签(MOB3@30)', '长期标签(MOB6@30)'], ... min_support=0.02, # 规则覆盖率>2% ... min_lift=1.5, # LIFT值>1.5 ... ) >>> miner.fit(df, features=['age', 'income', 'credit_score']) >>> rules = miner.get_rules(effectiveness='both') # 获取两标签均有效的规则 >>> report = miner.get_effectiveness_matrix() # 获取规则有效性矩阵
- fit(X, y=None, features=None)[源代码]
拟合多标签规则挖掘器.
- 参数:
X (DataFrame | ndarray) -- 输入数据 DataFrame,须包含 labels 指定的标签列
y -- 忽略
features (List[str] | None) -- 需要挖掘的特征列表,为 None 时自动选择数值特征
- 返回:
self
- 返回类型:
- get_rules(effectiveness='any', min_lift_per_label=None, top_n=None)[源代码]
获取筛选后的规则表.
- 参数:
effectiveness (str) -- 筛选条件 - 'both': 所有标签均有效 - 'any': 任一标签有效(默认) - 'short_only': 仅第一个标签有效 - 'long_only': 仅最后一个标签有效 - 'all': 不做筛选
min_lift_per_label (float | None) -- 覆盖最小 LIFT 阈值,为 None 时使用 self.min_lift
top_n (int | None) -- 返回前 N 条规则
- 返回:
规则 DataFrame
- 返回类型:
DataFrame
- class hscredit.report.mining.TreeRuleExtractor(algorithm='dt', target='target', exclude_cols=None, max_depth=5, min_samples_split=10, min_samples_leaf=5, n_estimators=10, max_features='sqrt', test_size=0.3, random_state=42, feature_trends=None, chi2_threshold=3.841, n_jobs=-1, parallel_backend=None, parallel_config=None, features=None, feature_map=None, **kwargs)[源代码]
基类:
BaseRuleMiner树模型规则提取器.
支持多种树模型的规则提取,包括决策树、随机森林、GBDT、XGBoost和孤立森林。
代码风格参考hscredit的binning模块,fit方法兼容scorecardpipeline风格。 支持通过**kwargs传入任意sklearn树模型参数。
- 参数:
algorithm (str) -- 算法类型,'dt', 'rf', 'chi2', 'gbdt', 'xgb', 'isf'
target (str) -- 目标变量列名,默认为'target'
exclude_cols (List[str] | None) -- 需要排除的列名列表
features (str | List[str] | None) -- 参与建树的字段名或字段名列表,默认None(使用全部候选字段)
feature_map (Dict[str, str] | None) -- 字段名到字段含义的映射,用于规则报告展示
max_depth (int) -- 树的最大深度,默认5
min_samples_split (int) -- 分裂节点最小样本数,默认10
min_samples_leaf (int) -- 叶子节点最小样本数,默认5
n_estimators (int) -- 森林中树的数量,默认10
max_features (str) -- 每棵树考虑的最大特征数,默认'sqrt'
test_size (float) -- 测试集比例,默认0.3
random_state (int) -- 随机种子,默认42
feature_trends (Dict[str, int] | None) -- 特征趋势字典,如{'age': 1}表示正相关
chi2_threshold (float) -- 卡方分箱阈值,默认3.841
kwargs -- 其他树模型参数,直接传递给底层sklearn模型: - DecisionTreeClassifier: criterion, splitter, max_leaf_nodes, etc. - RandomForestClassifier: bootstrap, oob_score, class_weight, etc. - GradientBoostingClassifier: learning_rate, subsample, loss, etc. - IsolationForest: contamination, max_samples, etc.
n_jobs (int | float | None)
parallel_backend (str | None)
parallel_config (Mapping[str, Any] | None)
参考样例
>>> extractor = TreeRuleExtractor(algorithm='dt', max_depth=5) # 决策树提取:从单棵树提取可解释规则 >>> extractor.fit(df) >>> rules = extractor.extract_rules() >>> extractor = TreeRuleExtractor( ... algorithm='rf', # 随机森林提取:多棵树投票,提取泛化性更好的规则 ... n_estimators=50, ... max_depth=10, ... class_weight='balanced', ... bootstrap=True, ... oob_score=True ... ) >>> extractor.fit(X, y) >>> rules = extractor.extract_rules() >>> extractor = TreeRuleExtractor( ... algorithm='isf', # 孤立森林异常检测:识别偏离正常模式的异常样本规则(无监督) ... contamination=0.05, ... max_samples=256 ... ) >>> extractor.fit(X) # 无监督,不需要y >>> anomaly_rules = extractor.extract_rules()
- VALID_ALGORITHMS = {'chi2', 'dt', 'gbdt', 'isf', 'rf', 'xgb'}
- fit(X, y=None, feature_names=None, **kwargs)[源代码]
在临时副本中拟合,成功后原子提交模型与编码状态。
- 参数:
X (DataFrame | ndarray)
y (ndarray | Series | None)
feature_names (str | List[str] | None)
- 返回类型:
- get_rule_objects(top_n=100, min_samples=10, min_confidence=0.0, target=None, datasets=None)[源代码]
获取Rule对象列表(与get_rules保持一致).
- 参数:
top_n (int)
min_samples (int)
min_confidence (float)
target (str | None)
datasets (DataFrame | None)
- 返回类型:
List[Rule]
- class hscredit.report.mining.DecisionTreeAnalyzer(target='target', features=None, tree_params=None, missing=None, n_jobs=-1, parallel_backend=None, parallel_config=None, feature_map=None, **kwargs)[源代码]
-
sklearn 决策树分析器。
在标准 sklearn DecisionTreeClassifier 基础上,提供决策树训练、 结构导出、规则提取、节点规则评估和 AUC / KS / LIFT 等模型指标计算。
参数
- 参数:
target (str) -- 目标变量列名(0=好样本,1=坏样本)
features (str | List[str] | None) -- 特征名或特征名列表(默认自动从数据中推断数值列)
feature_map (Dict[str, str] | None) -- 字段名到字段含义的映射,用于节点规则报告展示
tree_params (Dict[str, Any] | None) -- 决策树参数字典,默认值如下:
missing (float | None)
n_jobs (int | float | None)
parallel_backend (str | None)
parallel_config (Mapping[str, Any] | None)
kwargs (Any)
参考样例
>>> from hscredit.report.mining import DecisionTreeAnalyzer >>> analyzer = DecisionTreeAnalyzer(target='target', features=['age', 'income']) >>> analyzer.fit(df_train) >>> # 在测试集上评估 >>> metrics = analyzer.evaluate([('测试集', df_test)], metric_type='ks') >>> print(metrics) >>> # 获取规则表 >>> rules = analyzer.get_rules() >>> print(rules) >>> # 导出树图 >>> analyzer.export_tree('tree.dot')
- apply(df=None)[源代码]
返回每个样本所属叶子节点的编号。
- 参数:
df (DataFrame | None) -- 待评估数据(默认使用训练数据)
- 返回:
叶子节点编号数组
- 返回类型:
ndarray
参考样例
>>> leaf_ids = fitter.apply(df_test) >>> print(f"测试集样本分布在 {len(set(leaf_ids))} 个叶子节点")
- evaluate(test_data_list, metric_type='auc', top_rate=0.1)[源代码]
评估模型性能。
支持多种评估指标,计算训练集及多个测试集的指标值。
- 参数:
test_data_list (List[Tuple[str, DataFrame]]) -- 测试数据集列表,元素为 (数据集名称, DataFrame)
metric_type (str) --
评估指标类型
类型
说明
'auc'
ROC AUC 分数(使用 predict_proba 的正类概率)
'ks'
KS 统计量
'lift'
top 客群的 LIFT 值
'top'
top客群坏样本率(与 lift 等价)
top_rate (float) -- lift/top 指标计算时取 top 的比例(默认 10%)
- 返回:
评估结果列表,元素为 (数据集名称, 指标值)
- 返回类型:
List[Tuple[str, float]]
参考样例
>>> metrics = fitter.evaluate([('测试集', df_test)], metric_type='ks') >>> for name, value in metrics: ... print(f'{name}: {value:.4f}')
- export_tree(out_file=None, max_depth=None, class_names=None)[源代码]
导出决策树为 DOT 格式。
- 参数:
out_file (str | None) -- 输出 .dot 文件路径(可选,指定时同时写入文件)
max_depth (int | None) -- 最大显示深度(None=全部显示)
class_names (List[str] | None) -- 类别名列表,默认 ['好', '坏']
- 返回:
DOT 格式字符串
- 返回类型:
str
参考样例
>>> dot = fitter.export_tree('tree.dot') >>> with open('tree.dot') as f: ... print(f.read())
- fit(X, y=None, feature_names=None, tree_params=None, **kwargs)[源代码]
在临时副本中训练,并在全部成功后提交拟合状态。
- 参数:
X (DataFrame | ndarray)
y (ndarray | Series | None)
feature_names (str | List[str] | None)
tree_params (Dict[str, Any] | None)
kwargs (Any)
- 返回类型:
- get_rule_table(datasets=None, target=None, overdue=None, dpds=None, del_grey=False, leaf_only=False, **kwargs)[源代码]
获取决策树所有节点(分裂节点+叶子节点)的规则效果表。
从每个节点提取规则路径,解析为
Rule对象后调用Rule.report()在数据集上计算命中效果,取其中"命中"分箱对应的结果行汇总。各节点统计口径 与report()完全一致(均基于规则在数据集上的真实命中情况计算,而非 读取决策树内部存储的节点样本统计),便于与人工分裂结果横向对比。参数
- 参数:
datasets (DataFrame | None) -- 用于计算规则效果的数据集,默认 None(使用 fit 时的训练数据)
target (str | None) -- 目标变量列名,默认 None(使用拟合时的
self.target)overdue (str | List[str] | None) -- 逾期天数字段名,参考
Rule.report()dpds (int | List[int] | None) -- 逾期定义方式,参考
Rule.report()del_grey (bool) -- 是否删除灰度样本,默认 False
leaf_only (bool) -- 是否仅返回叶子节点,默认 False
kwargs (Any) -- 其余传递给
Rule.report()的参数(如amount、margins)
- 返回:
规则效果表,列结构与
report()一致(节点编号、是否叶子、 规则分类、指标名称、指标含义、分箱、样本总数、样本占比、好样本数、 坏样本数、坏样本率、LIFT值、坏账改善、风险拒绝比等)- 返回类型:
DataFrame
参考样例
>>> table = fitter.get_rule_table() # 在训练数据上评估 >>> table = fitter.get_rule_table(df_test) # 在新数据上评估
- get_rules()[源代码]
将树的叶子节点转换为 Rule 对象列表。
- 返回:
Rule 对象列表,每个 Rule 对应一个叶子节点
- 返回类型:
List[Rule]
参考样例
>>> rules = fitter.get_rules() >>> for rule in rules: ... report = rule.report(df_test, target='target')
- classmethod load(file_path)[源代码]
从 pickle 文件加载决策树。
- 参数:
file_path (str) -- 模型文件路径
- 返回:
加载后的 DecisionTreeAnalyzer 实例
- 返回类型:
参考样例
>>> analyzer2 = DecisionTreeAnalyzer.load('dt_model.pkl')
- plot(backend='matplotlib', save=None, title='', **kwargs)[源代码]
绘制当前决策树结构。
- 参数:
backend (str) -- 渲染后端,可选
matplotlib/pyecharts/graphvizsave (str | None) -- 保存路径(可选)
title (str) -- 图表标题
kwargs (Any) -- 其余参数透传给
hscredit.core.viz.plot_tree()
- 返回:
matplotlib Figure / pyecharts Chart / graphviz Source
- 返回类型:
Any
参考样例
>>> analyzer = DecisionTreeAnalyzer(target='target').fit(df, feature_names=['age', 'income']) >>> fig = analyzer.plot() >>> analyzer.plot(backend='graphviz', save='tree.pdf')
- predict(df=None)[源代码]
预测类别标签。
- 参数:
df (DataFrame | None) -- 待预测数据(默认使用训练数据)
- 返回:
预测结果数组
- 返回类型:
ndarray
- predict_proba(df=None)[源代码]
预测类别概率。
- 参数:
df (DataFrame | None) -- 待预测数据(默认使用训练数据)
- 返回:
类别概率数组,形状 (n_samples, n_classes)
- 返回类型:
ndarray
- report(datasets, target=None, overdue=None, dpds=None, del_grey=False, leaf_only=False, feature_map=None, **kwargs)[源代码]
在新数据集上评估决策树各节点规则的效果。
对每个节点(含分裂节点和叶子节点),将其规则路径解析为
Rule对象, 直接调用Rule.report()计算统计指标,取其中"命中"分箱对应的结果行汇总, 得到各节点在新数据上的效果报告。参数
- 参数:
datasets (DataFrame | List[DataFrame] | Dict[Any, DataFrame]) -- 待评估数据集,支持单个 DataFrame、DataFrame 列表、 或
{名称: DataFrame}字典;返回结果类型与输入保持一致target (str | None) -- 目标变量列名,默认为None(使用拟合时的
self.target)overdue (str | List[str] | None) -- 逾期天数字段名(可选,传入时以逾期天数>DPD定义坏样本, 支持多标签多DPD联合分析),参考
Rule.report()dpds (int | List[int] | None) -- 逾期定义方式,逾期天数 > DPD 为坏样本,默认为0; 传入列表时支持多DPD联合分析,参考
Rule.report()del_grey (bool) -- 是否删除逾期天数在(0, DPD]区间内的灰度样本,默认为False
leaf_only (bool) -- 是否仅评估叶子节点,默认 False(评估所有节点)
feature_map (Dict[str, str] | None) -- 字段名到字段含义的映射,显式传入时覆盖构造参数
kwargs (Any) -- 其余传递给
Rule.report()的参数(如amount、margins)
- 返回:
各节点效果评估表(DataFrame),结构与输入一致: 单个 DataFrame 输入返回单个 DataFrame,列表输入返回 DataFrame 列表, 字典输入返回
{名称: DataFrame}字典- 返回类型:
DataFrame | List[DataFrame] | Dict[Any, DataFrame]
参考样例
>>> result = fitter.report(df_test, target='target') >>> results = fitter.report([df_test1, df_test2], target='target') >>> results = fitter.report({'测试集1': df_test1, '测试集2': df_test2}, target='target') >>> # 多标签(逾期天数)联合分析 >>> result = fitter.report(df_test, overdue=['MOB1'], dpds=[7, 3, 0])
- class hscredit.report.mining.ManualTreeExtractor(target='target', max_depth=2, min_samples_split=10, min_samples_leaf=5, random_state=0, missing=None, n_jobs=-1, parallel_backend=None, parallel_config=None, features=None, feature_map=None, **kwargs)[源代码]
-
人工决策树提取器。
支持对 sklearn 决策树进行**人工指定分裂节点**后重新训练, 适合将业务经验注入数据驱动模型。
核心流程: 1. 用数据训练一棵基础决策树(或直接指定特征/阈值) 2. 人工在指定节点分裂(manual_split),指定特征和阈值 3. 获取规则表或在新数据集上评估效果
参数
- 参数:
target (str) -- 目标变量列名(坏样本标签,0=好,1=坏),默认 'target'
max_depth (int) -- 树的最大深度,默认 2
min_samples_split (int) -- 分裂节点最小样本数,默认 10
min_samples_leaf (int) -- 叶子节点最小样本数,默认 5
random_state (int) -- 随机种子,默认 0
features (str | List[str] | None) -- 参与建树的字段名或字段名列表,默认None(自动推断数值字段)
feature_map (Dict[str, str] | None) -- 字段名到字段含义的映射,用于节点规则报告展示
missing (float | None) -- 缺失值的等价填充数值(可选),参考
DecisionTreeAnalyzern_jobs (int | float | None)
parallel_backend (str | None)
parallel_config (Mapping[str, Any] | None)
kwargs (Any)
参考样例
>>> from hscredit.report.mining import ManualTreeExtractor >>> ext = ManualTreeExtractor(target='target', max_depth=2) >>> ext.fit(df, feature_names=['age', 'income']) >>> # 人工分裂:指定在某节点用某特征+阈值分裂 >>> ext.manual_split(df_sub, feature='age', threshold=35, node=1) >>> # 获取规则表 >>> print(ext.get_rule_table()) >>> # 在新数据上评估 >>> print(ext.report(df_test)) >>> # 获取 Rule 对象 >>> rules = ext.get_rules()
- delete_node(node)[源代码]
删除指定节点及其所有子节点,将该节点变为叶子。
参数
- 参数:
node (int) -- 待删除的节点 ID
- 返回:
self
- 返回类型:
参考样例
>>> ext.delete_node(node=3)
- display()[源代码]
在 Jupyter Notebook 中展示决策树图和规则表。
每次调用都会根据当前树结构重新生成,确保 manual_split / delete_node 后显示最新状态。
决策树图使用
hscredit.core.viz.plot_tree_matplotlib`(AntV G6 卡片式风格) 绘制:卡片节点 + 主题色标题栏 + 节点指标,按坏账率从浅蓝(低风险)到浅红(高风险) 着色,人工修改节点(manual_split)使用副主题色边框标记。规则表使用 :func:`style_rule_table()美化展示。参考样例
>>> ext = ManualTreeExtractor(target='target') >>> ext.fit(df, feature_names=['age', 'income']) >>> ext.display() # 在 Jupyter 中展示树图和规则表 >>> ext.manual_split(df, 'income', 5000, node=1).display()
- 返回类型:
- fit(X, y=None, feature_names=None, max_depth=None, min_samples_split=None, min_samples_leaf=None, **kwargs)[源代码]
串行构建临时树,全部成功后原子提交拟合状态。
- 参数:
X (DataFrame | ndarray)
y (ndarray | Series | None)
feature_names (str | List[str] | None)
max_depth (int | None)
min_samples_split (int | None)
min_samples_leaf (int | None)
kwargs (Any)
- 返回类型:
- classmethod from_sklearn(clf, feature_names=None, target='target', missing=None, feature_map=None)[源代码]
从已训练的 sklearn DecisionTreeClassifier 创建 ManualTreeExtractor。
适用于想先训练好树结构,再进行人工干预分裂的场景。
参数
- 参数:
clf (DecisionTreeClassifier) -- 已训练的 sklearn DecisionTreeClassifier
feature_names (List[str] | None) -- 特征名列表(默认从 clf.``feature_names_in_`` 推断)
target (str) -- 目标变量名(仅用于存储,不参与训练),默认 'target'
missing (float | None) -- 缺失值的等价填充数值(可选),参考
DecisionTreeAnalyzerfeature_map (Dict[str, str] | None)
- 返回:
ManualTreeExtractor 实例(已 fitted 状态)
- 返回类型:
参考样例
>>> from sklearn.tree import DecisionTreeClassifier >>> from hscredit.report.mining import ManualTreeExtractor >>> clf = DecisionTreeClassifier(max_depth=3, random_state=42) >>> clf.fit(X, y) >>> mte = ManualTreeExtractor.from_sklearn(clf, feature_names=feature_names) >>> mte.manual_split(df, feature='age', threshold=35, node=1)
- get_rule_table(datasets=None, target=None, overdue=None, dpds=None, del_grey=False, leaf_only=False, **kwargs)[源代码]
获取当前树各节点规则在数据集上的效果表。
从树的每个节点提取规则路径,解析为
Rule对象后调用Rule.report()在数据集上计算命中效果,取其中"命中"分箱对应的结果行汇总。这样人工分裂节点与 自动分裂节点的统计口径完全一致(均基于规则在数据集上的真实命中情况计算,而非 读取决策树内部存储的节点样本统计),与report()输出格式保持一致。参数
- 参数:
datasets (DataFrame | None) -- 用于计算规则效果的数据集,默认 None(使用 fit 时的训练数据)
target (str | None) -- 目标变量列名,默认 None(使用拟合时的
self.target)overdue (str | List[str] | None) -- 逾期天数字段名,参考
Rule.report()dpds (int | List[int] | None) -- 逾期定义方式,参考
Rule.report()del_grey (bool) -- 是否删除灰度样本,默认 False
leaf_only (bool) -- 是否仅返回叶子节点,默认 False
kwargs (Any) -- 其余传递给
Rule.report()的参数(如amount、margins)
- 返回:
规则效果表,列结构与
report()一致(节点编号、是否叶子、 规则分类、指标名称、指标含义、分箱、样本总数、样本占比、好样本数、 坏样本数、坏样本率、LIFT值、坏账改善、风险拒绝比等)。其中(参考Rule.report()的口径,将命中节点的样本视为"拒绝"客群):坏账改善 = (整体坏账率 - 拒绝后剩余客群坏账率) / 整体坏账率
风险拒绝比 = 坏账改善 / 样本占比
- 返回类型:
DataFrame
参考样例
>>> ext.manual_split(df, feature='age', threshold=35) >>> print(ext.get_rule_table()) # 在训练数据上评估 >>> print(ext.get_rule_table(df_test)) # 在新数据上评估
- get_rules()[源代码]
将当前树的叶子节点规则转换为 Rule 对象列表。
- 返回:
Rule 对象列表
- 返回类型:
List[Rule]
参考样例
>>> rules = ext.get_rules() >>> for r in rules: ... report = r.report(df, target='target')
- manual_split(data, feature, threshold=None, node=0)[源代码]
在指定节点人工分裂。
在 node 位置按 feature 和 threshold 进行分裂。 若 threshold 为 None,则用决策树自动计算最优分裂点。 支持链式调用。
参数
- 参数:
data (DataFrame) -- 用于计算分裂阈值的数据集(需包含分裂特征与目标列)。可直接传入 原始训练数据,方法会根据 node 的当前路径自动筛选该节点命中的样本
feature (str) -- 分裂特征名
threshold (float | None) -- 分裂阈值(None=自动计算最优阈值)
node (int) -- 分裂的节点 ID,默认 0(根节点)
- 返回:
self
- 返回类型:
参考样例
>>> # 人工指定阈值 >>> ext.manual_split(df, feature='age', threshold=35, node=1) >>> # 自动找最优阈值 >>> ext.manual_split(df, feature='income', threshold=None, node=2) >>> # 链式调用 >>> ext.manual_split(df, 'f1', 30).manual_split(df, 'f2', 20)
- plot(backend='matplotlib', save=None, title='', **kwargs)[源代码]
绘制当前人工决策树结构。
- 参数:
backend (str) -- 渲染后端,可选
matplotlib/pyecharts/graphvizsave (str | None) -- 保存路径(可选)
title (str) -- 图表标题
kwargs (Any) -- 其余参数透传给
hscredit.core.viz.plot_tree()
- 返回:
matplotlib Figure / pyecharts Chart / graphviz Source
- 返回类型:
Any
参考样例
>>> ext = ManualTreeExtractor(target='target').fit(df, feature_names=['age', 'income']) >>> fig = ext.plot() >>> ext.manual_split(df, 'age', 35).plot(save='tree.png')
- report(datasets, target=None, overdue=None, dpds=None, del_grey=False, leaf_only=False, feature_map=None, **kwargs)[源代码]
在新数据集上评估当前树各节点规则的效果。
对每个节点(含分裂节点和叶子节点),将其规则路径解析为
Rule对象, 直接调用Rule.report()计算统计指标,取其中"命中"分箱对应的结果行汇总, 得到各节点在新数据上的效果报告。参数
- 参数:
datasets (DataFrame | List[DataFrame] | Dict[Any, DataFrame]) -- 待评估数据集,支持单个 DataFrame、DataFrame 列表、 或
{名称: DataFrame}字典;返回结果类型与输入保持一致target (str | None) -- 目标变量列名,默认为None(使用拟合时的
self.target)overdue (str | List[str] | None) -- 逾期天数字段名(可选,传入时以逾期天数>DPD定义坏样本, 支持多标签多DPD联合分析),参考
Rule.report()dpds (int | List[int] | None) -- 逾期定义方式,逾期天数 > DPD 为坏样本,默认为0; 传入列表时支持多DPD联合分析,参考
Rule.report()del_grey (bool) -- 是否删除逾期天数在(0, DPD]区间内的灰度样本,默认为False
leaf_only (bool) -- 是否仅评估叶子节点,默认 False(评估所有节点)
feature_map (Dict[str, str] | None) -- 字段名到字段含义的映射,显式传入时覆盖构造参数
kwargs -- 其余传递给
Rule.report()的参数(如amount、margins)
- 返回:
各节点效果评估表(DataFrame),结构与输入一致: 单个 DataFrame 输入返回单个 DataFrame,列表输入返回 DataFrame 列表, 字典输入返回
{名称: DataFrame}字典- 返回类型:
DataFrame | List[DataFrame] | Dict[Any, DataFrame]
参考样例
>>> result = ext.report(df_test, target='target') >>> results = ext.report([df_test1, df_test2], target='target') >>> results = ext.report({'测试集1': df_test1, '测试集2': df_test2}, target='target') >>> # 多标签(逾期天数)联合分析 >>> result = ext.report(df_test, overdue=['MOB1'], dpds=[7, 3, 0])
- class hscredit.report.mining.RuleMetrics(target_positive=1, n_jobs=-1, parallel_backend=None, parallel_config=None)[源代码]
-
规则评估指标计算器.
提供全面的规则评估指标,支持训练集和测试集的对比分析。 所有指标计算统一收口到hscredit.core.metrics。
参考样例
>>> from hscredit.core.rules.mining import RuleMetrics >>> metrics = RuleMetrics() >>> result = metrics.evaluate_rule(rule, X_train, y_train, X_test, y_test) # 单规则评估:返回KS/AUC/IV等指标 >>> results = metrics.evaluate_rules(rules, X_train, y_train, X_test, y_test) # 批量规则评估:返回规则列表各自的指标
- 参数:
target_positive (int)
n_jobs (int | float | None)
parallel_backend (str | None)
parallel_config (Mapping[str, Any] | None)
- calculate_gini(y_true, y_score)[源代码]
计算Gini系数.
使用统一的Gini计算。
- 参数:
y_true (ndarray) -- 真实标签
y_score (ndarray) -- 预测分数
- 返回:
Gini系数
- 返回类型:
float
- calculate_iv(feature, target, n_bins=10)[源代码]
计算IV值(信息价值).
使用统一的IV计算。
- 参数:
feature (Series) -- 特征值
target (Series) -- 目标变量
n_bins (int) -- 分箱数
- 返回:
IV值
- 返回类型:
float
- calculate_ks(y_true, y_score)[源代码]
计算KS统计量.
使用统一的KS计算。
- 参数:
y_true (ndarray) -- 真实标签
y_score (ndarray) -- 预测分数
- 返回:
KS值
- 返回类型:
float
- evaluate_rule(rule, X_train, y_train, X_test=None, y_test=None, amount_train=None, amount_test=None)[源代码]
评估单个规则.
- 参数:
rule -- 规则对象(Rule或MinedRule)
X_train (DataFrame) -- 训练集特征
y_train (Series) -- 训练集标签
X_test (DataFrame | None) -- 测试集特征(可选)
y_test (Series | None) -- 测试集标签(可选)
amount_train (Series | None) -- 训练集金额(可选)
amount_test (Series | None) -- 测试集金额(可选)
- 返回:
评估指标字典
- 返回类型:
Dict[str, Any]