详解DBSCAN算法原理及其Python实现

# 距离矩阵
def disMat(data):
    arr = np.array(data)
    dMat = lambda arr : arr.reshape(1,-1) - arr.reshape(-1,1)
    # 此为单个轴的距离矩阵
    mats = [dMat(arr[:,i]) for i in range(arr.shape[1])]
    return np.linalg.norm(mats, axis=0)

import numpy as np

class DBSCAN(object):
    # e 最小距离, minPts 最少样本数量
    def __init__(self, e, minPts):
        self.e = e
        self.minPts = minPts

    # 获取点id的临近点
    def nearby(self, id):
        return np.where(self.mat[id] <= self.e)[0]

    def searchNearbyPts(self, points, group):
        for id in points:
            if id not in self.data:
                continue

            group.append(id)
            self.data.remove(id)

            # 查看id点的临近点
            nearbyPts = self.nearby(id)
            if len(nearbyPts) >= self.minPts:
                self.searchNearbyPts(nearbyPts, group)

    def fit(self, mat):
        self.mat = mat
        groups = list()

        keys = range(mat.shape[0])
        self.data = list(keys)

        for id in keys:
            if id not in self.data:
                continue

            # id点的临近点
            nearbyPts = self.nearby(id)
            if len(nearbyPts) < self.minPts:
                continue

            group = [id]
            self.data.remove(id)
            self.searchNearbyPts(nearbyPts, group)
            groups.append(group)

        # 加入飞点
        groups += self.data
        return groups

# 初始化数据
def initData(num, min, max):
    return [np.random.randint(min, max, size=2) for _ in range(num)]

def drawRes(data, groups):
    for gp in groups:
        xy = np.array([data[i] for i in gp])
        plt.scatter(xy[:,0], xy[:,1])
    plt.title(u'DBSCAN')
    plt.grid()
    plt.tight_layout()
    plt.show()

if __name__ == '__main__':
    np.random.seed(42)
    ds1 = initData(20, 0, 30)
    ds2 = initData(20, 40, 60)
    ds3 = initData(20, 70, 100)
    ds = ds1 + ds2 + ds3

    score_mat = disMat(ds)

    groups = DBSCAN(20, 3).fit(score_mat)
    drawRes(ds, groups)