K最近傍法を授業で実装したので、覚えておきたい点を備忘録としてまとめます。
K最近傍法は機械学習の教師あり学習手法の一つで、異常検知や顔認証などによく使われます。教師ありである理由は、周囲のデータを教師に、分類先を知りたいデータの分類を行うためです。
K最近傍法自体の解説は以下の記事がとても分かりやすいです。
k近傍法（knn）をわかりやすく Python を用いて基本から実装まで解説
データセット
データセットは、CIFAR10を用います。CIFAR10は60000枚の32×32のカラー画像で構成される、10クラスの画像のデータセットです。つまり、各クラスに6000枚の画像があります。
データセットについてはまず、訓練データセットと、テストデータセットの2つに分けます。訓練データセットは、最近傍の点を見つけるために利用されます。また、訓練データセットは、のちに20%を検証データセットとして扱います。検証データセットとは、ハイパーパラメータを調整するためのデータセットです。ハイパーパラメータとは、人間が調整しなければならないパラメータです。今回のK最近傍法だと、Kがハイパーパラメータです。様々なKでK最近傍法の交差検証を実装し、最も分類精度の平均値が高かったKを、最後のテストデータの分類時に使用します。
K最近傍法の実装
訓練データセットといっていますが、K最近傍法では、学習は行いません。K最近傍法は以下のような手順で実行されます。
①テストデータのあるデータに注目し、その点とすべての訓練データの点との距離をまず算出します。距離の算出には、ユークリッド距離（L2距離）を使用します。
※L1距離のことをマンハッタン距離という。
②そしてKの値をもとに、分類先を決めます。
実行環境
安定のGoogle Colab様です。
仕様ライブラリ
numpy, matplotlib, pickle, osを用います。
ソースコード解説
まずは下準備です。
import numpy as np
import matplotlib.pyplot as plt
import pickle
import os
%matplotlib inline
# helper functions for data loading
def load_CIFAR_batch(filename):
""" load single batch of cifar """
with open(filename, 'rb') as f:
datadict = pickle.load(f, encoding='latin1')
X = datadict['data']
Y = datadict['labels']
X = X.reshape(10000, 3, 32, 32).transpose(0,2,3,1).astype("float")
Y = np.array(Y)
return X, Y
def load_CIFAR10(ROOT):
""" load all of cifar """
xs = []
ys = []
for b in range(1,6):
f = os.path.join(ROOT, 'data_batch_%d' % (b, ))
X, Y = load_CIFAR_batch(f)
xs.append(X)
ys.append(Y)
Xtr = np.concatenate(xs)
Ytr = np.concatenate(ys)
del X, Y
Xte, Yte = load_CIFAR_batch(os.path.join(ROOT, 'test_batch'))
return Xtr, Ytr, Xte, Yte
CIFAR-10 datasetをダウンロードします。
import urllib.request
import tarfile
url = "https://www.cs.toronto.edu/~kriz/cifar-10-python.tar.gz"
urllib.request.urlretrieve(url, "cifar-10-python.tar.gz")
tar = tarfile.open("cifar-10-python.tar.gz", "r:gz")
tar.extractall()
tar.close()
次に、生データをロードし、それらを訓練&テストデータセットとして、読みます。もう一度言いますが、K最近傍法では、学習は行いません。訓練データセットとは、すでにクラスタが分かっているデータのことです。テストデータセットが、これから分類しようとしている画像たちです。
cifar10_dir = './cifar-10-batches-py'
X_train, y_train, X_test, y_test = load_CIFAR10(cifar10_dir)
# Checking the size of the training and testing data
print('Training data shape: ', X_train.shape)
print('Training labels shape: ', y_train.shape)
print('Test data shape: ', X_test.shape)
print('Test labels shape: ', y_test.shape)
Training data shape: (50000, 32, 32, 3)
Training labels shape: (50000,)
Test data shape: (10000, 32, 32, 3)
Test labels shape: (10000,)
のような表示がされるはずです。
テストデータには、既にラベルがついています。テストデータを訓練データを基に分類して、後で精度を出すために、ラベルがついています。
次に、データセットのサンプル画像を表示します。
classes = ['plane', 'car', 'bird', 'cat', 'deer', 'dog', 'frog', 'horse', 'ship', 'truck']
num_classes = len(classes)
samples_per_class = 7
for y, cls in enumerate(classes):
idxs = np.flatnonzero(y_train == y)
idxs = np.random.choice(idxs, samples_per_class, replace=False)
for i, idx in enumerate(idxs):
plt_idx = i * num_classes + y + 1
plt.subplot(samples_per_class, num_classes, plt_idx)
plt.imshow(X_train[idx].astype('uint8'))
plt.axis('off')
if i == 0:
plt.title(cls)
plt.show()
70枚の画像が表示されるはずです。
次に、50000枚の訓練データと、10000枚のテストデータのそれぞれ最初の10000枚と、1000枚をサンプリングします。これはコメントアウトでもあるように、メモリーエラーを防ぐためです。一気にCIFAR-10データセット全体を読み込むと、メモリが不足する可能性があるようです。
# Memory error prevention by subsampling data
num_training = 10000
mask = list(range(num_training))
X_train = X_train[mask]
y_train = y_train[mask]
num_test = 1000
mask = list(range(num_test))
X_test = X_test[mask]
y_test = y_test[mask]
次に、X_train, X_testの形を4次元配列から、2次元配列に変更します。
# reshaping data and placing into rows
X_train = np.reshape(X_train, (X_train.shape[0], -1))
X_test = np.reshape(X_test, (X_test.shape[0], -1))
print(X_train.shape, X_test.shape)
実行結果は、
(10000, 3072) (1000, 3072)
になるはずです。
32 * 32 * 3 = 3072ですね。
ついに、K-NNを実行します。ここが肝要な部分ですので、コメントアウトで解説したいと思います。
class KNearestNeighbor(object):
def __init__(self):
pass
def train(self, X, y):
self.X_train = X
self.y_train = y
def predict(self, X, k=1, num_loops=0):
dists = self.compute_distances(X)
return self.predict_labels(dists, k=k)
# k近傍法ではまず、対象の試験データと周りの訓練データ全てとの距離を計算します
def compute_distances(self, X):
num_test = X.shape[0]
num_train = self.X_train.shape[0]
# 距離を保持する配列distsを用意します。形状は(1000, 10000)です。各テストデータの点（1000個）から、訓練データ（10000個）までの距離を格納します。
dists = np.zeros((num_test, num_train))
for i in range(num_test):
for j in range(num_train):
dists[i,j] = np.sum((X[i] - self.X_train[j])**2)
return dists
# 最初はk=1で行います。
def predict_labels(self, dists, k=1):
num_test = dists.shape[0]
# 各テストデータ（1000点）のラベルを格納します。サイズは(1000,)です。
y_pred = np.zeros(num_test)
for i in range(num_test):
# 各テストデータから最も近い、訓練データのクラスを格納する配列です。
closest_y = []
sorted_dist = np.argsort(dists[i])
closest_y = list(self.y_train[sorted_dist[0:k]])
y_pred[i]= (np.argmax(np.bincount(closest_y)))
return y_pred
classifier = KNearestNeighbor()
classifier.train(X_train, y_train)
dists= classifier.compute_distances(X_test)
y_test_pred = classifier.predict_labels(dists, k=1)
num_correct = np.sum(y_test_pred == y_test)
accuracy = float(num_correct) / num_test
print('Got %d / %d correct => accuracy: %f' % (num_correct, num_test, accuracy))
Got 283 / 1000 correct => accuracy: 0.283000
という結果を得ました。1000個のテストデータのうち、283個を正しいクラスに分類できたということですね。
先ほどはk=1で実行しましたが、当然k=1が最適とは限りません。k_choices = [1, 3, 5, 10, 15, 50]として、最も性能の良いものを選択します。これがハイパーパラメータ（人間が調整するパラメータ）の調整ということですね。
ここでは、交差検証を行い、各kに対する精度を確認します。交差検証を行う理由は、精度にばらつきが生じないようにするためです。米国データサイエンティストのかめさんの記事がとても交差検証についてわかりやすく解説されていたため、画像を引用します。
画像内では、緑の部分がテストデータと記されていますが、本記事では検証（validation）データセットのことです。
引用：k-Fold Cross Validation(交差検証)を解説する【機械学習入門9】
num_folds = 5
k_choices = [1, 3, 5, 10, 15, 50]
X_train_folds = []
y_train_folds = []
X_train_folds = np.array_split(X_train,num_folds)
y_train_folds = np.array_split(y_train,num_folds)
k_to_accuracies = {}
# 各kについて交差検証
for k in k_choices:
k_to_accuracies[k] = []
# 交差検証開始（num_folds=5なので、5回回します）
for num_knn in range(0,num_folds):
# validation setを作ります。訓練データを5セットに分割し、1セットをの検証データとして扱います。
X_val = X_train_folds[num_knn]
y_val = y_train_folds[num_knn]
num_val = X_val.shape[0]
# 10000個あった、訓練データのうち、8000個が訓練データになります。2000個は検証データ
# concatenate関数は複数のndarrayを結合して新たなndarrayを作るための関数
# np.deleteの第3引数は削除対象の軸（次元）を意味する。0は行、1は列を表す。
# つまり、X_train_newという配列から、num_knn行目を削除するということ。
# （num_knn行目は検証データセット）
X_train_new = np.concatenate(np.delete(X_train_folds, num_knn, 0))
y_train_new = np.concatenate(np.delete(y_train_folds, num_knn, 0))
# run KNN on new training set and new valiation set
classifier = KNearestNeighbor()
classifier.train(X_train_new, y_train_new)
dists = classifier.compute_distances(X_val)
y_val_pred = classifier.predict_labels(dists, k)
num_correct = np.sum(y_val_pred == y_val)
accuracy = float(num_correct) / num_val
print('K = %d, fold %d, Got %d / %d correct => accuracy: %f' % (k, num_knn, num_correct, num_val, accuracy))
k_to_accuracies[k].append(accuracy)
以下実行結果です。
K = 1, fold 0, Got 577 / 2000 correct => accuracy: 0.288500
K = 1, fold 1, Got 568 / 2000 correct => accuracy: 0.284000
K = 1, fold 2, Got 565 / 2000 correct => accuracy: 0.282500
K = 1, fold 3, Got 549 / 2000 correct => accuracy: 0.274500
K = 1, fold 4, Got 554 / 2000 correct => accuracy: 0.277000
K = 3, fold 0, Got 575 / 2000 correct => accuracy: 0.287500
K = 3, fold 1, Got 548 / 2000 correct => accuracy: 0.274000
K = 3, fold 2, Got 557 / 2000 correct => accuracy: 0.278500
K = 3, fold 3, Got 535 / 2000 correct => accuracy: 0.267500
K = 3, fold 4, Got 531 / 2000 correct => accuracy: 0.265500
K = 5, fold 0, Got 589 / 2000 correct => accuracy: 0.294500
K = 5, fold 1, Got 568 / 2000 correct => accuracy: 0.284000
K = 5, fold 2, Got 595 / 2000 correct => accuracy: 0.297500
K = 5, fold 3, Got 550 / 2000 correct => accuracy: 0.275000
K = 5, fold 4, Got 557 / 2000 correct => accuracy: 0.278500
K = 10, fold 0, Got 605 / 2000 correct => accuracy: 0.302500
K = 10, fold 1, Got 574 / 2000 correct => accuracy: 0.287000
K = 10, fold 2, Got 568 / 2000 correct => accuracy: 0.284000
K = 10, fold 3, Got 532 / 2000 correct => accuracy: 0.266000
K = 10, fold 4, Got 570 / 2000 correct => accuracy: 0.285000
K = 15, fold 0, Got 588 / 2000 correct => accuracy: 0.294000
K = 15, fold 1, Got 594 / 2000 correct => accuracy: 0.297000
K = 15, fold 2, Got 550 / 2000 correct => accuracy: 0.275000
K = 15, fold 3, Got 548 / 2000 correct => accuracy: 0.274000
K = 15, fold 4, Got 552 / 2000 correct => accuracy: 0.276000
K = 50, fold 0, Got 547 / 2000 correct => accuracy: 0.273500
K = 50, fold 1, Got 578 / 2000 correct => accuracy: 0.289000
K = 50, fold 2, Got 568 / 2000 correct => accuracy: 0.284000
K = 50, fold 3, Got 512 / 2000 correct => accuracy: 0.256000
K = 50, fold 4, Got 524 / 2000 correct => accuracy: 0.262000
可視化のために、各kの値に対する性能をプロットします。
for k in k_choices:
accuracies = k_to_accuracies[k]
plt.scatter([k] * len(accuracies), accuracies)
# plot the trend line with error bars that correspond to standard deviation
accuracies_mean = np.array([np.mean(v) for k,v in sorted(k_to_accuracies.items())])
accuracies_std = np.array([np.std(v) for k,v in sorted(k_to_accuracies.items())])
plt.errorbar(k_choices, accuracies_mean, yerr=accuracies_std)
plt.title('Cross-validation on k')
plt.xlabel('k')
plt.ylabel('Cross-validation accuracy')
plt.show()
k=5で最も精度の平均値が高いことが分かります。
最後に最も性能の平均値が良かったk=5を最終的なkの値として、テストデータ（X_test, y_testのこと。1000個のデータです）で性能を確認します。
# Choosing best value of k based on cross-validation results
# put the best K you found and report test accuracy
best_k = 5
classifier = KNearestNeighbor()
classifier.train(X_train, y_train)
y_test_pred = classifier.predict(X_test, k=best_k)
# Computing and displaying the accuracy for best k found during cross-validation
num_correct = np.sum(y_test_pred == y_test)
accuracy = float(num_correct) / num_test
print('Got %d / %d correct => accuracy: %f' % (num_correct, num_test, accuracy))
結果は
Got 296 / 1000 correct => accuracy: 0.296000
となりました。
感想
意外と精度良くないんですね。kを変えたら結構精度変わると思ったのですが、kを変えてもあまり変わらなかったですね。それでも、30%近いのは良い方なのでしょうか。次は、教師無し学習（k-means, k-modes）を見ていきます。