K-means from scratch

Algorithm and implementation
#

K-means algorithm assigns labels to data points by iteratively updating centroids and clusters’ labels and using euclidean distance to calculate their proximity until convergence (or until a maximum number of iterations is reached).

Class structure
#

main attributes:
- the number of clusters K
- n_iterations
- clusters, which is the list of sample indices for each cluster initialized in the __init__
- centroid which is the center (or mean vector) for each cluster
predict method It is worth to note that there’s no fit method and we can do the predict straight away.

import numpy as np

class KMeans:
	def __init__(self, K=5, n_iterations=100):
		self.K = K
		self.n_iterations = n_iterations
		
		#list of sample indices for each cluster
		self.clusters = [[] for _ in range(self.K)]
		
		#mean vector for each cluster
		self.centroids = []
		
	def predict(self, X):
		pass

Inference
#

The predict method consists of the following steps:

ingesting the data points
initializing the centroids by assigning them to K randomly picked sample points
performing the optimization loop to create the clusters up to a maximum number of iterations or up to convergence (a sort of early stopping is implemented)
assign the label (or the respective cluster’s index) to the data points

It uses the following helper functions:

_create_clusters
_closest_centroids
_get_centroids
_is_converged
_get_cluster_labels

def predict(self, X):
	self.X = X
	self.n_samples, self.n_features = X.shape
	
	#initialize centroids drawing K indices from the full sample
	#and assigning the centroids to those points.
	#replace=False ensures that the same index is not drawn more than once
	random_samples_idxs = np.random.choice(self.n_samples, self.K, replace=False)
	self.centroids = [self.X[idx] for idx in random_samples_idxs]
	
	#optimize clusters
	for _ in range(self.n_iterations):
		#assign samples to the closest centroid (create clusters)
		self.clusters = self._create_clusters(self.centroids)
		
		#calculate new centroids from the clusters
		centroids_old = self.centroids
		self.centroids = self._get_centroids(self.clusters)
		
		if self._is_converged(centroids_old, self.centroids):
			break
			
	#classify the samples as the index of their cluster
	return self._get_cluster_labels(self.clusters)

Calculating euclidean distance
#

This 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))

Creating clusters
#

This method is used to assign the samples to the closest centroids.
It uses another helper function named _closest_centroid (defined at the next step).

def _create_clusters(self, centroids):
	clusters = [[] for _ in range(self.K)]
	for idx, sample in enumerate(self.X):
		centroid_idx = self._closest_centroid(sample, centroids)
		clusters[centroid_idx].append(idx)
	
	return clusters

Defining closest centroids
#

This function calculate the distance of the current sample to each centroid using the euclidean_distance function, and gets the closest.

def _closest_centroids(self, sample, centroids):
	distances = [euclidean_distance(sample, point) for point in centroids]
	closest_idx = np.argmin(distances)
	
	return closest_idx

Getting centroids
#

This method is used to assign the mean value of the clusters to the centroids.

def _get_centroids(self, clusters):
	centroids = np.zeros((self.K, self.n_features))
	for cluster_idx, cluster for enumerate(clusters):
		cluster_mean = np.mean(self.X[cluster], axis=0)
		centroids[cluster_idx] = cluster_mean
	return centroids

Defining convergence
#

This checks the distances between old and new centroids, for all centroids and returns True if their distances sum up to zero.

def _is_converged(self, centroids_old, centroids):
	distances = [euclidean_distance(centroids_old[i], centroids[i] for i in range(self.K))]
	return sum(distances) == 0

Assigning the labels
#

Each sample will get the label of the cluster it was assigned to.

def _get_cluster_labels(self, clusters):
	labels = np.empty(self.n_samples)
	
	for cluster_idx, cluster in enumerate(clusters):
		for sample_idx in cluster:
			labels[sample_idx] = cluster_idx
	
	return labels

Features scaling
#

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

Algorithm and implementation#

Class structure#

Inference#

Calculating euclidean distance#

Creating clusters#

Defining closest centroids#

Getting centroids#

Defining convergence#

Assigning the labels#

Features scaling#