聚类算法之推导流程

import numpy as np
import matplotlib.pyplot as plt
from sklearn.datasets import make_blobs
from sklearn.cluster import KMeans

# 1. 生成数据
print("=== 1. 生成数据 ===")
X, y_true = make_blobs(
    n_samples=300,
    centers=3,
    cluster_std=1.0,
    random_state=42
)
print(f"数据形状: {X.shape}")

# 2. 手动实现KMeans流程
print("\n=== 2. 手动实现KMeans流程 ===")

def kmeans_manual(X, k, max_iters=100, random_state=42):
    """手动实现KMeans算法"""
    np.random.seed(random_state)
    
    # 步骤1: 确定K值
    print(f"\n步骤1: 确定K值 = {k}")
    
    # 步骤2: 随机选择初始质心
    n_samples = X.shape[0]
    initial_indices = np.random.choice(n_samples, k, replace=False)
    centroids = X[initial_indices].copy()
    print(f"步骤2: 随机选择初始质心")
    print(f"  初始质心坐标:")
    for i, c in enumerate(centroids):
        print(f"    质心{i}: ({c[0]:.4f}, {c[1]:.4f})")
    
    # 记录每次迭代的结果
    history = [centroids.copy()]
    labels_history = []
    
    for iteration in range(max_iters):
        # 步骤3: 计算每个样本到K个质心的距离
        distances = np.zeros((n_samples, k))
        for i in range(k):
            distances[:, i] = np.sqrt(np.sum((X - centroids[i]) ** 2, axis=1))
        
        # 步骤4: 标记类别（选择最近的质心）
        labels = np.argmin(distances, axis=1)
        labels_history.append(labels.copy())
        
        # 步骤5: 重新计算质心
        new_centroids = np.zeros_like(centroids)
        for i in range(k):
            if np.sum(labels == i) > 0:
                new_centroids[i] = np.mean(X[labels == i], axis=0)
            else:
                new_centroids[i] = centroids[i]
        
        # 步骤7: 收敛判断
        if np.allclose(centroids, new_centroids):
            print(f"\n步骤7: 第{iteration+1}次迭代后收敛")
            print(f"  新的中心点和上一次一样，停止迭代")
            break
        
        centroids = new_centroids
        history.append(centroids.copy())
        
        if iteration < 5 or iteration % 10 == 0:
            print(f"\n迭代{iteration + 1}:")
            for i, c in enumerate(centroids):
                print(f"  质心{i}: ({c[0]:.4f}, {c[1]:.4f})")
    
    return centroids, labels, history, labels_history

# 运行手动KMeans
centroids, labels, history, labels_history = kmeans_manual(X, k=3)

# 3. 可视化迭代过程
print("\n=== 3. 可视化迭代过程 ===")
fig, axes = plt.subplots(2, 3, figsize=(15, 10))

# 显示前5次迭代和最终结果
iterations_to_show = [0, 1, 2, 3, 4, len(history)-1]

for idx, iter_num in enumerate(iterations_to_show):
    ax = axes[idx // 3, idx % 3]
    
    if iter_num < len(labels_history):
        labels = labels_history[iter_num]
    else:
        labels = labels_history[-1]
    
    centroids = history[min(iter_num, len(history)-1)]
    
    ax.scatter(X[:, 0], X[:, 1], c=labels, cmap='viridis', alpha=0.6)
    ax.scatter(centroids[:, 0], centroids[:, 1], c='red', marker='x', s=200, linewidths=3)
    ax.set_title(f'迭代{iter_num + 1}')

plt.tight_layout()
plt.savefig('kmeans_iterations.png', dpi=100)
print("迭代过程图已保存")

# 4. 最终结果
print("\n=== 4. 最终结果 ===")
print(f"最终质心坐标:")
for i, c in enumerate(centroids):
    print(f"  质心{i}: ({c[0]:.4f}, {c[1]:.4f})")

# 5. 与sklearn对比
print("\n=== 5. 与sklearn对比 ===")
kmeans_sklearn = KMeans(n_clusters=3, random_state=42, n_init=10)
labels_sklearn = kmeans_sklearn.fit_predict(X)

print(f"sklearn质心:")
for i, c in enumerate(kmeans_sklearn.cluster_centers_):
    print(f"  质心{i}: ({c[0]:.4f}, {c[1]:.4f})")

# 6. 可视化对比
plt.figure(figsize=(12, 5))

plt.subplot(1, 2, 1)
plt.scatter(X[:, 0], X[:, 1], c=labels, cmap='viridis', alpha=0.6)
plt.scatter(centroids[:, 0], centroids[:, 1], c='red', marker='x', s=200, linewidths=3)
plt.title('手动实现KMeans')

plt.subplot(1, 2, 2)
plt.scatter(X[:, 0], X[:, 1], c=labels_sklearn, cmap='viridis', alpha=0.6)
plt.scatter(kmeans_sklearn.cluster_centers_[:, 0], 
            kmeans_sklearn.cluster_centers_[:, 1],
            c='red', marker='x', s=200, linewidths=3)
plt.title('sklearn KMeans')

plt.tight_layout()
plt.savefig('kmeans_comparison.png', dpi=100)
print("对比图已保存")

# 7. KMeans流程总结
print("\n=== 7. KMeans流程总结 ===")
print("""
KMeans算法流程:
1. 确定K值（簇的数量）
2. 随机选择K个样本点作为初始质心
3. 计算每个样本到K个质心的距离
4. 选择最近的质心作为标记类别
5. 根据每个类别的样本点重新计算新的质心
6. 重复步骤3-5
7. 如果新的中心点和上次的中心点一样，停止
""")

# 8. 逐步演示
print("\n=== 8. 逐步演示 ===")

def kmeans_step_by_step(X, k=3, random_state=42):
    """逐步演示KMeans"""
    np.random.seed(random_state)
    n_samples = X.shape[0]
    
    print("步骤1: 确定K值")
    print(f"  K = {k}")
    
    print("\n步骤2: 随机选择初始质心")
    initial_indices = np.random.choice(n_samples, k, replace=False)
    centroids = X[initial_indices].copy()
    for i, c in enumerate(centroids):
        print(f"  初始质心{i}: ({c[0]:.4f}, {c[1]:.4f})")
    
    for iteration in range(3):  # 只演示前3次
        print(f"\n--- 迭代{iteration + 1} ---")
        
        print("步骤3: 计算每个样本到K个质心的距离")
        distances = np.zeros((n_samples, k))
        for i in range(k):
            distances[:, i] = np.sqrt(np.sum((X - centroids[i]) ** 2, axis=1))
        print(f"  距离矩阵形状: {distances.shape}")
        
        print("步骤4: 标记类别（选择最近的质心）")
        labels = np.argmin(distances, axis=1)
        for i in range(k):
            count = np.sum(labels == i)
            print(f"  簇{i}: {count}个样本")
        
        print("步骤5: 重新计算质心")
        new_centroids = np.zeros_like(centroids)
        for i in range(k):
            new_centroids[i] = np.mean(X[labels == i], axis=0)
            print(f"  新质心{i}: ({new_centroids[i][0]:.4f}, {new_centroids[i][1]:.4f})")
        
        print("步骤7: 收敛判断")
        if np.allclose(centroids, new_centroids):
            print("  收敛！停止迭代")
            break
        else:
            print("  未收敛，继续迭代")
        
        centroids = new_centroids

kmeans_step_by_step(X, k=3)