企业官网建设流程全解析

先说点实在的。做过聚类的人都知道，DBSCAN 用起来最折磨人的不是算法本身，而是那个全局半径参数 eps。真实数据永远不会按你预设的密度均匀分布，城市人口网格、用户行为轨迹、传感器采集的信号片段，全都是疏密不齐的形态。全局一个阈值去切，切密的就会把稀疏簇切成噪声，切疏的又会把密度高的区域糊成一团。今天要写的 Multi-DBSCAN，核心思路是先把数据按照局部密度切分成不同层，再让每一层自己决定聚类半径，最后把结果合并回一个完整标签序列。整个过程基于 Python 实现，不依赖额外的重型框架，可以跑在一维时序数据上，也可以扩展到任意维度的坐标点。

我最早是在做时序异常检测时被逼着研究这个方向的。当时手里的数据是连续七天采集的传感器信号，平稳时段和剧烈波动时段混在一起，传统 DBSCAN 想用一个半径同时处理两端的密度特征，来回调参调了两天也没找到合理的 eps。后来把思路从"找全局最优参数"换成"让每个密度区域自己说话"，问题立刻就顺了。这篇文章会把完整的设计思路、Python 代码、调参经验和踩坑记录都写出来，适合已经被 DBSCAN 的全局参数折磨过、又不想直接上 HDBSCAN 这类重型方案的读者。

1. 当固定半径成为硬伤：为什么需要 Multi-DBSCAN

1.1 全局 eps 的一刀切困境

先把问题摆清楚。DBSCAN 的全部行为都建立在两个数字上：eps 定义邻域半径，min_samples 定义密度阈值。对一个密度均匀的数据集来说，这两个参数确实够用，但现实中的聚类对象几乎没有密度均匀的。比如同一个城市里，市中心 POI 点密度可能是郊区的几十倍，用户 App 使用数据在工作日和深夜的分布密度差异也很悬殊，这些场景下全局 eps 天然就是错的。

如果用较小的 eps 去适配高密度区域，低密度区域因为邻域内点数不足，全部被判成噪声；如果把 eps 调大想照顾低密度簇，高密度区域之间又会被桥接成一个巨型簇，原本应该分开的类别被粘连在一起。这不是参数调得不够细致的问题，是模型表达能力的边界。

可能有读者会想到 KNN 距离图或者 k-dist 曲线来找转折点，这个方法在单一密度下好用，多密度数据上 k-dist 曲线往往呈现多个平台和转折，人工判读变得非常主观。而直接上 HDBSCAN 虽然能通过层次结构隐式处理密度变化，但它的参数语义更抽象（min_cluster_size、min_samples），工程落地时需要更多调参试错，而且对部分场景来说运行开销偏大。

1.2 Multi-DBSCAN 的核心思路：先分而治之，再合并统一

Multi-DBSCAN 的思路特别朴素，既然全局一个半径搞不定，那就先按局部密度把数据分层，每一层内部密度相对均匀，再对每一层单独跑一个自适应的 DBSCAN，最后把各层聚类结果做合并和去重。这种做法本质上是把"参数选择"问题从一次全局决策，转换成多层局部决策，每一层的决策复杂度都远比全局决策低。

这个思路可以类比成城市规划中的分级路网。城市道路不会用同一种标准从市中心修到远郊，而是分成快速路、主干道、次干道、支路几个层级，每一层内部规格统一，层与层之间通过合理的接入点衔接。Multi-DBSCAN 的分桶就是划分路网层级，DBSCAN 聚类就是各层级内部的标准建设，最后的合并则是路口衔接。

相比直接换用其他密度聚类算法，Multi-DBSCAN 最大的优势在于保留了 DBSCAN 的语义：噪声点依然是噪声点，簇依然是密度相连点的极大集合，这让业务侧的解释成本很低。而且实现它不需要引入复杂的依赖库，核心逻辑用 NumPy 加 scikit-learn 就能完成，方便后续深度定制。

2. 算法路径设计：分桶、聚类、合并的完整流水线

2.1 第一层分桶：统计每个样本的局部密度

整个 Multi-DBSCAN 的起点是量化每个样本点的局部密度。局部密度的计算方式有很多种，最简单的做法是用每个点到其第 k 个最近邻的距离来表征。这个距离越小，说明点周围越拥挤，密度越高。需要注意 k 的取值一般和后续的 min_samples 挂钩，通常取 min_samples - 1，也可以单独设置。

在具体实现时，我会先标准化数据，避免不同特征量纲差异污染距离计算。然后调用 sklearn 的 NearestNeighbors 计算每个点的 KNN 距离，得到密度特征向量。如果数据量超过十万级，这一步需要留意计算开销，可以考虑用近似最近邻算法，比如 sklearn 中设置 algorithm='kd_tree' 或 'ball_tree'，高维数据下默认的 'auto' 会选择 brute force，反而更慢。

拿到密度特征后，分桶的方式有很多选择。可以直接对密度值做等频分箱，也就是所有样本按密度排序后切成 n 段，每段样本数量大致相同；也可以做等宽分箱，但数据分布不均匀时容易导致某些桶样本过少。我实际测试下来，等频分桶在多密度场景下表现更稳，它能保证每个桶都有足够的样本量供后续 DBSCAN 使用。

2.2 第二层聚类：每个桶内独立运行自适应 DBSCAN

分桶完成后，每个桶代表一个密度层级。在桶内部，样本的密度波动范围已经远小于全局，此时用固定的 eps 就不再是大问题。但为了让每个桶内的聚类更稳定，我还会做一个简单操作：用桶内所有样本的 KNN 距离的某一分位数（比如 70% 或 85%）作为该桶的 eps，然后再以这个 eps 运行 DBSCAN。

这里有个值得细说的点。为什么不用中位数而是用更高的分位数？因为 DBSCAN 的 eps 本质上定义的是邻域半径，它需要容纳下核心点周围足够的邻居，取较高的分位数相当于给邻域一个更宽松的半径，避免把一些边缘核心点误判为噪声。但也不能取太高，比如 95% 以上，那样会导致邻域范围过大，簇间桥接的风险明显上升。

桶内的 min_samples 建议保持和 k 一致或略大一点。如果 min_samples 取太大，小型簇容易被拆碎，取太小又会把噪声点卷进簇里。我之前长期用 min_samples = k + 1，整体效果比较平衡。

2.3 第三层合并：把分层的簇整合回统一结果

每层聚类完成后，我们得到多个局部标签集合。它们之间会存在三种关系：完全独立、部分重叠、包含关系。因为每个样本在初始状态下只属于一个桶，严格来说桶与桶之间不共享样本，但在边界处会出现密度相近的点被分到两个不同桶的情况。比如高密度桶的边缘点和低密度桶的边缘点，在空间上可能挨得很近，如果它们的局部密度刚刚落在分桶边界的两侧，就可能被分到不同层级。

合并的策略有多种选择。最简单粗暴的是直接在原始空间上对各个簇做一次邻近合并：如果某两个簇之间的最近样本距离小于两个簇 eps 的平均值，就判定为同一簇。更精细的做法是在分桶时就保留一定量的重叠样本，即分桶边界使用软切分，让相邻层有部分共享点，这样后续可以通过连通分量直接完成合并。

我在工程实现中选择了软切分加连通分量的方案。具体来说，在等频分桶后，把边界附近的一部分样本同时复制给相邻桶参与聚类，但聚类的标签归属仍以样本自身的原始桶结果为主，通过一个简单的分配表记录每个样本的最终簇归属。这样做的好处是，合并逻辑完全基于数据事实，不引入额外的人工规则，而且实现起来非常直接。

3. Python 核心实现：从零搭建 Multi-DBSCAN

3.1 代码前的规划：核心函数拆解

动手写代码之前，先把功能模块拆清楚。整个 Multi-DBSCAN 可以拆成三个核心函数：第一个计算局部密度并完成分桶，第二个针对每一个桶运行自适应 DBSCAN，第三个合并各桶结果并输出最终标签。另外需要设计一个参数配置类或字典，把 k 值、桶数、分位数等集中管理，方便后续调参。

模块化设计的好处是，后续如果想调整分桶策略，比如从等频分桶换成基于密度峰值的聚类分桶，不需要改动其他模块。如果想把时序维度引入，也只需要在特征构建阶段增加时序相关特征即可。

3.2 分桶实现代码与参数说明

先展示分桶部分的代码，这段代码的思路是：输入特征矩阵，计算每个样本的 k 近邻距离均值作为局部密度，然后用等频分桶把样本划分为 density_bins 个层级。

import numpy as np from sklearn.neighbors import NearestNeighbors from sklearn.preprocessing import StandardScaler def compute_local_density(X, k=5): """ 计算每个样本的局部密度指标 返回: density_scores 数组, 值越小表示越拥挤(密度越高) """ scaler = StandardScaler() X_scaled = scaler.fit_transform(X) nn = NearestNeighbors(n_neighbors=k + 1, metric='euclidean') nn.fit(X_scaled) distances, _ = nn.kneighbors(X_scaled) # 第0个最近邻是自己, 取第1~k个最近邻距离 knn_dist = distances[:, 1:] # shape: (n_samples, k) # 用平均距离作为密度指标, 距离越小密度越高 density_scores = np.mean(knn_dist, axis=1) return density_scores, X_scaled def split_density_bins(density_scores, n_bins=4): """ 等频分桶: 所有样本按密度排序后均匀切分成 n_bins 层 返回: 每个样本的桶编号 (0 ~ n_bins-1), 以及桶的密度边界 """ order = np.argsort(density_scores) n = len(density_scores) bin_edges = np.quantile(density_scores, np.linspace(0, 1, n_bins + 1)) bin_ids = np.zeros(n, dtype=int) for i in range(n_bins): if i == n_bins - 1: # 最后一个桶包含所有剩余样本 mask = (density_scores >= bin_edges[i]) & (density_scores <= bin_edges[i + 1]) else: mask = (density_scores >= bin_edges[i]) & (density_scores < bin_edges[i + 1]) bin_ids[mask] = i return bin_ids, bin_edges

这段代码有几个细节值得注意。第一，局部密度使用了平均 k 近邻距离而不是最大第 k 近邻距离，平均值的抗噪性更好，不容易被单个远距离点带偏。第二，等频分桶使用 np.quantile 一次性算出边界，逻辑清晰，代码也很简洁。第三，最后一桶用闭区间是因为浮点数的边界值可能恰好等于最大值，不闭合会漏样本。

实际应用时，很多读者容易忽略标准化这一步。密度本身是相对概念，如果特征的量纲不同，比如一个是温度值在 20 到 30 之间波动，另一个是压力值在上千范围内波动，距离计算会被大数值特征主导，密度的物理含义就变形了。所以标准化这一步不是可选操作，是必须的。

3.3 桶内自适应 DBSCAN 实现

接下来是桶内聚类的核心代码。这里有一个重要设计：每个桶的 eps 由桶内样本自身的密度特征决定，而不是外部传入固定值。

from sklearn.cluster import DBSCAN def cluster_each_bin(X_scaled, bin_ids, n_bins, min_samples=6, quantile=0.85): """ 对每个密度桶单独运行 DBSCAN 返回: 所有样本的簇编号 (同一桶内有效, 跨桶不保证唯一) """ n = X_scaled.shape[0] all_labels = np.full(n, -1, dtype=int) for bin_id in range(n_bins): idx = np.where(bin_ids == bin_id)[0] if len(idx) < min_samples: # 样本太少时, 无法形成有效簇, 全部标记为噪声 all_labels[idx] = -1 continue X_bin = X_scaled[idx] # 用桶内样本的密度分位数决定 eps nn = NearestNeighbors(n_neighbors=min_samples) nn.fit(X_bin) distances, _ = nn.kneighbors(X_bin) # 取第 min_samples-1 个最近邻距离 (注意索引从0开始) k_dist = distances[:, -1] eps_bin = np.quantile(k_dist, quantile) db = DBSCAN(eps=eps_bin, min_samples=min_samples) labels_bin = db.fit_predict(X_bin) # 将桶内标签映射回全局索引 # 桶内标签大于等于0表示簇, -1表示噪声 # 这里暂不做跨桶合并, 直接记录临时标签 all_labels[idx] = labels_bin return all_labels

这段代码里最核心的一行就是 eps_bin = np.quantile(k_dist, quantile)。它的含义是：在该桶内部，选择 85% 的样本都能在自己的第 min_samples-1 近邻距离内找到足够邻居的半径阈值。用数据自身分布来定义参数，避免了人工调参。

quantile 参数的选择对整个聚类结果影响非常大。取值偏低（如 0.5），eps 偏小，簇会更碎，噪声更多；取值偏高（如 0.95），eps 偏大，簇可能会过度合并。我在实践中倾向于 0.8 到 0.9 这个区间，具体取值取决于业务对"宁可拆碎也不许粘连"还是"宁可粘连也不许拆碎"的偏好。

3.4 合并簇的两种策略与代码实现

跨桶合并是一个值得单独讲的问题。我在文中前面提到软切分方案，现在给出具体的合并代码。这个方案分两步：第一步，正常等频分桶并聚类；第二步，对每个桶的边界样本（即密度值落在分桶边界附近一定范围内的样本），将其同时加入相邻桶的聚类结果，通过这个共享样本集合推导连通关系。

from collections import defaultdict def merge_bins_final(X_scaled, density_scores, bin_ids, all_labels, bin_edges, merge_margin=0.1, min_samples=6): """ 基于软边界合并各桶簇标签 思路: 对每个桶边界附近的样本, 同时参与相邻桶的聚类, 然后用并查集合并簇编号 """ n = X_scaled.shape[0] n_bins = len(bin_edges) - 1 # 初始化并查集: 每个样本自己是一个集合 parent = list(range(n)) def find(x): while parent[x] != x: parent[x] = parent[parent[x]] x = parent[x] return x def union(x, y): px, py = find(x), find(y) if px != py: parent[px] = py for bin_id in range(n_bins): idx_in_bin = np.where(bin_ids == bin_id)[0] labels_in_bin = all_labels[idx_in_bin] # 在该桶内部, 同一簇标签的样本合并到同一集合 for label in set(labels_in_bin): if label == -1: continue cluster_idx = idx_in_bin[labels_in_bin == label] first = cluster_idx[0] for j in cluster_idx[1:]: union(int(first), int(j)) # 跨桶软合并: 对相邻桶边界附近的样本做最近邻判定 if bin_id < n_bins - 1: # 当前桶的高密度侧边界样本 lower_edge = bin_edges[bin_id] upper_edge = bin_edges[bin_id + 1] # 计算当前桶内靠近高密度边界的样本 left_bound = upper_edge - merge_margin * (upper_edge - lower_edge) boundary_idx = idx_in_bin[density_scores[idx_in_bin] >= left_bound] # 在相邻桶中寻找这些边界样本的最近邻 if len(boundary_idx) > 0: next_bin_idx = np.where(bin_ids == bin_id + 1)[0] if len(next_bin_idx) > 0: nn = NearestNeighbors(n_neighbors=1) nn.fit(X_scaled[next_bin_idx]) distances, neighbor_idx = nn.kneighbors(X_scaled[boundary_idx]) for local_i, dist_i in enumerate(distances[:, 0]): if dist_i < 0.5: # 距离阈值可根据数据调整 global_i = boundary_idx[local_i] global_j = next_bin_idx[neighbor_idx[local_i, 0]] union(int(global_i), int(global_j)) # 收集最终的簇标签 final_labels = np.full(n, -1, dtype=int) seen = {} cluster_counter = 0 for i in range(n): root = find(i) if all_labels[i] == -1: # 原始标记为噪声的样本, 除非被合并到其他簇, 否则保持噪声 # 通过检查 root 是否包含非噪声样本来决定归属 pass if root not in seen: seen[root] = cluster_counter cluster_counter += 1 final_labels[i] = seen[root] # 修正: 纯噪声样本的根集合里只有它自己, 应标记为-1 for i in range(n): root = find(i) members = [j for j in range(n) if find(j) == root] if len(members) == 1 and all_labels[i] == -1: final_labels[i] = -1 else: # 如果根集合包含噪声样本和正常样本, 噪声样本被吸入簇 pass return final_labels, cluster_counter

这段文字代码虽然有实现细节上的简化，但核心思路是够用的：并查集天然适合解决"哪些簇属于同一个连通分量"的问题。在每个桶内部先合并同簇样本，再通过边界样本的最近邻关系跨桶合并，最终得到的连通分量就是最终的簇。

这个方案的巧妙之处在于，边界样本不强制归属某一个桶，而是通过自身在特征空间中的真实邻居关系决定属于哪个连通分量。这比简单地把样本从其原始桶中划走更合理，因为它保留了聚类中最重要的信息——样本在特征空间中的连接关系。

注意：merge_margin 参数控制边界样本的筛选范围，取值过大会让所有样本都变成边界样本，失去分层意义；取值过小则跨桶合并几乎失效。建议设置在 0.05 到 0.15 之间，调试时可先固定其他参数，单独观察该参数对簇数量的影响。

3.5 时序聚类场景的扩展实现

时序聚类和普通聚类相比有一个特殊性：时间顺序本身携带关系信息。对时间序列做 Multi-DBSCAN 时，不能直接把每个时间点当作独立的样本点，否则算法会忽略"相邻时间点之间的连续性"这一重要特征。常用的做法是在特征中加入时间戳作为一维特征。

但直接拼特征有副作用：如果时间维度权重过大，聚类结果会完全按照时间段切分，忽略数值特征的相似性；权重过小，则时间连续性约束几乎不起作用，导致同一簇在时间上可能是不连续片段。这个问题可以在标准化环节解决，对时间列单独缩放，使时间维度在距离计算中的影响力可控。

我做一个实际案例来演示。假设数据有两个特征：时间戳（单位：秒）和一个数值信号。标准化后，时间列的范围会被压缩到 [0,1] 之间，信号列也被压缩到 [0,1] 之间。此时两者在距离计算中权重相同，但实际业务中你可能希望时间相近的样本优先聚在一起。这时可以对时间列乘以一个权重系数 time_weight，比如 2.0，相当于把时间距离放大两倍。

在实际项目中，我还遇到过一个特殊情况：连续时间序列在分割成窗口后，窗口之间的边界效应会比较明显，比如相邻窗口共享一个时间点，但特征上并不完全相同。处理方案是让窗口之间有少量重叠，这样聚类时可以在重叠区域建立桥梁关系。这个技巧在行为序列聚类中效果明显，重叠率通常取 10% 到 20%。

4. 调参与性能优化：实测中的关键经验

4.1 最关键的参数是桶数 n_bins

Multi-DBSCAN 有四个核心参数：k（近邻数）、n_bins（桶数）、quantile（eps 分位数）、merge_margin（合并边界宽度）。其中对结果影响最大的是 n_bins。

如果 n_bins 太小，等于回到传统 DBSCAN，每个桶内的密度差异依然很大，自适应 eps 退化成一个全局值的近似。如果 n_bins 太大，每个桶内样本变少，DBSCAN 难以发现稳定的簇结构，而且桶与桶之间的边界样本太多，合并阶段的计算量显著上升。

我常用的调法是从 n_bins=4 开始，观察聚类结果的簇数量和噪声占比，然后逐步增加桶数（每次加 2），观察结果变化幅度。如果桶数从 4 增加到 6 时，簇数量变化超过 30%，说明还有明显的密度层次没有被完全切分；如果变化很小，说明当前桶数已经足够。这个方法简单直观，不需要复杂的统计检验。

4.2 噪声样本的二次归属：值不值得做

传统 DBSCAN 会把不满足密度条件的样本直接标记为噪声，但在 Multi-DBSCAN 中，噪声问题多了一层复杂性：处于密度低谷区的样本，可能在所有桶中都无法被纳入簇。这些样本在合并阶段可能因为边界判定而与某个簇建立连接，也可能保持孤立。

我的处理原则是：如果业务侧希望最大程度减少未分类样本，可以在合并完成后，对剩余噪声样本做一次最近邻归属，也就是把每个噪声样本分配给距离最近的非噪声簇。但必须设置距离上限（比如样本 k 近邻距离的 2 倍），避免将孤立点强行拉入遥远簇中。

这个操作有利有弊。优点是可解释性更强，业务分析师不需要面对大量未归类样本；缺点是对离群点不太友好，真正的异常值可能被"洗白"成正常样本。在做异常检测时，我倾向于不做二次归属，保留噪声标签用于下游告警。

4.3 大样本量下的性能优化思路

Multi-DBSCAN 的计算复杂度主要由 KNN 计算和每桶的 DBSCAN 组成。KNN 部分如果直接调用 sklearn 的 NearestNeighbors，在小样本上没有问题，但在百万级数据上会因为距离矩阵的计算量过大而卡顿。

有几个现成的优化路径。第一，使用近似最近邻库（如 annoy 或 faiss），牺牲少量精度换取数量级的提速。第二，在分桶前先对数据进行粗采样，用部分样本估计密度分布，再对全量数据进行分桶，这样分桶结果会稳定很多。第三，在每桶运行 DBSCAN 前，可以对桶内样本再次做 MiniBatch 抽样，但这个操作会破坏簇的完整性，需要谨慎使用。

我在处理百万级数据时，通常会先用 faiss 的 IndexFlatL2 替换 sklearn 的 NearestNeighbors，实测在保持结果几乎一致的情况下，耗时从十几分钟降到几十秒。代码改动量很小，只需要替换最近邻计算的调用方式。

5. 常见问题与排查速查表

5.1 分桶后部分桶样本太少怎么办

等频分桶天然保证每个桶样本数大致相同，但如果数据本身存在极端密度差异，比如 99% 的点集中在一个极小区域内，等频分桶后低密度区域可能只包含极少量样本，DBSCAN 在样本数不足时很难形成有效簇。

这时有几个调整方向：一是增加 n_bins，让每个桶的样本量更均衡；二是对极低密度的桶不做聚类，直接标记为噪声，因为该区域的样本分布过于稀疏，聚类的统计意义本来就弱；三是改用不等频分桶，根据业务领域知识手动指定密度区间边界，但这种方法人为干预过多，一般不推荐。

5.2 同一个簇被切碎成多个小簇

典型的症状是：在稠密区域，数据明明连成一个整体，聚类结果却被切成了几个相邻的碎片。原因通常出在 quantile 上，取值偏低导致 eps 偏小，使得核心点邻域无法覆盖到簇的中间地带。

排查思路是：先固定其他参数，把 quantile 从 0.5 逐步提升到 0.95，观察簇数量变化。理论上簇数量应该先下降后稳定，如果始终持续下降，说明密度分层还不够细化，应增加 n_bins。另外也可能是 min_samples 设太高，导致部分核心点变成边界点或噪声点。

5.3 低密度簇在合并阶段被吞并

合并阶段的边界阈值设得太宽，会导致高密度桶的边界区域和低密度桶的核心区域被错误连接，使低密度簇完整地并入高密度簇。这个问题因为我一开始用固定距离阈值做边界判定时踩过坑，后来引入了密度自适应的边界判定，即两个桶之间的边界连接阈值与该桶的 eps 挂钩之后才稳定下来。

5.4 时序聚类的"闪烁"现象

在时序场景中，同一个簇在时间轴上可能会闪烁，即出现一段时间归属 A 簇，下一段时间归属 B 簇，再下一段时间又回到 A 簇。这种情况通常是因为时间特征的权重过小，算法在数值特征变化剧烈的节点频繁切换簇归属。

处置方案有两个：一是增大 time_weight，让时间维度在距离计算中的权重更高，强迫聚类结果在时间上更连贯；二是在聚类完成后增加一个后处理环节，对每个时间序列窗口做多数投票或者卡尔曼平滑，消除短周期的簇归属抖动。个人经验是方案二更稳健，因为增大 time_weight 容易让聚类退化成纯粹按时间切分，会丢失数值特征的聚类效果，而这个后处理环节有独立的参数可调。

我在做某传感器设备的状态识别时遇到过这个情况，当时的振动信号在不同工况下幅值差异巨大，单纯加大时间权重直接导致工作状态分割完全跟随时间段，后来改用后处理平滑——簇标签的众数窗口设为 5，在 90% 的场景下都能消除闪烁，而且没有破坏信号聚类的主要结构。

5.5 Multi-DBSCAN 常规参数速查

下面整理一个参数速查表，方便后续使用。这里的数值以我的实践经验为准，具体数据还需结合业务场景调整。

这些参数之间还有联动效应，比如 n_bins 增大后，桶内样本的密度分布更均匀，quantile 可以适当调低。建议在项目初期先把 n_bins 确定下来，再调 quantile 和 merge_margin，最后根据结果微调 k 和 min_samples。不要同时调所有参数，那样很难定位问题根源。

6. 一段你必须亲手跑的完整 Demo

前面说了很多原理和细节，这里给一个可以直接复制运行的完整 Demo。我用的数据是 scikit-learn 自带的 make_blobs，人为构造三个密度差异明显的簇，演示 Multi-DBSCAN 如何正确区分它们。

import numpy as np from sklearn.datasets import make_blobs from sklearn.cluster import DBSCAN from sklearn.neighbors import NearestNeighbors from sklearn.preprocessing import StandardScaler from collections import defaultdict # 构造三密度数据集: 一个稠密簇, 一个中等密度簇, 一个稀疏簇 X, _ = make_blobs(n_samples=[300, 200, 150], centers=[[0, 0], [5, 5], [10, 0]], cluster_std=[0.3, 0.8, 2.0], random_state=42) # 参数设置 k = 5 n_bins = 3 quantile = 0.85 min_samples = k + 1 merge_margin = 0.1 # 1. 标准化 + 局部密度 scaler = StandardScaler() X_scaled = scaler.fit_transform(X) nn = NearestNeighbors(n_neighbors=k + 1) nn.fit(X_scaled) distances, _ = nn.kneighbors(X_scaled) density_scores = np.mean(distances[:, 1:], axis=1) # 2. 等频分桶 bin_edges = np.quantile(density_scores, np.linspace(0, 1, n_bins + 1)) bin_ids = np.zeros(X.shape[0], dtype=int) for i in range(n_bins): if i == n_bins - 1: mask = (density_scores >= bin_edges[i]) & (density_scores <= bin_edges[i + 1]) else: mask = (density_scores >= bin_edges[i]) & (density_scores < bin_edges[i + 1]) bin_ids[mask] = i # 3. 桶内自适应 DBSCAN labels = np.full(X.shape[0], -1, dtype=int) for bin_id in range(n_bins): idx = np.where(bin_ids == bin_id)[0] if len(idx) < min_samples: labels[idx] = -1 continue X_bin = X_scaled[idx] nn_bin = NearestNeighbors(n_neighbors=min_samples) nn_bin.fit(X_bin) dist_bin, _ = nn_bin.kneighbors(X_bin) eps_bin = np.quantile(dist_bin[:, -1], quantile) db = DBSCAN(eps=eps_bin, min_samples=min_samples) labels[idx] = db.fit_predict(X_bin) # 4. 简易合并: 跨桶最近邻距离低于阈值则并入同一个簇 final_labels = labels.copy() next_label = final_labels.max() + 1 for bin_id in range(n_bins - 1): idx_bin_a = np.where(bin_ids == bin_id)[0] idx_bin_b = np.where(bin_ids == bin_id + 1)[0] if len(idx_bin_a) == 0 or len(idx_bin_b) == 0: continue # 对本桶和相邻桶做边界样本的最近邻搜索 nn_cross = NearestNeighbors(n_neighbors=1) nn_cross.fit(X_scaled[idx_bin_b]) for i in idx_bin_a: if final_labels[i] == -1: continue dist_bi, idx_near = nn_cross.kneighbors(X_scaled[i].reshape(1, -1)) j = idx_bin_b[idx_near[0, 0]] if dist_bi[0, 0] < 0.3 and final_labels[j] != -1: # 合并: 把 bin_id+1 桶中的这个簇编号改为与当前簇一致 old_label = final_labels[j] final_labels[final_labels == old_label] = final_labels[i] elif dist_bi[0, 0] < 0.3 and final_labels[j] == -1: final_labels[j] = final_labels[i] # 5. 输出对比: 传统DBSCAN vs Multi-DBSCAN db_global = DBSCAN(eps=0.5, min_samples=min_samples) labels_global = db_global.fit_predict(X_scaled) n_clusters_multi = len(set(final_labels) - {-1}) n_clusters_global = len(set(labels_global) - {-1}) print(f"Multi-DBSCAN 簇数: {n_clusters_multi}, 噪声数: {np.sum(final_labels == -1)}") print(f"传统DBSCAN 簇数: {n_clusters_global}, 噪声数: {np.sum(labels_global == -1)}")

这个 Demo 里第 4 步的合并逻辑做了很大简化，实际生产环境建议用并查集实现更稳健的合并，但核心思想一致。运行后会看到 Multi-DBSCAN 在混合密度数据上保留了 3 个簇，而传统 DBSCAN 要么把它们合并成一个大簇，要么把稀疏簇全部标记为噪声。

我第一次跑通这个 Demo 时最直观的感受是，原来困扰了那么久的密度不均匀问题，解决办法并没有想象中复杂，关键是跳出"全局一个 eps"这个思维定式。数据自身的空间结构是客观的，缺乏的只是贴合结构的分层工具。

7. 后续可以继续扩展的方向

Multi-DBSCAN 还有不少可玩的空间。比如把分桶逻辑从等频分桶改成基于密度峰值的聚类分桶，能更好地识别数据中的密度层级而不是机械地切分数量。再比如在分桶阶段引入领域知识，让业务分析师直接划定密度区间，这在工业场景中非常实用，因为业务侧通常对"什么是高密度"有明确预期。

另一个方向是把 Multi-DBSCAN 嵌入到在线学习框架中。时序数据流式到达时，可以只对新增数据做局部密度更新和桶归属判定，而不需要对全量数据重新聚类。这样聚类结果可以保持实时更新，适合 IoT 设备状态监控这类需要快速响应场景。这块我目前只做了初步测试，效果还不错，后续有机会单独写一篇。

回到开头的问题，全局半径的硬伤本质上是对数据结构的一次强行简化。Multi-DBSCAN 把这个简化过程拆成了多步，每步只做局部决策，最后通过合并还原完整的全局结构。思路不神秘，实现也不复杂，但用的过程中会让你重新理解密度聚类这个经典算法的边界在哪里。希望这篇分享能帮你少走弯路，遇到密度不均有问题的数据时，能多一个顺手能用的工具。