KNN from scratch

Table of Contents

Algorithm and implementation
#

KNN is a so-called lazy learner because it does not have a training step and it is immediately ready to predict, using euclidean distance to find the k points that are closest to the point we are predicting onto, and outputting a prediction based on these k points (either using the most common y class among them (classification) or averaging those closest points’ value of y (regression) ).

Class structure
#

main attributes:
- the number of closest points k
fit method where we just ingest the training data
predict method
_predict method that performs the prediction for a single data point value and include the distance computation used to rank the closest points

import numpy as np

class KNN:
	def __init__(self, k=3):
		self.k = k
		
	def fit(self, X, y):
		self.X_train = X
		self.y_train = y
		
	def predict(self, X):
		predictions = [self._predict(x) for x in X]
		return predictions
		
	def _predict(self, x):
		pass

Inference
#

The _predict method is the only one requiring proper implementation and performing:

computation of the distance between the point we predict onto and the training points
selection of the k closest points

def _predict(self, x):
	#compute the distances
	distances = [euclidean_distance(x, x_train) for x_train in self.X_train]
	
	#get the closest k
	k_indices = np.argsort(distances)[:self.k]
	k_nearest_labels = [self.y_train[i] for i in k_indices]
	
	#majority vote (classification)
	unique_labels, counts = np.unique(k_nearest_labels,return_counts=True)
	predicted_label = unique_labels[np.argmax(counts)]
	
	return predicted_label

Calculating euclidean distance
#

This global function will be used to calculate the distance between sample points and centroids.

def euclidean_distance(x1, x2):
	return np.sqrt(np.sum((x1 - x2)**2))

Features scaling
#

To better use the euclidean distance, standardization or normalization of the sample are recommended.

Efficient version with vector broadcasting
#

The version implemented above contains a nested loop (predict loops over the queries and _predict further loops over every training loop), and can be improved significantly in terms of efficiency by using broadcasting with numpy.

The first thing to do is to consider the shapes of the different datasets:

training points X_train : (N, D)
training lables y_train : (N, )
query points X_query : (M, D)
predictions (M, )
computation of pairwise squared distances : (M, N)
nearest indices/labels : (M, k)

The essential improvement is about the vectorized version of the euclidean distance function:

def euclidean_distance_vectorized(X, Z):    
    # Find the distance from each point 
    
    squaredX = np.matmul(X, X.T)
    diagX = np.diag(squaredX)
    XX = np.tile(diagX, (Z.shape[0], 1))
    XX_transposed = XX.T
    
    ZZ= np.tile(np.diag(np.matmul(Z, Z.T)), (X.shape[0],1))
    
    XZ=np.matmul(X, Z.T)
    
    D= np.sqrt(XX_transposed-2*XZ+ZZ)
    
    return D

Vectorized predict method
#

The vectorized implementation does not need the _predict method, introduce the vectorized euclidean distance, and also removes the

import numpy as np

class KNN:
	def __init__(self, k=3):
		self.k = k
		
	def fit(self, X, y):
		self.X_train = np.asarray(X)
		self.y_train = np.asarray(y)
		
	def predict(self, X):
        X_query = np.asarray(X, dtype=float)
        if X_query.ndim == 1:
	        X_query = X_query.reshape(1, -1)
	        
        squared_distances = euclidean_distance_vectorized(self.X_train, X_query)
        
        sorted_training_indices = np.argsort(squared_distances, axis=0)
        
        k_indices = sorted_training_indices[:self.k, :]
        
        predictions = []
        for query_idx in range(X_query.shape[0]):
	        neighbor_indices = k_indices[:, query_idx]
	        neighbor_labels = self.y_train[neighbor_indices]
	        unique_labels, counts = np.unique(neighbor_labels,return_counts=True)
	        predicted_label = unique_labels[np.argmax(counts)]
	        predictions.append(predicted_label)
	
		return np.asarray(predictions)

The last loop can also be removed for a vectorized implementation:

		...
		
		labels = self.y_train[k_indices] # (k, n_queries)
		classes = np.unique(labels)
		class_counts = (labels[:, None, :] == classes[None, :,None]).sum(axis=0) # (n_classes, n_queries)
		predictions = classes[np.argmax(class_counts, axis=0)]

		return np.asarray(predictions)

Algorithm and implementation#

Class structure#

Inference#

Calculating euclidean distance#

Features scaling#

Efficient version with vector broadcasting#

Vectorized predict method#