用Python手把手教你搞定K-Means聚类：从Excel数据读取到三维可视化（附完整代码）

当你面对一堆看似杂乱无章的三维数据时，是否曾想过如何从中发现隐藏的模式？K-Means聚类算法就是解决这类问题的利器。本文将带你从零开始，用Python实现一个完整的K-Means聚类项目，涵盖数据读取、预处理、算法实现到三维可视化的全流程。无论你是数据分析新手还是希望巩固基础的开发者，这篇"保姆级"教程都能让你收获满满。

1. 环境准备与数据读取

在开始之前，确保你的Python环境已安装以下必要库：

pip install numpy pandas matplotlib xlrd scikit-learn

我们将使用 xlrd 库来读取Excel数据。假设你有一个包含三维数据的Excel文件，格式如下：

特征1	特征2	特征3
1.2	3.4	5.6
2.3	4.5	6.7
...	...	...

读取数据的Python代码如下：

import xlrd
import numpy as np

def read_excel_data(file_path):
    workbook = xlrd.open_workbook(file_path)
    sheet = workbook.sheet_by_index(0)
    
    data = []
    for row_idx in range(1, sheet.nrows):  # 跳过标题行
        row_data = [
            sheet.cell_value(row_idx, 0),
            sheet.cell_value(row_idx, 1),
            sheet.cell_value(row_idx, 2)
        ]
        data.append(row_data)
    
    return np.array(data)

# 使用示例
raw_data = read_excel_data('your_data.xlsx')
print(f"数据形状: {raw_data.shape}")

注意：如果数据量较大，建议使用 pandas 的 read_excel 函数，它在处理大型文件时效率更高。

2. 数据预处理：标准化是关键

原始数据往往存在量纲不一致的问题，这对距离敏感的K-Means算法影响很大。我们使用 MinMaxScaler 将数据缩放到[0,1]范围：

from sklearn.preprocessing import MinMaxScaler

def normalize_data(data):
    scaler = MinMaxScaler(feature_range=(0, 1))
    normalized_data = scaler.fit_transform(data)
    return normalized_data, scaler

# 标准化数据
normalized_data, scaler = normalize_data(raw_data)

为什么标准化如此重要？考虑以下两个特征：

• 特征A范围：0-100
• 特征B范围：0-1

如果不标准化，特征A在距离计算中的权重会远大于特征B，导致聚类结果失真。

3. 从零实现K-Means算法

虽然 sklearn 提供了现成的K-Means实现，但自己编写能加深对算法的理解。以下是核心实现：

import numpy as np
import matplotlib.pyplot as plt
from mpl_toolkits.mplot3d import Axes3D

class KMeansCustom:
    def __init__(self, n_clusters=3, max_iter=300, tol=1e-4):
        self.n_clusters = n_clusters
        self.max_iter = max_iter
        self.tol = tol
        self.centroids = None
        self.labels = None
    
    def _initialize_centroids(self, X):
        # 随机选择初始质心
        random_idx = np.random.choice(X.shape[0], self.n_clusters, replace=False)
        self.centroids = X[random_idx]
        return self.centroids
    
    def _compute_distances(self, X):
        distances = np.zeros((X.shape[0], self.n_clusters))
        for i, centroid in enumerate(self.centroids):
            distances[:, i] = np.linalg.norm(X - centroid, axis=1)
        return distances
    
    def _assign_clusters(self, distances):
        self.labels = np.argmin(distances, axis=1)
        return self.labels
    
    def _update_centroids(self, X):
        new_centroids = np.zeros((self.n_clusters, X.shape[1]))
        for i in range(self.n_clusters):
            new_centroids[i] = np.mean(X[self.labels == i], axis=0)
        return new_centroids
    
    def fit(self, X):
        self._initialize_centroids(X)
        
        for _ in range(self.max_iter):
            distances = self._compute_distances(X)
            self._assign_clusters(distances)
            new_centroids = self._update_centroids(X)
            
            # 检查收敛
            centroid_shift = np.linalg.norm(new_centroids - self.centroids)
            if centroid_shift < self.tol:
                break
                
            self.centroids = new_centroids
        
        return self
    
    def predict(self, X):
        distances = self._compute_distances(X)
        return self._assign_clusters(distances)

算法核心步骤：

1. 随机初始化聚类中心
2. 计算每个点到各中心的距离
3. 将点分配到最近的中心
4. 重新计算中心位置
5. 重复2-4步直到收敛

4. 三维可视化：让结果一目了然

可视化是理解聚类结果的关键。我们使用 matplotlib 的3D绘图功能：

def plot_3d_clusters(data, labels, centroids=None):
    fig = plt.figure(figsize=(10, 8))
    ax = fig.add_subplot(111, projection='3d')
    
    # 为每个聚类设置不同颜色
    colors = ['r', 'g', 'b', 'c', 'm', 'y', 'k']
    for i in range(len(np.unique(labels))):
        cluster_data = data[labels == i]
        ax.scatter(
            cluster_data[:, 0], 
            cluster_data[:, 1], 
            cluster_data[:, 2],
            c=colors[i % len(colors)],
            label=f'Cluster {i}',
            alpha=0.6
        )
    
    # 绘制质心
    if centroids is not None:
        ax.scatter(
            centroids[:, 0],
            centroids[:, 1],
            centroids[:, 2],
            marker='*',
            s=300,
            c='k',
            label='Centroids'
        )
    
    ax.set_xlabel('Feature 1')
    ax.set_ylabel('Feature 2')
    ax.set_zlabel('Feature 3')
    ax.legend()
    plt.title('3D Cluster Visualization')
    plt.show()

# 使用自定义K-Means
kmeans = KMeansCustom(n_clusters=3)
kmeans.fit(normalized_data)
labels = kmeans.labels

# 可视化
plot_3d_clusters(normalized_data, labels, kmeans.centroids)

5. 关键参数调优与评估

5.1 如何选择K值？

K值的选择是K-Means的核心问题。常用的方法是肘部法则(Elbow Method)：

def find_optimal_k(data, max_k=10):
    distortions = []
    for k in range(1, max_k + 1):
        kmeans = KMeansCustom(n_clusters=k)
        kmeans.fit(data)
        # 计算每个点到其质心的平均距离
        mean_distortion = 0
        for i in range(k):
            cluster_data = data[kmeans.labels == i]
            if len(cluster_data) > 0:
                mean_distortion += np.mean(np.linalg.norm(
                    cluster_data - kmeans.centroids[i], axis=1
                ))
        distortions.append(mean_distortion / k)
    
    plt.plot(range(1, max_k + 1), distortions, 'bx-')
    plt.xlabel('Number of clusters (k)')
    plt.ylabel('Average Distortion')
    plt.title('The Elbow Method showing the optimal k')
    plt.show()

find_optimal_k(normalized_data)

5.2 评估聚类质量

常用的内部评估指标包括轮廓系数(Silhouette Score)：

from sklearn.metrics import silhouette_score

def evaluate_clusters(data, labels):
    score = silhouette_score(data, labels)
    print(f"轮廓系数: {score:.3f}")
    return score

evaluate_clusters(normalized_data, labels)

轮廓系数范围在[-1,1]之间：

• 接近1表示样本与同类样本距离近，与其他类样本距离远
• 接近0表示样本在两个类的边界上
• 接近-1表示样本可能被分配到了错误的类

6. 完整项目实战：用户行为分析案例

假设我们有一组用户行为数据，包含三个维度：

1. 页面停留时间(秒)
2. 点击次数
3. 购买金额(元)

# 模拟数据生成
np.random.seed(42)
n_samples = 300

# 生成三个簇的数据
cluster1 = np.random.normal(loc=[0.2, 0.2, 0.2], scale=0.05, size=(n_samples, 3))
cluster2 = np.random.normal(loc=[0.5, 0.8, 0.3], scale=0.1, size=(n_samples, 3))
cluster3 = np.random.normal(loc=[0.8, 0.3, 0.7], scale=0.07, size=(n_samples, 3))

user_data = np.vstack([cluster1, cluster2, cluster3])

# 聚类分析
kmeans = KMeansCustom(n_clusters=3)
kmeans.fit(user_data)

# 评估
evaluate_clusters(user_data, kmeans.labels)

# 可视化
plot_3d_clusters(user_data, kmeans.labels, kmeans.centroids)

在实际项目中，你可能需要：

• 尝试不同的初始化方法（如k-means++）
• 处理异常值
• 结合业务知识解释聚类结果

7. 常见问题与解决方案

7.1 空簇问题

有时某个簇可能不包含任何样本，导致计算新质心时出错。解决方法：

def _update_centroids_safe(self, X):
    new_centroids = np.zeros((self.n_clusters, X.shape[1]))
    for i in range(self.n_clusters):
        if np.sum(self.labels == i) > 0:  # 检查簇是否为空
            new_centroids[i] = np.mean(X[self.labels == i], axis=0)
        else:
            # 为空时重新随机初始化
            new_centroids[i] = X[np.random.choice(X.shape[0])]
    return new_centroids

7.2 处理不同形状的簇

K-Means假设簇是凸形的且大小相似。对于非凸形簇，可能需要考虑其他算法如DBSCAN。

7.3 大数据集优化

对于大型数据集，可以考虑：

• 使用Mini-Batch K-Means
• 降维后再聚类
• 分布式实现

from sklearn.cluster import MiniBatchKMeans

mbkmeans = MiniBatchKMeans(n_clusters=3, batch_size=100)
mbkmeans.fit(large_data)

8. 进阶技巧与扩展

8.1 特征工程

好的特征能显著提升聚类效果。可以尝试：

• 添加多项式特征
• 使用PCA降维
• 结合领域知识构造新特征

from sklearn.decomposition import PCA

# 降维可视化
pca = PCA(n_components=2)
reduced_data = pca.fit_transform(normalized_data)

plt.scatter(reduced_data[:, 0], reduced_data[:, 1], c=labels)
plt.title('PCA Reduced Cluster Visualization')
plt.show()

8.2 半监督学习

如果有少量标签数据，可以尝试约束聚类：

from sklearn.semi_supervised import LabelSpreading

# 假设我们有5%的标签数据
n_total = len(normalized_data)
n_labeled = int(0.05 * n_total)
labeled_indices = np.random.choice(n_total, n_labeled, replace=False)
labels_partial = -np.ones(n_total)  # -1表示无标签
labels_partial[labeled_indices] = labels[labeled_indices]

label_prop_model = LabelSpreading(kernel='knn', n_neighbors=10)
label_prop_model.fit(normalized_data, labels_partial)
refined_labels = label_prop_model.predict(normalized_data)

8.3 实时聚类

对于流式数据，可以实现增量式K-Means：

class IncrementalKMeans:
    def __init__(self, n_clusters=3, decay=0.9):
        self.n_clusters = n_clusters
        self.decay = decay  # 控制旧质心的影响衰减速度
        self.centroids = None
        self.counts = np.zeros(n_clusters)  # 每个簇的样本计数
    
    def partial_fit(self, X):
        if self.centroids is None:
            self.centroids = X[:self.n_clusters]
        
        for sample in X:
            distances = np.linalg.norm(sample - self.centroids, axis=1)
            closest = np.argmin(distances)
            
            # 更新计数和质心
            self.counts[closest] += 1
            learning_rate = 1 / self.counts[closest]
            self.centroids[closest] = (1 - learning_rate) * self.centroids[closest] + learning_rate * sample
        
        return self