"""人工决策树提取器.
提供决策树规则挖掘的核心工具集，支持：
- **DecisionTreeAnalyzer**：标准 sklearn 决策树训练、结构导出、规则提取与指标评估
- **ManualTreeExtractor**：人工干预决策树节点分裂（业务经验注入模型）
**参考样例**
>>> # DecisionTreeAnalyzer：训练决策树并评估
>>> from hscredit.report.mining import DecisionTreeAnalyzer
>>> analyzer = DecisionTreeAnalyzer(target='target', features=['age', 'income'])
>>> analyzer.fit(df_train)
>>> metrics = analyzer.evaluate(df_test_list=[('测试', df_test)], metric_type='ks')
>>> print(metrics)
>>> # ManualTreeExtractor：人工分裂
>>> from hscredit.report.mining import ManualTreeExtractor
>>> ext = ManualTreeExtractor(target='target')
>>> ext.fit(df, feature_names=['age', 'income'])
>>> ext.manual_split(df, feature='age', threshold=35)
>>> print(ext.get_rule_table())
"""
import copy
import pickle
from typing import Any, Dict, List, Optional, Tuple, Union
import numpy as np
import pandas as pd
import sklearn
from sklearn.tree import DecisionTreeClassifier, export_graphviz
from ...core.rules.rule import Rule
from ...exceptions import InputValidationError
from ...utils.pandas_extensions import style_rule_table
from ...utils.parallel import (
ParallelizableMixin,
_current_parallel_budget,
parallel_execute,
resolve_n_jobs,
)
from .base import (
FeatureNames,
_mining_workload,
check_features_valid,
format_feature_context,
normalize_feature_names,
resolve_feature_map,
)
def _effective_nested_n_jobs(n_jobs):
"""在活跃父预算中限制内部显式并发，根调用保留公开配置。"""
budget = _current_parallel_budget()
if budget.depth == 0:
return n_jobs
if n_jobs is None or n_jobs in (1, 1.0):
return 1
resolved = resolve_n_jobs(n_jobs, available_budget=budget.available) or 1
return min(resolved, budget.available)
def _commit_transactional_state(instance, working, parameter_names):
"""提交临时拟合状态，并将显式构造参数重新绑定到调用方原始对象。"""
parameter_references = {name: getattr(instance, name) for name in parameter_names}
instance.__dict__.clear()
instance.__dict__.update(working.__dict__)
for name, value in parameter_references.items():
setattr(instance, name, value)
def _tree_metric_dataset_worker(task):
"""计算一个独立数据集的树模型指标。"""
analyzer, name, data, metric_type, top_rate = task
if analyzer.target not in data.columns:
raise ValueError(f"测试集 '{name}' 缺少目标列: {analyzer.target}")
probability = analyzer.predict_proba(data)[:, 1]
value = analyzer._calc_metric(
probability,
data[analyzer.target].values,
metric_type,
top_rate,
)
return name, value
def _node_hit_worker(task):
"""计算一个独立节点规则在单个数据集上的命中报告。"""
node_id, is_leaf, rule, data, target, overdue, dpds, del_grey, kwargs = task
table = rule.report(
data,
target=target,
overdue=overdue,
dpds=dpds,
del_grey=del_grey,
desc=rule.description,
**kwargs,
)
is_multi = isinstance(table.columns, pd.MultiIndex)
group = "分箱详情" if is_multi else None
bin_col = (group, "分箱") if is_multi else "分箱"
node_col = (group, "节点编号") if is_multi else "节点编号"
leaf_col = (group, "是否叶子") if is_multi else "是否叶子"
hit = table[table[bin_col] == "命中"].copy()
hit[node_col] = node_id
hit[leaf_col] = "是" if is_leaf else "否"
front_cols = [node_col, leaf_col]
other_cols = [column for column in hit.columns if column not in front_cols]
return hit[front_cols + other_cols]
def _tree_report_dataset_worker(task):
"""计算一个独立数据集的整棵树节点报告。"""
analyzer, data, target, overdue, dpds, del_grey, leaf_only, feature_map, kwargs = task
return analyzer._report_one_dataset(
data,
target=target,
overdue=overdue,
dpds=dpds,
del_grey=del_grey,
leaf_only=leaf_only,
feature_map=feature_map,
report_kwargs=kwargs,
)
def _sklearn_supports_native_missing() -> bool:
"""判断当前 sklearn 版本是否支持决策树原生缺失值（>= 1.3）。
sklearn 1.3 起 DecisionTree 支持训练/预测含缺失值的数据，并在 ``tree_`` 上
暴露 ``missing_go_to_left`` 记录缺失样本的路由方向；旧版本不支持，含缺失数据
需先填充（通过 ``missing`` 参数指定等价填充数值）。
"""
try:
major, minor = (int(p) for p in sklearn.__version__.split(".")[:2])
except Exception:
return False
return (major, minor) >= (1, 3)
# 当前 sklearn 是否支持决策树原生缺失值（模块加载时计算一次）
_SKLEARN_NATIVE_MISSING = _sklearn_supports_native_missing()
# ============================================================================
# 指标计算 — 优先使用 hscredit.core.metrics 中的统一实现
# ============================================================================
try:
from ...core.metrics import ks as _ks
from ...core.metrics import auc as _auc
from ...core.metrics import badrate as _badrate
except ImportError:
# 降级：内联最小实现（仅在 metrics 未注册时使用）
def _ks(y_true, y_prob):
from sklearn.metrics import roc_curve
fpr, tpr, _ = roc_curve(y_true, y_prob)
return float((tpr - fpr).max())
def _auc(y_true, y_prob):
from sklearn.metrics import roc_auc_score
return float(roc_auc_score(y_true, y_prob))
def _badrate(y_true, mask):
if mask.sum() == 0:
return 0.0
return float(y_true[mask].mean())
def _lift_local(y_true, y_score, n_bins=10):
"""LIFT 计算：取 top-n% 中的坏样本率相对总体坏样本率的倍数。
:param y_true: 真实标签数组
:param y_score: 预测分数（分数越高越"坏"）
:param n_bins: 取最高分样本的比例分母，默认 10 即 top 10%
:return: LIFT 值
"""
df = pd.DataFrame({'y': y_true, 's': y_score}).sort_values('s', ascending=False)
top_n = max(1, int(len(df) * n_bins / 100))
top_bad_rate = df.head(top_n)['y'].mean()
overall_bad_rate = df['y'].mean()
return float(top_bad_rate / overall_bad_rate) if overall_bad_rate > 0 else 0.0
def _lift_table_local(y_true, y_score, n_bins=10):
"""LIFT 表格：按分数分箱计算各箱的坏账率和 LIFT 值。
:param y_true: 真实标签数组
:param y_score: 预测分数
:param n_bins: 分箱数，默认 10
:return: 含 LIFT 值的 DataFrame
"""
df = pd.DataFrame({'y': y_true, 's': y_score})
df['bin'] = pd.qcut(df['s'], n_bins, labels=False, duplicates='drop')
result = df.groupby('bin').agg(y=('y', 'mean'), count=('y', 'count'))
result['lift'] = result['y'] / df['y'].mean()
return result.reset_index()
# ============================================================================
# 决策树工具函数
# ============================================================================
def _impute_features(data: pd.DataFrame, feature_list: List[str], missing: Optional[float]) -> pd.DataFrame:
"""按 missing 填充特征缺失值；missing 为 None 时原样返回。
用于训练/预测前将缺失等价为指定数值，使其与规则的缺失处理口径一致，
同时让不支持原生缺失的旧版 sklearn 也能正常训练。
:param data: 输入数据
:param feature_list: 需要填充的特征列
:param missing: 缺失填充数值，None 表示不填充
:return: 仅特征列被填充后的 DataFrame（其余列保持不变）
"""
if missing is None:
return data
filled = data.copy()
filled[feature_list] = filled[feature_list].fillna(missing)
return filled
def _prepare_training_features(
data: pd.DataFrame, feature_list: List[str], missing: Optional[float], owner: str
) -> np.ndarray:
"""根据 sklearn 版本与 missing 参数准备训练特征矩阵。
- 指定 ``missing``：将缺失等价为该数值后训练（任意 sklearn 版本均可）；
- 未指定且 sklearn 支持原生缺失（>=1.3）：保留缺失，交由决策树原生处理；
- 未指定且旧版 sklearn 且数据含缺失：抛出明确提示，引导设置 ``missing``。
:param data: 训练数据（已过滤目标缺失）
:param feature_list: 特征列
:param missing: 缺失等价填充数值，None 表示按 sklearn 版本自动处理
:param owner: 调用方名称（用于错误信息）
:return: 训练特征矩阵 ndarray
:raises InputValidationError: 旧版 sklearn 且含缺失但未指定 missing 时
"""
feat = data[feature_list]
if missing is not None:
return feat.fillna(missing).values
if bool(feat.isna().any().any()) and not _SKLEARN_NATIVE_MISSING:
raise InputValidationError(
f"{owner}：训练数据包含缺失值，当前 sklearn 版本（{sklearn.__version__} < 1.3）"
"不支持决策树原生缺失值；请在初始化时传入 missing 指定缺失值的等价填充数值"
"（例如 missing=-999）"
)
return feat.values
def _resolve_fit_data(
X: Union[pd.DataFrame, np.ndarray],
y: Optional[Union[pd.Series, np.ndarray, List]] = None,
features: FeatureNames = None,
target: str = "target",
) -> Tuple[pd.DataFrame, List[str]]:
"""统一解析 sklearn 风格 (X, y) 与 scorecardpipeline 风格 (df) 两种调用方式。
- ``y`` 不为 None：sklearn 风格，``X`` 为特征矩阵（DataFrame 或 ndarray），
``y`` 为目标变量，目标列由 ``y`` 提供；
- ``y`` 为 None：scorecardpipeline 风格，``X`` 为含目标列的 DataFrame，从中提取目标。
:param X: 特征矩阵或含目标列的 DataFrame
:param y: 目标变量（可选）；传入时优先按 sklearn 风格解析
:param features: 特征名列表（可选）。未指定时：scorecardpipeline 风格取除目标外的
数值列；sklearn 风格的 DataFrame 取全部列，ndarray 取 ``feature_0`` 等默认名
:param target: 目标列名
:return: ``(含目标列的 DataFrame, 特征名列表)``
:raises InputValidationError: 入参不满足任一风格时
"""
selected = normalize_feature_names(features, "feature_names")
if y is not None:
# sklearn 风格：X=特征，y=标签
if isinstance(X, pd.DataFrame):
data = X.copy()
inferred = [c for c in data.columns if c != target]
else:
arr = np.asarray(X)
if arr.ndim == 1:
arr = arr.reshape(-1, 1)
inferred = selected if selected is not None else [f"feature_{i}" for i in range(arr.shape[1])]
if len(inferred) != arr.shape[1]:
raise InputValidationError(
"ndarray 输入无法按字段名选择子集；feature_names 数量必须与特征列数一致"
)
data = pd.DataFrame(arr, columns=inferred)
y_arr = np.asarray(y.values if isinstance(y, pd.Series) else y).ravel()
if len(y_arr) != len(data):
raise InputValidationError(f"特征与标签数量不匹配：{len(data)} != {len(y_arr)}")
data = data.copy()
data[target] = y_arr
feats = selected if selected is not None else inferred
else:
# scorecardpipeline 风格：X=含目标列的 DataFrame
if not isinstance(X, pd.DataFrame):
raise InputValidationError(
"未传入 y 时按 scorecardpipeline 风格解析：X 需为包含目标列的 DataFrame；"
"若按 sklearn 风格请同时传入 y"
)
if target not in X.columns:
raise InputValidationError(
f"目标列 '{target}' 未在数据中找到；请传入 y（sklearn 风格）"
f"或在 DataFrame 中包含 '{target}' 列（scorecardpipeline 风格）"
)
data = X
if selected is not None:
feats = selected
else:
feats = [c for c in X.columns if c != target and pd.api.types.is_numeric_dtype(X[c])]
if target in feats:
raise ValueError(f"目标列 '{target}' 不能作为特征")
check_features_valid(data.drop(columns=[target], errors="ignore"), feats)
return data, feats
def _rule_generator(
clf, feature_name_list: List[str], missing: Optional[float] = None
) -> pd.DataFrame:
"""从训练好的决策树（或模拟树对象）提取规则 DataFrame。
遍历每个非根节点，根据其父节点路径构建分裂条件。
:param clf: 已训练的 sklearn 决策树分类器，或包含 tree_ 属性的模拟对象
:param feature_name_list: 特征名列表
:param missing: 缺失值的等价填充数值（可选），按 sklearn 版本自动适配：
- ``None``（默认）：新版 sklearn（>=1.3）读取 ``tree_.missing_go_to_left``
按决策树学到的方向路由缺失；旧版 sklearn 无原生缺失支持，按"缺失走右"
（与 pandas eval 中 ``NaN`` 比较恒为 False、人工分裂缺失归右一致）。
- 数值：将缺失等价为该数值参与比较（如训练前已将缺失填充为 -999，则传入
``missing=-999``），按该数值是否满足各分裂条件决定缺失样本流向，
适用于不支持原生缺失的旧版 sklearn。
:return: 规则 DataFrame，含列：
node / if_leaf / rule_list / node_path / node_samples / node_value / impurity
.. note::
当某节点缺失样本被路由到当前路径方向时，对应条件追加 ``| (特征 != 特征)``
（``NaN != NaN`` 为 True，用于在 pandas eval 中识别缺失），确保规则在数据集
上的命中样本数与决策树节点样本数完全一致。
"""
children_left = list(clf.tree_.children_left)
children_right = list(clf.tree_.children_right)
feature = list(clf.tree_.feature)
threshold = list(clf.tree_.threshold)
node_samples = list(clf.tree_.n_node_samples)
node_values = list(clf.tree_.value)
node_impurity = list(clf.tree_.impurity)
# 缺失值路由方向：1=缺失走左子节点(<=)，0=缺失走右子节点(>)。
# 仅当未指定 missing 且 sklearn 支持原生缺失（>=1.3）时读取决策树学到的方向，
# 否则按 0 处理（缺失走右，与 pandas eval、人工分裂缺失归右一致）。
if missing is None and _SKLEARN_NATIVE_MISSING:
missing_go_to_left = list(getattr(clf.tree_, "missing_go_to_left", []))
else:
missing_go_to_left = []
if len(missing_go_to_left) < len(feature):
missing_go_to_left = missing_go_to_left + [0] * (len(feature) - len(missing_go_to_left))
def _edge_includes_nan(node: int, operator: str) -> bool:
"""判断该分裂边方向上缺失样本是否随之流动。
指定 missing 时按"缺失等价为该数值"判断（数值是否满足该边条件）；
否则按决策树的缺失值路由方向（missing_go_to_left）判断。
"""
if missing is not None:
return (missing <= threshold[node]) if operator == "<=" else (missing > threshold[node])
return (missing_go_to_left[node] == 1) if operator == "<=" else (missing_go_to_left[node] == 0)
def _find_father_path(node: int, father_path: List[str] = None) -> List[str]:
"""递归查找从根节点到目标节点的路径描述。"""
if father_path is None:
father_path = []
if node in children_left:
father_node = children_left.index(node)
node_path = f"{father_node},<="
elif node in children_right:
father_node = children_right.index(node)
node_path = f"{father_node},>"
else:
father_node = 0
node_path = "None"
path = copy.copy(father_path)
path.append(node_path)
if father_node > 0:
return _find_father_path(node=father_node, father_path=path)
return path
def _father_path_to_rule(father_path: List[str]) -> Tuple[List[str], List]:
"""将路径描述转换为特征分裂规则列表。
对同一特征的多个条件取交集（合并 max/min 阈值），并按缺失值路由方向
标记是否需要纳入缺失样本（特征级别取交集：仅当该特征所有相关分裂节点的
缺失值都沿当前路径方向流动时，缺失样本才会到达当前节点）。
返回 (路径ID列表, 规则列表[[特征名, 操作符, 阈值, 是否含缺失], ...])
"""
edges = []
# 特征级"缺失是否随路径流动"标记：对该特征的所有边取逻辑与
nan_follows_feat: Dict[str, bool] = {}
for node_tmp in father_path:
node = int(node_tmp.split(",")[0])
operator = node_tmp.split(",")[1]
feat = feature_name_list[feature[node]]
edges.append([feat, operator, threshold[node]])
# 该边方向上缺失样本是否随之流动（按 missing 或 missing_go_to_left 判定）
nan_follows_feat[feat] = nan_follows_feat.get(feat, True) and _edge_includes_nan(node, operator)
# 按特征聚合：同一特征的多个条件取交集
rule_df = pd.DataFrame(edges, columns=["feature_name", "operator", "threshold"])
grouped = rule_df.groupby(["feature_name", "operator"], observed=True).agg(
{"threshold": ["max", "min"]}
)
final_rule = []
for idx in grouped.index:
feat = idx[0]
thres = (
grouped.loc[idx, ("threshold", "min")]
if idx[1] == "<="
else grouped.loc[idx, ("threshold", "max")]
)
final_rule.append([feat, idx[1], thres, nan_follows_feat.get(feat, False)])
return father_path, final_rule
result = {
"node": [],
"if_leaf": [],
"rule_list": [],
"node_path": [],
"node_samples": [],
"node_value": [],
"impurity": [],
}
for i in range(1, len(feature)):
result["node"].append(i)
result["if_leaf"].append(True if feature[i] == -2 else False)
father_path = _find_father_path(i)
rule_path, final_rule = _father_path_to_rule(father_path)
result["rule_list"].append(final_rule)
result["node_path"].append(rule_path)
result["node_samples"].append(node_samples[i])
# node_value 格式：(n_samples, n_classes) 的比例值
# 兼容 numpy array（sklearn 原始）和 Python list（手动构造）
n = node_samples[i]
raw_val = node_values[i]
if hasattr(raw_val[0], "tolist"):
vals = [round(v * n) for v in raw_val[0].tolist()]
else:
vals = [round(v * n) for v in raw_val[0]]
result["node_value"].append(vals)
result["impurity"].append(node_impurity[i])
return pd.DataFrame(result)
def _export_dot_data(
clf,
feature_list: List[str],
class_names: Optional[List[str]] = None,
out_file: Optional[str] = None,
max_depth: Optional[int] = None,
filled: bool = True,
node_ids: bool = True,
proportion: bool = True,
precision: int = 3,
) -> str:
"""导出决策树为 DOT 格式字符串。
:param clf: 决策树分类器
:param feature_list: 特征名列表
:param class_names: 类别名列表
:param out_file: 输出 .dot 文件路径（可选）
:param max_depth: 最大显示深度
:param filled: 是否填充颜色
:param node_ids: 是否显示节点 ID
:param proportion: 是否显示样本比例
:param precision: 数值精度
:return: DOT 格式字符串
"""
dot_data = export_graphviz(
decision_tree=clf,
feature_names=feature_list,
class_names=class_names,
out_file=out_file,
max_depth=max_depth,
label="all",
filled=filled,
leaves_parallel=True,
impurity=True,
node_ids=node_ids,
proportion=proportion,
rotate=False,
rounded=True,
special_characters=True,
precision=precision,
)
if out_file is not None:
with open(out_file, "r") as f:
return f.read()
return dot_data
# ============================================================================
# 树结构节点操作
# ============================================================================
def _add_nodes_to_tree(
node: int,
split_list_left: List[int],
split_list_right: List[int],
feature: List[int],
threshold: List[float],
node_samples: List[int],
node_values: List,
node_impurity: List[float],
split_list_left_new: List[int],
split_list_right_new: List[int],
feature_new: List[int],
threshold_new: List[float],
node_samples_new: List[int],
node_values_new: List,
node_impurity_new: List[float],
missing_go_to_left: Optional[List[int]] = None,
missing_go_to_left_new: Optional[List[int]] = None,
) -> Tuple[List, List, List, List, List, List, List, List]:
"""向现有树结构的指定节点插入一棵子树。
用于在决策树中指定节点处插入新的分裂分支。
新子树的节点 ID 会自动偏移以避免与原树冲突。
:param node: 目标节点 ID（插入位置）
:param split_list_left: 原树左子节点列表
:param split_list_right: 原树右子节点列表
:param feature: 原树分裂特征列表
:param threshold: 原树分裂阈值列表
:param node_samples: 原树节点样本数列表
:param node_values: 原树节点值列表
:param node_impurity: 原树节点不纯度列表
:param split_list_left_new: 新子树左子节点列表
:param split_list_right_new: 新子树右子节点列表
:param feature_new: 新子树分裂特征列表
:param threshold_new: 新子树分裂阈值列表
:param node_samples_new: 新子树节点样本数列表
:param node_values_new: 新子树节点值列表
:param node_impurity_new: 新子树节点不纯度列表
:param missing_go_to_left: 原树缺失值路由方向列表
:param missing_go_to_left_new: 新子树缺失值路由方向列表
:return: 更新后的树结构元组
"""
split_list_left_new = list(split_list_left_new)
split_list_right_new = list(split_list_right_new)
feature_new = list(feature_new)
threshold_new = list(threshold_new)
node_samples_new = list(node_samples_new)
node_values_new = list(node_values_new)
node_impurity_new = list(node_impurity_new)
if missing_go_to_left is None:
missing_go_to_left = [0] * len(feature)
if missing_go_to_left_new is None:
missing_go_to_left_new = [0] * len(feature_new)
missing_go_to_left_new = list(missing_go_to_left_new)
if node == 0:
return (
split_list_left_new,
split_list_right_new,
feature_new,
threshold_new,
node_samples_new,
node_values_new,
node_impurity_new,
missing_go_to_left_new,
)
# 为避免节点 ID 冲突，将新子树节点 ID 偏移
add_n = len(feature) - 1
split_list_left_new = [i + add_n if i != -1 else i for i in split_list_left_new]
split_list_right_new = [i + add_n if i != -1 else i for i in split_list_right_new]
# 替换目标节点
split_list_left[node] = split_list_left_new[0]
split_list_right[node] = split_list_right_new[0]
feature[node] = feature_new[0]
threshold[node] = threshold_new[0]
node_samples[node] = node_samples_new[0]
node_values[node] = node_values_new[0]
node_impurity[node] = node_impurity_new[0]
missing_go_to_left[node] = missing_go_to_left_new[0]
# 追加新子树剩余节点
split_list_left += split_list_left_new[1:]
split_list_right += split_list_right_new[1:]
feature += feature_new[1:]
threshold += threshold_new[1:]
node_samples += node_samples_new[1:]
node_values += node_values_new[1:]
node_impurity += node_impurity_new[1:]
missing_go_to_left += missing_go_to_left_new[1:]
return (
split_list_left,
split_list_right,
feature,
threshold,
node_samples,
node_values,
node_impurity,
missing_go_to_left,
)
def _delete_nodes(
node: int,
split_list_left: List[int],
split_list_right: List[int],
feature: List[int],
threshold: List[float],
node_samples: List[int],
node_values: List,
node_impurity: List[float],
missing_go_to_left: Optional[List[int]] = None,
) -> Tuple[List, List, List, List, List, List, List, List]:
"""删除树中指定节点及其所有子节点，将该节点变为叶子节点。
:param node: 待删除节点 ID
:param split_list_left: 左子节点列表
:param split_list_right: 右子节点列表
:param feature: 分裂特征列表
:param threshold: 分裂阈值列表
:param node_samples: 节点样本数列表
:param node_values: 节点值列表
:param node_impurity: 节点不纯度列表
:param missing_go_to_left: 缺失值路由方向列表
:return: 更新后的树结构元组
"""
snd = node
if missing_go_to_left is None:
missing_go_to_left = [0] * len(feature)
def _del_children_iter(
start_node: int, left_list: List[int], right_list: List[int]
) -> Tuple[List[int], List[int]]:
"""递归删除子树：将左右子节点指针置为 -1。"""
left_list = list(left_list)
right_list = list(right_list)
next_node = left_list[start_node]
if next_node != -1:
if start_node == snd:
left_list[start_node] = -1
else:
left_list[start_node] = 0
left_list, right_list = _del_children_iter(next_node, left_list, right_list)
else:
if start_node == snd:
left_list[start_node] = -1
else:
left_list[start_node] = 0
next_node = right_list[start_node]
if next_node != -1:
if start_node == snd:
right_list[start_node] = -1
else:
right_list[start_node] = 0
left_list, right_list = _del_children_iter(next_node, left_list, right_list)
else:
if start_node == snd:
right_list[start_node] = -1
else:
right_list[start_node] = 0
return left_list, right_list
def _remap_nodes(node_list: List[int]) -> Tuple[Dict[int, int], List[int]]:
"""重新编号节点，构建节点 ID 映射。
返回 (ID映射字典, 被删除节点列表)。
"""
remap: Dict[int, int] = {i: i for i in range(len(node_list))}
remap[-1] = -1
removed: List[int] = []
for i, v in enumerate(node_list):
if v == 0:
removed.append(i)
for j in remap:
if i < j:
remap[j] = remap[j] - 1
return remap, removed
split_list_left, split_list_right = _del_children_iter(node, split_list_left, split_list_right)
node_map, _ = _remap_nodes(split_list_left)
split_left_new: List[int] = []
split_right_new: List[int] = []
feat_new: List[int] = []
thresh_new: List[float] = []
n_samples_new: List[int] = []
val_new: List = []
impur_new: List[float] = []
mgl_new: List[int] = []
for a, b, c, d, f, g, h, k, m in zip(
split_list_left,
split_list_right,
feature,
threshold,
node_samples,
node_values,
node_impurity,
missing_go_to_left,
list(range(len(split_list_left))),
):
if a == 0 and b == 0:
continue
elif a * b == 0:
raise ValueError(f"树结构异常，请检查节点 {m} 的输入")
split_left_new.append(node_map[a])
split_right_new.append(node_map[b])
if m == node:
feat_new.append(-2)
thresh_new.append(-2.0)
else:
feat_new.append(c)
thresh_new.append(d)
n_samples_new.append(f)
val_new.append(g)
impur_new.append(h)
mgl_new.append(k)
return (
split_left_new,
split_right_new,
feat_new,
thresh_new,
n_samples_new,
val_new,
impur_new,
mgl_new,
)
# ============================================================================
# 辅助函数
# ============================================================================
def _find_subtree_node_ids(tree_info, root_node: int) -> List[int]:
"""返回以 root_node 为根的子树中所有节点 ID（包括根节点及所有后代）。
:param tree_info: 树信息对象（_TreeInfo 或含 children_left/children_right 的对象）
:param root_node: 子树根节点 ID
:return: 节点 ID 列表
"""
result = [root_node]
stack = [root_node]
while stack:
node = stack.pop()
left = tree_info.children_left[node]
right = tree_info.children_right[node]
for child in (left, right):
if child != -1:
result.append(child)
stack.append(child)
return result
def _node_hit_report(
df_rules: pd.DataFrame,
format_rule,
data: pd.DataFrame,
target: str,
overdue: Optional[Union[str, List[str]]],
dpds: Optional[Union[int, List[int]]],
del_grey: bool,
leaf_only: bool,
feature_map: Optional[Dict[str, str]] = None,
n_jobs=-1,
parallel_backend=None,
parallel_config=None,
**kwargs: Any,
) -> pd.DataFrame:
"""对单个数据集，汇总每个节点规则 :meth:`Rule.report` 中"命中"分箱的结果行。
供 :class:`DecisionTreeAnalyzer` 和 :class:`ManualTreeExtractor` 的 ``report()`` 方法共用。
:param df_rules: 规则 DataFrame（含 node / if_leaf / rule_list 列）
:param format_rule: 将 rule_list 解析为 :class:`Rule` 对象的函数（空规则返回 None）
:param data: 待评估数据集
:param target: 目标变量列名
:param overdue: 逾期天数字段名，参考 :meth:`Rule.report`
:param dpds: 逾期定义方式，参考 :meth:`Rule.report`
:param del_grey: 是否删除灰度样本
:param leaf_only: 是否仅评估叶子节点
:return: 各节点效果评估 DataFrame
"""
rules_df = df_rules[df_rules["if_leaf"]] if leaf_only else df_rules
tasks = []
labels = []
for _, row in rules_df.iterrows():
node_id = int(row["node"])
is_leaf = bool(row["if_leaf"])
rule_list = row["rule_list"]
rule = format_rule(rule_list)
if rule is None:
rule = Rule(expr="True", name="空规则", description="空规则")
rule.n_jobs = 1 if n_jobs is None or n_jobs in (1, 1.0) else -1
rule.parallel_backend = parallel_backend
rule.parallel_config = parallel_config
tasks.append(
(
node_id,
is_leaf,
rule,
data,
target,
overdue,
dpds,
del_grey,
dict(kwargs),
)
)
labels.append(f"节点 {node_id}")
hit_frames = parallel_execute(
_node_hit_worker,
tasks,
n_jobs=n_jobs,
parallel_backend=parallel_backend,
parallel_config=parallel_config,
task_labels=labels,
default_backend="threading",
has_parallel_children=overdue is not None,
workload=_mining_workload(
data,
len(tasks),
operation="决策树节点规则报告",
cost_per_item=12.0,
has_parallel_children=overdue is not None,
),
)
if not hit_frames:
return pd.DataFrame()
result = pd.concat(hit_frames, ignore_index=True)
sort_col = ("分箱详情", "节点编号") if isinstance(result.columns, pd.MultiIndex) else "节点编号"
resolved_map = resolve_feature_map(feature_map)
if resolved_map is not None:
node_context = {}
for _, row in rules_df.iterrows():
used_features = [condition[0] for condition in row["rule_list"]]
node_context[int(row["node"])] = format_feature_context(used_features, resolved_map)
field_col = ("分箱详情", "入参字段") if isinstance(result.columns, pd.MultiIndex) else "入参字段"
meaning_col = ("分箱详情", "字段含义") if isinstance(result.columns, pd.MultiIndex) else "字段含义"
result[field_col] = result[sort_col].map(lambda node: node_context.get(int(node), ("", ""))[0])
result[meaning_col] = result[sort_col].map(lambda node: node_context.get(int(node), ("", ""))[1])
return result.sort_values(sort_col).reset_index(drop=True)
# ============================================================================
# 树信息内部类（封装 sklearn 树结构的原始数组）
# ============================================================================
class _TreeInfo:
"""封装决策树原始结构数据的内部类。
模拟 sklearn 的 tree_ 属性，方便手动构建和操作树结构，
同时兼容 sklearn 原生的 tree_ 数组格式。
"""
def __init__(self, feature_list: List[str], n_classes: int):
self.children_left: List[int] = []
self.children_right: List[int] = []
self.feature: List[int] = []
self.threshold: List[float] = []
self.feature_names = list(feature_list)
self.n_features_in_ = len(feature_list)
self.n_node_samples: List[int] = []
self.value: List = []
self.impurity: List[float] = []
# 缺失值路由方向：每节点 1=缺失走左(<=)，0=缺失走右(>)
self.missing_go_to_left: List[int] = []
self.n_outputs = 1
self.n_classes = n_classes
class _SimTree:
"""模拟 sklearn tree_ 对象的简易封装。
用于将 _TreeInfo 结构适配为 _rule_generator 可识别的格式。
"""
def __init__(
self,
children_left: List[int],
children_right: List[int],
feature: List[int],
threshold: List[float],
n_node_samples: List[int],
value: List,
impurity: List[float],
missing_go_to_left: Optional[List[int]] = None,
):
# 将自身作为 .tree_ 属性暴露，兼容 _rule_generator(clf.tree_) 的访问方式
self.tree_ = self
self.children_left = children_left
self.children_right = children_right
self.feature = feature
self.threshold = threshold
self.n_node_samples = n_node_samples
self.value = value
self.impurity = impurity
# 缺失值路由方向，缺省按 0（缺失走右，与 pandas eval 行为一致）
self.missing_go_to_left = (
list(missing_go_to_left) if missing_go_to_left is not None else [0] * len(feature)
)
# ============================================================================
# DecisionTreeAnalyzer：标准 sklearn 决策树分析器
# ============================================================================
[文档]
class DecisionTreeAnalyzer(ParallelizableMixin):
"""sklearn 决策树分析器。
在标准 sklearn DecisionTreeClassifier 基础上，提供决策树训练、
结构导出、规则提取、节点规则评估和 AUC / KS / LIFT 等模型指标计算。
**参数**
:param target: 目标变量列名（0=好样本，1=坏样本）
:param features: 特征名或特征名列表（默认自动从数据中推断数值列）
:param feature_map: 字段名到字段含义的映射，用于节点规则报告展示
:param tree_params: 决策树参数字典，默认值如下：
============ ================================
参数 默认值
============ ================================
criterion 'gini'
splitter 'best'
max_depth 2
min_samples_split 2
min_samples_leaf 1
random_state 0
============ ================================
**参考样例**
>>> 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')
"""
def __init__(
self,
target: str = "target",
features: FeatureNames = None,
tree_params: Optional[Dict[str, Any]] = None,
missing: Optional[float] = None,
n_jobs: Optional[Union[int, float]] = -1,
parallel_backend: Optional[str] = None,
parallel_config: Optional[Dict[str, Any]] = None,
feature_map: Optional[Dict[str, str]] = None,
**kwargs: Any,
):
"""初始化决策树训练器。
:param target: 目标变量列名（0=好样本，1=坏样本）
:param features: 特征名列表（默认自动从数据中推断数值列）
:param tree_params: 决策树参数字典，默认值如下：
============ ================================
参数 默认值
============ ================================
criterion 'gini'
splitter 'best'
max_depth 2
min_samples_split 2
min_samples_leaf 1
random_state 0
============ ================================
:param missing: 缺失值的等价填充数值（可选），按 sklearn 版本自动适配缺失处理。
``None``（默认）时：新版 sklearn（>=1.3）按决策树学到的缺失值路由方向处理；
旧版 sklearn 无原生缺失支持，含缺失数据需传入本参数。传入数值时，训练/预测/
规则评估均将缺失等价为该数值，使规则命中样本数与决策树节点样本数完全一致，
并兼容不支持原生缺失的旧版 sklearn（例如训练前已将缺失填充为 -999，则传入
``missing=-999``）。
:param kwargs: sklearn DecisionTreeClassifier 的其他参数，直接透传给底层分类器。
例如：`ccp_alpha=0.01`、`class_weight='balanced'`、`min_weight_fraction_leaf=0.1` 等。
"""
self.target = target
self.features = features
self.tree_params = tree_params
self.missing = missing
self.n_jobs = n_jobs
self.parallel_backend = parallel_backend
self.parallel_config = parallel_config
self.feature_map = feature_map
self._sklearn_kwargs: Dict[str, Any] = kwargs
self.features_: List[str] = []
# 默认树参数
self._default_params = {
"criterion": "gini",
"splitter": "best",
"max_depth": 2,
"min_samples_split": 2,
"min_samples_leaf": 1,
"min_weight_fraction_leaf": 0.0,
"max_features": None,
"random_state": 0,
"max_leaf_nodes": None,
"min_impurity_decrease": 0.0,
"class_weight": None,
"ccp_alpha": 0.0,
}
# 内部状态
self.clf: Optional[DecisionTreeClassifier] = None
self._data: Optional[pd.DataFrame] = None
self._df_rules: Optional[pd.DataFrame] = None
self._is_fitted: bool = False
self._dot_data: Optional[str] = None
self.__tree_info_cache: Optional[_TreeInfo] = None
self._manual_split_nodes: set = set() # 兼容 tree_viz._extract_tree_from_mte()
@property
def _tree_info(self) -> _TreeInfo:
"""提供 _TreeInfo 接口代理，兼容 tree_viz 等工具的 _tree_info 访问模式。
注意：DecisionTreeAnalyzer 底层是 sklearn 树，不支持 manual_split 干预。
"""
self._check_fitted()
if self.__tree_info_cache is None:
n_classes = getattr(self.clf, "n_classes_", getattr(self.clf.tree_, "n_classes", 2))
n_classes = int(np.asarray(n_classes).ravel()[0])
self.__tree_info_cache = _TreeInfo(
self.features_,
n_classes,
)
tree = self.clf.tree_
self.__tree_info_cache.children_left = list(tree.children_left)
self.__tree_info_cache.children_right = list(tree.children_right)
self.__tree_info_cache.feature = list(tree.feature)
self.__tree_info_cache.threshold = list(tree.threshold)
self.__tree_info_cache.n_node_samples = list(tree.n_node_samples)
self.__tree_info_cache.value = [list(v) for v in tree.value]
self.__tree_info_cache.impurity = list(tree.impurity)
self.__tree_info_cache.missing_go_to_left = list(
getattr(tree, "missing_go_to_left", [0] * len(tree.feature))
)
return self.__tree_info_cache
# -------------------------------------------------------------------------
# 训练
# -------------------------------------------------------------------------
[文档]
def fit(
self,
X: Union[pd.DataFrame, np.ndarray],
y: Optional[Union[pd.Series, np.ndarray]] = None,
feature_names: FeatureNames = None,
tree_params: Optional[Dict[str, Any]] = None,
**kwargs: Any,
) -> "DecisionTreeAnalyzer":
"""在临时副本中训练，并在全部成功后提交拟合状态。"""
legacy_features = kwargs.pop("features", None)
if feature_names is not None and legacy_features is not None:
raise ValueError("feature_names 与兼容参数 features 不能同时传入")
resolved_feature_names = feature_names if feature_names is not None else legacy_features
working = copy.deepcopy(self)
working._fit_inplace(
X,
y,
feature_names=resolved_feature_names,
tree_params=tree_params,
**kwargs,
)
_commit_transactional_state(
self,
working,
(
"target",
"features",
"tree_params",
"missing",
"n_jobs",
"parallel_backend",
"parallel_config",
"feature_map",
),
)
return self
def _fit_inplace(
self,
X: Union[pd.DataFrame, np.ndarray],
y: Optional[Union[pd.Series, np.ndarray]] = None,
feature_names: FeatureNames = None,
tree_params: Optional[Dict[str, Any]] = None,
**kwargs: Any,
) -> "DecisionTreeAnalyzer":
"""训练决策树。
支持两种 API 风格：
- **sklearn 风格**：``fit(X, y)``，``X`` 为特征矩阵（DataFrame/ndarray），``y`` 为标签；
- **scorecardpipeline 风格**：``fit(df, feature_names=[...])``，``df`` 含目标列，
目标列名由初始化 ``target`` 指定。
:param X: 特征矩阵（sklearn 风格）或含目标列的 DataFrame（scorecardpipeline 风格）
:param y: 目标变量（可选）；传入时按 sklearn 风格解析，优先于 ``X`` 中的目标列
:param feature_names: 特征名或特征名列表（默认：scorecardpipeline 风格取除 target 外的数值列，
未指定时回退到构造参数 ``features``）
:param tree_params: 决策树参数字典（与构造参数合并，覆盖默认参数）
:param kwargs: sklearn DecisionTreeClassifier 的其他参数，直接透传给底层分类器。
优先级最高，会覆盖默认参数、tree_params 和构造参数中的同名值。
:return: self
**参考样例**
>>> # scorecardpipeline 风格
>>> analyzer = DecisionTreeAnalyzer(target='target')
>>> analyzer.fit(df_train, feature_names=['age', 'income', 'loan'])
>>> # sklearn 风格
>>> DecisionTreeAnalyzer(target='target').fit(X_train, y_train)
>>> # 使用 ccp_alpha 后剪枝
>>> DecisionTreeAnalyzer(target='target').fit(df_train, ccp_alpha=0.01)
"""
self._parallel_execute(
_tree_metric_dataset_worker,
[],
default_backend="threading",
workload=_mining_workload(None, 0, operation="决策树并行配置校验"),
)
# 解析双 API：sklearn 风格 (X, y) 或 scorecardpipeline 风格 (df)
resolved_features = feature_names if feature_names is not None else self.features
df, self.features_ = _resolve_fit_data(X, y, resolved_features, self.target)
# 过滤缺失数据
self._data = df.loc[df[self.target].notna(), self.features_ + [self.target]].copy()
# 合并参数：默认参数 → 构造参数 → 调用参数 → kwargs（优先级最高）
params = {**self._default_params, **(self.tree_params or {})}
if tree_params:
params = {**params, **tree_params}
params = {**params, **self._sklearn_kwargs, **kwargs}
# 训练（按 sklearn 版本与 missing 自动处理缺失：指定 missing 则等价填充，
# 新版 sklearn 原生支持缺失，旧版含缺失且未指定 missing 时给出明确提示）
self.clf = DecisionTreeClassifier(**params)
X = _prepare_training_features(self._data, self.features_, self.missing, "DecisionTreeAnalyzer")
y = self._data[self.target].values
self.clf.fit(X, y)
self._is_fitted = True
self.__tree_info_cache = None
self._df_rules = _rule_generator(self.clf, self.features_, missing=self.missing)
return self
# -------------------------------------------------------------------------
# 预测
# -------------------------------------------------------------------------
[文档]
def predict(self, df: Optional[pd.DataFrame] = None) -> np.ndarray:
"""预测类别标签。
:param df: 待预测数据（默认使用训练数据）
:return: 预测结果数组
"""
self._check_fitted()
data = df if df is not None else self._data
data = _impute_features(data, self.features_, self.missing)
return self.clf.predict(data[self.features_].values)
[文档]
def predict_proba(self, df: Optional[pd.DataFrame] = None) -> np.ndarray:
"""预测类别概率。
:param df: 待预测数据（默认使用训练数据）
:return: 类别概率数组，形状 (n_samples, n_classes)
"""
self._check_fitted()
data = df if df is not None else self._data
data = _impute_features(data, self.features_, self.missing)
return self.clf.predict_proba(data[self.features_].values)
[文档]
def apply(self, df: Optional[pd.DataFrame] = None) -> np.ndarray:
"""返回每个样本所属叶子节点的编号。
:param df: 待评估数据（默认使用训练数据）
:return: 叶子节点编号数组
**参考样例**
>>> leaf_ids = fitter.apply(df_test)
>>> print(f"测试集样本分布在 {len(set(leaf_ids))} 个叶子节点")
"""
self._check_fitted()
data = df if df is not None else self._data
data = _impute_features(data, self.features_, self.missing)
return self.clf.apply(data[self.features_].values)
# -------------------------------------------------------------------------
# 评估
# -------------------------------------------------------------------------
[文档]
def evaluate(
self,
test_data_list: List[Tuple[str, pd.DataFrame]],
metric_type: str = "auc",
top_rate: float = 0.1,
) -> List[Tuple[str, float]]:
"""评估模型性能。
支持多种评估指标，计算训练集及多个测试集的指标值。
:param test_data_list: 测试数据集列表，元素为 (数据集名称, DataFrame)
:param metric_type: 评估指标类型
========= ==========================================
类型 说明
========= ==========================================
'auc' ROC AUC 分数（使用 predict_proba 的正类概率）
'ks' KS 统计量
'lift' top 客群的 LIFT 值
'top' top客群坏样本率（与 lift 等价）
========= ==========================================
:param top_rate: lift/top 指标计算时取 top 的比例（默认 10%）
:return: 评估结果列表，元素为 (数据集名称, 指标值)
**参考样例**
>>> metrics = fitter.evaluate([('测试集', df_test)], metric_type='ks')
>>> for name, value in metrics:
... print(f'{name}: {value:.4f}')
"""
self._check_fitted()
if metric_type not in ("auc", "ks", "lift", "top"):
raise ValueError(f"不支持的指标类型: {metric_type}，可选值: auc/ks/lift/top")
datasets = [("训练集", self._data)] + list(test_data_list)
tasks = [
(self, name, data, metric_type, top_rate)
for name, data in datasets
]
return self._parallel_execute(
_tree_metric_dataset_worker,
tasks,
task_labels=[name for name, _ in datasets],
default_backend="threading",
has_parallel_children=False,
workload=_mining_workload(
self._data,
len(tasks),
operation="决策树数据集指标评估",
cost_per_item=10.0,
),
)
def _calc_metric(
self,
y_prob: np.ndarray,
y_true: np.ndarray,
metric_type: str,
top_rate: float,
) -> float:
"""计算单条数据的指定指标。"""
if metric_type == "auc":
return _auc(y_true, y_prob)
elif metric_type == "ks":
return _ks(y_true, y_prob)
elif metric_type in ("lift", "top"):
return _lift_local(y_true, y_prob, n_bins=int(top_rate * 100))
[文档]
def report(
self,
datasets: Union[pd.DataFrame, List[pd.DataFrame], Dict[Any, pd.DataFrame]],
target: Optional[str] = None,
overdue: Optional[Union[str, List[str]]] = None,
dpds: Optional[Union[int, List[int]]] = None,
del_grey: bool = False,
leaf_only: bool = False,
feature_map: Optional[Dict[str, str]] = None,
**kwargs: Any,
) -> Union[pd.DataFrame, List[pd.DataFrame], Dict[Any, pd.DataFrame]]:
"""在新数据集上评估决策树各节点规则的效果。
对每个节点（含分裂节点和叶子节点），将其规则路径解析为 :class:`Rule` 对象，
直接调用 :meth:`Rule.report` 计算统计指标，取其中"命中"分箱对应的结果行汇总，
得到各节点在新数据上的效果报告。
**参数**
:param datasets: 待评估数据集，支持单个 DataFrame、DataFrame 列表、
或 ``{名称: DataFrame}`` 字典；返回结果类型与输入保持一致
:param target: 目标变量列名，默认为None（使用拟合时的 ``self.target``）
:param overdue: 逾期天数字段名（可选，传入时以逾期天数>DPD定义坏样本，
支持多标签多DPD联合分析），参考 :meth:`Rule.report`
:param dpds: 逾期定义方式，逾期天数 > DPD 为坏样本，默认为0；
传入列表时支持多DPD联合分析，参考 :meth:`Rule.report`
:param del_grey: 是否删除逾期天数在(0, DPD]区间内的灰度样本，默认为False
:param leaf_only: 是否仅评估叶子节点，默认 False（评估所有节点）
:param feature_map: 字段名到字段含义的映射，显式传入时覆盖构造参数
:param kwargs: 其余传递给 :meth:`Rule.report` 的参数（如 ``amount``、``margins``）
:return: 各节点效果评估表（DataFrame），结构与输入一致：
单个 DataFrame 输入返回单个 DataFrame，列表输入返回 DataFrame 列表，
字典输入返回 ``{名称: 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])
"""
self._check_fitted()
if isinstance(datasets, dict):
names = list(datasets)
data_values = list(datasets.values())
elif isinstance(datasets, list):
names = None
data_values = list(datasets)
else:
names = None
data_values = [datasets]
resolved_feature_map = resolve_feature_map(self.feature_map, feature_map)
tasks = [
(
self,
data,
target or self.target,
overdue,
dpds,
del_grey,
leaf_only,
resolved_feature_map,
dict(kwargs),
)
for data in data_values
]
results = self._parallel_execute(
_tree_report_dataset_worker,
tasks,
task_labels=names if names is not None else list(range(len(tasks))),
default_backend="threading",
has_parallel_children=True,
workload=_mining_workload(
max(data_values, key=len, default=None),
len(tasks),
operation="决策树多数据集报告",
cost_per_item=16.0,
has_parallel_children=True,
),
)
if isinstance(datasets, dict):
return dict(zip(names, results))
if isinstance(datasets, list):
return results
return results[0]
def _report_one_dataset(
self,
data,
*,
target,
overdue,
dpds,
del_grey,
leaf_only,
feature_map,
report_kwargs,
):
"""计算单个数据集的节点报告，供模块级 worker 调用。"""
return _node_hit_report(
self._df_rules,
self._format_rule,
data,
target=target,
overdue=overdue,
dpds=dpds,
del_grey=del_grey,
leaf_only=leaf_only,
feature_map=feature_map,
n_jobs=_effective_nested_n_jobs(self.n_jobs),
parallel_backend=self.parallel_backend,
parallel_config=self.parallel_config,
**report_kwargs,
)
[文档]
def get_leaf_node_ids(self) -> List[int]:
"""获取所有叶子节点的 ID 列表。"""
self._check_fitted()
return self._df_rules[self._df_rules["if_leaf"]]["node"].tolist()
# -------------------------------------------------------------------------
# 规则提取
# -------------------------------------------------------------------------
[文档]
def get_rules(self) -> List[Rule]:
"""将树的叶子节点转换为 Rule 对象列表。
:return: Rule 对象列表，每个 Rule 对应一个叶子节点
**参考样例**
>>> rules = fitter.get_rules()
>>> for rule in rules:
... report = rule.report(df_test, target='target')
"""
self._check_fitted()
rules: List[Rule] = []
leaf_rules = self._df_rules[self._df_rules["if_leaf"]]
for _, row in leaf_rules.iterrows():
rule = self._format_rule(row["rule_list"])
if rule is None:
# 叶子节点必然带有规则路径，空规则（根节点）理论上不会出现，稳妥跳过
continue
rule.name = f"DecisionTree_N{int(row['node'])}"
rules.append(rule)
return rules
[文档]
def get_rule_table(
self,
datasets: Optional[pd.DataFrame] = None,
target: Optional[str] = None,
overdue: Optional[Union[str, List[str]]] = None,
dpds: Optional[Union[int, List[int]]] = None,
del_grey: bool = False,
leaf_only: bool = False,
**kwargs: Any,
) -> pd.DataFrame:
"""获取决策树所有节点（分裂节点+叶子节点）的规则效果表。
从每个节点提取规则路径，解析为 :class:`Rule` 对象后调用 :meth:`Rule.report`
在数据集上计算命中效果，取其中"命中"分箱对应的结果行汇总。各节点统计口径
与 :meth:`report` 完全一致（均基于规则在数据集上的真实命中情况计算，而非
读取决策树内部存储的节点样本统计），便于与人工分裂结果横向对比。
**参数**
:param datasets: 用于计算规则效果的数据集，默认 None（使用 fit 时的训练数据）
:param target: 目标变量列名，默认 None（使用拟合时的 ``self.target``）
:param overdue: 逾期天数字段名，参考 :meth:`Rule.report`
:param dpds: 逾期定义方式，参考 :meth:`Rule.report`
:param del_grey: 是否删除灰度样本，默认 False
:param leaf_only: 是否仅返回叶子节点，默认 False
:param kwargs: 其余传递给 :meth:`Rule.report` 的参数（如 ``amount``、``margins``）
:return: 规则效果表，列结构与 :meth:`report` 一致（节点编号、是否叶子、
规则分类、指标名称、指标含义、分箱、样本总数、样本占比、好样本数、
坏样本数、坏样本率、LIFT值、坏账改善、风险拒绝比等）
**参考样例**
>>> table = fitter.get_rule_table() # 在训练数据上评估
>>> table = fitter.get_rule_table(df_test) # 在新数据上评估
"""
self._check_fitted()
data = datasets if datasets is not None else self._data
if data is None:
raise InputValidationError(
"get_rule_table 需要数据集计算规则效果：请传入 datasets 参数，"
"或先通过 fit() 提供训练数据"
)
return self.report(
data,
target=target,
overdue=overdue,
dpds=dpds,
del_grey=del_grey,
leaf_only=leaf_only,
**kwargs,
)
@staticmethod
def _rule_list_to_text(rule_list: List) -> str:
"""将规则列表格式化为可读的中文规则串（如 "age <= 35.0000 且 income > 5000"）。"""
if not rule_list:
return "空规则"
parts = []
for item in rule_list:
feat = item[0]
op = item[1]
thres = f"{item[2]:.4f}" if isinstance(item[2], float) else str(item[2])
# 第 4 个元素为是否纳入缺失样本，True 时追加"(含缺失)"标记
include_nan = bool(item[3]) if len(item) > 3 else False
text = f"{feat} {op} {thres}"
parts.append(f"{text}(含缺失)" if include_nan else text)
return " 且 ".join(parts)
def _format_rule(self, rule_list: List) -> Optional[Rule]:
"""将规则列表解析为 :class:`Rule` 对象。
- ``expr``：pandas eval 表达式（用于 predict / report 等规则评估）
- ``name`` / ``description``：可读的中文规则串（用于表格展示）
根节点对应的空规则返回 None。
:param rule_list: 规则列表，元素为 ``[特征名, 操作符, 阈值, 是否含缺失]``
（第 4 个元素标记该条件是否需纳入缺失样本，由决策树缺失值路由方向决定）
:return: 解析得到的 Rule 对象；空规则返回 None
"""
if not rule_list:
return None
text = self._rule_list_to_text(rule_list)
expr = self._rule_to_expr(rule_list)
return Rule(expr=expr, name=text, description=text)
def _rule_to_expr(self, rule_list: List) -> str:
"""将规则列表转换为 pandas eval 表达式。
当规则需纳入缺失样本时（决策树将缺失值路由到当前路径方向），对应条件追加
``| (特征 != 特征)``（NaN != NaN 为 True，用于在 pandas eval 中识别缺失），
确保命中样本数与决策树节点样本数完全一致。
"""
if not rule_list:
return "True"
parts = []
for item in rule_list:
feat, op, thres = item[0], item[1], item[2]
include_nan = bool(item[3]) if len(item) > 3 else False
feat_esc = f"`{feat}`" if not str(feat).isidentifier() else str(feat)
cond = f"({feat_esc} {op} {repr(float(thres))})"
if include_nan:
cond = f"({cond} | ({feat_esc} != {feat_esc}))"
parts.append(cond)
return " & ".join(parts)
# -------------------------------------------------------------------------
# 可视化与导出
# -------------------------------------------------------------------------
[文档]
def export_tree(
self,
out_file: Optional[str] = None,
max_depth: Optional[int] = None,
class_names: Optional[List[str]] = None,
) -> str:
"""导出决策树为 DOT 格式。
:param out_file: 输出 .dot 文件路径（可选，指定时同时写入文件）
:param max_depth: 最大显示深度（None=全部显示）
:param class_names: 类别名列表，默认 ['好', '坏']
:return: DOT 格式字符串
**参考样例**
>>> dot = fitter.export_tree('tree.dot')
>>> with open('tree.dot') as f:
... print(f.read())
"""
self._check_fitted()
if class_names is None:
class_names = ["好", "坏"]
return _export_dot_data(
self.clf,
self.features_,
class_names=class_names,
out_file=out_file,
max_depth=max_depth,
)
[文档]
def plot(
self,
backend: str = "matplotlib",
save: Optional[str] = None,
title: str = "",
**kwargs: Any,
) -> Any:
"""绘制当前决策树结构。
:param backend: 渲染后端，可选 ``matplotlib`` / ``pyecharts`` / ``graphviz``
:param save: 保存路径（可选）
:param title: 图表标题
:param kwargs: 其余参数透传给 :func:`hscredit.core.viz.plot_tree`
:return: matplotlib Figure / pyecharts Chart / graphviz Source
**参考样例**
>>> analyzer = DecisionTreeAnalyzer(target='target').fit(df, feature_names=['age', 'income'])
>>> fig = analyzer.plot()
>>> analyzer.plot(backend='graphviz', save='tree.pdf')
"""
self._check_fitted()
from ...core.viz.tree_plots import plot_tree
kwargs.setdefault("feature_names", self.features_)
return plot_tree(self, backend=backend, save=save, title=title, **kwargs)
[文档]
def save(
self,
file_path: str,
include_data: bool = True,
) -> None:
"""将决策树保存为 pickle 文件。
:param file_path: 保存路径
:param include_data: 是否包含训练数据（默认 True，保存后可直接 load 并 evaluate）
**参考样例**
>>> fitter.save('dt_model.pkl')
"""
self._check_fitted()
payload = {
"clf": self.clf,
"features": self.features_,
"target": self.target,
"tree_params": self.tree_params,
"missing": self.missing,
"feature_map": self.feature_map,
}
if include_data and self._data is not None:
payload["_data"] = self._data
with open(file_path, "wb") as f:
pickle.dump(payload, f)
[文档]
@classmethod
def load(cls, file_path: str) -> "DecisionTreeAnalyzer":
"""从 pickle 文件加载决策树。
:param file_path: 模型文件路径
:return: 加载后的 DecisionTreeAnalyzer 实例
**参考样例**
>>> analyzer2 = DecisionTreeAnalyzer.load('dt_model.pkl')
"""
with open(file_path, "rb") as f:
payload = pickle.load(f)
instance = cls(
target=payload["target"],
features=payload.get("features", payload.get("feature_list")),
tree_params=payload["tree_params"],
missing=payload.get("missing"),
feature_map=payload.get("feature_map"),
)
instance.clf = payload["clf"]
instance.features_ = list(payload.get("features", payload.get("feature_list", [])))
instance._is_fitted = True
if "_data" in payload:
instance._data = payload["_data"]
instance.__tree_info_cache = None
instance._df_rules = _rule_generator(
instance.clf, instance.features_, missing=instance.missing
)
return instance
# -------------------------------------------------------------------------
# 辅助方法
# -------------------------------------------------------------------------
def _check_fitted(self) -> None:
"""检查是否已训练。"""
if not self._is_fitted:
raise RuntimeError("请先调用 fit() 方法训练决策树")
def __repr__(self) -> str:
if self._is_fitted:
n_leaves = int(self._df_rules["if_leaf"].sum()) if self._df_rules is not None else 0
return (
f"DecisionTreeAnalyzer(target='{self.target}', "
f"features={self.features_}, "
f"leaves={n_leaves})"
)
return "DecisionTreeAnalyzer(not fitted)"
# ============================================================================
# ManualTreeExtractor：人工干预决策树节点分裂
# ============================================================================