k邻近算法实现

我会带着你远行 2023-02-18 12:55 9阅读 0赞

###  k邻近算法实现 ###

##### 前言 #####

k邻近算法是经典的机器学习算法，也是较为简单的一种。它的基本思想非常简单，去寻找与之最相近的一些点，根据这些点的性质，判断新点的性质。实现k邻近一般有两种方法，暴力计算和基于KD树的算法。

##### 暴力计算 #####

简而言之，直接遍历所有点，计算新点与这些点的距离，然后选出距离最相近的k个点。再根据这k个点的性质来做出判断。

##### 基于KD树的实现 #####

首先，需要了解KD树的思路。由于已经存在讲解非常好的博客，这里不在叙述KD树的思路，可以参考博客：[KD树算法思路][KD]  
[KD树算法图解][KD 1]  
本博客主要介绍如何用代码实现k邻近算法。  
由于机器学习都是大都是基于Python实现的，所以放弃c++,开始尝试用Python来实现。

###### 建立KD树 ######

从KD树的算法思路，我们可以很容易的知道，KD树其实是一颗二叉树，因此建树我们使用递归的方式来建立。本次代码假设输入数据是二维的，实际情况是多维的，但由二维扩展到多维是很容易的，所以先实现二维的。选择深度depth来判断当前划分的维度（depth为偶数是，以第一维度划分，depth为奇数时，以第二维度划分）。

def build_kd_tree(self,nums,depth):   # 递归建立kd_tree
        n = len(nums)
        if n <= 0:
          return None
        aix = depth%2
        sorted_nums = sorted(nums,key=lambda num :num[aix])
        root_num = n//2
        return {
            'root': sorted_nums[root_num],
            'left': self.build_kd_tree(sorted_nums[:root_num],depth+1),
            'right': self.build_kd_tree(sorted_nums[root_num+1:],depth+1)
        }

代码返回的是一个字典形式的数据结构，其实就是根节点，左右子节点。

###### 计算距离 ######

def distance(self,point1,point2):
        if point1 is None or point2 is None:
          return 0
        x1,y1 = point1
        x2,y2 = point2
        return math.sqrt((x2-x1)**2+(y2-y1)**2)

这个无需多言，就是两点间的距离。当然，很多k邻近算法并不使用这种方法来计算距离。

###### 判断哪个点距离更近 ######

def closer_point(self,point,p1,p2):
        d1 = distance(point,p1)
        d2 = distance(point,p2)
        if p1 is None:
          return (p2,d2)
        elif p2 is None:
          return (p1,d1)
        else:
          if(distance(point,p1)>distance(point,p2)):
            return (p2,d2)
          else:
            return (p1,d1)

目标点point，分别计算到p1,p2两点的距离即可。

###### 寻找最邻近的k个点 ######

def find_best(self,rt,point,depth):
        aix = depth%2 # 判断划分维度
        if rt is None:
          return None
        if point[aix] < rt['root'][aix]:  # 与根节点比较大小，从而判断搜索的方向
          next_branch = rt['left']
          oppsite_branch = rt['right']
        else:
          next_branch = rt['right']
          oppsite_branch = rt['left']
        best,closer_dis = self.closer_point(point,self.find_best(next_branch,point,depth+1),rt['root']) # 递归搜索,这里的best其实目的是寻找与point最近的一个点
        # 下面都是回溯的过程
        dis = abs(point[aix]-rt['root'][aix])  # 计算目标点到划分线的距离，从而判断是否要向另一个分支搜索
        print('closer_dis,dis',closer_dis,dis)
        sorted_store2 = sorted(self.store.items(),key=lambda x:x[1])  # 将已经找到的点，按距离大小排序
        cmp_d = 0.0
        if len(sorted_store2)>=self.topk:  # 如果已经找到3个点
          cmp_d = sorted_store2[self.topk-1][1] # 记录最远的一个点的距离
        if dis < cmp_d or len(sorted_store2)<3: # 如果比最远的距离要小，那么考虑搜索另一个分支
          best,closer_dis = self.closer_point(point,self.find_best(oppsite_branch,point,depth+1),best) # 同样的递归搜索
        
        if best in self.store and self.store[best] > closer_dis:  # 记录best
          self.store[best] = closer_dis
        else:
          self.store[best] = closer_dis
        self.dic[best]=1  
        if best!=rt['root'] and self.dic[rt['root']]==0:  # 回溯过程中，记录相关点，这里不再进行判断，而是直接加入字典中，比较大小的事情，留到最后在做。
          self.dic[rt['root']]=1 # 标记
          self.store[rt['root']]=self.distance(point,rt['root']) 
        return best

这部分我实现的可能不是很完美，正常来说，应该用个优先队列就好了，但Python的一些结构，我用的不是很熟练，先用字典实现一下吧。这里很难理解的一点就是，为什么我要去找best，其实其他点只不过是我找best的过程中途径的一些点，他们有可能成为最近的k个点之一而已，记录它们是在回溯过程中实现的。

###### 预处理 ######

def fit(self,X,Y):
        self.data = dict(zip(X,Y))
        self.kdtree = self.build_kd_tree(X,0)
        for t in X:   # 标记，为了搜索的时候不重复
          self.dic[t]=0

将数据和标签对应，并且建立kd树.

###### 预测 ######

def predict(self,point):
        best = self.find_best(self.kdtree,point,0)
        print('best: ',best)
        print('store ',self.store)
        sorted_store = sorted(self.store.items(),key= lambda x:x[1])[:self.topk]
        print('sortd_store ',sorted_store)
        counter = defaultdict(int) # 构造字典
        for t,score in sorted_store:  # 统计次数出现最多的
          counter[self.data[t]]+=1
        sorted_counter = sorted(counter.items(),key=lambda x:x[1],reverse=True)
        print('sorted_counter ',sorted_counter)
        return sorted_counter[0][0]

###### 实验 ######

if __name__ == '__main__':
    
        # 训练数据
        points = [(6.27, 5.50), (1.24, -2.86), (-6.88, -5.40), (-2.96, -0.5), (17.05, -12.79), (7.75, -22.68),(10.80,-5.03)]
        labels = ['A', 'A', 'B', 'B', 'C', 'C','C']
    
        knn = KNN(topk=3)
        # 开始训练
        knn.fit(points, labels)
        # 预测
        label = knn.predict((-1,-5))
        print(label)

实验数据与知乎那篇文章中，画图分析用的数据相同。相互对比一下即可知道，该算法能正确的进行分类。

##### 总结 #####

上述代码实现的仅仅是二维数据，那么如果是n维数据呢？那就需要在节点上多加上一个维度的标志了。之后，可以按照从0到(n-1)维在回到0维的方式进行划分。当然，正常做法不是按顺序进行划分，而是每次选择方差最大的维度进行划分，因为这样能保证二叉树的高度尽可能小。  
上述实现仅仅作为一个参考，最好还是阅读比较成熟的源码。

##### 完整代码 #####

import math
    from collections import defaultdict
    class KNN(object):
      def __init__(self,topk):   # 初始化
        self.data = None
        self.store = {}
        self.topk=topk
        self.dic = {}
    
      def build_kd_tree(self,nums,depth):   # 递归建立kd_tree
        n = len(nums)
        if n <= 0:
          return None
        aix = depth%2
        sorted_nums = sorted(nums,key=lambda num :num[aix])
        root_num = n//2
        return {
            'root': sorted_nums[root_num],
            'left': self.build_kd_tree(sorted_nums[:root_num],depth+1),
            'right': self.build_kd_tree(sorted_nums[root_num+1:],depth+1)
        }
      
      def distance(self,point1,point2):
        if point1 is None or point2 is None:
          return 0
        x1,y1 = point1
        x2,y2 = point2
        return math.sqrt((x2-x1)**2+(y2-y1)**2)
      
      def closer_point(self,point,p1,p2):
        d1 = self.distance(point,p1)
        d2 = self.distance(point,p2)
        if p1 is None:
          return (p2,d2)
        elif p2 is None:
          return (p1,d1)
        else:
          if(self.distance(point,p1)>self.distance(point,p2)):
            return (p2,d2)
          else:
            return (p1,d1)
    
      def find_best(self,rt,point,depth):
        aix = depth%2 # 判断划分维度
        if rt is None:
          return None
        if point[aix] < rt['root'][aix]:  # 与根节点比较大小，从而判断搜索的方向
          next_branch = rt['left']
          oppsite_branch = rt['right']
        else:
          next_branch = rt['right']
          oppsite_branch = rt['left']
        best,closer_dis = self.closer_point(point,self.find_best(next_branch,point,depth+1),rt['root']) # 递归搜索,这里的best其实目的是寻找与point最近的一个点
        # 下面都是回溯的过程
        dis = abs(point[aix]-rt['root'][aix])  # 计算目标点到划分线的距离，从而判断是否要向另一个分支搜索
        print('closer_dis,dis',closer_dis,dis)
        sorted_store2 = sorted(self.store.items(),key=lambda x:x[1])  # 将已经找到的点，按距离大小排序
        cmp_d = 0.0
        if len(sorted_store2)>=self.topk:  # 如果已经找到3个点
          cmp_d = sorted_store2[self.topk-1][1] # 记录最远的一个点的距离
        if dis < cmp_d or len(sorted_store2)<3: # 如果比最远的距离要小，那么考虑搜索另一个分支
          best,closer_dis = self.closer_point(point,self.find_best(oppsite_branch,point,depth+1),best) # 同样的递归搜索
        
        if best in self.store and self.store[best] > closer_dis:  # 记录best
          self.store[best] = closer_dis
        else:
          self.store[best] = closer_dis
        self.dic[best]=1  
        if best!=rt['root'] and self.dic[rt['root']]==0:  # 回溯过程中，记录相关点，这里不再进行判断，而是直接加入字典中，比较大小的事情，留到最后在做。
          self.dic[rt['root']]=1 # 标记
          self.store[rt['root']]=self.distance(point,rt['root']) 
        return best
    
      def fit(self,X,Y):
        self.data = dict(zip(X,Y))
        self.kdtree = self.build_kd_tree(X,0)
        for t in X:   # 标记，为了搜索的时候不重复
          self.dic[t]=0
    
      def predict(self,point):
        best = self.find_best(self.kdtree,point,0)
        print('best: ',best)
        print('store ',self.store)
        sorted_store = sorted(self.store.items(),key= lambda x:x[1])[:self.topk]
        print('sortd_store ',sorted_store)
        counter = defaultdict(int) # 构造字典
        for t,score in sorted_store:  # 统计次数出现最多的
          counter[self.data[t]]+=1
        sorted_counter = sorted(counter.items(),key=lambda x:x[1],reverse=True)
        print('sorted_counter ',sorted_counter)
        return sorted_counter[0][0]
    if __name__ == '__main__':
    
        # 训练数据
        points = [(6.27, 5.50), (1.24, -2.86), (-6.88, -5.40), (-2.96, -0.5), (17.05, -12.79), (7.75, -22.68),(10.80,-5.03)]
        labels = ['A', 'A', 'B', 'B', 'C', 'C','C']
    
        knn = KNN(topk=3)
        # 开始训练
        knn.fit(points, labels)
        # 预测
        label = knn.predict((-1,-5))
        print(label)

[KD]: https://www.joinquant.com/view/community/detail/dd60bd4e89761b916fe36dc4d14bb272
[KD 1]: https://zhuanlan.zhihu.com/p/23966698