机器学习：k近邻算法（Python）

一、k近邻算法的定义

二、KD树结点信息封装 

kdtree_node.py

 
class KDTreeNode:
    """
    KD树结点信息封装
    """
    def __init__(self, instance_node=None, instance_label=None, instance_idx=None,
                 split_feature=None, left_child=None, right_child=None, kdt_depth=None):
        """
        用于封装kd树的结点信息结构
        :param instance_node: 实例点，一个样本
        :param instance_label: 实例点对应的类别标记
        :param instance_idx: 该实例点对应的样本索引，用于kd树的可视化
        :param split_feature: 划分的特征属性，x^(i)
        :param left_child: 左子树，小于划分点的
        :param right_child: 右子树，大于切分点的
        :param kdt_depth: kd树的深度
        """
        self.instance_node = instance_node
        self.instance_label = instance_label
        self.instance_idx = instance_idx
        self.split_feature = split_feature
        self.left_child = left_child
        self.right_child = right_child
        self.kdt_depth = kdt_depth
 

三、距离度量的工具类

distUtils.py

import numpy as np
 
 
class DistanceUtils:
    """
    距离度量的工具类，此处仅实现闵可夫斯基距离
    """
    def __init__(self, p=2):
        self.p = p  # 默认欧式距离，p=1曼哈顿距离，p=np。inf是切比雪夫距离
 
    def distance_func(self, xi, xj):
        """
        特征空间中两个样本示例的距离计算
        :param xi: k维空间某个样本示例
        :param xj: k维空间某个样本示例
        :return:
        """
        xi, xj = np.asarray(xi), np.asarray(xj)
        if self.p == 1 or self.p == 2:
            return (((np.abs(xi - xj)) ** self.p).sum()) ** (1 / self.p)
        elif self.p == np.inf:
            return np.max(np.abs(xi - xj))
        elif self.p == "cos":  # 余弦距离或余弦相似度
            return xi.dot(xj) / np.sqrt((xi ** 2).sum()) / np.sqrt((xj ** 2).sum())
        else:
            raise ValueError("目前仅支持p=1、p=2、p=np.inf或余弦距离四种距离...")
 

四、K近邻算法的实现

knn_kdtree.py

import numpy as np
from kdtree_node import KDTreeNode
from distUtils import DistanceUtils
import heapq  # 堆结构，实现堆排序
from collections import Counter  # 集合中的计数功能
import networkx as nx  # 网络图，可视化
import matplotlib.pyplot as plt
 
 
class KNearestNeighborKDTree:
    """
    K近邻算法的实现，基于KD树结构
    1. fit: 特征向量空间的划分，即构建KD树（建立KNN算法模型）
    2. predict: 预测，近邻搜索
    3. 可视化kd树
    """
    def __init__(self, k: int=5, p=2, view_kdt=False):
        """
        KNN算法的初始化必要参数
        :param k: 近邻数
        :param p: 距离度量标准
        :param view_kdt: 是否可视化KD树
        """
        self.k = k  # 预测，近邻搜索时，使用的参数，表示近邻树
        self.p = p  # 预测，近邻搜索时，使用的参数，表示样本的近邻度
        self.view_kdt = view_kdt
        self.dis_utils = DistanceUtils(self.p)  # 距离度量的类对象
        self.kdt_root: KDTreeNode() = None  # KD树的根节点
        self.k_dimension = 0  # 特征空间维度，即样本的特征属性数
        self.k_neighbors = []  # 用于记录某个测试样本的近邻实例点
 
    def fit(self, x_train, y_train):
        """
        递归创建KD树，即对特征向量空间进行划分，递归调用进行创建
        :param x_train: 训练样本集
        :param y_train: 训练样本目标集合
        :return:
        """
        if self.k < 1:
            raise ValueError("k must be greater than 0 and be int.")
        x_train, y_train = np.asarray(x_train), np.asarray(y_train)
        self.k_dimension = x_train.shape[1]  # 特征维度
        idx_array = np.arange(x_train.shape[0])  # 训练样本索引编号
        self.kdt_root = self._build_kd_tree(x_train, y_train, idx_array, 0)
        if self.view_kdt:
            self.draw_kd_tree()  # 可视化kd树
 
    def _build_kd_tree(self, x_train, y_train, idx_array, kdt_depth):
        """
        递归创建KD树，KD树是二叉树，严格区分左子树右子树，表示对k维空间的一个划分
        :param x_train: 递归划分的训练样本子集
        :param y_train: 递归划分的训练样本目标子集
        :param idx_array: 递归划分的样本索引
        :param depth: kd树的深度
        :return:
        """
        if x_train.shape[0] == 0:  # 递归出口
            return
 
        split_dimension = kdt_depth % self.k_dimension  # 数据的划分维度x^(i)
        sorted(x_train, key=lambda x: x[split_dimension])  # 按某个划分维度排序
        median_idx = x_train.shape[0] // 2  # 中位数所对应的数据的索引
        median_node = x_train[median_idx]  # 切分点作为当前子树的根节点
        # 划分左右子树区域
        left_instances, right_instances = x_train[:median_idx], x_train[median_idx + 1:]
        left_labels, right_labels = y_train[:median_idx], y_train[median_idx + 1:]
        left_idx, right_idx = idx_array[:median_idx], idx_array[median_idx + 1:]
        # 递归调用
        left_child = self._build_kd_tree(left_instances, left_labels, left_idx, kdt_depth + 1)
        right_child = self._build_kd_tree(right_instances, right_labels, right_idx, kdt_depth + 1)
        kdt_new_node = KDTreeNode(median_node, y_train[median_idx], idx_array[median_idx],
                                  split_dimension, left_child, right_child, kdt_depth)
        return kdt_new_node
 
    def _search_kd_tree(self, kd_tree: KDTreeNode, x_test):
        """
        kd树的递归搜索算法，后序遍历，搜索k个最近邻实例点
        数据结构：堆排序，搜索过程中，维护一个小根堆
        :param kd_tree: 已构建的kd树
        :param x_test: 单个测试样本
        :return:
        """
        if kd_tree is None:  # 递归出口
            return
 
        # 计算测试样本与当前kd子树的根结点的距离（相似度）
        distance = self.dis_utils.distance_func(kd_tree.instance_node, x_test)
        # 1. 如果不够k个样本，继续递归
        # 2. 如果搜索了k个样本，但是k个样本未必是最近邻的。
        # 当计算的当前实例点的距离小于k个样本的最大距离，则递归，大于最大距离，没必要递归
        if (len(self.k_neighbors) < self.k) or (distance < self.k_neighbors[-1]["distance"]):
            self._search_kd_tree(kd_tree.left_child, x_test)  # 递归左子树
            self._search_kd_tree(kd_tree.right_child, x_test)  # 递归右子树
            # 在整个搜索路径上的kd树的结点，存储在self.k_neighbors中，包含三个值
            # 当前实例点，类别，距离
            self.k_neighbors.append({
                "node": kd_tree.instance_node,  # 结点
                "label": kd_tree.instance_label,  # 当前实例的类别
                "distance": distance  # 当前实例点与测试样本的距离
            })
            # 按照距离进行排序，选择最小的k个最近邻样本实例，更新最近邻距离
            # 小根堆，k_neighbors中第一个结点是距离测试样本最近的
            self.k_neighbors = heapq.nsmallest(self.k, self.k_neighbors,
                                               key=lambda d: d["distance"])
 
    def predict(self, x_test):
        """
        KD树的近邻搜索，即测试样本的预测
        :param x_test: 测试样本，ndarray: (n * k)
        :return:
        """
        x_test = np.asarray(x_test)
        if self.kdt_root is None:
            raise ValueError("KDTree is None, Please fit KDTree...")
        elif x_test.shape[1] != self.k_dimension:
            raise ValueError("Test Sample dimension unmatched KDTree's dimension.")
        else:
            y_test_hat = []  # 用于存储测试样本的预测类别
            for i in range(x_test.shape[0]):
                self.k_neighbors = []  # 调用递归搜索，则包含了k个最近邻的实例点
                self._search_kd_tree(self.kdt_root, x_test[i])
                # print(self.k_neighbors)
                y_test_labels = []
                # 取每个近邻样本的类别标签
                for k in range(self.k):
                    y_test_labels.append(self.k_neighbors[k]["label"])
                # 按分类规则（多数表决法）
                # print(y_test_labels)
                counter = Counter(y_test_labels)
                idx = int(np.argmax(list(counter.values())))
                y_test_hat.append(list(counter.keys())[idx])
        return np.asarray(y_test_hat)
 
    def _create_kd_tree(self, graph, kdt_node: KDTreeNode, pos=None, x=0, y=0, layer=1):
        """
        递归可视化KD树，递归构造树的结点、边。
        :param graph: 有向图对象，递归中逐步增加结点和左子树右子树
        :param kdt_node: 递归创建KD树的结点
        :param pos: 可视化中树结点位置，初始化(0, 0)绘制根结点
        :param x: 对应pos中的横坐标，随着递归，更新
        :param y: 对应pos中的纵坐标，随着递归，更新
        :param layer: kd树的层次
        :return:
        """
        if pos is None:
            pos = {}
        pos[str(kdt_node.instance_idx)] = (x, y)
        if kdt_node.left_child:
            # 父结点指向左子树
            graph.add_edge(str(kdt_node.instance_idx), str(kdt_node.left_child.instance_idx))
            l_x, l_y = x - 1 / 2 ** layer, y - 1  # 下一个树结点位置的计算
            l_layer = layer + 1  # 树的层次 + 1
            self._create_kd_tree(graph, kdt_node.left_child, x=l_x, y=l_y, pos=pos, layer=l_layer)  # 递归
        if kdt_node.right_child:
            # 父结点指向右子树
            graph.add_edge(str(kdt_node.instance_idx), str(kdt_node.right_child.instance_idx))
            r_x, r_y = x + 1 / 2 ** layer, y - 1
            r_layer = layer + 1
            self._create_kd_tree(graph, kdt_node.right_child, x=r_x, y=r_y, pos=pos, layer=r_layer)  # 递归
        return graph, pos
 
    def draw_kd_tree(self):
        """
        可视化kd树
        :return:
        """
        directed_graph = nx.DiGraph()  # 初始化一个有向图，树
        graph, pos = self._create_kd_tree(directed_graph, self.kdt_root)
        fig, ax = plt.subplots(figsize=(20, 10))  # 比例可以根据树的深度适当调节
        nx.draw_networkx(graph, pos, ax=ax, node_size=500, font_color="w", font_size=15,
                         arrowsize=20)
        plt.tight_layout()
        plt.show()

 五、K近邻算法的测试

test_knn_1.py

import numpy as np
from sklearn.datasets import load_iris, load_breast_cancer
from knn_kdtree import KNearestNeighborKDTree
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, accuracy_score
import matplotlib.pyplot as plt
from sklearn.model_selection import StratifiedKFold
from sklearn.preprocessing import StandardScaler
 
 
iris = load_iris()
X, y = iris.data, iris.target
 
# bc_data = load_breast_cancer()
# X, y = bc_data.data, bc_data.target
X = StandardScaler().fit_transform(X)
 
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=0, stratify=y)
k_neighbors = np.arange(3, 21)
 
# acc = []
# for k in k_neighbors:
#     knn = KNearestNeighborKDTree(k=k)
#     knn.fit(X_train, y_train)
#     y_test_hat = knn.predict(X_test)
#     # print(classification_report(y_test, y_test_hat))
#     acc.append(accuracy_score(y_test, y_test_hat))
 
accuracy_scores = []  # 存储每个alpha阈值下的交叉验证均分
for k in k_neighbors:
    scores = []
    k_fold = StratifiedKFold(n_splits=10).split(X, y)
    for train_idx, test_idx in k_fold:
        # knn = KNearestNeighborKDTree(k=k, p="cos")
        knn = KNearestNeighborKDTree(k=k)
        knn.fit(X[train_idx], y[train_idx])
        y_test_pred = knn.predict(X[test_idx])
        scores.append(accuracy_score(y[test_idx], y_test_pred))
        del knn
    print("k = %d：" % k, np.mean(scores))
    accuracy_scores.append(np.mean(scores))
 
plt.figure(figsize=(7, 5))
plt.plot(k_neighbors, accuracy_scores, "ko-", lw=1)
plt.grid(ls=":")
plt.xlabel("K Neighbors", fontdict={"fontsize": 12})
plt.ylabel("Accuracy Scores", fontdict={"fontsize": 12})
plt.title("KNN(KDTree) Testing Scores under different K Neighbors", fontdict={"fontsize": 14})
plt.show()
 
# knn = KNearestNeighborKDTree(k=3)
# knn.fit(X_train, y_train)
# knn.draw_kd_tree()