使用sklearn.cluster.DBSCAN如何仅扩大邻域查找范围避免噪声误判与簇合并

scikit-learn 原生的 sklearn.cluster.DBSCAN 没有提供单独拆分「聚类合并阈值」和「噪声邻域查找阈值」的参数，你需要的效果可以通过双阈值DBSCAN后处理实现，不需要改动原始聚类阶段的规则，具体实现方式如下：

实现方案1：手动后处理（你提出的逻辑的标准化实现） #

你提到的kd-tree查找最近非噪声点的逻辑，scikit-learn已经有内置的NearestNeighbors工具类封装了相关优化（支持kd-tree、ball-tree等近邻查找算法），不需要手动实现kd-tree，完整代码示例如下：

from sklearn.cluster import DBSCAN
from sklearn.neighbors import NearestNeighbors
import numpy as np

# 1. 用原始参数跑DBSCAN，保证不会合并无关聚类
original_eps = 0.3  # 替换为你原本的eps参数
original_min_samples = 5 # 替换为你原本的min_samples参数
db = DBSCAN(eps=original_eps, min_samples=original_min_samples)
labels = db.fit_predict(X) # X为你的特征数据集

# 2. 分离噪声点和已聚类的核心/边界点
clustered_mask = labels != -1
clustered_points = X[clustered_mask]
clustered_labels = labels[clustered_mask]
noise_points = X[~clustered_mask]

# 3. 用扩大的邻域阈值为符合条件的噪声点分配簇
larger_eps = 0.6 # 替换为你想要的邻域查找阈值
if len(clustered_points) > 0 and len(noise_points) > 0:
    nn = NearestNeighbors(n_neighbors=1, algorithm='kd_tree')
    nn.fit(clustered_points)
    distances, indices = nn.kneighbors(noise_points)
    # 距离在扩大阈值内的噪声点分配到对应簇，否则保留为噪声
    for i, (dist, idx) in enumerate(zip(distances, indices)):
        if dist[0] <= larger_eps:
            labels[~clustered_mask][i] = clustered_labels[idx[0]]

实现方案2：封装为自定义类（兼容原生DBSCAN调用方式） #

如果需要多次复用该逻辑，可以继承原生DBSCAN类封装双阈值逻辑，调用方式和原生API完全一致：

class TwoEpsDBSCAN(DBSCAN):
    def __init__(self, eps_cluster=0.5, eps_noise=1.0, min_samples=5, **kwargs):
        # eps_cluster为聚类阶段的阈值，保证簇不合并
        super().__init__(eps=eps_cluster, min_samples=min_samples, **kwargs)
        # eps_noise为噪声点邻域查找的扩大阈值
        self.eps_noise = eps_noise

    def fit_predict(self, X, y=None, sample_weight=None):
        labels = super().fit_predict(X, y, sample_weight)
        clustered_mask = labels != -1
        # 没有聚类结果或没有噪声点直接返回
        if not np.any(clustered_mask) or not np.any(~clustered_mask):
            return labels
        
        clustered_points = X[clustered_mask]
        clustered_labels = labels[clustered_mask]
        noise_points = X[~clustered_mask]
        
        nn = NearestNeighbors(n_neighbors=1, algorithm='kd_tree')
        nn.fit(clustered_points)
        distances, indices = nn.kneighbors(noise_points)
        
        for i, (dist, idx) in enumerate(zip(distances, indices)):
            if dist[0] <= self.eps_noise:
                labels[~clustered_mask][i] = clustered_labels[idx[0]]
        return labels

# 调用示例
db = TwoEpsDBSCAN(eps_cluster=0.3, eps_noise=0.6, min_samples=5)
labels = db.fit_predict(X)

补充说明 #

目前scikit-learn没有内置该双阈值DBSCAN的实现，上述两种方案的核心逻辑都复用了sklearn内置的优化实现，性能和原生DBSCAN几乎没有差异，完全可以满足需求。

内容的提问来源于stack exchange，提问作者DottyPhone