[Python数据挖掘] DBSCAN聚类算法原理底层实现

旧城等待， 2021-11-15 12:40 425阅读 0赞

注明：本文主要内容亦曾整理作为本人所选课程《模式识别与机器学习》的期末课程报告。

**\[问题背景\]**

关于DBSCAN聚类原理的动画演示，我在B站的投稿《[DBSCAN聚类原理 动画演示][DBSCAN_]》中已作了介绍，有兴趣的朋友可以先看一下链接中的1分半钟动画演示。

在[\[Python数据挖掘\] sklearn-DBSCAN聚类][Python_ sklearn-DBSCAN]一文中，我们调用sklearn库的DBSCAN聚类方法，成功对以下ANSI编码的grade.txt文本文档数据集进行了聚类：

YZN,133,108,76
    ZHY,96,145,101
    WYZ,132,107,60
    DHY,100,102,120
    CYH,139,99,93
    LHY,73,149,81
    ZHY,85,148,93
    TQP,39,138,85
    ZZL,145,112,71
    HJC,101,116,118
    XZY,99,98,117

**\[问题分析\]**

下面从底层实现的角度，只需用到math库，无需使用numpy，逐步实现DBSCAN聚类算法。

针对grade.txt文本文档进行数据预处理并聚类的实例代码：

（grade.txt需要与代码的.py文件位于同一目录下）

import math
    
    class DBSCAN:
        'DBSCAN类'
        eps = 0
        min_samples = 0
        labels_ = []
        
    
        def __init__(self, eps, min_samples):
            self.eps = eps
            self.min_samples = min_samples
    
        def distance(self, data1, data2): # 计算距离
            distance = 0
            for i in range(len(data1)):
                distance = distance + pow(data1[i]-data2[i], 2)
            distance = math.sqrt(distance)
            return distance
    
        def spread(self, i, Dict, labels_):
            for m in Dict[i]:
                if (labels_[m] == -1):
                    labels_[m] = labels_[i] 
                    self.spread(m, Dict, labels_)
        
        def fit(self, Data): # 聚类
            # 初始化
            flag = []
            for t in range(len(Data)):
                flag.append(-1) # 核心点(1)、边界点(0)、噪声点(-1)标记 # 默认标注为噪声点
                
            Dict = {}
            for t in range(len(Data)): # 生成相邻点关系
                Dict[t] = []
                
            # 样本点标注    
            for i in range(len(Data)):
                self.labels_.append(-1)
                samples = 0
                for j in range(len(Data)): 
                    if (i == j):
                        pass
                    else:
                        if (self.distance(Data[i], Data[j]) <= self.eps): 
                            samples = samples + 1 
                            if (flag[j] != 1): 
                                flag[j] = 0 
                            Dict[i].append(j) 
                if (samples >= self.min_samples): 
                    flag[i] = 1 
    
            # 样本点聚类
            for i in range(len(Data)):
                
                if (flag[i] == 1 and self.labels_[i] == -1):
                    self.labels_[i] = i
                    self.spread(i, Dict, self.labels_)
                else:
                        pass
            sort = []
            for t in self.labels_:
                if ((t not in sort) and (t != -1)):
                    sort.append(t)
            for td in range(len(self.labels_)):
                if (self.labels_[td] != -1):
                    for index in range(len(sort)):
                        if (sort[index] == self.labels_[td]):
                            self.labels_[td] = index
     
    def loadData(filePath):
        fr = open(filePath, 'r+')
        lines = fr.readlines()
        retName = []
        retData = []
        for line in lines:
            items = line.strip().split(',')
            retName.append(items[0])
            retData.append([float(items[i]) for i in range(1,len(items))])
        return retName, retData
     
    if __name__ == '__main__':
        Name, Data = loadData('grade.txt')
        
        eps = 21
        min_samples = 2
        db = DBSCAN(eps, min_samples)
        db.fit(Data) # 分开写，否则会return一个None，提示None不包含labels_
        label = db.labels_
        outputName = []
        noise = []
     
        for i in range(max(label)+1):
            outputName.append([])
        for i in range(len(Name)):
            if(label[i] != -1):
                outputName[label[i]].append(Name[i])
            else:
                noise.append(Name[i])
        print(outputName)
        print("噪声点:", noise)

运行结果为：

[['YZN', 'WYZ', 'CYH', 'ZZL'], ['ZHY', 'LHY', 'ZHY'], ['DHY', 'HJC', 'XZY']]
    噪声点: ['TQP']

可以看到，这一聚类结果与我们用sklearn的DBSCAN得到的聚类结果是一致的。

DBSCAN算法的核心实现部分：

import math
    
    class DBSCAN:
        'DBSCAN类'
        eps = 0
        min_samples = 0
        labels_ = []
        
    
        def __init__(self, eps, min_samples): # 参数初始化
            self.eps = eps
            self.min_samples = min_samples
    
        def distance(self, data1, data2): # 计算距离
            distance = 0
            for i in range(len(data1)):
                distance = distance + pow(data1[i]-data2[i], 2)
            distance = math.sqrt(distance)
            return distance
    
        def spread(self, i, Dict, labels_): # 传播标记
            for m in Dict[i]:
                if (labels_[m] == -1):
                    labels_[m] = labels_[i] 
                    self.spread(m, Dict, labels_)
        
        def fit(self, Data): # 聚类
            # 初始化
            flag = []
            for t in range(len(Data)):
                flag.append(-1) # 核心点(1)、边界点(0)、噪声点(-1)标记 # 默认标注为噪声点
                
            Dict = {}
            for t in range(len(Data)): # 生成相邻点关系
                Dict[t] = []
                
            # 样本点标注    
            for i in range(len(Data)):
                self.labels_.append(-1)
                samples = 0
                for j in range(len(Data)): 
                    if (i == j):
                        pass
                    else:
                        if (self.distance(Data[i], Data[j]) <= self.eps): 
                            samples = samples + 1 
                            if (flag[j] != 1): 
                                flag[j] = 0 
                            Dict[i].append(j) 
                if (samples >= self.min_samples): 
                    flag[i] = 1 
    
            # 样本点聚类
            for i in range(len(Data)):
                
                if (flag[i] == 1 and self.labels_[i] == -1):
                    self.labels_[i] = i
                    self.spread(i, Dict, self.labels_)
                else:
                        pass
            sort = []
            for t in self.labels_:
                if ((t not in sort) and (t != -1)):
                    sort.append(t)
            for td in range(len(self.labels_)):
                if (self.labels_[td] != -1):
                    for index in range(len(sort)):
                        if (sort[index] == self.labels_[td]):
                            self.labels_[td] = index

[DBSCAN_]: https://www.bilibili.com/video/av54693046
[Python_ sklearn-DBSCAN]: https://blog.csdn.net/swordtraveller/article/details/91954952