如何在自定义ODKM类中优雅获取KMeans聚类的各项计算指标和结果

改造思路 #

• 在类初始化阶段新增多个实例属性，存储拟合、预测过程产生的所有核心聚类数据
• 改造fit方法，将聚类标签、中心、距离矩阵、effect值计算结果同步存储
• 改造predict方法，缓存最近一次预测的样本标签和得分结果
• 新增KM_summary方法，支持两种格式的信息输出：全量配置字典、可直接拼接进原数据集的样本级结果表

import math
from math import pow
import numpy as np
import pandas as pd
from sklearn.cluster import KMeans


class ODKM:
    
    def __init__(self,n_clusters=15,effectiveness=500,max_iter=2, random_state=42):
        self.n_clusters=n_clusters
        self.effectiveness=effectiveness
        self.max_iter=max_iter
        self.random_state = random_state # 新增固定随机种子，保证结果可复现
        # 原有存储属性
        self.kmeans = {}
        self.cluster_score = {}
        # 新增聚类信息存储属性
        self.column_labels = {} # 存储训练集每列的聚类标签
        self.column_centers = {} # 存储每列的聚类中心
        self.column_dist = {} # 存储每列的聚类中心距离矩阵
        self.column_effect = {} # 存储每列的effect计算矩阵
        self.last_predict_labels = {} # 存储最近一次预测的样本聚类标签
        self.last_score_array = None # 存储最近一次预测的ODKM得分
        
    def fit(self, data):
        length = len(data)
        for column in data.columns:
            kmeans = KMeans(
                n_clusters=self.n_clusters,
                max_iter=self.max_iter,
                random_state=self.random_state
            )
            self.kmeans[column]=kmeans
            kmeans.fit(data[column].values.reshape(-1,1))
            # 存储聚类标签、中心
            self.column_labels[column] = kmeans.labels_
            self.column_centers[column] = kmeans.cluster_centers_.flatten()
            # 初始化当前列的距离、effect矩阵
            self.column_dist[column] = np.zeros((self.n_clusters, self.n_clusters))
            self.column_effect[column] = np.zeros((self.n_clusters, self.n_clusters))
            
            assign = pd.DataFrame(kmeans.predict(data[column].values.reshape(-1,1)),columns=['cluster'])
            cluster_score=assign.groupby('cluster').apply(len).apply(lambda x:x/length)
            ratio=cluster_score.copy()
        
            sorted_centers = sorted(kmeans.cluster_centers_)
            max_distance = ( sorted_centers[-1] - sorted_centers[0] )[ 0 ]
        
            for i in range(self.n_clusters):
                for k in range(self.n_clusters):
                    if i != k:
                        dist = np.abs(kmeans.cluster_centers_[i] - kmeans.cluster_centers_[k])/max_distance
                        effect = ratio[k]*(1/pow(self.effectiveness,dist))
                        cluster_score[i] = cluster_score[i]+effect
                        # 存储当前dist、effect值
                        self.column_dist[column][i][k] = dist
                        self.column_effect[column][i][k] = effect
                        
            self.cluster_score[column] = cluster_score
                    
    def predict(self, data):
        length = len(data)
        score_array = np.zeros(length)
        # 清空上一次预测的标签缓存
        self.last_predict_labels = {}
        for column in data.columns:
            kmeans = self.kmeans[ column ]
            cluster_score = self.cluster_score[ column ]
            
            assign = kmeans.predict( data[ column ].values.reshape(-1,1) )
            # 存储当前列的预测标签
            self.last_predict_labels[f'{column}_cluster_label'] = assign
            
            for i in range(length):
                score_array[i] = score_array[i] + math.log10( cluster_score[assign[i]] )
        # 存储当前预测得分
        self.last_score_array = score_array
        return score_array
    
    def fit_predict(self,data):
        self.fit(data)
        return self.predict(data)
    
    def KM_summary(self, return_type='dict'):
        """
        导出聚类全量信息
        :param return_type: 可选值
            'dict'：返回全量聚类配置的嵌套字典
            'predict_df'：返回最近一次预测的样本级标签+得分DataFrame，可直接拼接原数据集
        """
        if return_type == 'dict':
            return {
                'cluster_centers': self.column_centers,
                'train_sample_labels': self.column_labels,
                'cluster_distance_matrix': self.column_dist,
                'cluster_effect_matrix': self.column_effect,
                'cluster_score': self.cluster_score
            }
        elif return_type == 'predict_df':
            predict_res = pd.DataFrame(self.last_predict_labels)
            predict_res['ODKM_Score'] = self.last_score_array
            return predict_res

使用示例 #

import pandas as pd

df = pd.DataFrame(data={'attr1':[1,1,1,1,2,2,2,2,2,2,2,2,3,5,5,6,6,7,7,7,7,7,7,7,15],
                         'attr2':[1,1,1,1,2,2,2,2,2,2,2,2,3,5,5,6,6,7,7,7,13,13,13,14,15]})


odkm_model = ODKM(n_clusters=3, max_iter=1)
result = odkm_model.fit_predict(df)

# 直接将聚类标签、得分合并到原DataFrame
df = pd.concat([df, odkm_model.KM_summary(return_type='predict_df')], axis=1)

# 单独获取全量聚类信息做可视化
cluster_info = odkm_model.KM_summary(return_type='dict')
# 示例：获取attr1的聚类中心
attr1_centers = cluster_info['cluster_centers']['attr1']

扩展说明 #

如果需要单独获取某列的KMeans原生对象，直接调用odkm_model.kmeans['attr1']即可，原生的cluster_centers_、labels_等属性都可以正常访问，和无封装的使用习惯完全一致。

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