"""基于sklearn的风控模型.
提供 RandomForest、ExtraTrees、GradientBoosting、SVM 和 DecisionTreeClassifier 等模型的统一封装。
**参考样例**
>>> from hscredit.core.models import RandomForest, GradientBoosting
>>> model = RandomForest(n_estimators=100, max_depth=10) # 随机森林模型
>>> model.fit(X_train, y_train) # 训练模型
>>> proba = model.predict_proba(X_test) # 预测概率
"""
import inspect
from typing import Any, Dict, List, Optional, Tuple, Union
import numpy as np
import pandas as pd
from sklearn.ensemble import (
RandomForestClassifier,
ExtraTreesClassifier,
GradientBoostingClassifier,
)
from sklearn.svm import SVC
from sklearn.tree import DecisionTreeClassifier as SklearnDecisionTreeClassifier
from ....exceptions import ValidationError
from ..base import BaseRiskModel
class SklearnRiskModel(BaseRiskModel):
"""基于 sklearn 集成分类器的风控模型基类。
将任意 sklearn 分类器（通过 ``estimator_class`` 传入）封装为 hscredit 统一接口，
继承 :class:`~hscredit.core.models.base.BaseRiskModel`，提供 ``fit`` /
``predict`` / ``predict_proba`` / ``get_feature_importances`` / ``evaluate`` /
``save_model`` / ``load_model`` 及 scorecardpipeline 风格（``fit(df)`` 自动提取
``target`` 列）。具体模型由子类 :class:`RandomForest` /
:class:`ExtraTrees` / :class:`GradientBoosting` 指定。
**参数**
:param estimator_class: sklearn 分类器类（如 ``RandomForestClassifier``），由子类传入
:param objective: 任务类型，默认 ``'binary'``（二分类）
:param eval_metric: 评估指标名或列表，默认 ``None``
:param validation_fraction: 验证集占比，默认 ``0.2``
:param random_state: 随机种子，默认 ``None``
:param n_jobs: 并行任务数，默认 ``-1``（用满 CPU；``GradientBoosting`` 不支持，自动忽略）
:param verbose: 是否输出训练日志，默认 ``False``
:param scorecard_params: 概率评分卡部分覆盖参数，默认 PDO=50、基准分=600、范围0-1000
:param kwargs: 透传给底层 sklearn 分类器的其他超参数
**属性**
- ``feature_importances_``: 特征重要性数组（兼容 sklearn）
- ``feature_names_in_`` / ``n_features_in_``: 输入特征名/数量
- ``classes_``: 类别标签
**参考样例**
>>> from hscredit.core.models import RandomForest
>>> model = RandomForest(n_estimators=200, max_depth=8)
>>> model.fit(X_train, y_train)
>>> proba = model.predict_proba(X_test)[:, 1]
>>> model.get_feature_importances().head()
**引用**
底层实现见 sklearn ``ensemble`` 模块：
https://scikit-learn.org/stable/modules/ensemble.html
"""
def __init__(
self,
estimator_class,
objective: str = "binary",
eval_metric: Union[str, List[str], None] = None,
validation_fraction: float = 0.2,
random_state: Optional[int] = None,
n_jobs: int = -1,
verbose: bool = False,
scorecard_params: Optional[Dict[str, Any]] = None,
**kwargs,
):
super().__init__(
objective=objective,
eval_metric=eval_metric,
early_stopping_rounds=None, # sklearn不支持早停
validation_fraction=validation_fraction,
random_state=random_state,
n_jobs=n_jobs,
verbose=verbose,
scorecard_params=scorecard_params,
**kwargs,
)
self._estimator_class = estimator_class
def fit(
self,
X: Union[np.ndarray, pd.DataFrame],
y: Optional[Union[np.ndarray, pd.Series]] = None,
sample_weight: Optional[np.ndarray] = None,
eval_set: Optional[List[Tuple]] = None,
**fit_params,
) -> "SklearnRiskModel":
"""训练模型.
支持两种调用方式:
1. 常规方式: fit(X, y)
2. scorecardpipeline风格: fit(X) 在init中指定target
"""
# 准备数据（支持从X中提取target）
X, y, sample_weight = self._prepare_data(X, y, sample_weight, extract_target=True, training=True)
self._validate_probability_scorecard_labels(y)
# 保存特征信息
self.n_features_in_ = X.shape[1]
# _prepare_data 已在内部设置 feature_names_in_（DataFrame 或人工命名）
self.classes_ = np.unique(y)
# 只向底层模型传递其真实支持的统一参数，避免 SVC/决策树收到 n_jobs 等未知参数。
params = self.kwargs.copy()
supported = inspect.signature(self._estimator_class).parameters
public_params = self.get_params(deep=False)
for name in supported:
if name in public_params:
params[name] = public_params[name]
# 创建模型
self._model = self._estimator_class(**params)
# 训练
if sample_weight is not None:
self._model.fit(X, y, sample_weight=sample_weight)
else:
self._model.fit(X, y)
# 底层模型已经完成拟合；先提交状态，确保 eval_set 走统一评估入口时
# 能通过严格的布尔训练状态检查。
self._is_fitted = True
self._fit_probability_scorecard(X, y)
# 保存评估结果
self._evals_result = {}
if eval_set:
for i, (X_val, y_val) in enumerate(eval_set):
scores = self.evaluate(X_val, y_val)
self._evals_result[f"validation_{i}"] = scores
return self
def predict(self, X: Union[np.ndarray, pd.DataFrame]) -> np.ndarray:
"""预测类别标签。
:param X: 特征矩阵，DataFrame 或 ndarray
:return: 预测类别数组（0/1）
:raises NotFittedError: 模型尚未训练时
"""
self._require_fitted()
X = self._prepare_data(X)[0]
return self._model.predict(X)
def predict_proba(self, X: Union[np.ndarray, pd.DataFrame]) -> np.ndarray:
"""预测各类别概率。
:param X: 特征矩阵，DataFrame 或 ndarray
:return: 概率数组，shape ``(n_samples, 2)``，第 1 列为正类（坏样本）概率
:raises NotFittedError: 模型尚未训练时
"""
self._require_fitted()
X = self._prepare_data(X)[0]
return self._model.predict_proba(X)
def get_feature_importances(self, importance_type: str = "gain") -> pd.Series:
"""获取特征重要性（基于底层模型的不纯度下降）。
:param importance_type: 重要性类型，默认 ``'gain'``（sklearn 树模型仅支持基于
不纯度的 ``feature_importances_``，该参数为与 boosting 模型接口对齐而保留）
:return: 以特征名为索引、按重要性降序的 Series
:raises NotFittedError: 模型尚未训练时
"""
self._require_fitted()
importances = self._native_feature_importances()
# 创建Series
importance_series = pd.Series(importances, index=self.feature_names_in_, name="importance").sort_values(
ascending=False
)
self._feature_importances = importance_series
return importance_series
@property
def feature_importances_(self) -> np.ndarray:
"""特征重要性属性 (兼容sklearn风格).
直接在包装类上暴露重要性，兼容sklearn RFE/SFS等组件的 importance_getter。
"""
self._require_fitted()
return self._native_feature_importances()
def _native_feature_importances(self) -> np.ndarray:
"""返回底层模型真实提供的重要性或线性系数绝对值。"""
if hasattr(self._model, "feature_importances_"):
return np.asarray(self._model.feature_importances_, dtype=float)
if hasattr(self._model, "coef_"):
coefficient = self._model.coef_
if hasattr(coefficient, "toarray"):
coefficient = coefficient.toarray()
coefficient = np.abs(np.asarray(coefficient, dtype=float))
return coefficient if coefficient.ndim == 1 else coefficient.mean(axis=0)
raise ValidationError(
"当前非线性模型没有原生逐字段特征重要性，请使用 permutation importance 或模型解释工具"
)
def save_model(self, path: str):
"""保存底层sklearn模型（pickle格式）.
:param path: 保存路径
"""
from ....utils import save_pickle
self._require_fitted()
save_pickle(self._model, path)
self._save_score_transformer_sidecar(path)
def load_model(self, path: str) -> "SklearnRiskModel":
"""加载底层sklearn模型（pickle格式）.
:param path: 模型路径
:return: self
"""
from ....utils import load_pickle
self._model = load_pickle(path)
self._is_fitted = True
self.classes_ = getattr(self._model, "classes_", np.array([0, 1]))
if hasattr(self._model, "n_features_in_"):
self.n_features_in_ = self._model.n_features_in_
if not hasattr(self, "feature_names_in_"):
n_feat = getattr(self, "n_features_in_", 0)
self.feature_names_in_ = [f"feature_{i}" for i in range(n_feat)]
self._load_score_transformer_sidecar(path)
return self
[文档]
class RandomForest(SklearnRiskModel):
"""随机森林风控模型.
基于sklearn的RandomForestClassifier封装。
**参数**
:param n_estimators: 树的数量，默认100
:param max_depth: 树最大深度，默认None
:param min_samples_split: 节点分裂最小样本数，默认2
:param min_samples_leaf: 叶子节点最小样本数，默认1
:param max_features: 最大特征数，默认'sqrt'
:param bootstrap: 是否使用自助采样，默认True
:param class_weight: 类别权重，默认None
:param criterion: 分裂标准，默认'gini'
:param random_state: 随机种子，默认None
:param n_jobs: 并行任务数，默认-1
:param verbose: 是否输出详细信息，默认False
"""
def __init__(
self,
n_estimators: int = 100,
max_depth: Optional[int] = None,
min_samples_split: Union[int, float] = 2,
min_samples_leaf: Union[int, float] = 1,
max_features: Union[str, int, float] = "sqrt",
bootstrap: bool = True,
class_weight: Optional[Union[str, Dict]] = None,
criterion: str = "gini",
random_state: Optional[int] = None,
n_jobs: int = -1,
verbose: bool = False,
scorecard_params: Optional[Dict[str, Any]] = None,
**kwargs,
):
super().__init__(
estimator_class=RandomForestClassifier,
random_state=random_state,
n_jobs=n_jobs,
verbose=verbose,
scorecard_params=scorecard_params,
**kwargs,
)
self.n_estimators = n_estimators
self.max_depth = max_depth
self.min_samples_split = min_samples_split
self.min_samples_leaf = min_samples_leaf
self.max_features = max_features
self.bootstrap = bootstrap
self.class_weight = class_weight
self.criterion = criterion
# 更新kwargs
self.kwargs.update(
{
"n_estimators": n_estimators,
"max_depth": max_depth,
"min_samples_split": min_samples_split,
"min_samples_leaf": min_samples_leaf,
"max_features": max_features,
"bootstrap": bootstrap,
"class_weight": class_weight,
"criterion": criterion,
}
)
[文档]
class GradientBoosting(SklearnRiskModel):
"""梯度提升树风控模型.
基于sklearn的GradientBoostingClassifier封装。
**参数**
:param n_estimators: 树的数量，默认100
:param learning_rate: 学习率，默认0.1
:param max_depth: 树最大深度，默认3
:param min_samples_split: 节点分裂最小样本数，默认2
:param min_samples_leaf: 叶子节点最小样本数，默认1
:param subsample: 样本采样比例，默认1.0
:param max_features: 最大特征数，默认None
:param criterion: 分裂标准，默认'friedman_mse'
:param random_state: 随机种子，默认None
:param verbose: 是否输出详细信息，默认False
"""
def __init__(
self,
n_estimators: int = 100,
learning_rate: float = 0.1,
max_depth: int = 3,
min_samples_split: Union[int, float] = 2,
min_samples_leaf: Union[int, float] = 1,
subsample: float = 1.0,
max_features: Optional[Union[str, int, float]] = None,
criterion: str = "friedman_mse",
validation_fraction: float = 0.1,
n_iter_no_change: Optional[int] = None,
random_state: Optional[int] = None,
verbose: bool = False,
scorecard_params: Optional[Dict[str, Any]] = None,
**kwargs,
):
super().__init__(
estimator_class=GradientBoostingClassifier,
random_state=random_state,
n_jobs=1, # GBT不支持n_jobs
verbose=verbose,
scorecard_params=scorecard_params,
**kwargs,
)
self.n_estimators = n_estimators
self.learning_rate = learning_rate
self.max_depth = max_depth
self.min_samples_split = min_samples_split
self.min_samples_leaf = min_samples_leaf
self.subsample = subsample
self.max_features = max_features
self.criterion = criterion
self.validation_fraction = validation_fraction
self.n_iter_no_change = n_iter_no_change
# 更新kwargs
self.kwargs.update(
{
"n_estimators": n_estimators,
"learning_rate": learning_rate,
"max_depth": max_depth,
"min_samples_split": min_samples_split,
"min_samples_leaf": min_samples_leaf,
"subsample": subsample,
"max_features": max_features,
"criterion": criterion,
"validation_fraction": validation_fraction,
"n_iter_no_change": n_iter_no_change,
}
)
[文档]
def fit(
self,
X: Union[np.ndarray, pd.DataFrame],
y: Optional[Union[np.ndarray, pd.Series]] = None,
sample_weight: Optional[np.ndarray] = None,
eval_set: Optional[List[Tuple]] = None,
**fit_params,
) -> "GradientBoosting":
"""训练模型.
支持两种调用方式:
1. 常规方式: fit(X, y)
2. scorecardpipeline风格: fit(X) 在init中指定target
"""
result = super().fit(X, y, sample_weight, eval_set, **fit_params)
# 保存训练过程中的损失
if hasattr(self._model, "train_score_"):
self._evals_result["train"] = {"loss": self._model.train_score_}
if hasattr(self._model, "validation_score_") and self._model.validation_score_:
self._evals_result["validation"] = {"loss": self._model.validation_score_}
# 最佳迭代次数
if hasattr(self._model, "n_estimators_"):
self._best_iteration = self._model.n_estimators_
return result
[文档]
class SVM(SklearnRiskModel):
"""基于 sklearn SVC 的概率型支持向量机模型。
**参数**
:param C: 正则强度倒数，默认 ``1.0``
:param kernel: 核函数，默认 ``'rbf'``
:param degree: 多项式核次数，默认 ``3``
:param gamma: 核系数，默认 ``'scale'``
:param probability: 是否启用概率估计，只允许 ``True``
:param class_weight: 类别权重，默认 ``None``
:param random_state: 随机种子，默认 ``None``
:param n_jobs: hscredit 包装层并行预算，不传给 SVC，默认 ``1``
:param verbose: 是否输出训练日志，默认 ``False``
**属性**
- ``classes_``: 训练类别标签
- ``feature_names_in_``: 训练字段名称
- ``tuner``: 最近一次调优使用的 ModelTuner
**参考样例**
>>> from hscredit.core.models import SVM
>>> model = SVM(C=1.0, kernel="rbf", random_state=42)
>>> model.fit(X_train, y_train)
>>> probability = model.predict_proba(X_test)[:, 1]
"""
def __init__(
self,
C: float = 1.0,
kernel: str = "rbf",
degree: int = 3,
gamma: Union[str, float] = "scale",
coef0: float = 0.0,
shrinking: bool = True,
probability: bool = True,
tol: float = 1e-3,
class_weight: Optional[Union[str, Dict]] = None,
max_iter: int = -1,
random_state: Optional[int] = None,
n_jobs: int = 1,
verbose: bool = False,
scorecard_params: Optional[Dict[str, Any]] = None,
**kwargs,
):
if probability is not True:
raise ValidationError("SVM 必须设置 probability=True 以提供统一概率预测")
super().__init__(
estimator_class=SVC,
random_state=random_state,
n_jobs=n_jobs,
verbose=verbose,
scorecard_params=scorecard_params,
**kwargs,
)
self.C = C
self.kernel = kernel
self.degree = degree
self.gamma = gamma
self.coef0 = coef0
self.shrinking = shrinking
self.probability = True
self.tol = tol
self.class_weight = class_weight
self.max_iter = max_iter
self.kwargs.update(
{
"C": C,
"kernel": kernel,
"degree": degree,
"gamma": gamma,
"coef0": coef0,
"shrinking": shrinking,
"probability": True,
"tol": tol,
"class_weight": class_weight,
"max_iter": max_iter,
}
)
[文档]
def set_params(self, **params):
"""设置 sklearn 参数，同时禁止关闭统一概率能力。"""
if "probability" in params and params["probability"] is not True:
raise ValidationError("SVM 必须设置 probability=True 以提供统一概率预测")
return super().set_params(**params)
[文档]
class DecisionTreeClassifier(SklearnRiskModel):
"""基于 sklearn DecisionTreeClassifier 的统一风控模型。
**参数**
:param criterion: 节点划分质量指标，默认 ``'gini'``
:param splitter: 节点划分策略，默认 ``'best'``
:param max_depth: 最大树深，默认 ``None``
:param min_samples_split: 节点分裂最小样本数，默认 ``2``
:param min_samples_leaf: 叶节点最小样本数，默认 ``1``
:param max_features: 每次分裂考虑的最大特征数，默认 ``None``
:param class_weight: 类别权重，默认 ``None``
:param ccp_alpha: 最小代价复杂度剪枝系数，默认 ``0.0``
:param random_state: 随机种子，默认 ``None``
:param n_jobs: hscredit 包装层并行预算，不传给底层决策树，默认 ``1``
**属性**
- ``feature_importances_``: 决策树原生特征重要性
- ``classes_``: 训练类别标签
- ``feature_names_in_``: 训练字段名称
**参考样例**
>>> from hscredit.core.models import DecisionTreeClassifier
>>> model = DecisionTreeClassifier(max_depth=4, min_samples_leaf=20, random_state=42)
>>> model.fit(X_train, y_train)
>>> probability = model.predict_proba(X_test)[:, 1]
"""
def __init__(
self,
criterion: str = "gini",
splitter: str = "best",
max_depth: Optional[int] = None,
min_samples_split: Union[int, float] = 2,
min_samples_leaf: Union[int, float] = 1,
max_features: Optional[Union[str, int, float]] = None,
class_weight: Optional[Union[str, Dict]] = None,
ccp_alpha: float = 0.0,
random_state: Optional[int] = None,
n_jobs: int = 1,
scorecard_params: Optional[Dict[str, Any]] = None,
**kwargs,
):
super().__init__(
estimator_class=SklearnDecisionTreeClassifier,
random_state=random_state,
n_jobs=n_jobs,
verbose=False,
scorecard_params=scorecard_params,
**kwargs,
)
self.criterion = criterion
self.splitter = splitter
self.max_depth = max_depth
self.min_samples_split = min_samples_split
self.min_samples_leaf = min_samples_leaf
self.max_features = max_features
self.class_weight = class_weight
self.ccp_alpha = ccp_alpha
self.kwargs.update(
{
"criterion": criterion,
"splitter": splitter,
"max_depth": max_depth,
"min_samples_split": min_samples_split,
"min_samples_leaf": min_samples_leaf,
"max_features": max_features,
"class_weight": class_weight,
"ccp_alpha": ccp_alpha,
}
)