"""F检验筛选器.
使用单因素方差分析（ANOVA F-Test）评估特征与目标变量的线性相关性，
筛选出组间差异显著的特征。适用于分类问题中的特征筛选。
基于 sklearn.feature_selection.f_classif 实现。
**参考样例**
>>> from hscredit.core.selectors import FTestSelector
>>> import pandas as pd
>>> import numpy as np
>>> np.random.seed(42)
>>> X = pd.DataFrame(np.random.randn(1000, 5), columns=[f'f{i}' for i in range(5)]) # 5个特征
>>> y = pd.Series(np.random.randint(0, 2, 1000)) # 目标变量
>>> selector = FTestSelector(k=3) # 选择F检验得分最高的前3个特征
>>> selector.fit(X, y)
>>> print(selector.selected_features_)
"""
from typing import Union, List, Optional, Dict, Any
import numpy as np
import pandas as pd
from sklearn.feature_selection import f_classif, SelectKBest, SelectPercentile
from .base import BaseFeatureSelector
def _compute_f_test_feature(task):
"""计算单个特征的 ANOVA F 得分和 p 值。"""
feature, values, y = task
scores, p_values = f_classif(values.reshape(-1, 1), y)
return feature, scores[0], p_values[0]
[文档]
class FTestSelector(BaseFeatureSelector):
"""F检验筛选器.
使用F检验（ANOVA）评估特征与目标变量的相关性。
适用于分类问题。
F值解释:
- 值越大: 特征与目标变量越相关
**参数**
:param threshold: F值阈值，默认为0.0
:param k: 保留的特征数，默认为'all'
:param percentile: 保留的特征百分比，默认为None
:param target: 目标变量列名，默认为'target'
**参考样例**
::
>>> from hscredit.core.selectors import FTestSelector
>>> import pandas as pd
>>> import numpy as np
>>> np.random.seed(42)
>>> X = pd.DataFrame(np.random.randn(1000, 5), columns=[f'f{i}' for i in range(5)])
>>> y = pd.Series(np.random.randint(0, 2, 1000))
>>> selector = FTestSelector(k=3)
>>> selector.fit(X, y)
>>> print(selector.selected_features_)
**注意**
F 检验只能捕捉特征与目标的**线性**相关，非线性关系可能漏检（此时改用
:class:`MutualInfoSelector`）。``k``/``percentile``/``threshold`` 可组合限制选中数量。
**引用**
基于 sklearn ``f_classif``（ANOVA F 值）：
https://scikit-learn.org/stable/modules/generated/sklearn.feature_selection.f_classif.html
"""
method_name = "F检验筛选"
def __init__(
self,
threshold: float = 0.0,
k: Union[int, str] = "all",
percentile: Optional[int] = None,
target: str = "target",
include: Optional[List[str]] = None,
exclude: Optional[List[str]] = None,
force_drop: Optional[List[str]] = None,
n_jobs: Optional[Union[int, float]] = -1,
binner: Optional[Any] = None,
binning_params: Optional[Dict[str, Any]] = None,
parallel_backend: Optional[str] = None,
parallel_config: Optional[Dict[str, Any]] = None,
):
super().__init__(
target=target,
threshold=threshold,
include=include,
exclude=exclude,
force_drop=force_drop,
n_jobs=n_jobs,
binner=binner,
binning_params=binning_params,
parallel_backend=parallel_backend,
parallel_config=parallel_config,
)
self.k = k
self.percentile = percentile
def _fit_impl(
self,
X: pd.DataFrame,
y: Optional[Union[pd.Series, np.ndarray]],
) -> None:
"""拟合F检验筛选器。
:param X: 输入特征DataFrame
:param y: 目标变量
"""
if y is None:
if self.target not in X.columns:
raise ValueError(f"需要传入y或X中包含{self.target}列")
y = X[self.target].values
X = X.drop(columns=self.target)
self._get_feature_names(X)
if isinstance(self.k, (int, np.integer)) and not isinstance(self.k, (bool, np.bool_)):
if int(self.k) <= 0:
raise ValueError("k 必须大于 0")
elif self.k != "all":
raise ValueError("k 必须是大于 0 的整数或 'all'")
if self.percentile is not None and (
isinstance(self.percentile, (bool, np.bool_))
or not isinstance(self.percentile, (int, np.integer))
or not 0 < int(self.percentile) <= 100
):
raise ValueError("percentile 必须是 (0, 100] 范围内的整数")
# 处理类别变量
X_encoded = X.copy()
for col in X.columns:
if X[col].dtype == "object":
X_encoded[col] = pd.factorize(X[col])[0]
# 缺失值处理：f_classif 不接受 NaN，使用列中位数填充，整列缺失时回退为 0，
# 保持与 chi2/mutual_info 等筛选器对原始信贷数据的鲁棒性一致
if X_encoded.isna().any().any():
X_encoded = X_encoded.fillna(X_encoded.median(numeric_only=True)).fillna(0)
self._validate_parallel_configuration()
# f_classif 原生支持二维矩阵，整表计算可复用中心化/校验开销。
f_scores, _ = f_classif(X_encoded.values, np.asarray(y))
# 处理NaN
f_scores = np.nan_to_num(f_scores, nan=0.0)
self.scores_ = pd.Series(f_scores, index=X.columns)
# 选择特征
selected_mask = f_scores >= self.threshold
if self.percentile is not None:
selector = SelectPercentile(percentile=self.percentile)
selector.fit(X_encoded.values, y)
selected_mask &= selector.get_support()
if isinstance(self.k, (int, np.integer)) and not isinstance(self.k, (bool, np.bool_)):
top_k = min(int(self.k), len(X.columns))
ranking = np.argsort(-f_scores, kind="stable")
top_mask = np.zeros(len(X.columns), dtype=bool)
top_mask[ranking[:top_k]] = True
selected_mask &= top_mask
self.selected_features_ = X.columns[selected_mask].tolist()
self._drop_reason = f"F值 < {self.threshold}"