基于GridSearchCV优化KMeans用于异常值检测的技术咨询

问题核心 #

直接用GridSearchCV优化KMeans的聚类数k时，默认会使用KMeans的无监督评分（基于聚类内平方和inertia），完全忽略你的真实异常标签。这就是为什么最优k会出现不符合预期的结果——它在优化聚类效果，而非异常检测的分类指标。

解决方案一：自定义评分函数 #

sklearn的自定义评分函数必须接收estimator、X、y_true三个参数，你不需要额外传入聚类中心或标签，因为训练后的KMeans模型自带cluster_centers_（聚类中心）和labels_（样本聚类标签）属性，可以直接从estimator中提取。

代码示例

import numpy as np
from sklearn.cluster import KMeans
from sklearn.model_selection import GridSearchCV
from sklearn.metrics import f1_score, recall_score

def kmeans_outlier_f1(estimator, X, y_true):
    # 从训练好的KMeans模型中获取聚类中心和样本标签
    centers = estimator.cluster_centers_
    sample_labels = estimator.labels_
    
    # 计算每个样本到所属聚类中心的欧氏距离
    distances = [np.linalg.norm(X[i] - centers[sample_labels[i]]) for i in range(len(X))]
    
    # 用均值±1.5倍标准差作为异常值阈值
    mean_dist = np.mean(distances)
    std_dist = np.std(distances)
    lower_thresh = mean_dist - 1.5 * std_dist
    upper_thresh = mean_dist + 1.5 * std_dist
    
    # 标记异常值：超出阈值的为1（异常），否则0（正常）
    y_pred = np.where((distances < lower_thresh) | (distances > upper_thresh), 1, 0)
    
    # 返回F1分数（可替换为recall_score等你关注的指标）
    return f1_score(y_true, y_pred)

# 假设X是特征数据，y是真实异常标签（0=正常，1=异常）
param_grid = {'n_clusters': [2, 4, 6, 8]}  # 待测试的k值范围
kmeans = KMeans(random_state=42)

# 传入自定义评分函数，cv根据数据量设置（示例用3折）
grid_search = GridSearchCV(
    estimator=kmeans,
    param_grid=param_grid,
    scoring=kmeans_outlier_f1,
    cv=3
)
grid_search.fit(X, y)

# 查看最优结果
print(f"最优k值: {grid_search.best_params_['n_clusters']}")
print(f"最优F1分数: {grid_search.best_score_:.4f}")

解决方案二：封装自定义异常检测估计器 #

把KMeans的聚类逻辑和异常值标记逻辑封装成符合sklearn接口的估计器，让它的predict方法直接返回异常标签，这样可以直接使用sklearn内置的分类评分函数（如f1_score）。

代码示例

import numpy as np
from sklearn.base import BaseEstimator, OutlierMixin
from sklearn.cluster import KMeans
from sklearn.model_selection import GridSearchCV
from sklearn.metrics import make_scorer, f1_score

class KMeansOutlierDetector(BaseEstimator, OutlierMixin):
    def __init__(self, n_clusters=2, threshold_factor=1.5):
        self.n_clusters = n_clusters
        self.threshold_factor = threshold_factor
        self.kmeans = KMeans(n_clusters=n_clusters)
        self.mean_dist_ = None
        self.std_dist_ = None

    def fit(self, X, y=None):
        # 训练KMeans模型
        self.kmeans.fit(X)
        # 计算训练样本到聚类中心的距离，保存阈值参数
        centers = self.kmeans.cluster_centers_
        sample_labels = self.kmeans.labels_
        distances = [np.linalg.norm(X[i] - centers[sample_labels[i]]) for i in range(len(X))]
        self.mean_dist_ = np.mean(distances)
        self.std_dist_ = np.std(distances)
        return self

    def predict(self, X):
        # 预测样本的聚类标签，计算距离并标记异常
        sample_labels = self.kmeans.predict(X)
        centers = self.kmeans.cluster_centers_
        distances = [np.linalg.norm(X[i] - centers[sample_labels[i]]) for i in range(len(X))]
        lower_thresh = self.mean_dist_ - self.threshold_factor * self.std_dist_
        upper_thresh = self.mean_dist_ + self.threshold_factor * self.std_dist_
        # 返回0=正常，1=异常（适配真实标签格式）
        return np.where((distances < lower_thresh) | (distances > upper_thresh), 1, 0)

# 定义评分器（直接用f1_score即可）
f1_scorer = make_scorer(f1_score)

# 设置参数网格和网格搜索
param_grid = {'n_clusters': [2, 4, 6, 8]}
detector = KMeansOutlierDetector()
grid_search = GridSearchCV(
    estimator=detector,
    param_grid=param_grid,
    scoring=f1_scorer,
    cv=3
)
grid_search.fit(X, y)

# 查看最优结果
print(f"最优k值: {grid_search.best_params_['n_clusters']}")
print(f"最优F1分数: {grid_search.best_score_:.4f}")

关键说明 #

• 两种方案都核心是让GridSearchCV基于异常检测的分类指标（如F1、召回率）来选择最优k，而非KMeans的无监督聚类指标。
• 若你更关注召回率，只需把评分函数中的f1_score替换为recall_score即可。
• 由于你提到数据量小，cv参数可以设置为较小的值（如2或3），避免过拟合。

内容的提问来源于stack exchange，提问作者Blackandwhite23