hscredit.report.mining.tree_extractor 源代码
"""决策树规则提取模块.
支持从多种树模型中提取规则，包括：
- 决策树 (Decision Tree)
- 随机森林 (Random Forest)
- 卡方决策树 (Chi-square Tree)
- GBDT
- 孤立森林 (Isolation Forest)
"""
import copy
import numpy as np
import pandas as pd
from typing import Union, List, Dict, Optional, Any
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder
import warnings
from .base import (
BaseRuleMiner,
FeatureNames,
_mining_workload,
format_feature_context,
resolve_feature_map,
)
from ...core.rules.rule import Rule
from ...utils.parallel import _current_parallel_budget, resolve_n_jobs, validate_parallel_config
# 从 hscredit.core.models 统一导入 sklearn 模型
from ...core.models import (
DecisionTreeClassifier,
RandomForest,
GradientBoosting,
)
def _normalize_tree_algorithm(algorithm: str) -> str:
"""校验树算法名称，同时保留构造参数的原始 sklearn 契约。"""
resolved = algorithm.lower()
if resolved == "xgb":
resolved = "gbdt"
warnings.warn("'xgb'算法已弃用，请使用'gbdt'", DeprecationWarning)
if resolved not in TreeRuleExtractor.VALID_ALGORITHMS:
raise ValueError(
f"不支持的算法: {resolved}，可选: {TreeRuleExtractor.VALID_ALGORITHMS}"
)
return resolved
def _tree_extract_worker(task):
"""从一棵独立树提取规则。"""
extractor, tree, tree_id = task
return extractor._extract_from_tree(tree, tree_id=tree_id)
def _gbdt_tree_extract_worker(task):
"""提取并评估一棵独立 GBDT 基学习器的规则。"""
extractor, tree, tree_id = task
accepted = []
for rule in extractor._extract_from_tree(tree, tree_id=tree_id):
mask = extractor._apply_conditions(rule["conditions"], extractor.X_train_)
hit_count = mask.sum()
if hit_count >= 5:
hit_bad = extractor.y_train_[mask].sum()
badrate = hit_bad / hit_count
if badrate >= 0.05:
rule["sample_count"] = int(hit_count)
rule["class_probability"] = badrate
accepted.append(rule)
return accepted
def _isolation_tree_extract_worker(task):
"""从一棵独立孤立树提取异常路径规则。"""
extractor, estimator, tree_id, scores, anomaly_mask = task
return extractor._extract_from_isolation_tree(
estimator, tree_id, scores, anomaly_mask
)
def _tree_rule_report_worker(task):
"""构造并评估一条独立树规则。"""
ordinal, rule_item, expression, datasets, target, feature_map = task
rule = Rule(
expr=expression,
name=f"TreeRule_{rule_item.get('rule_id', ordinal)}",
description=expression,
weight=float(rule_item.get("importance", 0)),
n_jobs=1,
)
metadata = dict(rule_item)
if feature_map is not None:
used_features = [condition["feature"] for condition in rule_item.get("conditions", [])]
input_fields, field_meanings = format_feature_context(used_features, feature_map)
metadata.update({"入参字段": input_fields, "字段含义": field_meanings})
if datasets is not None:
report_df = rule.report(datasets=datasets, target=target)
hit_rows = (
report_df[report_df["分箱"] == "命中"]
if "分箱" in report_df.columns
else pd.DataFrame()
)
if not hit_rows.empty:
hit = hit_rows.iloc[0].to_dict()
metadata.update(
{
"命中样本数": hit.get("样本总数"),
"命中样本占比": hit.get("样本占比"),
"命中坏样本率": hit.get("坏样本率"),
"命中LIFT值": hit.get("LIFT值"),
"坏账改善": hit.get("坏账改善"),
"风险拒绝比": hit.get("风险拒绝比"),
}
)
rule.metadata_ = metadata
rule.metric_score_ = metadata.get("命中LIFT值", metadata.get("importance", 0))
return rule
[文档]
class TreeRuleExtractor(BaseRuleMiner):
"""树模型规则提取器.
支持多种树模型的规则提取，包括决策树、随机森林、GBDT、XGBoost和孤立森林。
代码风格参考hscredit的binning模块，fit方法兼容scorecardpipeline风格。
支持通过**kwargs传入任意sklearn树模型参数。
:param algorithm: 算法类型，'dt', 'rf', 'chi2', 'gbdt', 'xgb', 'isf'
:param target: 目标变量列名，默认为'target'
:param exclude_cols: 需要排除的列名列表
:param features: 参与建树的字段名或字段名列表，默认None（使用全部候选字段）
:param feature_map: 字段名到字段含义的映射，用于规则报告展示
:param max_depth: 树的最大深度，默认5
:param min_samples_split: 分裂节点最小样本数，默认10
:param min_samples_leaf: 叶子节点最小样本数，默认5
:param n_estimators: 森林中树的数量，默认10
:param max_features: 每棵树考虑的最大特征数，默认'sqrt'
:param test_size: 测试集比例，默认0.3
:param random_state: 随机种子，默认42
:param feature_trends: 特征趋势字典，如{'age': 1}表示正相关
:param chi2_threshold: 卡方分箱阈值，默认3.841
:param 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.
**参考样例**
>>> 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 = {'dt', 'rf', 'chi2', 'gbdt', 'xgb', 'isf'}
def __init__(
self,
algorithm: str = 'dt',
target: str = 'target',
exclude_cols: Optional[List[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: Optional[Dict[str, int]] = None,
chi2_threshold: float = 3.841,
n_jobs: Optional[Union[int, float]] = -1,
parallel_backend: Optional[str] = None,
parallel_config: Optional[Dict[str, Any]] = None,
features: FeatureNames = None,
feature_map: Optional[Dict[str, str]] = None,
**kwargs
):
super().__init__(
target=target,
exclude_cols=exclude_cols,
n_jobs=n_jobs,
parallel_backend=parallel_backend,
parallel_config=parallel_config,
features=features,
feature_map=feature_map,
)
self.algorithm = algorithm
self._resolved_algorithm = _normalize_tree_algorithm(algorithm)
self.max_depth = max_depth
self.min_samples_split = min_samples_split
self.min_samples_leaf = min_samples_leaf
self.n_estimators = n_estimators
self.max_features = max_features
self.test_size = test_size
self.random_state = random_state
self.feature_trends = feature_trends
self.chi2_threshold = chi2_threshold
self.model_kwargs = kwargs # 存储额外的模型参数
self.model_ = None
self.encoders_ = {}
self.feature_names_ = []
self.rules_ = []
self.is_fitted_ = False
[文档]
def fit(
self,
X: Union[pd.DataFrame, np.ndarray],
y: Optional[Union[pd.Series, np.ndarray]] = None,
feature_names: FeatureNames = None,
**kwargs
) -> 'TreeRuleExtractor':
"""在临时副本中拟合，成功后原子提交模型与编码状态。"""
working = copy.deepcopy(self)
working._fit_inplace(X, y, feature_names=feature_names, **kwargs)
self._commit_fitted_state(working)
return self
def _fit_inplace(
self,
X: Union[pd.DataFrame, np.ndarray],
y: Optional[Union[pd.Series, np.ndarray]] = None,
feature_names: FeatureNames = None,
**kwargs
) -> 'TreeRuleExtractor':
"""拟合提取器.
:param X: 训练数据
:param y: 目标变量（监督学习需要）
:param feature_names: 本次拟合使用的字段名或字段名列表，优先于构造参数 ``features``
:param kwargs: 额外参数，可覆盖初始化参数
:return: self
"""
self._reset_fitted_state()
# 更新参数
for key, value in kwargs.items():
if hasattr(self, key):
setattr(self, key, value)
elif key in self.model_kwargs:
self.model_kwargs[key] = value
self._resolved_algorithm = _normalize_tree_algorithm(self.algorithm)
validate_parallel_config(self.parallel_backend, self.parallel_config)
self._effective_model_workers()
X, y = self._check_input_data(X, y, feature_names=feature_names)
# 保存特征名
self.feature_names_ = list(X.columns)
# 编码类别型特征
X_encoded = self._encode_categorical_features(X)
# 初始化模型
self.model_ = self._initialize_model()
if self._resolved_algorithm == 'isf':
# 孤立森林不需要y
self.model_.fit(X_encoded)
self.X_train_ = X_encoded
self.X_ = X
else:
# 监督学习需要y
if y is None:
raise ValueError(f"算法 '{self._resolved_algorithm}' 需要目标变量y")
# 划分训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(
X_encoded, y,
test_size=self.test_size,
random_state=self.random_state
)
self.X_train_ = X_train
self.X_test_ = X_test
self.y_train_ = y_train
self.y_test_ = y_test
self.X_ = X
# 卡方分箱预处理
if self._resolved_algorithm == 'chi2':
X_train = self._chi2_preprocess(X_train, y_train)
X_test = self._chi2_preprocess(X_test, y_test, fit=False)
# 训练模型
self.model_.fit(X_train, y_train)
# 保存训练后的数据
self.X_train_ = X_train
self.X_test_ = X_test
self.is_fitted_ = True
return self
def _reset_fitted_state(self) -> None:
"""清除上一轮模型派生状态；仅在事务 working 副本中调用。"""
self.model_ = None
self.encoders_ = {}
self.feature_names_ = []
self.rules_ = []
self.is_fitted_ = False
for attribute in (
"X_",
"X_train_",
"X_test_",
"y_train_",
"y_test_",
"chi2_bins_",
):
self.__dict__.pop(attribute, None)
def _effective_model_workers(self) -> int:
"""解析底层树模型的工作数，并遵守活跃的父层预算。"""
budget = _current_parallel_budget()
workers = resolve_n_jobs(
self.n_jobs,
available_budget=budget.available,
) or 1
if budget.depth > 0:
workers = min(workers, budget.available)
return max(1, workers)
def _encode_categorical_features(self, X: pd.DataFrame) -> pd.DataFrame:
"""编码类别型特征.
:param X: 输入数据
:return: 编码后的数据
"""
X_encoded = X.copy()
self.encoders_ = {}
for col in X.columns:
if not pd.api.types.is_numeric_dtype(X[col]):
le = LabelEncoder()
X_encoded[col] = le.fit_transform(X[col].astype(str))
self.encoders_[col] = le
return X_encoded.fillna(0)
def _initialize_model(self):
"""初始化模型，支持通过**kwargs传入任意参数.
从 hscredit.core.models 导入模型，通过 get_native_model() 获取底层 sklearn 模型。
"""
# 构建基础参数字典
base_params = {
'random_state': self.random_state
}
if self._resolved_algorithm == 'dt':
base_params.update({
'max_depth': self.max_depth,
'min_samples_split': self.min_samples_split,
'min_samples_leaf': self.min_samples_leaf,
'n_jobs': self._effective_model_workers(),
})
base_params.update(self.model_kwargs)
return DecisionTreeClassifier(**base_params)
elif self._resolved_algorithm in ['rf', 'chi2']:
base_params.update({
'n_estimators': self.n_estimators,
'max_depth': self.max_depth,
'min_samples_split': self.min_samples_split,
'min_samples_leaf': self.min_samples_leaf,
'max_features': self.max_features,
'n_jobs': self._effective_model_workers()
})
base_params.update(self.model_kwargs)
# 使用 hscredit 的 RandomForest
return RandomForest(**base_params)
elif self._resolved_algorithm == 'gbdt':
base_params.update({
'n_estimators': self.n_estimators,
'max_depth': self.max_depth,
'min_samples_split': self.min_samples_split,
'min_samples_leaf': self.min_samples_leaf,
'max_features': self.max_features,
})
base_params.update(self.model_kwargs)
# 使用 hscredit 的 GradientBoosting
return GradientBoosting(**base_params)
elif self._resolved_algorithm == 'isf':
# 孤立森林直接使用 sklearn（hscredit 暂无 IsolationForestRiskModel）
from sklearn.ensemble import IsolationForest
base_params.update({
'n_estimators': self.n_estimators,
'max_samples': min(256, 1000), # 限制最大样本数
'contamination': 0.1,
'n_jobs': self._effective_model_workers()
})
base_params.update(self.model_kwargs)
return IsolationForest(**base_params)
def _chi2_preprocess(
self,
X: pd.DataFrame,
y: pd.Series,
fit: bool = True
) -> pd.DataFrame:
"""卡方分箱预处理.
:param X: 特征数据
:param y: 目标变量
:param fit: 是否拟合
:return: 分箱后的数据
"""
from scipy.stats import chi2_contingency
if fit:
self.chi2_bins_ = {}
X_binned = X.copy()
for col in X.columns:
if not pd.api.types.is_numeric_dtype(X[col]):
continue
if fit:
# 初始等频分箱
from sklearn.preprocessing import KBinsDiscretizer
discretizer = KBinsDiscretizer(
n_bins=10,
encode='ordinal',
strategy='quantile'
)
try:
bins = discretizer.fit_transform(X[[col]].dropna()).flatten()
bin_edges = discretizer.bin_edges_[0].copy()
# 合并相似分箱
while len(bin_edges) > 2:
chi2_scores = []
for i in range(len(bin_edges) - 2):
merged_bins = bins.copy()
merged_bins[merged_bins == i + 1] = i
try:
contingency = pd.crosstab(merged_bins, y)
chi2_stat, _, _, _ = chi2_contingency(contingency)
chi2_scores.append((i, chi2_stat))
except Exception:
chi2_scores.append((i, float('inf')))
if not chi2_scores:
break
min_idx, min_chi2 = min(chi2_scores, key=lambda x: x[1])
if min_chi2 < self.chi2_threshold:
bins[bins > min_idx] -= 1
bin_edges = np.delete(bin_edges, min_idx + 1)
else:
break
self.chi2_bins_[col] = bin_edges
except Exception:
continue
# 应用分箱
if col in getattr(self, 'chi2_bins_', {}):
X_binned[col] = pd.cut(X[col], bins=self.chi2_bins_[col], labels=False)
return X_binned.fillna(0)
[文档]
def extract_rules(self) -> List[Dict[str, Any]]:
"""提取规则.
:return: 规则列表
"""
self._check_fitted()
if self._resolved_algorithm == 'dt':
self.rules_ = self._extract_from_tree(self.model_, tree_id=0)
elif self._resolved_algorithm in ['rf', 'chi2']:
self.rules_ = self._extract_from_forest()
elif self._resolved_algorithm == 'gbdt':
self.rules_ = self._extract_from_gbdt()
elif self._resolved_algorithm == 'isf':
self.rules_ = self._extract_from_isolation_forest()
# 过滤规则
self.rules_ = self._filter_rules(self.rules_)
# 去重
self.rules_ = self._deduplicate_rules(self.rules_)
# 计算重要性
for rule in self.rules_:
rule['importance'] = self._calculate_rule_importance(rule)
# 排序
self.rules_.sort(key=lambda x: x.get('importance', 0), reverse=True)
return self.rules_
def _extract_from_tree(
self,
tree_model,
tree_id: int = 0
) -> List[Dict[str, Any]]:
"""从单棵树提取规则.
:param tree_model: 树模型 (sklearn Tree 或 hscredit RiskModel)
:param tree_id: 树ID
:return: 规则列表
"""
# 如果是 hscredit 模型，获取底层 sklearn 模型
if hasattr(tree_model, 'get_native_model'):
native_model = tree_model.get_native_model()
tree = native_model.tree_
else:
tree = tree_model.tree_
rules = []
def recurse(node_id, conditions):
"""递归遍历树."""
if tree.feature[node_id] == -2: # 叶子节点
# 计算叶子节点的实际坏账率
badrate = self._calculate_leaf_badrate(conditions)
# 使用 n_node_samples 获取节点样本数（版本无关）。
# sklearn>=1.3 起 tree_.value 存储的是归一化比例（每节点求和为 1），
# 不能再用 value.sum() 还原样本数，否则恒为 1。
total = tree.n_node_samples[node_id]
if total == 0:
return
predicted_class = 1 if badrate > 0.5 else 0
rule = {
'rule_id': len(rules),
'conditions': conditions.copy(),
'predicted_class': predicted_class,
'class_name': 'bad' if predicted_class == 1 else 'good',
'class_probability': badrate,
'sample_count': int(total),
'tree_id': tree_id
}
rules.append(rule)
else:
# 非叶子节点
feature = self.feature_names_[tree.feature[node_id]]
threshold = tree.threshold[node_id]
# 左子树 (<=)
left_conditions = conditions + [{
'feature': feature,
'threshold': threshold,
'operator': '<='
}]
recurse(tree.children_left[node_id], left_conditions)
# 右子树 (>)
right_conditions = conditions + [{
'feature': feature,
'threshold': threshold,
'operator': '>'
}]
recurse(tree.children_right[node_id], right_conditions)
recurse(0, [])
return rules
def _get_native_model(self):
"""获取底层 sklearn 模型.
:return: 底层 sklearn 模型对象
"""
if hasattr(self.model_, 'get_native_model'):
return self.model_.get_native_model()
return self.model_
def _extract_from_forest(self) -> List[Dict[str, Any]]:
"""从随机森林提取规则.
:return: 规则列表
"""
native_model = self._get_native_model()
tasks = [(self, tree, index) for index, tree in enumerate(native_model.estimators_)]
extracted = self._parallel_execute(
_tree_extract_worker,
tasks,
task_labels=[f"树 {index}" for index in range(len(tasks))],
default_backend="loky",
has_parallel_children=False,
workload=_mining_workload(
self.X_train_,
len(tasks),
operation="随机森林规则提取",
cost_per_item=8.0,
capability="process_safe",
),
)
return [rule for tree_rules in extracted for rule in tree_rules]
def _extract_from_gbdt(self) -> List[Dict[str, Any]]:
"""从GBDT提取规则.
:return: 规则列表
"""
native_model = self._get_native_model()
tasks = [
(self, native_model.estimators_[index, 0], index)
for index in range(native_model.n_estimators_)
]
extracted = self._parallel_execute(
_gbdt_tree_extract_worker,
tasks,
task_labels=[f"树 {index}" for index in range(len(tasks))],
default_backend="loky",
has_parallel_children=False,
workload=_mining_workload(
self.X_train_,
len(tasks),
operation="GBDT规则提取",
cost_per_item=8.0,
capability="process_safe",
),
)
return [rule for tree_rules in extracted for rule in tree_rules]
def _extract_from_isolation_forest(self) -> List[Dict[str, Any]]:
"""从孤立森林提取规则.
:return: 规则列表
"""
native_model = self._get_native_model()
# 计算异常分数
scores = native_model.score_samples(self.X_train_)
threshold = np.percentile(scores, 10) # 取异常分数最低的10%
anomaly_mask = scores < threshold
tasks = [
(self, estimator, tree_idx, scores, anomaly_mask)
for tree_idx, estimator in enumerate(native_model.estimators_)
]
extracted = self._parallel_execute(
_isolation_tree_extract_worker,
tasks,
task_labels=[f"树 {index}" for index in range(len(tasks))],
default_backend="loky",
has_parallel_children=False,
workload=_mining_workload(
self.X_train_,
len(tasks),
operation="孤立森林规则提取",
cost_per_item=8.0,
capability="process_safe",
),
)
return [rule for tree_rules in extracted for rule in tree_rules]
def _extract_from_isolation_tree(self, estimator, tree_idx, scores, anomaly_mask):
"""从一棵孤立树按既有深度和纯度约束提取规则。"""
tree = estimator.tree_
rules = []
def extract_path(node_id, conditions, depth):
if depth > 3: # 限制深度
return
if tree.feature[node_id] == -2: # 叶子
mask = self._apply_conditions(conditions, self.X_train_)
hit_count = mask.sum()
if hit_count >= 5:
anomaly_count = anomaly_mask[mask].sum()
purity = anomaly_count / hit_count
if purity >= 0.3: # 异常纯度要求
rule = {
'rule_id': len(rules),
'conditions': conditions.copy(),
'predicted_class': 1,
'class_name': 'anomaly',
'class_probability': purity,
'sample_count': int(hit_count),
'tree_id': tree_idx,
'anomaly_score': scores[mask].mean()
}
rules.append(rule)
else:
feature = self.feature_names_[tree.feature[node_id]]
thresh = tree.threshold[node_id]
# 左子树
left_cond = conditions + [{
'feature': feature,
'threshold': thresh,
'operator': '<='
}]
extract_path(tree.children_left[node_id], left_cond, depth + 1)
# 右子树
right_cond = conditions + [{
'feature': feature,
'threshold': thresh,
'operator': '>'
}]
extract_path(tree.children_right[node_id], right_cond, depth + 1)
extract_path(0, [], 0)
return rules
def _apply_conditions(
self,
conditions: List[Dict],
X: pd.DataFrame
) -> pd.Series:
"""应用条件到数据.
:param conditions: 条件列表
:param X: 数据
:return: 布尔掩码
"""
mask = pd.Series(True, index=X.index)
for cond in conditions:
feature = cond['feature']
threshold = cond['threshold']
operator = cond['operator']
if operator == '<=':
mask &= X[feature] <= threshold
elif operator == '>':
mask &= X[feature] > threshold
elif operator == '<':
mask &= X[feature] < threshold
elif operator == '>=':
mask &= X[feature] >= threshold
elif operator == '==':
mask &= X[feature] == threshold
return mask
def _calculate_leaf_badrate(self, conditions: List[Dict]) -> float:
"""计算叶子节点的坏账率.
:param conditions: 条件列表
:return: 坏账率
"""
if not hasattr(self, 'y_train_'):
return 0.5
mask = self._apply_conditions(conditions, self.X_train_)
hit_count = mask.sum()
if hit_count == 0:
return 0.0
hit_bad = self.y_train_[mask].sum()
return hit_bad / hit_count
def _filter_rules(self, rules: List[Dict]) -> List[Dict]:
"""根据feature_trends过滤规则.
:param rules: 规则列表
:return: 过滤后的规则列表
"""
if not self.feature_trends:
return rules
filtered = []
for rule in rules:
valid = True
for cond in rule['conditions']:
feature = cond['feature']
operator = cond['operator']
if feature in self.feature_trends:
trend = self.feature_trends[feature]
# 正相关：只保留>方向的规则
if trend == 1 and operator in ['<=', '<']:
valid = False
break
# 负相关：只保留<=方向的规则
if trend == -1 and operator in ['>', '>=']:
valid = False
break
if valid:
filtered.append(rule)
return filtered
def _deduplicate_rules(
self,
rules: List[Dict],
similarity_threshold: float = 0.9
) -> List[Dict]:
"""规则去重.
:param rules: 规则列表
:param similarity_threshold: 相似度阈值
:return: 去重后的规则列表
"""
if not rules:
return []
unique_rules = []
for rule in rules:
# 生成规则签名
conditions = rule['conditions']
signature = '|'.join([
f"{c['feature']}{c['operator']}{c['threshold']:.4f}"
for c in sorted(conditions, key=lambda x: x['feature'])
])
# 检查是否已存在相似规则
is_duplicate = False
for existing in unique_rules:
existing_sig = '|'.join([
f"{c['feature']}{c['operator']}{c['threshold']:.4f}"
for c in sorted(existing['conditions'], key=lambda x: x['feature'])
])
# 计算相似度
if signature == existing_sig:
is_duplicate = True
break
if not is_duplicate:
unique_rules.append(rule)
return unique_rules
def _calculate_rule_importance(self, rule: Dict) -> float:
"""计算规则重要性.
:param rule: 规则字典
:return: 重要性分数
"""
sample_count = rule.get('sample_count', 0)
class_prob = rule.get('class_probability', 0)
# 坏样本权重更高
weight = 2.0 if rule.get('class_name') in ['bad', 'anomaly'] else 1.0
return sample_count * class_prob * weight
[文档]
def get_rules(
self,
top_n: int = 100,
min_samples: int = 10,
min_confidence: float = 0.0,
target: Optional[str] = None,
datasets: Optional[pd.DataFrame] = None
) -> List[Rule]:
"""获取挖掘规则（直接返回Rule对象）.
当传入 datasets 时，规则效果通过 Rule.report 计算并回填到 rule.metadata_。
"""
if not self.rules_:
self.extract_rules()
target_col = target or self.target
feature_map = resolve_feature_map(self.feature_map)
tasks = []
labels = []
for rule_item in self.rules_[:top_n]:
if rule_item['sample_count'] < min_samples:
continue
if rule_item['class_probability'] < min_confidence:
continue
expr = self._rule_to_string(rule_item)
ordinal = len(tasks)
tasks.append((ordinal, rule_item, expr, datasets, target_col, feature_map))
labels.append(f"规则 {rule_item.get('rule_id', ordinal)}")
return self._parallel_execute(
_tree_rule_report_worker,
tasks,
task_labels=labels,
default_backend="threading",
has_parallel_children=False,
workload=_mining_workload(
datasets if datasets is not None else self.X_train_,
len(tasks),
operation="树规则报告",
cost_per_item=10.0,
),
)
[文档]
def get_rule_objects(
self,
top_n: int = 100,
min_samples: int = 10,
min_confidence: float = 0.0,
target: Optional[str] = None,
datasets: Optional[pd.DataFrame] = None
) -> List[Rule]:
"""获取Rule对象列表（与get_rules保持一致）."""
return self.get_rules(
top_n=top_n,
min_samples=min_samples,
min_confidence=min_confidence,
target=target,
datasets=datasets
)
[文档]
def get_rules_dataframe(
self,
top_n: int = 100,
datasets: Optional[pd.DataFrame] = None,
target: Optional[str] = None,
min_samples: int = 10,
min_confidence: float = 0.0
) -> pd.DataFrame:
"""获取规则DataFrame（以rule_expr+Rule.report命中结果为准）."""
rule_objs = self.get_rules(
top_n=top_n,
min_samples=min_samples,
min_confidence=min_confidence,
datasets=datasets,
target=target,
)
if not rule_objs:
return pd.DataFrame()
data = []
for i, rule in enumerate(rule_objs):
md = getattr(rule, 'metadata_', {}) or {}
report_row = {
'规则编号': md.get('rule_id', i),
'规则表达式': rule.expr,
'命中样本数': md.get('命中样本数'),
'命中样本占比': md.get('命中样本占比'),
'命中坏样本率': md.get('命中坏样本率'),
'命中LIFT值': md.get('命中LIFT值'),
'坏账改善': md.get('坏账改善'),
'风险拒绝比': md.get('风险拒绝比'),
}
if '入参字段' in md:
report_row = {
'规则编号': report_row.pop('规则编号'),
'入参字段': md['入参字段'],
'字段含义': md['字段含义'],
**report_row,
}
data.append(report_row)
return pd.DataFrame(data)
def _rule_to_string(self, rule: Dict) -> str:
"""将规则转换为可用于Rule的表达式字符串.
:param rule: 规则字典
:return: pandas eval规则表达式
"""
conditions = rule['conditions']
if not conditions:
return "True"
parts = []
for c in conditions:
feature = c['feature']
feature_expr = f"`{feature}`" if not str(feature).isidentifier() else str(feature)
threshold = c['threshold']
if isinstance(threshold, np.generic):
threshold = threshold.item()
parts.append(f"({feature_expr} {c['operator']} {repr(threshold)})")
return " & ".join(parts)
[文档]
def get_feature_importance(self) -> pd.DataFrame:
"""获取特征重要性.
:return: 特征重要性DataFrame
"""
# 获取底层模型
native_model = self._get_native_model()
if not hasattr(native_model, 'feature_importances_'):
raise ValueError(f"算法 '{self._resolved_algorithm}' 不支持特征重要性")
importance = native_model.feature_importances_
result = pd.DataFrame({
'feature': self.feature_names_,
'importance': importance
}).sort_values('importance', ascending=False)
feature_map = resolve_feature_map(self.feature_map)
if feature_map is not None:
result.insert(1, '入参字段', result['feature'])
result.insert(2, '字段含义', result['feature'].map(feature_map).fillna(''))
return result
def _check_fitted(self):
"""检查是否已拟合."""
if not self.is_fitted_:
raise RuntimeError("请先调用fit()方法")