DBSCAN
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn import datasets
参考链接
DBSCAN 基本定义
- 邻域 ：半径，密度，
- 核心对象： 符合邻域要求
- 密度直达： 邻域内的对象
- 密度可达： 核心对象在邻域内
- 边界，在邻域内，却不是核心对象
- 噪音：不在邻域内
- 任何一个簇内的点都是核心点， 边界点将被抛弃？
- 那边界点和噪音点有什么区别？
DBSCAN 算法流程
x,y=datasets.make_blobs()
for i in list(set(y)):
plt.plot(x[y==i][:,0],x[y==i][:,1],'*',label=i)
plt.legend()
plt.show()
def dist(a,b):
'''
计算点a,b的距离
'''
return np.sqrt(np.sum((a-b)**2))
def nPoints_eps(data,nIndex,r):
'''
计算数据集data 中nIndex距离r内的数据点
data:所有数据
nIndex:数据点索引
return: 该点周围的数据
'''
epsIndex=[] #用来保存邻域的数据索引
for i,d in enumerate(data):
if dist(d,data[nIndex])<r:
epsIndex.append(i)
return epsIndex
1. 手动循环一遍
clusterResult=np.zeros(y.shape)# 等于0 表示未被访问，其他的表示为第几个簇 ，-1为噪音
MinPts=3 #密度大于3
r=3 # 距离为3
search=set() # 等待搜寻的数据索引
K=1 # 代表第几个簇
nIndex=np.random.choice(np.where(clusterResult==0)[0]) # 从未读取的数据中随机选一个
nPoints=nPoints_eps(x,nIndex,r) # 第一个数据
if len(nPoints)>MinPts:# 如果数据点大于密度,否则，就重选一个
search=search.union(set(nPoints))
clusterResult[nIndex]=K
while len(search):
# print(len(search))
nIndex=np.random.choice(list(search))#cong中任意选择一个
nPoints=nPoints_eps(x,nIndex,r) # 第n个数据
if len(nPoints)>MinPts:
search=search.union(set(nPoints)).intersection(np.where(clusterResult==0)[0])
clusterResult[nIndex]=K
search.remove(nIndex)
else:#标记维噪音
clusterResult[nIndex]=-1
search.remove(nIndex)
# 如果没有数据了，就要进行下一个簇了
#重复以上步骤
K+=1
nIndex=np.random.choice(np.where(clusterResult==0)[0]) # 从未读取的数据中随机选一个
nPoints=nPoints_eps(x,nIndex,r) # 第一个数据
if len(nPoints)>MinPts:# 如果数据点大于密度,否则，就重选一个
search=search.union(set(nPoints))
clusterResult[nIndex]=K
while len(search):
#print(len(search))
nIndex=np.random.choice(list(search))#cong中任意选择一个
nPoints=nPoints_eps(x,nIndex,r) # 第n个数据
if len(nPoints)>MinPts:
search=search.union(set(nPoints)).intersection(np.where(clusterResult==0)[0])
clusterResult[nIndex]=K
search.remove(nIndex)
else:#标记维噪音
clusterResult[nIndex]=-1
search.remove(nIndex)
K+=1
nIndex=np.random.choice(np.where(clusterResult==0)[0]) # 从未读取的数据中随机选一个
nPoints=nPoints_eps(x,nIndex,r) # 第一个数据
if len(nPoints)>MinPts:# 如果数据点大于密度,否则，就重选一个
search=search.union(set(nPoints))
clusterResult[nIndex]=K
while len(search):
#print(len(search))
nIndex=np.random.choice(list(search))#cong中任意选择一个
nPoints=nPoints_eps(x,nIndex,r) # 第n个数据
if len(nPoints)>MinPts:
search=search.union(set(nPoints)).intersection(np.where(clusterResult==0)[0])
clusterResult[nIndex]=K
search.remove(nIndex)
else:#标记维噪音
clusterResult[nIndex]=-1
search.remove(nIndex)
clusterResult
array([3., 1., 1., 2., 3., 1., 3., 3., 2., 1., 3., 3., 1., 3., 1., 1., 2.,
2., 2., 2., 3., 1., 3., 1., 1., 2., 2., 1., 2., 3., 2., 2., 2., 3.,
2., 2., 2., 1., 3., 1., 3., 1., 3., 2., 3., 3., 2., 3., 1., 1., 3.,
3., 1., 2., 2., 1., 2., 1., 1., 2., 3., 1., 3., 3., 3., 1., 1., 1.,
3., 3., 2., 1., 3., 1., 1., 2., 3., 2., 3., 2., 3., 3., 2., 3., 1.,
1., 2., 2., 2., 3., 1., 2., 1., 1., 2., 2., 3., 2., 1., 1.])
2. 完整来一遍
def selectOne(data,clusterResult,K):
'''
保证从数据集中选择一个不是噪音的点,用于每一个簇开始时
data： 数据集
clusterResult：聚类结果
K:第几个簇
'''
for nIndex in np.where(clusterResult==0)[0]:
'''
从未读取的数据中读取
'''
nPoints=nPoints_eps(x,nIndex,r) # 数据索引
if len(nPoints)<MinPts:
clusterResult[nIndex]=-1
else:
clusterResult[nIndex]=K
return nIndex
x,y=datasets.make_blobs()
clusterResult=np.zeros(y.shape)# 等于0 表示未被访问，其他的表示为第几个簇 ，-1为噪音
MinPts=3 #密度大于3
r=2 # 距离为3
K=1 # 代表第几个簇
while (clusterResult==0).sum():
'''
只要还有未读取的数据就要继续读取
'''
# 一个新的簇
nIndex=selectOne(x,clusterResult,K)
search=set(nPoints_eps(x,nIndex,r)) # 等待搜寻的数据索引
while len(search):
nIndex=np.random.choice(list(search))#cong中任意选择一个
nPoints=nPoints_eps(x,nIndex,r) # 第n个数据
if len(nPoints)>MinPts:
search=search.union(set(nPoints)).intersection(np.where(clusterResult==0)[0])
clusterResult[nIndex]=K
search.remove(nIndex)
else:#标记为噪音
clusterResult[nIndex]=-1
search.remove(nIndex)
K+=1
for i in list(set(clusterResult)):
plt.plot(x[clusterResult==i][:,0],x[clusterResult==i][:,1],'*',label=i)
plt.legend()
plt.show()
