机器学习之 K-近邻算法

摘要：近邻算法通过测量不同特征值之间的距离方法进行分类。对于近邻算法来说，它是一个特殊的没有模型的算法，但是我们将其训练数据集看作是模型。算法优缺点近邻算法是一个比较简单的算法，有其优点但也有缺点。

k-近邻算法通过测量不同特征值之间的距离方法进行分类。

对于一个存在标签的训练样本集，输入没有标签的新数据后，将新数据的每个特征与样本集中数据对应的特征进行比较，根据算法选择样本数据集中前k个最相似的数据，选择k个最相似数据中出现次数最多的分类，作为新数据的分类。

这里只是对单个新数据的预测，对同时多个新数据的预测放在后文中。

假定存在训练样本集 X_train（X_train.shape=(10, 2)），对应的标记 y_train（y_train.shape=(10,)，包含0、1)，使用 matplotlib.pyplot 作图表示如下（绿色的点表示标记0，红色的点表示标记1）：

现有一个新的数据：x（x = np.array([3.18557125, 6.03119673])），作图表示如下（蓝色的点）：

首先，使用欧拉距离公式计算 x 到 X_train 中每个样本的距离：

import math

distances = [math.sqrt(np.sum((x_train - x) ** 2)) for x_train in X_train]

第二，对 distances 进行升序操作，使用 np.argsort() 方法返回排序后的索引，而不会对原数据的顺序有任何影响：

import numpy as np

nearest = np.argsort(distances)

第三，取 k 个距离最近的样本对应的标记：

topK_y = [y_train[i] for i in nearest[:k]]

最后，对这 k 个距离最近的样本对应的标记进行统计，找出占比最多标记即为 x 的预测分类，此例的预测分类为0：

from collections import Counter

votes = Counter(topK_y)
votes.most_common(1)[0][0]

将上面的代码封装到一个方法中：

import numpy as np
import math

from collections import Counter


def kNN(k, X_train, y_train, x):
    distances = [math.sqrt(np.sum((x_train - x) ** 2)) for x_train in X_train]
    nearest = np.argsort(distances)

    topK_y = [y_train[i] for i in nearest[:k]]
    votes = Counter(topK_y)
    return votes.most_common(1)[0][0]

一个典型的机器学习算法流程是将训练数据集通过机器学习算法训练（fit）出模型，通过这个模型来预测输入样例的结果。

对于 k-近邻算法来说，它是一个特殊的没有模型的算法，但是我们将其训练数据集看作是模型。Scikit Learn 中就是怎么处理的。

Scikit Learn 中 k-邻近算法在 neighbors 模块中，初始化时传入参数 n_neighbors 为 6，即为上面的 k：

from sklearn.neighbors import KNeighborsClassifier

kNN_classifier = KNeighborsClassifier(n_neighbors=6)

fit() 方法根据训练数据集“训练”分类器，该方法会返回分类器本身：

kNN_classifier.fit(X_train, y_train)

predict() 方法预测输入的结果，该方法要求传入的参数类型为矩阵。因此，这里先对 x 进行 reshape 操作：

X_predict = x.reshape(1, -1)
y_predict = kNN_classifier.predict(X_predict)

y_predict 值为0，与前面实现的 kNN 方法结果一致。

定义一个 KNNClassifier 类，其构造器方法传入参数 k，表示预测时选取的最相似数据的个数：

class KNNClassifier:
    def __init__(self, k):
        self.k = k
        self._X_train = None
        self._y_train = None

fit() 方法训练分类器，并且返回分类器本身：

def fit(self, X_train, y_train):
    self._X_train = X_train
    self._y_train = y_train
    return self

predict() 方法对待测数据集进行预测，参数 X_predict 类型为矩阵。该方法使用列表解析式对 X_predict 进行了遍历，对每个待测数据调用了一次 _predict() 方法。

def predict(self, X_predict):
    y_predict = [self._predict(x) for x in X_predict]
    return np.array(y_predict)

def _predict(self, x):
    distances = [math.sqrt(np.sum((x_train - x) ** 2))
                 for x_train in self._X_train]
    nearest = np.argsort(distances)

    topK_y = [self._y_train[i] for i in nearest[:self.k]]
    votes = Counter(topK_y)

    return votes.most_common(1)[0][0]

上面通过训练样本集训练出了模型，但是并不知道这个模型的好坏，还存在两个问题。

如果模型很坏，预测的结果就不是我们想要的。同时实际情况中，很难拿到真实的标记（label），无法对模型进行检验。

训练模型时训练样本没有包含所有的标记。

对于第一个问题，通常将样本集中一定比例（如20%）的数据作为测试数据，其余数据作为训练数据。

以 Scikit Learn 中提供的鸢尾花数据为例，其包含了150个样本。

import numpy as np
from sklearn import datasets

iris = datasets.load_iris()
X = iris.data
y = iris.target

现在将样本分为20%示例测试数据和80%比例训练数据：

test_ratio = 0.2
test_size = int(len(X) * test_ratio)

X_train = X[test_size:]
y_train = y[test_size:]

X_test = X[:test_size]
y_test = y[:test_size]

将 X_train 和 y_train 作为训练数据用于训练模型，X_test 和 y_test 作为测试数据验证模型准确性。

对于第二个问题，还是以 Scikit Learn 中提供的鸢尾花数据为例，其标记 y 的内容为：

array([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
       0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
       0, 0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1,
       1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1,
       1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2,
       2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2,
       2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2])

发现0、1、2是以顺序存储的，在将样本划分为训练数据和测试数据过程中，如果训练数据中才对标记只包含0、1，这样的训练数据对于模型的训练将是致命的。以此，应将样本数据先进行随机处理。

np.random.permutation() 方法传入一个整数 n，会返回一个区间在 [0, n) 且随机排序的一维数组。将 X 的长度作为参数传入，返回 X 索引的随机数组：

shuffle_indexes = np.random.permutation(len(X))

将随机化的索引数组分为训练数据的索引与测试数据的索引两部分：

test_ratio = 0.2
test_size = int(len(X) * test_ratio)

test_indexes = shuffle_indexes[:test_size]
train_indexes = shuffle_indexes[test_size:]

再通过两部分的索引将样本数据分为训练数据和测试数据：

X_train = X[train_indexes]
y_train = y[train_indexes]

X_test = X[test_indexes]
y_test = y[test_indexes]

可以将两个问题的解决方案封装到一个方法中，seed 表示随机数种子，作用在 np.random 中：

import numpy as np

def train_test_split(X, y, test_ratio=0.2, seed=None):
    if seed:
        np.random.seed(seed)
    shuffle_indexes = np.random.permutation(len(X))

    test_size = int(len(X) * test_ratio)
    test_indexes = shuffle_indexes[:test_size]
    train_indexes = shuffle_indexes[test_size:]

    X_train = X[train_indexes]
    y_train = y[train_indexes]

    X_test = X[test_indexes]
    y_test = y[test_indexes]

    return X_train, X_test, y_train, y_test

Scikit Learn 中封装了 train_test_split() 方法，放在了 model_selection 模块中：

from sklearn.model_selection import train_test_split

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

通过 train_test_split() 方法对样本数据进行了预处理后，开始训练模型，并且对测试数据进行验证：

from sklearn.neighbors import KNeighborsClassifier

kNN_classifier = KNeighborsClassifier(n_neighbors=6)
kNN_classifier.fit(X_train, y_train)
y_predict = kNN_classifier.predict(X_test)

y_predict 是对测试数据 X_test 的预测结果，其中与 y_test 相等的个数除以 y_test 的个数就是该模型的正确率，将其和 y_test 进行比较可以算出模型的正确率：

def accuracy_score(y_true, y_predict):
    return sum(y_predict == y_true) / len(y_true)

调用该方法，返回一个小于等于1的浮点数：

accuracy_score(y_test, y_predict)

同样在 Scikit Learn 的 metrics 模块中封装了 accuracy_score() 方法：

from sklearn.metrics import accuracy_score

accuracy_score(y_test, y_predict)

Scikit Learn 中的 KNeighborsClassifier 类的父类 ClassifierMixin 中有一个 score() 方法，里面就调用了 accuracy_score() 方法，将测试数据 X_test 和 y_test 作为参数传入该方法中，可以直接计算出算法正确率。

class ClassifierMixin(object):
    def score(self, X, y, sample_weight=None):
        from .metrics import accuracy_score
        return accuracy_score(y, self.predict(X), sample_weight=sample_weight)

前文中提到的 k 是一种超参数，超参数是在算法运行前需要决定的参数。 Scikit Learn 中 k-近邻算法包含了许多超参数，在初始化构造函数中都有指定：

def __init__(self, n_neighbors=5,
             weights="uniform", algorithm="auto", leaf_size=30,
             p=2, metric="minkowski", metric_params=None, n_jobs=None,
             **kwargs):
    # code here

这些超参数的含义在源代码和官方文档[scikit-learn.org]中都有说明。

k-近邻算法是一个比较简单的算法，有其优点但也有缺点。

优点是思想简单，但效果强大， 天然的适合多分类问题。

缺点是效率低下，比如一个训练集有 m 个样本，n 个特征，则预测一个新的数据的算法复杂度为 O(m*n)；同时该算法可能产生维数灾难，当维数很大时，两个点之间的距离可能也很大，如 (0,0,0,...,0) 和 (1,1,1,...,1)（10000维）之间的距离为100。

Github | ML-Algorithms-Action