KMeans算法实战：用Python从零开始实现一个简单的聚类模型（附完整代码）

ujm567890

726人浏览 · 2026-02-18 06:43:21

ujm567890 · 2026-02-18 06:43:21 发布

KMeans算法实战：用Python从零开始实现一个简单的聚类模型（附完整代码）

当你第一次面对一堆杂乱无章的数据时，是否感到无从下手？想象一下，你手上有1000个客户的消费记录，或者10000张图片的像素数据，如何从中发现隐藏的模式？这就是KMeans算法大显身手的时候。本文将带你从零开始，用Python实现这个经典的聚类算法，不仅理解其工作原理，还能亲手构建一个可运行的模型。无论你是数据分析师、机器学习爱好者，还是正在学习Python的学生，这篇实战指南都将为你打开无监督学习的大门。

1. 准备工作与环境搭建

在开始编码之前，我们需要确保开发环境准备就绪。推荐使用Python 3.7或更高版本，并安装以下必要的库：

pip install numpy matplotlib scikit-learn

这些库将帮助我们完成数据处理、算法实现和结果可视化。特别说明的是，虽然scikit-learn已经提供了成熟的KMeans实现，但为了深入理解算法原理，我们将从零开始编写核心代码。

数据集选择：为了演示KMeans的工作原理，我们将使用两种类型的数据：

人工生成的二维数据点（便于可视化理解）
经典的鸢尾花(Iris)数据集（真实世界数据）

创建一个新的Python文件（如kmeans_from_scratch.py），导入以下基础模块：

import numpy as np
import matplotlib.pyplot as plt
from sklearn.datasets import make_blobs, load_iris

2. KMeans算法核心原理拆解

KMeans算法的魅力在于其简洁而强大的思想：通过迭代寻找数据中的自然分组。让我们深入理解它的工作机制。

2.1 算法流程分解

KMeans的运行过程可以概括为以下四个关键步骤：

初始化中心点：随机选择K个数据点作为初始簇中心
分配数据点：计算每个点到各中心的距离，分配到最近的中心
更新中心点：重新计算每个簇的均值作为新中心
迭代优化：重复2-3步直到中心点不再变化或达到最大迭代次数

这个过程的数学本质是最小化簇内平方误差（SSE）：

SSE = ΣΣ ||x - μ||²

其中x是数据点，μ是所属簇的中心。

2.2 关键参数解析

在实现算法前，我们需要明确几个重要参数：

参数名	说明	典型值
K	要形成的簇数量	根据数据特点选择
max_iter	最大迭代次数	100-300
tol	收敛阈值（中心点移动距离）	1e-4
init	初始化方法	'random'或'k-means++'

注意：K值的选择对结果影响很大。实践中可以使用肘部法则或轮廓系数来确定最佳K值。

3. 从零实现KMeans算法

现在，让我们动手实现算法核心部分。我们将创建一个KMeans类，包含完整的算法逻辑。

3.1 类结构与初始化

class KMeans:
    def __init__(self, n_clusters=3, max_iter=100, tol=1e-4, random_state=None):
        self.n_clusters = n_clusters
        self.max_iter = max_iter
        self.tol = tol
        self.random_state = random_state
        self.centroids = None
        self.labels_ = None
        
    def _initialize_centroids(self, X):
        np.random.seed(self.random_state)
        random_idx = np.random.permutation(X.shape[0])
        centroids = X[random_idx[:self.n_clusters]]
        return centroids

初始化方法设置了基本参数，_initialize_centroids负责随机选择初始中心点。

3.2 核心算法实现

def fit(self, X):
    # 初始化中心点
    self.centroids = self._initialize_centroids(X)
    
    for _ in range(self.max_iter):
        # 分配数据点到最近中心
        distances = np.sqrt(((X - self.centroids[:, np.newaxis])**2).sum(axis=2))
        self.labels_ = np.argmin(distances, axis=0)
        
        # 更新中心点
        new_centroids = np.array([X[self.labels_ == k].mean(axis=0) 
                                 for k in range(self.n_clusters)])
        
        # 检查收敛条件
        if np.allclose(self.centroids, new_centroids, atol=self.tol):
            break
            
        self.centroids = new_centroids
        
    return self

这段代码完整实现了KMeans的核心逻辑。fit方法接受数据矩阵X，执行迭代优化过程。

3.3 辅助功能实现

为了让我们的实现更实用，添加预测和可视化方法：

def predict(self, X):
    distances = np.sqrt(((X - self.centroids[:, np.newaxis])**2).sum(axis=2))
    return np.argmin(distances, axis=0)

def plot_clusters(self, X):
    plt.figure(figsize=(8, 6))
    for k in range(self.n_clusters):
        cluster_data = X[self.labels_ == k]
        plt.scatter(cluster_data[:, 0], cluster_data[:, 1], 
                   label=f'Cluster {k}', alpha=0.7)
    plt.scatter(self.centroids[:, 0], self.centroids[:, 1], 
               marker='x', s=200, c='black', label='Centroids')
    plt.title('KMeans Clustering Results')
    plt.xlabel('Feature 1')
    plt.ylabel('Feature 2')
    plt.legend()
    plt.grid(True)
    plt.show()

4. 实战应用与效果评估

现在，让我们用实现好的算法解决实际问题，并评估其表现。

4.1 人工数据测试

首先生成测试数据：

# 生成样本数据
X, y = make_blobs(n_samples=300, centers=4, 
                 cluster_std=0.6, random_state=42)

# 运行我们的KMeans实现
kmeans = KMeans(n_clusters=4, random_state=42)
kmeans.fit(X)
kmeans.plot_clusters(X)

这段代码会生成4个明显分组的簇，并展示我们的算法如何正确识别它们。

4.2 真实数据测试：鸢尾花数据集

# 加载鸢尾花数据集
iris = load_iris()
X_iris = iris.data[:, :2]  # 只取前两个特征方便可视化
y_iris = iris.target

# 应用我们的KMeans
kmeans_iris = KMeans(n_clusters=3, random_state=42)
kmeans_iris.fit(X_iris)

# 可视化结果
kmeans_iris.plot_clusters(X_iris)

虽然鸢尾花数据有3个类别，但仅使用两个特征时，某些类别可能会有重叠，这正可以展示KMeans的局限性。

4.3 性能评估指标

为了量化算法表现，我们实现几个评估指标：

def calculate_sse(X, labels, centroids):
    sse = 0
    for k in range(len(centroids)):
        cluster_data = X[labels == k]
        sse += np.sum((cluster_data - centroids[k])**2)
    return sse

def calculate_silhouette(X, labels):
    n = len(X)
    silhouette_scores = np.zeros(n)
    
    for i in range(n):
        # 计算a(i): i点到同簇其他点的平均距离
        a = np.mean([np.linalg.norm(X[i] - X[j]) 
                    for j in np.where(labels == labels[i])[0] if j != i])
        
        # 计算b(i): i点到其他各簇的最小平均距离
        b = min([np.mean([np.linalg.norm(X[i] - X[j]) 
                         for j in np.where(labels == k)[0]])
                for k in set(labels) if k != labels[i]])
        
        silhouette_scores[i] = (b - a) / max(a, b)
    
    return np.mean(silhouette_scores)

使用这些指标评估我们的模型：

print(f"SSE: {calculate_sse(X, kmeans.labels_, kmeans.centroids):.2f}")
print(f"Silhouette Score: {calculate_silhouette(X, kmeans.labels_):.2f}")

5. 高级话题与优化技巧

虽然我们的基础实现已经可以工作，但在实际应用中还需要考虑更多因素。

5.1 初始化方法改进

随机初始化可能导致次优结果。实现KMeans++初始化方法：

def _initialize_centroids_plus(self, X):
    np.random.seed(self.random_state)
    centroids = [X[np.random.randint(X.shape[0])]]
    
    for _ in range(1, self.n_clusters):
        distances = np.array([min([np.linalg.norm(x - c)**2 for c in centroids]) 
                            for x in X])
        prob = distances / distances.sum()
        next_centroid = X[np.random.choice(X.shape[0], p=prob)]
        centroids.append(next_centroid)
    
    return np.array(centroids)

这种方法能显著提高收敛速度和结果质量。

5.2 处理空簇问题

在迭代过程中，可能出现某个簇没有数据点的情况。我们需要处理这种边界情况：

def _update_centroids(self, X, labels):
    new_centroids = []
    for k in range(self.n_clusters):
        if np.sum(labels == k) == 0:  # 空簇处理
            new_centroids.append(X[np.random.randint(X.shape[0])])
        else:
            new_centroids.append(X[labels == k].mean(axis=0))
    return np.array(new_centroids)

5.3 并行计算优化

对于大数据集，我们可以利用numpy的广播特性进行向量化计算：

def _compute_distances(self, X):
    # 向量化计算所有点到所有中心的距离
    return np.sqrt(((X[:, np.newaxis, :] - self.centroids)**2).sum(axis=2))

这种方法比循环计算效率高得多，尤其适合大规模数据。

6. 完整代码整合

将所有部分整合成一个完整的、可直接运行的Python脚本：

import numpy as np
import matplotlib.pyplot as plt
from sklearn.datasets import make_blobs, load_iris

class KMeans:
    def __init__(self, n_clusters=3, max_iter=100, tol=1e-4, 
                 init='random', random_state=None):
        self.n_clusters = n_clusters
        self.max_iter = max_iter
        self.tol = tol
        self.init = init
        self.random_state = random_state
        self.centroids = None
        self.labels_ = None
        
    def _initialize_centroids(self, X):
        np.random.seed(self.random_state)
        if self.init == 'random':
            random_idx = np.random.permutation(X.shape[0])
            return X[random_idx[:self.n_clusters]]
        elif self.init == 'k-means++':
            centroids = [X[np.random.randint(X.shape[0])]]
            for _ in range(1, self.n_clusters):
                distances = np.array([min([np.linalg.norm(x - c)**2 for c in centroids]) 
                                    for x in X])
                prob = distances / distances.sum()
                next_centroid = X[np.random.choice(X.shape[0], p=prob)]
                centroids.append(next_centroid)
            return np.array(centroids)
    
    def _compute_distances(self, X):
        return np.sqrt(((X[:, np.newaxis, :] - self.centroids)**2).sum(axis=2))
    
    def _update_centroids(self, X, labels):
        new_centroids = []
        for k in range(self.n_clusters):
            if np.sum(labels == k) == 0:
                new_centroids.append(X[np.random.randint(X.shape[0])])
            else:
                new_centroids.append(X[labels == k].mean(axis=0))
        return np.array(new_centroids)
    
    def fit(self, X):
        self.centroids = self._initialize_centroids(X)
        
        for _ in range(self.max_iter):
            distances = self._compute_distances(X)
            self.labels_ = np.argmin(distances, axis=1)
            
            new_centroids = self._update_centroids(X, self.labels_)
            
            if np.allclose(self.centroids, new_centroids, atol=self.tol):
                break
                
            self.centroids = new_centroids
        
        return self
    
    def predict(self, X):
        distances = self._compute_distances(X)
        return np.argmin(distances, axis=1)
    
    def plot_clusters(self, X):
        plt.figure(figsize=(8, 6))
        for k in range(self.n_clusters):
            cluster_data = X[self.labels_ == k]
            plt.scatter(cluster_data[:, 0], cluster_data[:, 1], 
                       label=f'Cluster {k}', alpha=0.7)
        plt.scatter(self.centroids[:, 0], self.centroids[:, 1], 
                   marker='x', s=200, c='black', label='Centroids')
        plt.title('KMeans Clustering Results')
        plt.xlabel('Feature 1')
        plt.ylabel('Feature 2')
        plt.legend()
        plt.grid(True)
        plt.show()

def calculate_sse(X, labels, centroids):
    sse = 0
    for k in range(len(centroids)):
        cluster_data = X[labels == k]
        sse += np.sum((cluster_data - centroids[k])**2)
    return sse

def calculate_silhouette(X, labels):
    n = len(X)
    silhouette_scores = np.zeros(n)
    
    for i in range(n):
        a = np.mean([np.linalg.norm(X[i] - X[j]) 
                    for j in np.where(labels == labels[i])[0] if j != i])
        
        b = min([np.mean([np.linalg.norm(X[i] - X[j]) 
                         for j in np.where(labels == k)[0]])
                for k in set(labels) if k != labels[i]])
        
        silhouette_scores[i] = (b - a) / max(a, b)
    
    return np.mean(silhouette_scores)

# 示例使用
if __name__ == "__main__":
    # 生成测试数据
    X, y = make_blobs(n_samples=300, centers=4, 
                     cluster_std=0.6, random_state=42)
    
    # 运行KMeans
    kmeans = KMeans(n_clusters=4, init='k-means++', random_state=42)
    kmeans.fit(X)
    kmeans.plot_clusters(X)
    
    # 评估结果
    print(f"SSE: {calculate_sse(X, kmeans.labels_, kmeans.centroids):.2f}")
    print(f"Silhouette Score: {calculate_silhouette(X, kmeans.labels_):.2f}")