无监督学习:没有标签也能找结构

无监督学习:没有标签也能找结构

本节摘要:无监督学习(Unsupervised Learning)没有标签、没有老师,算法自己从数据里找结构。真实世界里标签昂贵:医院有数百万病历却没人逐一标疾病类别,电商有数百万用户会话却没人手标客户分群,安全团队有海量日志却没人标出每处异常。无监督学习把相似点分组、挖隐藏结构、捞出异常——监督学习是拿着答案的教材学习,无监督学习是盯着原始数据直到模式自己浮现。本节将从零实现三大聚类算法:K-Means(球形簇的主力)、DBSCAN(密度聚类,能找任意形状并自动检测噪声)、高斯混合模型 GMM(软分配,EM 算法),用肘部法和轮廓系数选 K,把每个算法的边界、优劣、何时该用哪个讲透,并展示如何用聚类做异常检测。

学习目标

阅读完本节,你应当能够:

1. 从零实现 K-Means、DBSCAN、高斯混合模型,对比它们的聚类行为。
2. 用轮廓系数(Silhouette Score)和肘部法(Elbow Method)评估簇质量、选最优 K。
3. 解释 DBSCAN 何时胜过 K-Means,识别哪个算法处理非球形簇和离群点。
4. 用聚类方法构建异常检测流水线,标记偏离正常模式的点。

一、问题与直觉

到目前每节都假设有标注数据:「这是输入,这是正确输出」。真实世界标签昂贵。医院有数百万病历,没人手动标每个的疾病类别。电商有数百万用户会话,没人手标客户分群。安全团队有网络日志,没人标出每个异常。

无监督学习不被告诉要找什么,自己找模式。它把相似点分组、发现隐藏结构、浮现异常。监督学习是拿着答案的教材学习,无监督学习是盯着原始数据直到模式自己浮现。

代价是:没有标签,你无法直接衡量「对」或「错」。你需要不同的工具来评估算法找到的结构是否有意义。

聚类:把相似的东西归一起

聚类把每个点分到一个组(簇),使同组内点彼此相似度高于其他组的点。问题永远是:「相似」是什么意思?

K-Means:主力算法

K-Means 把数据切成恰好 K 个簇。每个簇有个质心(重心),每个点属于最近的质心。

Lloyd 算法:

1. 随机选 K 个点作初始质心
2. 把每个点分到最近的质心
3. 把每个质心重算为其所属点的均值
4. 重复 2~3 直到分配不再变化

目标函数(惯性 inertia)度量每个点到其所属质心的总平方距离。K-Means 最小化它,但只找到局部最小。不同初始化会给出不同结果。

选 K

两种标准方法:

肘部法:对 K = 1, 2, 3, ..., n 跑 K-Means,画 inertia 对 K 的曲线,找「肘部」——再多加簇也不会显著降低 inertia 的拐点。

轮廓系数:对每个点,度量它到自己簇的相似度(a)与到最近其他簇的相似度(b)。轮廓系数 = (b - a) / max(a, b),范围 -1(分错簇)到 +1(分得好)。对所有点平均得到全局分。

DBSCAN:基于密度的聚类

K-Means 假设簇是球形的,且要你预先定 K。DBSCAN 两个假设都不要,它把簇定义为被稀疏区隔开的稠密区。

两个参数:

• eps:邻域半径
• min_samples:形成稠密区所需最少点数

三类点:

• 核心点:eps 距离内至少有 min_samples 个点
• 边界点:在某核心点 eps 内,但自己不是核心点
• 噪声点:既非核心也非边界,就是离群点

DBSCAN 把彼此在 eps 内的核心点连进同一簇。边界点加入附近核心点的簇。噪声点不属于任何簇。

优势:找任意形状的簇、自动确定簇数、识别离群点。劣势:对密度差异大的簇吃力。

层次聚类

构建一棵嵌套簇的树(树状图)。

凝聚法(自底向上):

1. 起初每个点是自己的簇
2. 合并最近的两个簇
3. 重复直到只剩一个簇
4. 在期望层级切开树状图,得到 K 个簇

簇间「近度」可这样度量:

• 单链接:两簇任意两点间的最小距离
• 全链接:两簇任意两点间的最大距离
• 平均链接:所有点对的平均距离
• Ward 法:使簇内总方差增加最小的合并

高斯混合模型(GMM)

K-Means 给硬分配:每个点恰属一个簇。GMM 给软分配:每个点有属于每个簇的概率。

GMM 假设数据由 K 个高斯分布混合而成,每个有自己的均值和协方差。期望最大化(EM)算法在两步间交替:

• E 步:算每个点属于每个高斯的概率
• M 步:更新每个高斯的均值、协方差、混合权重以最大化数据似然

GMM 能建模椭圆簇(不像 K-Means 只能球形),并天然处理重叠簇。

何时用哪个

用聚类做异常检测

聚类天然支持异常检测:

• K-Means:离任何质心都远的点是异常
• DBSCAN:噪声点天然就是异常
• GMM:在所有高斯下概率都低的点是异常

二、从零实现

第 1 步:从零实现 K-Means

import math
import random


def euclidean_distance(a, b):
    return math.sqrt(sum((ai - bi) ** 2 for ai, bi in zip(a, b)))


def kmeans(data, k, max_iterations=100, seed=42):
    random.seed(seed)
    n_features = len(data[0])

    centroids = random.sample(data, k)

    for iteration in range(max_iterations):
        clusters = [[] for _ in range(k)]
        assignments = []

        for point in data:
            distances = [euclidean_distance(point, c) for c in centroids]
            nearest = distances.index(min(distances))
            clusters[nearest].append(point)
            assignments.append(nearest)

        new_centroids = []
        for cluster in clusters:
            if len(cluster) == 0:
                new_centroids.append(random.choice(data))
                continue
            centroid = [
                sum(point[j] for point in cluster) / len(cluster)
                for j in range(n_features)
            ]
            new_centroids.append(centroid)

        if all(
            euclidean_distance(old, new) < 1e-6
            for old, new in zip(centroids, new_centroids)
        ):
            print(f"  Converged at iteration {iteration + 1}")
            break

        centroids = new_centroids

    return assignments, centroids
​

第 2 步:肘部法与轮廓系数

def compute_inertia(data, assignments, centroids):
    total = 0.0
    for point, cluster_id in zip(data, assignments):
        total += euclidean_distance(point, centroids[cluster_id]) ** 2
    return total


def silhouette_score(data, assignments):
    n = len(data)
    if n < 2:
        return 0.0

    clusters = {}
    for i, c in enumerate(assignments):
        clusters.setdefault(c, []).append(i)

    if len(clusters) < 2:
        return 0.0

    scores = []
    for i in range(n):
        own_cluster = assignments[i]
        own_members = [j for j in clusters[own_cluster] if j != i]

        if len(own_members) == 0:
            scores.append(0.0)
            continue

        a = sum(euclidean_distance(data[i], data[j]) for j in own_members) / len(own_members)

        b = float("inf")
        for cluster_id, members in clusters.items():
            if cluster_id == own_cluster:
                continue
            avg_dist = sum(euclidean_distance(data[i], data[j]) for j in members) / len(members)
            b = min(b, avg_dist)

        if max(a, b) == 0:
            scores.append(0.0)
        else:
            scores.append((b - a) / max(a, b))

    return sum(scores) / len(scores)


def find_best_k(data, max_k=10):
    print("Elbow method:")
    inertias = []
    for k in range(1, max_k + 1):
        assignments, centroids = kmeans(data, k)
        inertia = compute_inertia(data, assignments, centroids)
        inertias.append(inertia)
        print(f"  K={k}: inertia={inertia:.2f}")

    print("\nSilhouette scores:")
    for k in range(2, max_k + 1):
        assignments, centroids = kmeans(data, k)
        score = silhouette_score(data, assignments)
        print(f"  K={k}: silhouette={score:.4f}")

    return inertias
​

第 3 步:从零实现 DBSCAN

def dbscan(data, eps, min_samples):
    n = len(data)
    labels = [-1] * n
    cluster_id = 0

    def region_query(point_idx):
        neighbors = []
        for i in range(n):
            if euclidean_distance(data[point_idx], data[i]) <= eps:
                neighbors.append(i)
        return neighbors

    visited = [False] * n

    for i in range(n):
        if visited[i]:
            continue
        visited[i] = True

        neighbors = region_query(i)

        if len(neighbors) < min_samples:
            labels[i] = -1
            continue

        labels[i] = cluster_id
        seed_set = list(neighbors)
        seed_set.remove(i)

        j = 0
        while j < len(seed_set):
            q = seed_set[j]

            if not visited[q]:
                visited[q] = True
                q_neighbors = region_query(q)
                if len(q_neighbors) >= min_samples:
                    for nb in q_neighbors:
                        if nb not in seed_set:
                            seed_set.append(nb)

            if labels[q] == -1:
                labels[q] = cluster_id

            j += 1

        cluster_id += 1

    return labels
​

第 4 步:高斯混合模型(EM 算法)

def gmm(data, k, max_iterations=100, seed=42):
    random.seed(seed)
    n = len(data)
    d = len(data[0])

    indices = random.sample(range(n), k)
    means = [list(data[i]) for i in indices]
    variances = [1.0] * k
    weights = [1.0 / k] * k

    def gaussian_pdf(x, mean, variance):
        d = len(x)
        coeff = 1.0 / ((2 * math.pi * variance) ** (d / 2))
        exponent = -sum((xi - mi) ** 2 for xi, mi in zip(x, mean)) / (2 * variance)
        return coeff * math.exp(max(exponent, -500))

    for iteration in range(max_iterations):
        responsibilities = []
        for i in range(n):
            probs = []
            for j in range(k):
                probs.append(weights[j] * gaussian_pdf(data[i], means[j], variances[j]))
            total = sum(probs)
            if total == 0:
                total = 1e-300
            responsibilities.append([p / total for p in probs])

        old_means = [list(m) for m in means]

        for j in range(k):
            r_sum = sum(responsibilities[i][j] for i in range(n))
            if r_sum < 1e-10:
                continue

            weights[j] = r_sum / n

            for dim in range(d):
                means[j][dim] = sum(
                    responsibilities[i][j] * data[i][dim] for i in range(n)
                ) / r_sum

            variances[j] = sum(
                responsibilities[i][j]
                * sum((data[i][dim] - means[j][dim]) ** 2 for dim in range(d))
                for i in range(n)
            ) / (r_sum * d)
            variances[j] = max(variances[j], 1e-6)

        shift = sum(
            euclidean_distance(old_means[j], means[j]) for j in range(k)
        )
        if shift < 1e-6:
            print(f"  GMM converged at iteration {iteration + 1}")
            break

    assignments = []
    for i in range(n):
        assignments.append(responsibilities[i].index(max(responsibilities[i])))

    return assignments, means, weights, responsibilities
​

第 5 步:生成测试数据并运行

def make_blobs(centers, n_per_cluster=50, spread=0.5, seed=42):
    random.seed(seed)
    data = []
    true_labels = []
    for label, (cx, cy) in enumerate(centers):
        for _ in range(n_per_cluster):
            x = cx + random.gauss(0, spread)
            y = cy + random.gauss(0, spread)
            data.append([x, y])
            true_labels.append(label)
    return data, true_labels


def make_moons(n_samples=200, noise=0.1, seed=42):
    random.seed(seed)
    data = []
    labels = []
    n_half = n_samples // 2
    for i in range(n_half):
        angle = math.pi * i / n_half
        x = math.cos(angle) + random.gauss(0, noise)
        y = math.sin(angle) + random.gauss(0, noise)
        data.append([x, y])
        labels.append(0)
    for i in range(n_half):
        angle = math.pi * i / n_half
        x = 1 - math.cos(angle) + random.gauss(0, noise)
        y = 1 - math.sin(angle) - 0.5 + random.gauss(0, noise)
        data.append([x, y])
        labels.append(1)
    return data, labels


if __name__ == "__main__":
    centers = [[2, 2], [8, 3], [5, 8]]
    data, true_labels = make_blobs(centers, n_per_cluster=50, spread=0.8)

    print("=== K-Means on 3 blobs ===")
    assignments, centroids = kmeans(data, k=3)
    print(f"  Centroids: {[[round(c, 2) for c in cent] for cent in centroids]}")
    sil = silhouette_score(data, assignments)
    print(f"  Silhouette score: {sil:.4f}")

    print("\n=== Elbow Method ===")
    find_best_k(data, max_k=6)

    print("\n=== DBSCAN on 3 blobs ===")
    db_labels = dbscan(data, eps=1.5, min_samples=5)
    n_clusters = len(set(db_labels) - {-1})
    n_noise = db_labels.count(-1)
    print(f"  Found {n_clusters} clusters, {n_noise} noise points")

    print("\n=== GMM on 3 blobs ===")
    gmm_assignments, gmm_means, gmm_weights, _ = gmm(data, k=3)
    print(f"  Means: {[[round(m, 2) for m in mean] for mean in gmm_means]}")
    print(f"  Weights: {[round(w, 3) for w in gmm_weights]}")
    gmm_sil = silhouette_score(data, gmm_assignments)
    print(f"  Silhouette score: {gmm_sil:.4f}")

    print("\n=== DBSCAN on moons (non-spherical clusters) ===")
    moon_data, moon_labels = make_moons(n_samples=200, noise=0.1)
    moon_db = dbscan(moon_data, eps=0.3, min_samples=5)
    n_moon_clusters = len(set(moon_db) - {-1})
    n_moon_noise = moon_db.count(-1)
    print(f"  Found {n_moon_clusters} clusters, {n_moon_noise} noise points")

    print("\n=== K-Means on moons (will fail to separate) ===")
    moon_km, moon_centroids = kmeans(moon_data, k=2)
    moon_sil = silhouette_score(moon_data, moon_km)
    print(f"  Silhouette score: {moon_sil:.4f}")
    print("  K-Means splits moons poorly because they are not spherical")

    print("\n=== Anomaly detection with DBSCAN ===")
    anomaly_data = list(data)
    anomaly_data.append([20.0, 20.0])
    anomaly_data.append([-5.0, -5.0])
    anomaly_data.append([15.0, 0.0])
    anomaly_labels = dbscan(anomaly_data, eps=1.5, min_samples=5)
    anomalies = [
        anomaly_data[i]
        for i in range(len(anomaly_labels))
        if anomaly_labels[i] == -1
    ]
    print(f"  Detected {len(anomalies)} anomalies")
    for a in anomalies[-3:]:
        print(f"    Point {[round(v, 2) for v in a]}")
​

三、框架对比

用 scikit-learn,同样算法是一行代码:

from sklearn.cluster import KMeans, DBSCAN, AgglomerativeClustering
from sklearn.mixture import GaussianMixture
from sklearn.metrics import silhouette_score as sklearn_silhouette

km = KMeans(n_clusters=3, random_state=42).fit(data)
db = DBSCAN(eps=1.5, min_samples=5).fit(data)
agg = AgglomerativeClustering(n_clusters=3).fit(data)
gmm_model = GaussianMixture(n_components=3, random_state=42).fit(data)
​

从零版让你看清库在算什么:K-Means 在分配与重算间迭代,DBSCAN 从稠密种子向外长簇,GMM 在期望与最大化间交替。库版本加了数值稳定、更聪明的初始化(K-Means++)和 GPU 加速,但核心逻辑一样。

四、可复用产物

本节产出 K-Means、DBSCAN、GMM 的可用从零实现,代码可复用为更高级无监督方法的基础。聚类逻辑可直接接到特征工程、异常检测、客户分群等下游任务。

五、练习

1. 实现 K-Means++ 初始化:第一个质心随机选,后续每个质心按到最近已有质心平方距离的比例概率选取,对比与随机初始化的收敛速度。
2. 给代码加层次凝聚聚类,实现 Ward 链接,产出树状图(嵌套合并列表),在不同层级切开并与 K-Means 结果对比。
3. 构建简单异常检测流水线:在同一数据上跑 DBSCAN 和 GMM,标记两者都同意是离群的点(DBSCAN 噪声 + GMM 低概率),测重合并讨论分歧场景。

本节要点回顾

1. 无监督无标签找结构:把相似点分组、挖隐藏模式、捞异常,无法直接评对错,需特殊评估工具。
2. K-Means 是主力:选 K 个质心、分配、重算、循环,最小化惯性只到局部最小,需多次初始化。
3. 选 K 两种方法:肘部法看 inertia 拐点,轮廓系数(b-a)/max(a,b) 越接近 1 越好。
4. DBSCAN 密度聚类:eps 和 min_samples 两参数,核心/边界/噪声三类点,找任意形状、自动定簇数、天然检测离群。
5. 层次聚类建树:自底向上合并,单/全/平均/Ward 链接,切树状图得 K 簇,O(n²) 内存限小数据。
6. GMM 软分配:假设数据由 K 个高斯混合,EM 在 E 步算归属概率、M 步更新参数,能建模椭圆和重叠簇。
7. 何时用哪个:球形已知 K 用 K-Means,任意形状要离群检测用 DBSCAN,要软分配和重叠用 GMM,小数据要树状图用层次。
8. 聚类天然做异常检测:K-Means 看离质心距离、DBSCAN 看噪声、GMM 看低概率。
9. 库版同核加优化:K-Means++ 初始化、数值稳定、GPU 加速,但分配-重算/种子生长/EM 交替的本质不变。

下一节,我们转向特征工程与选择——好特征比花哨算法更重要,把原始数据变成模型能用、能学好的表示。