一、原理
1. 概述
K近鄰法（k-nearest neighbors,KNN）是一種有監督的學習算法，也是機器學習中最簡單、且不那么依靠各類假設的算法（基本上所有算法都會有假設的前提條件，在數據分布符合算法的假設條件時，其效果往往會更好）。
1.1 核心思想
物以類聚，人以群分。俗話說，“看一個男人好不好，就看他身邊的朋友絕對沒錯”，對我們要學習和預測的樣本來說，道理也是一樣的。我們要判斷一個樣本屬於什么類別，可以通過圍在他身邊的樣本來判斷，在特征空間內與這個樣本距離最近的 k 個樣本應該與待預測樣本是一伙的，所以如果這 k 個樣本大多屬於類別A，那么認為待預測樣本也屬於類別A。
1.2 用途
KNN可以用於分類（二分類多分類都可以，且算法不需要作修改），也可以用於回歸，用於分類還是回歸主要在於算法輸出結果的計算方式，分類問題多是通過對待預測樣本附近的k個樣本對所屬類別進行投票，根據投票結果來決定待分類樣本的類別，即多數表決法（少數服從多數）；回歸問題是對待預測樣本最近的k個樣本的輸出進行平均，均值作為待預測樣本的輸出，即平均法。
1.3 KNN優缺點
優點：
- 對數據的分布沒有假設，可以適用於各種分布形式的數據集，不過一般來說密集一些的數據更好，太稀疏的數據更難控制 k 的取值，易被誤導；
- 算法思想簡單，容易理解和實現，而且可以分類也可以回歸；
缺點：
- KNN基本沒有根據數據訓練和學習的過程，每次分類都要跑一次整個分類過程，速度慢；
- 對樣本不均衡情況比較敏感，容易被樣本量大的類別干擾（可以考慮使用距離來加權，增大距離最近的數據的影響）；
- 暴力實現計算量大，KD樹等實現耗內存。
2. 算法流程及實現
2.1 KNN算法流程（以KNN分類算法為例）
輸入：訓練集\(T=\{(x_{1}, y_{1}),(x_{2}, y_{2}),...(x_{n}, y_{n})\}\)，其中\(x\)為樣本的特征向量，為樣本的類別
輸出：樣本 x 所屬的類別 y
（1）選定距離度量方法，在訓練集T中找出與待預測樣本 x 最接近的 k個樣本點，包含這 k個點的 x 的鄰域記作；
（2）在中根據分類決策規則（如多數表決法）來決定 x 所屬的類別 y。
由上述KNN算法流程可以發現，在運算中存在四個需要關注的重點：
- 距離度量方式的選擇；
- k值的選取；
- 怎樣找出這k個近鄰的樣本點；
- 分類決策規則的選擇。
其中度量方式、k值、分類決策規則稱為KNN算法的三要素，會影響算法的效果，非常重要；而怎樣找出這k個近鄰的樣本點則直接決定了KNN算法的實現后的具體計算過程，影響算法的復雜度。接下來就詳細分析這四個要點：
（1）距離度量方式：一般來說距離度量方式主要包括歐式距離、曼哈頓距離、閔可夫斯基距離，歐式距離是我們比較常用的，這是比較基礎的內容，本文不再贅述。
（2）k值的選取：k值的選取沒有很好的辦法，一般才用交叉驗證來選擇合適的k值，比如使用gridsearchcv工具。k值的選擇會對預測的結果造成比較大的影響，k值越小，模型越復雜（k越大模型越簡單，比如k＝n，所有待預測樣本都被划分為同一類了，模型就很簡單），這時模型的偏差bias減小，方差variance增大，模型容易過擬合，所以對k值的選取要慎重。
（3）分類決策規則的選擇，分類問題大多才用多數表決法。
（4）怎樣找出這k個近鄰的樣本點，常用的方法有三種：
a. 暴力實現，即直接遍歷得到訓練集中所有樣本點與預測點的距離，然后排序取前k個最近鄰點，所以其計算的時間復雜度為O(n)；
b. KD樹，使用暴力實現沒有利用到數據本身蘊含的結構信息，在樣本量較大時效率太低，計算復雜度高，在實際工程應用中，對成百上千個特征幾百萬的樣本量計算比較困難，因此可以通過索引樹，對搜索空間進行層次划分，以此來優化計算，其時間復雜度為O(logn)，詳細內容在后面會講到；
c. 球數，球樹和KD樹類似，但不會像KD樹一樣做一些多余的計算，主要區別在於KD樹得到的是節點樣本組成的超矩形體，而球樹得到的是節點樣本組成的最小超球體，這個超球體要比對應的KD樹的超矩形體小，這樣在做最近鄰搜索的時候，可以避免一些無謂的搜索。
接下來，對基於KD樹實現的 KNN 進行詳述，為什么寫KD樹不寫球樹呢？一是兩者差不多，二是KD樹看起來跟凱文杜蘭特好像有什么莫名其妙的聯系。
2.2 KD樹
KD樹（K-dimension tree），即K個特征維度的樹，這個樹的功能就是可以幫我們快速找到跟目標點相近的數據點，想想這個功能，是不是很像我們玩的猜數字的游戲，一個人偷偷想一個數字（目標數據點），其他人不停的猜數字縮小范圍，這不就跟我們為目標點找最相近的 k 個點是一樣一樣的嗎？
所以，再想想我們玩猜數字常用的套路：二分法，這也是KD樹的核心思想：分而治之。因此，KD樹是一個二叉樹，他的每個節點會將所有的數字點在某個維度上一分為二，設想一下，如果數據空間在各個維度上被我們划分成了很多個小的part，那么我們要確定一個新的數據點相近的點，只要找到這個點所在的part不就八九不離十了。
不過這個樹應該在哪個維度上、怎么分呢？當然是選擇能把數字分的最開的維度來開刀（比如使用方差大的，數據方差大說明沿該坐標軸方向上數據點分散的比較開，這個方向上，進行數據分割可以獲得最好的分辨率）。這是不是跟決策樹的思路很像？當然，決策樹是可以直接用於分類的，特征選擇更為復雜（比如用信息增益比），而且每個特征只會用到一次，並且並不要求特征是數值類型的，不過其基本思想還是很像的，好了不說決策樹了，來看看怎么實現一個KD樹。
2.2.1 KD樹的構建
先舉個常見的簡單例子：假設有六個二維數據點\(\{(2,3),(5,4),(9,6),(4,7),(8,1),(7,2)\}\)，數據點位於二維空間中。根據上面的介紹，我們需要根據這些數據點把數據空間划分成很多個小的part。事實上，六個二維數據點生成的Kd-樹的圖，即划分結果為：
這是怎么做到的呢？
- 找到划分的特征( split )。6個數據點在 x，y 維度上的數據方差分別為6.97，5.37，所以在x軸上方差更大，用第1維特征建樹；
- 確定划分點（7,2）( Node-Data )。根據 x 維上的值將數據排序，6個數據的中值(所謂中值，即中間大小的值)為6，所以划分點的數據是（7,2）（選（5,4）也可以）。這樣，該節點的分割超平面就是通過（7,2）並垂直於：划分點維度的直線 x=7；
- 確定左子空間 ( left ) 和右子空間 ( right )。分割超平面 x=7 將整個空間分為兩部分：x<=7 的部分為左子空間，包含3個節點 {(2,3),(5,4),(4,7)}；另一部分為右子空間，包含2個節點={(9,6)，(8,1)}；
- 迭代。用同樣的辦法划分左子樹的節點{(2,3),(5,4),(4,7)}和右子樹的節點{(9,6)，(8,1)}。
我們可以看出，如下圖：點 (7,2) 為根結點，(5,4) 和 (9,6) 則為根結點的左右子結點，而 (2,3)，(4,7) 則為 (5,4) 的左右子，最后，(8,1) 為 (9,6) 的左子。如此，便形成了這樣一棵k-d樹。
從以上過程和結果的描述，我們可以總結出KD樹的基本數據結構和構建流程：
k-d樹的數據結構：
Kd-tree構建流程的偽代碼：
輸入：數據點集DataSet，和其所在的空間
輸出：Kd，類型為Kd-tree1 if DataSet is null , return null；
2 else 調用Node-Data生成程序：
a 計算 Split。對於所有數據（高維向量），統計他們在每個維度上的方差，方差最大值所對應的維度就是 Split 域的值；
b 計算 Node-Data。數據點集 DataSet 按照第 split 維的值排序，最靠近中位數的那個數據點被選為 Node-Data；
3 dataleft = { d 屬於 DataSet & d[:split] <= Node-data[:split] }；
Left-Range = { Range && dataleft }；
dataright = {d 屬於 DataSet & d[:split] > Node-data[:split] }；
Right-Range = { Range && dataright }；
4 left = 由（dataleft, Left-Range）建立的Kd-tree；
設置 left 為 KD樹的Left；
right =由（dataright,Right-Range）建立的Kd-tree；
設置 right 為KD樹的Right；
5 在left 和 right 上重復上面的過程，直到數據全部分完。
根據以上偽代碼就可以寫出構建KD樹的程序，Python實現如下：
Kd-tree構建流程的Python代碼：
class KDNode(object):
def __init__(self, node_data, split, left, right):
self.node_data = node_data
self.split = split
self.left = left
self.right = right
class KDTree(object):
def __init__(self, dataset):
self.dim = len(dataset[0])
self.tree = self.generate_kdtree(dataset)
def generate_kdtree(self, dataset):
'''
遞歸生成一棵樹
'''
if not dataset:
return None
else:
split, data_node = self.cal_node_data(dataset)
left = [d for d in dataset if d[split] < data_node[split]]
right = [d for d in dataset if d[split] > data_node[split]]
return KDNode(data_node, split, self.generate_kdtree(left), self.generate_kdtree(right))
def cal_node_data(self, dataset):
std_lst = [] # 用標准差代替方差，都一樣
for i in range(self.dim):
std_i = np.std([d[i] for d in dataset])
std_lst.append(std_i)
split = std_lst.index(max(std_lst))
dataset.sort(key=lambda x: x[split])
indx=int((len(dataset) + 2) / 2)-1
data_node = dataset[indx]
return split, data_node
2.2.2 使用KD樹得到 k 近鄰
（1）使用KD樹得到最近鄰
1）順着二叉樹搜索，直到找到葉子節點中的最近鄰節點，這個過程會確定一條搜索路經；
2）因為二叉樹每次划分數據空間都是以某一個特征為標准進行的，所以綜合考慮所有特征，根據二叉樹搜索到的子節點不一定是最近鄰的，不過最近鄰點肯定位於以查詢點為圓心且通過葉子節點的圓域內（不然就會比這個葉子節點離得遠了），所以根據搜索路徑進行回溯操作，找到最近鄰點。
舉兩個例子，在之前建立的Kd樹中搜索（3, 4.5）的最近鄰點：
- 二叉樹搜索查找（3, 4.5），得到搜索路經 <(7,2) - (5,4) - (4,7)>；
- 首先以（4,7）作為當前最近鄰點，計算其到查詢點（2.1,3.1）的距離為2.69；
- 然后回溯到其父節點（5,4），並判斷在該父節點的其他子節點空間中是否有距離查詢點更近的數據點。以（3, 4.5）為圓心，以2.69為半徑畫圓，如下圖所示。發現該圓和超平面y = 4交割，因此計算（5,4）與（3, 4.5）的距離為2.06，小於2.69，更新最近鄰為（5,4），更新距離為2.06，畫一個綠色的圓；
- 因為畫的圓進入到了（5,4）分割的另一部分區域，所以需要在這個區域查找。發現（2,3）結點與目標點距離為1.8，比（5,4）更近，更新最近鄰為（2，3），以（3, 4.5）為圓心畫一個藍色的圓。
- 再回溯到（7,2），藍色的圓與x = 7超平面不相交，因此不用進入（7,2）右子空間進行查找。至此，搜索路徑中的節點已經全部回溯完，結束整個搜索，返回最近鄰點（2,3），最近距離為1.8。
這是找到最近鄰的過程，只找到了最近的一個點，那么怎么找 k 近鄰個點呢？
（2）使用KD樹得到 k 近鄰
李航老師在《統計學習方法》中只提了一句，按上面的思路就能找到 k 近鄰，我們就改造一下上面的思路：上面是找1個，現在要找k個，所以我們可以維護一個長度為k的有序數據集合，在搜索過程中發現更小的距離，就替換掉集合里的最大值，這樣搜索結束后就得到了 k 近鄰，有序數據集合我們用優先隊列來實現似乎非常的合理，ok，下面來試試看：
class NodeDis(object):
'''
自定義將節點與其和目標距離綁定的類，並自定義根據距離比較對象的大小，因為要從大的開始pop，所以self.distance > other.distance
'''
def __init__(self, node, dis):
self.node = node
self.distance = dis
def __lt__(self, other):
return self.distance > other.distance
def search_k_neighbour(kdtree, target, k):
k_queue = PriorityQueue()
return search_path(kdtree, target, k, k_queue)
def search_path(kdtree, target, k, k_queue):
'''
遞歸找到整個樹中的k近鄰
'''
if kdtree is None:
return NodeDis([], np.inf)
path = []
while kdtree:
if target[kdtree.split] <= kdtree.node_data[kdtree.split]:
path.append((kdtree.node_data, kdtree.split, kdtree.right))
kdtree = kdtree.left
else:
path.append((kdtree.node_data, kdtree.split, kdtree.left))
kdtree = kdtree.right
path.reverse()
radius = np.inf
for i in path:
node_data = i[0]
split = i[1]
opposite_tree = i[2]
# 先判斷圈定的區域與分割軸是否相交
distance_axis = abs(node_data[split] - target[split])
if distance_axis > radius:
break
else:
distance = cal_Euclidean_dis(node_data, target)
k_queue.put(NodeDis(node_data, distance))
if k_queue.qsize() > k:
k_queue.get()
radius = k_queue.queue[-1].distance
# print(radius,[i.distance for i in k_queue.queue])
search_path(opposite_tree, target, k, k_queue)
return k_queue
def cal_Euclidean_dis(point1, point2):
return np.sqrt(np.sum((np.array(point1) - np.array(point2)) ** 2))
對例子中的數據進行測試，輸出（3, 4.5）的3個近鄰點結果如下圖：
2.3 基於暴力搜索與KD樹的KNN分類
根據暴力搜索和上一節中使用KD樹得到k近鄰的方法，我們來實現一個KNN分類器：
KNN的python實現
import numpy as np
from collections import Counter
import kd_tree # 上一節實現的KD樹
class KNNClassfier(object):
def __init__(self, k, distance='Euclidean',kdtree=True):
'''
初始化確定距離度量方式和k、是否使用KD樹
'''
self.distance = distance
self.k = k
self.kdtree=kdtree
def get_k_neighb(self, new_point, train_data, labels):
'''
暴力搜索得到k近鄰的k個label
'''
distance_lst = [(self.cal_Euclidean_dis(new_point, point), label) for point, label in zip(train_data, labels)]
distance_lst.sort(key=lambda x: x[0], reverse=False)
k_labels = [i[1] for i in distance_lst[:self.k]]
return k_labels
def fit(self, train_data):
'''
使用KD樹的話，可以事先將樹訓練好，好像有了點學習的過程一樣，要提高效率的話可以將構建好的樹保存起來
'''
kdtree = kd_tree.KDTree(train_data)
return kdtree
def get_k_neighb_kdtree(self, new_point, train_data, labels, kdtree):
'''
通過KD樹搜索得到k近鄰的k個label
'''
result = kd_tree.search_k_neighbour(kdtree.tree, new_point, self.k)
data_dict={data:label for data,label in zip(train_data, labels)}
k_labels=[data_dict[data.node] for data in result.queue]
return k_labels
def predict(self, new_point, train_data, labels,kdtree):
if self.kdtree:
k_labels = self.get_k_neighb_kdtree(new_point, train_data, labels,kdtree)
else:
k_labels = self.get_k_neighb(new_point, train_data, labels)
# print(k_labels)
return self.decision_rule(k_labels)
def decision_rule(self, k_labels):
'''
分類決策規則：投票
'''
label_count = Counter(k_labels)
new_label = None
max = 0
for label, count in label_count.items():
if count > max:
new_label = label
max = count
return new_label
def cal_Euclidean_dis(self, point1, point2):
return np.sqrt(np.sum((np.array(point1) - np.array(point2)) ** 2))
分類結果如下，兩種方式實現的KNN的結果是一樣的（圖中的兩個正方形點為待分類點）：
