"""逻辑回归模型模块.
提供扩展的逻辑回归模型，继承自 sklearn.linear_model.LogisticRegression，
增加了统计信息计算功能，包括标准误差、z统计量、p值、置信区间和VIF等。
核心功能:
- 继承 sklearn LogisticRegression 的所有功能
- 自动计算统计信息（标准误差、z值、p值）
- 支持 VIF（方差膨胀因子）计算
- 提供 summary() 方法输出回归结果表
参考实现:
- scorecardpipeline.model.ITLubberLogisticRegression
- skorecard.linear_model.LogisticRegression
示例:
>>> from hscredit.core.models import LogisticRegression
>>> from sklearn.datasets import make_classification
>>> X, y = make_classification(n_samples=1000, n_features=10, random_state=42)
>>> model = LogisticRegression(calculate_stats=True, C=1.0, max_iter=1000)
>>> model.fit(X, y)
>>> summary = model.summary()
>>> print(summary)
"""
from .._lifecycle import record_training
import numpy as np
import pandas as pd
import scipy.special
import scipy.stats
import inspect
from typing import Any, Dict, Union, Optional, List
from sklearn.linear_model import LogisticRegression as SklearnLogisticRegression
from sklearn.utils.validation import check_is_fitted
from ....utils.overdue import validate_overdue_operator
from ....utils.serialization import ArtifactSerializableMixin
from ..scorecard_support import _ProbabilityScoreCardMixin
from .._contracts import (
InferenceExportMixin, align_model_features, extract_target, record_feature_schema, validate_sample_weight,
)
_SKLEARN_LOGISTIC_PARAMS = set(inspect.signature(SklearnLogisticRegression.__init__).parameters)
[文档]
class LogisticRegression(_ProbabilityScoreCardMixin, InferenceExportMixin, ArtifactSerializableMixin, SklearnLogisticRegression):
artifact_kind = "风险模型"
"""扩展逻辑回归模型.
继承 sklearn.linear_model.LogisticRegression，增加统计信息计算功能。
在训练时自动计算以下统计信息（当 calculate_stats=True 时）：
- cov_matrix_: 参数的协方差矩阵
- std_err_intercept_: 截距的标准误差
- std_err_coef_: 系数的标准误差
- z_intercept_: 截距的z统计量
- z_coef_: 系数的z统计量
- p_val_intercept_: 截距的p值
- p_val_coef_: 系数的p值
- vif_: 方差膨胀因子（VIF）
可通过 summary() 方法获取格式化的回归结果表。
**参数**
:param penalty: 正则化类型，可选 'l1', 'l2', 'elasticnet', 'none'，默认 'l2'
- 'newton-cg', 'lbfgs', 'sag': 仅支持 'l2'
- 'liblinear', 'saga': 支持 'l1', 'l2'
- 'saga': 支持 'elasticnet'
:param calculate_stats: 是否在训练时计算统计信息，默认 True
设置为 False 可提高训练速度，但无法使用 summary() 方法
:param statistics_level: none/coef/full，分别不计算统计、系数推断、不省略VIF；
None 时沿用 calculate_stats。coef 的 VIF 列为 NaN
:param max_dense_bytes: 统计设计矩阵与协方差的预计字节上限，默认256MiB；
超限跳过诊断并记录 statistics_status_，不会中止有效模型预测
:param weight_type: frequency 为频数权重，标准误使用加权信息矩阵；cost 为成本权重，
暂不提供显著性推断。正则化模型统计标记为近似，VIF仍为未加权设计矩阵诊断
:param history_policy: summary/diagnostics/full；默认摘要不复制逐行训练参数
:param max_history: 有界训练历史条数，默认20
:param dual: 是否使用对偶形式，默认 False
仅当 solver='liblinear' 时有效
:param tol: 优化算法的收敛容差，默认 1e-4
:param C: 正则化强度的倒数，默认 1.0
必须是正浮点数，值越小正则化越强
:param fit_intercept: 是否拟合截距项，默认 True
:param intercept_scaling: 截距缩放系数，默认 1.0
仅当 solver='liblinear' 且 fit_intercept=True 时有效
:param class_weight: 类别权重，默认 None
可选 'balanced' 或自定义字典
'balanced' 使用 n_samples / (n_classes * np.bincount(y))
:param random_state: 随机数种子，默认 None
仅在 solver 为 'sag', 'saga', 'liblinear' 时有效
:param solver: 优化算法，默认 'lbfgs'
- 'newton-cg': 牛顿共轭梯度法
- 'lbfgs': 拟牛顿法
- 'liblinear': 坐标下降法
- 'sag': 随机平均梯度下降
- 'saga': SAGA 随机优化算法
:param max_iter: 最大迭代次数，默认 100
:param multi_class: 多分类策略，默认 'auto'
可选 'auto', 'ovr', 'multinomial'
:param verbose: 日志详细程度，默认 0
:param warm_start: 是否使用上次结果初始化，默认 False
:param n_jobs: 并行计算的CPU核心数，默认 None
-1 表示使用所有可用核心
:param l1_ratio: 弹性网络混合参数，默认 None
0 <= l1_ratio <= 1，仅当 penalty='elasticnet' 时有效
:param positive_woe_coef: WOE 模型系数正向化策略，默认 'auto'
- 'auto': 仅当输入 DataFrame 被标记为 hscredit 的 WOE 编码结果时启用
- True: 始终启用。会将负系数对应列乘以 -1，并把系数改为正值
- False: 禁用，保持 sklearn 原始系数符号
:param target: scorecardpipeline 风格的目标列名，默认 None
:param scorecard_params: 概率评分卡部分覆盖参数，默认 PDO=50、基准分=600、范围0-1000
可用 base_bad_rate 显式指定 (0,1) 内的业务先验；优先于训练样本坏率。
**属性**
- coef_: 模型系数，形状 (n_classes, n_features) 或 (n_features,)
- intercept_: 截距项，形状 (n_classes,) 或 (1,)
- classes_: 类别标签数组
- n_features_in_: 训练时的特征数量
- feature_names_in_: 训练时的特征名称（当输入为DataFrame时）
- cov_matrix_: 参数的协方差矩阵
- std_err_coef_: 系数的标准误差
- std_err_intercept_: 截距的标准误差
- z_coef_: 系数的z统计量
- z_intercept_: 截距的z统计量
- p_val_coef_: 系数的p值
- p_val_intercept_: 截距的p值
- vif_: 方差膨胀因子数组
- woe_coef_signs_: WOE 列方向调整向量，1 表示不变，-1 表示该列已翻转
- raw_coef_: 原始拟合系数（正向化前，仅在启用 positive_woe_coef 后提供）
**参考样例**
基本使用::
>>> from hscredit.core.models import LogisticRegression
>>> import pandas as pd
>>> import numpy as np
>>>
>>> # 创建示例数据
>>> np.random.seed(42)
>>> X = pd.DataFrame({
... 'age': np.random.randint(18, 65, 1000),
... 'income': np.random.randint(3000, 50000, 1000),
... })
>>> y = (X['age'] + X['income'] / 1000 > 50).astype(int)
>>>
>>> # 训练模型
>>> model = LogisticRegression(calculate_stats=True, max_iter=1000)
>>> model.fit(X, y)
>>>
>>> # 查看统计摘要
>>> summary = model.summary()
>>> print(summary[['Coef.', 'Std.Err', 'P>|z|', 'VIF']])
使用样本权重::
>>> sample_weight = np.where(y == 1, 2.0, 1.0) # 增加正样本权重
>>> model.fit(X, y, sample_weight=sample_weight)
筛选显著特征::
>>> # 获取 p < 0.05 的显著特征
>>> sig_features = model.get_significant_features(alpha=0.05)
**注意事项**
- 当 calculate_stats=True 时，会计算标准误差、z值、p值和VIF
- 统计计算基于高斯假设，使用协方差矩阵估计
- VIF 计算需要拟合截距项（fit_intercept=True）
- VIF > 10 通常表示存在严重的多重共线性
"""
def __init__(
self,
penalty: str = "l2",
calculate_stats: bool = True,
dual: bool = False,
tol: float = 1e-4,
C: float = 1.0,
fit_intercept: bool = True,
intercept_scaling: float = 1,
class_weight: Optional[Union[dict, str]] = None,
random_state: Optional[int] = None,
solver: str = "lbfgs",
max_iter: int = 100,
multi_class: str = "auto",
verbose: int = 0,
warm_start: bool = False,
n_jobs: Optional[int] = None,
l1_ratio: Optional[float] = None,
positive_woe_coef: Union[bool, str] = 'auto',
target: Optional[str] = None,
scorecard_params: Optional[Dict[str, Any]] = None,
history_policy: str = "summary",
max_history: int = 20,
statistics_level: Optional[str] = None,
max_dense_bytes: Optional[int] = 268435456,
weight_type: str = "frequency",
):
init_kwargs = {
"penalty": penalty,
"calculate_stats": calculate_stats,
"dual": dual,
"tol": tol,
"C": C,
"fit_intercept": fit_intercept,
"intercept_scaling": intercept_scaling,
"class_weight": class_weight,
"random_state": random_state,
"solver": solver,
"max_iter": max_iter,
"multi_class": multi_class,
"verbose": verbose,
"warm_start": warm_start,
"n_jobs": n_jobs,
"l1_ratio": l1_ratio,
}
init_kwargs = {k: v for k, v in init_kwargs.items() if k in _SKLEARN_LOGISTIC_PARAMS}
super().__init__(**init_kwargs)
self.calculate_stats = calculate_stats
self.multi_class = multi_class
self.positive_woe_coef = positive_woe_coef
self.target = target
self.history_policy = history_policy
self.max_history = max_history
self.statistics_level = statistics_level
self.max_dense_bytes = max_dense_bytes
self.weight_type = weight_type
self._initialize_scorecard_params(scorecard_params)
self.tuner = None
def __sklearn_is_fitted__(self) -> bool:
"""按真实训练产物判断拟合状态，避免构造期配置属性造成误判。"""
return self.__dict__.get("_is_fitted", True) and all(
hasattr(self, attr) for attr in ("coef_", "intercept_", "classes_")
)
[文档]
@record_training
def fit(
self,
X: Union[pd.DataFrame, np.ndarray],
y: Optional[Union[pd.Series, np.ndarray]] = None,
sample_weight: Optional[np.ndarray] = None,
**kwargs
) -> "LogisticRegression":
"""训练逻辑回归模型.
支持两种调用方式:
1. 常规方式: fit(X, y)
2. scorecardpipeline风格: 在__init__中指定target，然后fit(X)
在 sklearn LogisticRegression.fit() 的基础上，
当 calculate_stats=True 时额外计算统计信息。
**参数**
:param X: 训练数据，形状 (n_samples, n_features)
支持 numpy array 或 pandas DataFrame
如果是DataFrame且y为None，会尝试从X中提取target列
:param y: 目标变量，形状 (n_samples,)，可选
二分类时为 0/1 或 -1/1
如果为None且init中指定了target，则从X中提取
:param sample_weight: 样本权重，形状 (n_samples,)
默认 None，所有样本权重为1
:param kwargs: 其他传递给父类 fit 方法的参数
**返回**
:return: self，训练好的模型实例
**异常**
:raises ValueError: 输入数据格式不正确
:raises AssertionError: calculate_stats=False 时无法计算统计信息
**参考样例**
基本拟合::
>>> model = LogisticRegression(calculate_stats=True)
>>> model.fit(X_train, y_train)
使用样本权重::
>>> sample_weight = np.where(y_train == 1, 2.0, 1.0)
>>> model.fit(X_train, y_train, sample_weight=sample_weight)
scorecardpipeline风格::
>>> model = LogisticRegression(target='label')
>>> model.fit(X_train) # 从X_train中提取'label'列作为y
"""
X, y = extract_target(X, y, self.target)
record_feature_schema(self, X)
sample_weight = validate_sample_weight(sample_weight, len(y))
self._validate_probability_scorecard_labels(y)
level = self.statistics_level if self.statistics_level is not None else ("full" if self.calculate_stats else "none")
if level not in ("none", "coef", "full"):
raise ValueError("statistics_level 必须为 none、coef、full 或 None")
if self.weight_type not in ("frequency", "cost"):
raise ValueError("weight_type 必须为 frequency 或 cost")
if self.max_dense_bytes is not None and (isinstance(self.max_dense_bytes, (bool, np.bool_)) or not isinstance(self.max_dense_bytes, (int, np.integer)) or self.max_dense_bytes < 1):
raise ValueError("max_dense_bytes 必须为正整数或 None")
self.statistics_status_ = {"状态": "未执行", "级别": level, "权重口径": self.weight_type}
self._statistics_requires_weights_ = sample_weight is not None
if self.warm_start and hasattr(self, "raw_coef_"):
self.coef_ = self.raw_coef_.copy()
for name in (
"woe_coef_signs_",
"raw_coef_",
"cov_matrix_",
"std_err_coef_",
"std_err_intercept_",
"p_val_coef_",
"p_val_intercept_",
"z_coef_",
"z_intercept_",
"vif_",
"names_",
):
self.__dict__.pop(name, None)
# 保存特征名
if isinstance(X, pd.DataFrame):
self.feature_names_in_ = X.columns.tolist()
else:
self.feature_names_in_ = None
# 如果不计算统计信息，直接调用父类方法
apply_positive_woe_coef = self._should_apply_positive_woe_coef(X)
if level == "none":
fitted_model = super().fit(X, y, sample_weight=sample_weight, **kwargs)
self._is_fitted = True
if apply_positive_woe_coef:
self.ensure_positive_woe_coefficients()
self._fit_probability_scorecard(X, y, sample_weight=sample_weight)
return fitted_model
# 准备特征名列表（用于summary输出）
if self.feature_names_in_ is not None:
self.names_ = (["const"] if self.fit_intercept else []) + self.feature_names_in_
else:
self.names_ = (["const"] if self.fit_intercept else []) + [f"x{i}" for i in range(X.shape[1])]
# 调用父类fit方法
super().fit(X, y, sample_weight=sample_weight, **kwargs)
self._is_fitted = True
if apply_positive_woe_coef:
self.ensure_positive_woe_coefficients()
self._compute_fitted_statistics(X, sample_weight=sample_weight, level=level)
self._fit_probability_scorecard(X, y, sample_weight=sample_weight)
return self
def _compute_fitted_statistics(self, X, sample_weight=None, level="full"):
"""预测拟合成功后单独计算有预算的诊断，诊断失败不破坏模型。"""
import warnings
if self.coef_.shape[0] != 1 or (sample_weight is not None and self.weight_type == "cost") or self.class_weight is not None:
self.statistics_status_.update(状态="不适用", 原因="当前仅支持二分类、无 class_weight 的频数权重统计；成本权重推断未实现")
return
n, p = X.shape
width = p + int(self.fit_intercept)
estimate = n * width * 8 * 2 + width * width * 8 * 3
self.statistics_status_["预计字节"] = estimate
if self.max_dense_bytes is not None and estimate > self.max_dense_bytes:
self.statistics_status_.update(状态="超出预算", 原因=f"统计预计 {estimate} 字节，超过 max_dense_bytes={self.max_dense_bytes}")
warnings.warn(self.statistics_status_["原因"], UserWarning)
return
try:
prepared = self._prepare_input_for_model(X)
probabilities = self._predict_proba_from_prepared_input(prepared)
prepared = np.asarray(self._convert_sparse_matrix(prepared), dtype=float)
design = np.hstack([np.ones((n, 1)), prepared]) if self.fit_intercept else prepared
self._compute_statistics(design, probabilities, sample_weight=sample_weight, compute_vif=level == "full")
penalized = self.penalty not in (None, "none")
self.statistics_status_.update(状态="近似" if penalized else "完成", 原因="正则化估计下的未惩罚信息矩阵近似，不作严格显著性推断" if penalized else "非正则二分类频数权重信息矩阵")
except (ValueError, TypeError, ArithmeticError, MemoryError, np.linalg.LinAlgError) as exc:
for name in ("cov_matrix_", "std_err_coef_", "std_err_intercept_", "p_val_coef_", "p_val_intercept_", "z_coef_", "z_intercept_", "vif_"):
self.__dict__.pop(name, None)
self.statistics_status_.update(状态="失败", 原因=f"{type(exc).__name__}: {exc}")
warnings.warn(f"模型训练成功，但统计信息计算失败: {exc}", UserWarning)
def _should_apply_positive_woe_coef(
self,
X: Union[pd.DataFrame, np.ndarray]
) -> bool:
"""判断是否需要对 WOE 模型做系数正向化."""
if self.positive_woe_coef is True:
return True
if self.positive_woe_coef is False:
return False
if self.positive_woe_coef != 'auto':
raise ValueError("positive_woe_coef 仅支持 True/False/'auto'")
return isinstance(X, pd.DataFrame) and X.attrs.get('hscredit_encoding') == 'woe'
[文档]
def ensure_positive_woe_coefficients(
self,
X: Optional[Union[pd.DataFrame, np.ndarray]] = None,
sample_weight: Optional[np.ndarray] = None,
) -> "LogisticRegression":
"""将 WOE 逻辑回归的负系数归一为正，并保持预测结果不变.
做法是：对负系数对应的输入列乘以 -1，同时把系数改成绝对值。
这样线性预测值保持不变，但模型摘要和评分卡解释更符合 WOE 场景。
如果提供 X 重新计算诊断，仍遵守统计预算；原训练带权时需显式提供对应权重。
"""
check_is_fitted(self)
if getattr(self, 'woe_coef_signs_', None) is None:
coef_vector = np.asarray(self.coef_[0], dtype=float)
self.raw_coef_ = self.coef_.copy()
self.woe_coef_signs_ = np.where(coef_vector < 0, -1.0, 1.0)
if np.any(self.woe_coef_signs_ < 0):
self.coef_ = self.coef_.copy()
self.coef_[0] = np.abs(coef_vector)
# 坐标变换 X'=X*S、beta'=S*beta；无须原始数据也能精确转换
# 已有协方差与 z，不能留下正系数/负 z 的矛盾摘要。
signs = self.woe_coef_signs_
if hasattr(self, "cov_matrix_"):
coordinates = np.r_[1., signs] if self.fit_intercept else signs
self.cov_matrix_ = self.cov_matrix_ * coordinates[:, None] * coordinates[None, :]
if hasattr(self, "z_coef_"):
self.z_coef_ = self.z_coef_ * signs
elif X is None:
return self
level = self.statistics_level if self.statistics_level is not None else ("full" if self.calculate_stats else "none")
if X is not None and level != "none":
if getattr(self, "_statistics_requires_weights_", False) and sample_weight is None:
raise ValueError("原模型带权训练，重新计算统计信息时必须提供对应 sample_weight")
weights = validate_sample_weight(sample_weight, len(X))
self._compute_fitted_statistics(X, sample_weight=weights, level=level)
return self
def _prepare_input_for_model(
self,
X: Union[pd.DataFrame, np.ndarray]
) -> Union[pd.DataFrame, np.ndarray]:
"""按 WOE 方向调整输入，使正向化后的系数仍保持原始预测结果."""
X = align_model_features(self, X)
X_model = X
signs = getattr(self, 'woe_coef_signs_', None)
if signs is None:
return X_model
if isinstance(X_model, pd.DataFrame):
adjusted = X_model.copy()
for feature_index, sign in enumerate(signs):
if sign >= 0 or feature_index >= adjusted.shape[1]:
continue
adjusted.iloc[:, feature_index] = adjusted.iloc[:, feature_index] * sign
return adjusted
if scipy.sparse.issparse(X_model):
return X_model.multiply(signs).tocsr()
adjusted = np.asarray(X_model).copy()
adjusted = adjusted * signs
return adjusted
def _predict_proba_from_prepared_input(
self,
X_model: Union[pd.DataFrame, np.ndarray]
) -> np.ndarray:
"""对已经完成 WOE 方向处理的输入计算概率，避免重复变换."""
decision = super().decision_function(X_model)
if np.ndim(decision) == 1:
positive_proba = scipy.special.expit(decision)
return np.column_stack([1.0 - positive_proba, positive_proba])
return scipy.special.softmax(decision, axis=1)
[文档]
def predict_proba(self, X: Union[pd.DataFrame, np.ndarray]) -> np.ndarray:
"""预测各类别概率（评分卡建模常用，输入通常为 WOE 数据）。
当启用 ``positive_woe_coef`` 时，会按训练阶段确定的 WOE 列方向自动调整输入，
保证概率与评分卡分数方向一致。
:param X: 特征矩阵（一般为 WOE 编码后数据），DataFrame 或 ndarray
:return: 概率数组，shape ``(n_samples, 2)``，第 1 列为正类（坏样本）概率
"""
X_model = self._prepare_input_for_model(X)
return self._predict_proba_from_prepared_input(X_model)
[文档]
def predict_log_proba(self, X: Union[pd.DataFrame, np.ndarray]) -> np.ndarray:
"""预测各类别的对数概率（``ln(predict_proba)``，数值已做下限裁剪）。
:param X: 特征矩阵，DataFrame 或 ndarray
:return: 对数概率数组，shape ``(n_samples, 2)``
"""
probabilities = self.predict_proba(X)
return np.log(np.clip(probabilities, 1e-15, 1.0))
[文档]
def decision_function(self, X: Union[pd.DataFrame, np.ndarray]) -> np.ndarray:
"""计算线性决策值（log-odds），必要时自动应用 WOE 列方向调整。
:param X: 特征矩阵，DataFrame 或 ndarray
:return: 决策函数值数组（``> 0`` 判为正类）
"""
X_model = self._prepare_input_for_model(X)
return super().decision_function(X_model)
[文档]
def predict(self, X: Union[pd.DataFrame, np.ndarray]) -> np.ndarray:
"""预测类别标签（阈值为决策值 0），必要时自动应用 WOE 列方向调整。
:param X: 特征矩阵，DataFrame 或 ndarray
:return: 预测类别数组
"""
X_model = self._prepare_input_for_model(X)
decision = super().decision_function(X_model)
if np.ndim(decision) == 1:
return np.where(decision > 0, self.classes_[1], self.classes_[0])
return self.classes_[np.argmax(decision, axis=1)]
def _compute_statistics(
self,
X_design: np.ndarray,
pred_probs: np.ndarray,
sample_weight: Optional[np.ndarray] = None,
compute_vif: bool = True,
) -> None:
"""计算模型统计信息.
计算协方差矩阵、标准误差、z统计量、p值和VIF。
**参数**
:param X_design: 设计矩阵（已添加截距列）
:param pred_probs: 预测概率，形状 (n_samples, n_classes)
"""
# 计算协方差矩阵: (X' * W * X)^(-1)
p = np.prod(pred_probs, axis=1)
if sample_weight is not None:
p = p * np.asarray(sample_weight, dtype=float)
# 使用伪逆矩阵处理奇异矩阵问题
# 当存在多重共线性时，矩阵可能不可逆，使用 pinv 更稳健
try:
XTWX = (X_design * p[..., np.newaxis]).T @ X_design
# 检查矩阵是否可逆
cond_number = np.linalg.cond(XTWX)
if cond_number > 1e10: # 条件数过大，矩阵接近奇异
import warnings
warnings.warn(
f"协方差矩阵条件数过大 ({cond_number:.2e})，存在严重多重共线性。\n"
"建议使用 VIFSelector 或 CorrSelector 剔除高度相关的特征。\n"
"将使用伪逆矩阵计算统计量。",
UserWarning
)
self.cov_matrix_ = np.linalg.pinv(XTWX)
else:
self.cov_matrix_ = np.linalg.inv(XTWX)
except np.linalg.LinAlgError as e:
import warnings
warnings.warn(
f"协方差矩阵计算失败: {e}\n"
"可能存在多重共线性问题，将使用伪逆矩阵。",
UserWarning
)
# 使用伪逆矩阵作为备选
XTWX = (X_design * p[..., np.newaxis]).T @ X_design
self.cov_matrix_ = np.linalg.pinv(XTWX)
# 计算标准误差
std_err = np.sqrt(np.diag(self.cov_matrix_)).reshape(1, -1)
# 分离截距和系数的标准误差
if self.fit_intercept:
self.std_err_intercept_ = std_err[:, 0]
self.std_err_coef_ = std_err[:, 1:][0]
# 计算z统计量
self.z_intercept_ = self.intercept_ / self.std_err_intercept_
self.z_coef_ = self.coef_ / self.std_err_coef_
# 计算p值（基于高斯分布的双侧检验）
self.p_val_intercept_ = scipy.stats.norm.sf(abs(self.z_intercept_)) * 2
self.p_val_coef_ = scipy.stats.norm.sf(abs(self.z_coef_)) * 2
else:
# 没有截距时设为NaN
self.std_err_intercept_ = np.array([np.nan])
self.std_err_coef_ = std_err[0]
self.z_intercept_ = np.array([np.nan])
self.z_coef_ = self.coef_ / self.std_err_coef_
self.p_val_intercept_ = np.array([np.nan])
self.p_val_coef_ = scipy.stats.norm.sf(abs(self.z_coef_)) * 2
# 计算VIF（方差膨胀因子）
self.vif_ = self._compute_vif(X_design) if compute_vif else np.full(X_design.shape[1], np.nan)
# 检查是否存在多重共线性问题
if hasattr(self, 'vif_') and self.vif_ is not None:
# 检查是否有无限大的 VIF（完全共线性）
inf_vif_count = np.sum(np.isinf(self.vif_))
# 检查是否有高 VIF（> 10）
high_vif_count = np.sum((self.vif_ > 10) & (self.vif_ != np.inf))
if inf_vif_count > 0 or high_vif_count > 0:
import warnings
if inf_vif_count > 0:
warnings.warn(
f"检测到 {inf_vif_count} 个特征存在完全共线性（VIF=inf）。\n"
"这表明某些特征可以由其他特征完全线性表示。\n"
"建议使用 VIFSelector 或 CorrSelector 剔除这些特征。",
UserWarning
)
elif high_vif_count > 0:
warnings.warn(
f"检测到 {high_vif_count} 个特征存在严重多重共线性（VIF > 10）。\n"
"这可能会影响模型系数的稳定性和解释性。\n"
"建议使用 VIFSelector 或 CorrSelector 降低特征相关性。",
UserWarning
)
def _compute_vif(self, X_design: np.ndarray) -> np.ndarray:
"""计算方差膨胀因子 (VIF).
VIF 用于检测多重共线性。VIF > 10 通常表示存在严重的多重共线性。
**参数**
:param X_design: 设计矩阵（已添加截距列）
**返回**
:return: vif，VIF值数组，长度与X_design列数相同
"""
from statsmodels.stats.outliers_influence import variance_inflation_factor
try:
vif = [variance_inflation_factor(X_design, i) for i in range(X_design.shape[1])]
return np.array(vif)
except Exception:
# 如果计算失败，返回NaN
return np.full(X_design.shape[1], np.nan)
@staticmethod
def _convert_sparse_matrix(X: Union[pd.DataFrame, np.ndarray]) -> Union[pd.DataFrame, np.ndarray]:
"""转换稀疏矩阵为密集矩阵.
**参数**
:param X: 输入数据，可能是稀疏矩阵
**返回**
:return: X，密集矩阵或原DataFrame
"""
import scipy.sparse
if scipy.sparse.issparse(X):
return X.toarray()
return X
[文档]
def get_feature_importances(self, importance_type: str = 'coef') -> pd.Series:
"""获取特征重要性.
对于逻辑回归模型，使用系数绝对值作为特征重要性。
:param importance_type: 重要性类型，默认'coef'
- 'coef': 系数绝对值
- 'p_value': 基于p值的重要性 (1 - p_value)
- 'z_score': z统计量绝对值
:return: 特征重要性Series
"""
check_is_fitted(self, 'coef_')
# 获取特征名称
if self.feature_names_in_ is None:
feature_names = [f'feature_{i}' for i in range(self.coef_.shape[1])]
else:
feature_names = self.feature_names_in_
# 根据类型计算重要性
if importance_type == 'coef':
importances = np.abs(self.coef_[0]) # 取绝对值
elif importance_type == 'p_value':
if not hasattr(self, 'p_val_coef_'):
raise ValueError("未计算p值，请在fit时设置calculate_stats=True")
importances = 1 - self.p_val_coef_[0] # p值越小越重要
elif importance_type == 'z_score':
if not hasattr(self, 'z_coef_'):
raise ValueError("未计算z统计量，请在fit时设置calculate_stats=True")
importances = np.abs(self.z_coef_[0])
else:
raise ValueError(f"不支持的重要性类型: {importance_type}")
# 创建Series
importance_series = pd.Series(
importances,
index=feature_names,
name='importance'
).sort_values(ascending=False)
self._feature_importances = importance_series
return importance_series
@property
def feature_importances_(self) -> np.ndarray:
"""特征重要性属性 (兼容sklearn风格).
直接从内部模型获取系数绝对值，避免缓存逻辑在clone后出错。
"""
check_is_fitted(self, 'coef_')
return np.abs(self.coef_[0])
[文档]
def summary(self) -> pd.DataFrame:
"""获取回归结果的统计摘要.
返回包含系数、标准误差、z统计量、p值、置信区间和VIF的DataFrame。
**返回**
:return: summary_df，统计摘要表，包含以下列：
- Coef.: 回归系数
- Std.Err: 标准误差
- z: z统计量（系数/标准误差）
- P>|z|: p值
- [0.025: 95%置信区间下限
- 0.975]: 95%置信区间上限
- VIF: 方差膨胀因子
**异常**
:raises AssertionError: 如果训练时 calculate_stats=False
**参考样例**
基本使用::
>>> model = LogisticRegression(calculate_stats=True)
>>> model.fit(X, y)
>>> summary = model.summary()
>>> print(summary)
筛选显著变量::
>>> # 筛选 p < 0.05 的显著变量
>>> significant = summary[summary['P>|z|'] < 0.05]
>>> print(significant[['Coef.', 'P>|z|', 'VIF']])
"""
check_is_fitted(self)
if not hasattr(self, "std_err_coef_"):
msg = f"统计信息未计算（{getattr(self, 'statistics_status_', {})}）。解决方法:\n"
msg += " 1. 使用 model.fit(X, y) 重新训练（calculate_stats=True）\n"
msg += " 2. 初始化时设置 calculate_stats=True"
raise AssertionError(msg)
# 构建统计表
data = {
"Coef.": np.concatenate([self.intercept_.flatten(), self.coef_.flatten()]) if self.fit_intercept else self.coef_.flatten(),
"Std.Err": np.concatenate([self.std_err_intercept_.flatten(), self.std_err_coef_.flatten()]) if self.fit_intercept else self.std_err_coef_.flatten(),
"z": np.concatenate([self.z_intercept_.flatten(), self.z_coef_.flatten()]) if self.fit_intercept else self.z_coef_.flatten(),
"P>|z|": np.concatenate([self.p_val_intercept_.flatten(), self.p_val_coef_.flatten()]) if self.fit_intercept else self.p_val_coef_.flatten(),
}
summary_df = pd.DataFrame(data, index=self.names_)
# 计算95%置信区间
summary_df["[0.025"] = summary_df["Coef."] - 1.96 * summary_df["Std.Err"]
summary_df["0.975]"] = summary_df["Coef."] + 1.96 * summary_df["Std.Err"]
# 添加VIF
summary_df["VIF"] = self.vif_
summary_df.attrs["统计状态"] = dict(getattr(self, "statistics_status_", {}))
return summary_df
[文档]
def summary_with_desc(self, feature_map: Optional[dict] = None) -> pd.DataFrame:
"""获取带特征描述的统计摘要.
在 summary() 基础上增加特征描述列。
**参数**
:param feature_map: 特征描述字典，格式为 {特征名: 描述}，默认 None
**返回**
:return: summary_df，带描述的统计摘要表，增加 Features 和 Describe 列
**参考样例**
使用特征描述::
>>> feature_map = {
... 'age': '年龄',
... 'income': '收入',
... }
>>> summary = model.summary_with_desc(feature_map)
>>> print(summary[['Features', 'Describe', 'Coef.', 'P>|z|']])
"""
summary_df = self.summary().reset_index().rename(columns={"index": "Features"})
if feature_map is not None and len(feature_map) > 0:
summary_df.insert(
loc=1,
column="Describe",
value=[feature_map.get(c, "") for c in summary_df["Features"]]
)
return summary_df
[文档]
def get_significant_features(
self,
alpha: float = 0.05,
include_intercept: bool = False
) -> pd.DataFrame:
"""获取统计显著的特征.
根据p值筛选显著特征。
**参数**
:param alpha: 显著性水平，默认 0.05
p < alpha 的特征被认为是显著的
:param include_intercept: 是否包含截距项，默认 False
**返回**
:return: significant_df，显著特征的统计信息
**参考样例**
筛选显著特征::
>>> # 获取 p < 0.01 的显著特征
>>> sig_features = model.get_significant_features(alpha=0.01)
>>> print(sig_features)
"""
summary = self.summary()
# 筛选显著特征
mask = summary["P>|z|"] < alpha
# 是否包含截距
if not include_intercept:
mask = mask & (summary.index != "const")
return summary[mask]
[文档]
def check_multicollinearity(self, threshold: float = 10.0) -> pd.DataFrame:
"""检查多重共线性.
基于VIF值检测多重共线性。VIF > threshold 表示存在共线性问题。
**参数**
:param threshold: VIF阈值，默认 10.0
VIF > threshold 被认为存在多重共线性
**返回**
:return: high_vif_df，VIF值超过阈值的特征信息
**参考样例**
检查共线性::
>>> # 检查 VIF > 5 的特征
>>> collinear = model.check_multicollinearity(threshold=5.0)
>>> print(collinear)
"""
summary = self.summary()
# 排除截距
mask = (summary.index != "const") & (summary["VIF"] > threshold)
return summary[mask][["Coef.", "VIF"]].sort_values("VIF", ascending=False)
# ==================== 统一API（与 BaseRiskModel 对齐） ====================
[文档]
def evaluate(
self,
X: Union[pd.DataFrame, np.ndarray],
y: Union[pd.Series, np.ndarray],
sample_weight: Optional[np.ndarray] = None,
metrics: Optional[List[str]] = None,
score_direction: str = 'auto',
) -> dict:
"""评估模型性能（与 BaseRiskModel.evaluate 接口一致）.
:param X: 特征矩阵
:param y: 真实标签
:param sample_weight: 样本权重
:param metrics: 评估指标列表
:param score_direction: AUC/Gini分数方向，同 ``metrics.auc``
:return: 评估结果字典
"""
from ..base import BaseRiskModel, _evaluate_binary_predictions
check_is_fitted(self)
return _evaluate_binary_predictions(
y,
self.predict_proba(X)[:, 1],
self.predict(X),
metrics=BaseRiskModel.DEFAULT_METRICS if metrics is None else metrics,
sample_weight=sample_weight,
score_direction=score_direction,
)
[文档]
def predict_score(self, X: Union[pd.DataFrame, np.ndarray]) -> np.ndarray:
"""使用训练坏好比对应的标准概率评分卡预测风险评分.
:param X: 特征矩阵
:return: 风险评分 (0-1000)
"""
return self._predict_probability_score(X)
[文档]
def report(
self,
datasets=None,
X_train=None,
y_train=None,
X_test=None,
y_test=None,
overdue=None,
dpds=None,
excel_path=None,
verbose=True,
*,
overdue_operator: Optional[str] = None,
**kwargs
):
"""生成风控建模报告（与 :meth:`BaseRiskModel.report` 接口一致）。
在训练/测试集上汇总模型表现（KS、AUC、分数分布、逾期率等），可选导出 Excel。
:param datasets: 数据集字典（如 ``{'train': (X, y), 'test': (X, y)}``），与
``X_train``/``y_train``/``X_test``/``y_test`` 二选一
:param X_train: 训练特征，与 ``datasets`` 二选一
:param y_train: 训练标签
:param X_test: 测试特征，可选
:param y_test: 测试标签，可选
:param overdue: 逾期天数字段名（多标签分析时使用），可选
:param dpds: 逾期定义 DPD（如 ``[7, 3]``），可选
:param excel_path: 报告导出的 Excel 路径，提供则落盘，可选
:param verbose: 是否打印进度，默认 ``True``
:param kwargs: 透传给报告生成器的其他参数
:return: ``ModelReport`` 实例
:raises NotFittedError: 模型尚未训练时
:param overdue_operator: 逾期标签比较符，支持 ``>``、``>=``、``<``、``<=``，默认 ``>``。
满足比较条件记为坏样本(1)，否则为好样本(0)。
"""
if overdue_operator is not None:
validate_overdue_operator(overdue_operator)
check_is_fitted(self)
from ....report import auto_model_report
return auto_model_report(
model=self,
datasets=datasets,
X_train=X_train,
y_train=y_train,
X_test=X_test,
y_test=y_test,
overdue=overdue,
dpds=dpds,
overdue_operator=overdue_operator,
excel_path=excel_path,
verbose=verbose,
**kwargs
)
[文档]
def tune(
self,
X: Union[pd.DataFrame, np.ndarray],
y: Optional[Union[pd.Series, np.ndarray]] = None,
search_space: Optional[dict] = None,
fixed_params: Optional[dict] = None,
metric='ks',
direction='maximize',
n_trials: int = 100,
cv: int = 5,
timeout: Optional[int] = None,
verbose: bool = False,
sample_weight: Optional[np.ndarray] = None,
show_progress_bar: Optional[bool] = None,
**kwargs
) -> 'LogisticRegression':
"""超参数调优并返回最佳模型（与 BaseRiskModel.tune 接口一致）.
集成 ModelTuner（基于 Optuna），默认使用逻辑回归的自适应搜索空间
（C / penalty / class_weight / max_iter / solver）。
:param X: 特征矩阵或包含 target 的 DataFrame
:param y: 目标变量，可选
:param search_space: 参数搜索空间，默认使用自适应空间
:param fixed_params: 固定参数
:param metric: 优化指标，默认 'ks'
:param direction: 优化方向，默认 'maximize'
:param n_trials: 搜索次数，默认 100
:param cv: 交叉验证折数，默认 5
:param timeout: 超时时间(秒)
:param verbose: 是否输出详细信息
:param sample_weight: 样本权重，可选
:param show_progress_bar: 是否显示进度条；默认跟随 verbose，可显式覆盖
:param kwargs: 其他传递给 ModelTuner 的参数（如 sampler/storage）
:return: 使用最佳参数训练好的模型实例
**参考样例**
>>> model = LogisticRegression()
>>> best_model = model.tune(X_train, y_train, n_trials=50)
>>> proba = best_model.predict_proba(X_test)
"""
from ..tuning import ModelTuner
fit_options = {name: kwargs.pop(name) for name in ("groups", "catch", "gc_after_trial", "evaluation_weight") if name in kwargs}
fit_options.update(kwargs.pop("optimize_kwargs", {}))
kwargs.setdefault("random_state", self.random_state)
if show_progress_bar is None:
show_progress_bar = bool(verbose)
model_params = self.get_params(deep=False)
model_params["scorecard_params"] = self.scorecard_params
tuner = ModelTuner(
model_class=self.__class__,
search_space=search_space,
fixed_params=fixed_params,
model_params=model_params,
metric=metric,
direction=direction,
target=self.target or 'target',
cv=cv,
verbose=verbose,
**kwargs
)
self.tuner = tuner
tuner.fit(
X,
y,
n_trials=n_trials,
timeout=timeout,
sample_weight=sample_weight,
show_progress_bar=show_progress_bar,
**fit_options,
)
best_model = tuner.get_best_model()
best_model.tuner = tuner
return best_model
[文档]
def save(self, path: str, engine: str = 'auto', **kwargs) -> str:
"""保存模型到文件（与 BaseRiskModel.save 接口一致）.
:param path: 保存路径
:param engine: 序列化引擎
:return: 保存路径
"""
check_is_fitted(self)
from .._lifecycle import atomic_save_pickle
path_str = str(path)
eng = engine
if eng == 'auto':
path_lower = path_str.lower()
if path_lower.endswith('.dill') or path_lower.endswith('.dill.gz'):
eng = 'dill'
elif path_lower.endswith('.cloudpickle'):
eng = 'cloudpickle'
else:
eng = "cloudpickle"
atomic_save_pickle(self, path_str, engine=eng, **kwargs)
return path_str
[文档]
@classmethod
def load(cls, path: str, engine: str = 'auto', **kwargs) -> 'LogisticRegression':
"""从文件加载模型（与 BaseRiskModel.load 接口一致）.
:param path: 模型文件路径
:param engine: 序列化引擎
:return: 加载的模型实例
"""
from ....utils import load_pickle
model = load_pickle(path, engine=engine, **kwargs)
if not isinstance(model, LogisticRegression):
raise TypeError(f"加载的对象类型为 {type(model).__name__}，不是 LogisticRegression")
return model
[文档]
def get_model_info(self) -> dict:
"""获取模型摘要信息（与 :meth:`BaseRiskModel.get_model_info` 接口一致）。
:return: 含 ``model_type`` / ``objective`` / ``n_features`` / ``n_classes`` /
``params`` 等键的字典（LR 无 boosting 迭代，故 ``best_iteration``/``best_score`` 为 None）
:raises NotFittedError: 模型尚未训练时
"""
check_is_fitted(self)
info = {
'model_type': self.__class__.__name__,
'objective': 'binary:logistic',
'eval_metric': None,
'n_features': self.n_features_in_ if hasattr(self, 'n_features_in_') else self.coef_.shape[1],
'n_classes': len(self.classes_),
'best_iteration': None,
'best_score': None,
'params': self.get_params(),
}
return info
[文档]
def get_native_model(self):
"""获取底层原生模型（与 :meth:`BaseRiskModel.get_native_model` 接口一致）。
本类直接继承 sklearn ``LogisticRegression`` 并未包装额外底层模型，故返回自身，
便于与 boosting 模型（返回各自原生 booster）的统一调用。
:return: self
:raises NotFittedError: 模型尚未训练时
"""
check_is_fitted(self)
return self
[文档]
def get_native_params(self):
"""返回当前 sklearn 版本支持的原生逻辑回归参数。"""
return {name: value for name, value in self.get_params(deep=False).items() if name in _SKLEARN_LOGISTIC_PARAMS}
def __getstate__(self):
"""支持 pickle 序列化.
确保 dill/joblib 等序列化引擎能正确处理继承自 sklearn 的类。
"""
state = self.__dict__.copy()
return state
def __setstate__(self, state):
"""支持 pickle 反序列化."""
self.__dict__.update(state)
from .._lifecycle import restore_model_defaults
restore_model_defaults(self)
# 确保父类状态正确恢复
if not hasattr(self, 'classes_'):
# 如果模型未拟合，不需要额外处理
pass