Exercício 12: Algoritmo K-Means

K-Means

Este caderno segue o conteúdo da aula sobre clustering, que demonstra o algoritmo k-means.

import numpy as np
from matplotlib import pyplot as plt

from sklearn.datasets import make_blobs
np.random.seed(7)

Execute a célula de código a seguir para carregar as bibliotecas necessárias para o trabalho.

A próxima célula contém código que cria um conjunto de dados. Consiste em 100 instâncias com dois atributos numéricos cada. Execute-o para criar o conjunto de dados.

def create_data():
  X, _ = make_blobs(n_samples=100, n_features=2, centers=4, cluster_std=1.5, random_state=6)
  return X

X = create_data()

Podes exibir graficamente o conjunto de dados criado executando a próxima célula. O primeiro atributo é mostrado ao longo do eixo x e o segundo atributo ao longo do eixo y.

plt.xlabel('Attribute 1')
plt.ylabel('Attribute 2')

plt.scatter(X[:, 0], X[:, 1])
plt.show()

A próxima célula contém configurações que serão usadas doravante: o número de clusters 'k' e suas cores.

k = 4
cluster_colors = ['orange', 'yellow', 'purple', 'green']

A função 'calculate_distance' calcula a distância euclidiana entre dois pontos de um plano. Ela será usada posteriormente para calcular a distância entre centróides e instâncias.

def calculate_distance(x1, x2):
  return np.sqrt((x1[0]-x2[0])**2 + (x1[1]-x2[1])**2)

A função 'generate_centroids' gera instâncias aleatórias 'k' necessárias para a inicialização do algoritmo k-means.

def generate_centroids(X, k):
    N = X.shape[0]
    indices = np.random.randint(low=0, high=N, size=k)
    return X[indices]
centroids = generate_centroids(X, 4)

A função 'show_centroids' permite ver onde os centróides estão localizados em relação aos clusters.

def show_centroids(X, centroids, cluster_colors=cluster_colors):
  plt.xlabel('Attribute 1')
  plt.ylabel('Attribute 2')

  plt.scatter(X[:, 0], X[:, 1])

  for i, centroid in enumerate(centroids):
    plt.scatter(centroid[0], centroid[1], color=cluster_colors[i], marker='*')

  plt.show()
show_centroids(X, centroids)

A função 'divide_data' divide instâncias em clusters. Para cada caso, ela primeiro calcula as distâncias para os centróides. Em seguida, ela identifica o centróide mais próximo da instância (a distância para esse centróide é a menor) e atribui a instância ao cluster determinado por esse centróide. Para distinguir agrupamentos, usaremos os números 0, 1, 2, ..., k-1.

def divide_data(X, centroids, k):

  # initialize the list of cluster labels
  cluster_labels = []

  # iterate through the dataset instance by instance
  for x in X:

    # initialize the list of distances to centroids
    distances_to_centroids = []

    # then for each centroid ...
    for centroid in centroids:
      # ... calculate the distance between the instance and the centroid
      d = calculate_distance(x, centroid)

      # ... and add it to the list of distances
      distances_to_centroids.append(d)

    # when we have visited all centroids,
    # choose the centroid closest to the instance x
    label = np.argmin(distances_to_centroids)

    # conclude that the instance belongs to the cluster
    # determined by that centroid
    cluster_labels.append(label)

  # the result of the function is an array of cluster labels
  return np.array(cluster_labels)

cluster_labels = divide_data(X, centroids, k)

A próxima célula contém uma função que calcula o número de instâncias por cluster. Podes executá-la e ver a distribuição numérica dos clusters.

def show_number_of_instances_per_cluster(k, cluster_labels, cluster_colors=cluster_colors):
  plt.bar(np.arange(0, k), np.bincount(cluster_labels), color=cluster_colors)
  plt.xticks(np.arange(0, k), np.arange(0, k))
  plt.show()

show_number_of_instances_per_cluster(k, cluster_labels)

A função 'show_clusters' mostra a divisão dos dados por cluster. Os centróides do aglomerado são mostrados como estrelas negras devido à visibilidade.

def show_clusters(X, centroids, cluster_labels, cluster_colors=cluster_colors):
  plt.xlabel('Attribute 1')
  plt.ylabel('Attribute 2')

  for i, x in enumerate(X):
    instance_color = cluster_colors[cluster_labels[i]]
    plt.scatter(x[0], x[1], color=instance_color)

  for centroid in centroids:
    plt.scatter(centroid[0], centroid[1], color='black', marker='*')

  plt.show()

show_clusters(X, centroids, cluster_labels)

A função 'calculate_new_centroids' atualiza os centróides do cluster. Ele faz isso fazendo a média dos valores de todas as instâncias que pertencem a um cluster e declarando o valor resultante como o novo centróide.

def calculate_new_centroids(X, cluster_labels, k):

  # initialize the list of new centroids
  new_centroids = []

  # for each cluster
  for i in range(0, k):

    # ... extract the instances that belong to it
    instance_indices = cluster_labels == i
    instances_in_cluster = X[instance_indices]

    # then calculate the new centroid value
    # by averaging all instances in the cluster
    new_centroid = np.average(instances_in_cluster, axis=0)

    # add the calculated new centroid to the list of all centroids
    new_centroids.append(new_centroid)

  # the result of the function is an array of new centroids
  return np.array(new_centroids)

new_centroids = calculate_new_centroids(X, np.array(cluster_labels), k)

Para verificar onde os novos centróides estão localizados, chamaremos a função 'show_centroids'.

show_centroids(X, new_centroids)

A função 'execute_clustering' combina todas as etapas pelas quais passamos individualmente:

* gera centróides iniciais

* em iterações, divide instâncias em clusters e, em seguida, calcula novos centróides.

def execute_clustering(X, k, epsilon=1e-4, max_iterations=300):

  # step of initializing centroids
  centroids = generate_centroids(X, k)

  # in each iteration of the loop
  for i in range(0, max_iterations):

    # step 1: dividing instances into clusters
    cluster_labels = divide_data(X, centroids, k)

    # step 2: calculating new centroids
    new_centroids = calculate_new_centroids(X, cluster_labels, k)

    # checking stopping criteria
    # if they are met, we stop the algorithm
    if np.linalg.norm(new_centroids - centroids) < epsilon:
      break
    # otherwise, we move to the next iteration
    centroids = new_centroids.copy()

  # the result of the function is the final cluster labels and centroid values
  return cluster_labels, new_centroids

final_cluster_labels, final_centroids = execute_clustering(X, k)

Podemos primeiro verificar a distribuição numérica das instâncias nos clusters finais.

show_number_of_instances_per_cluster(k, final_cluster_labels)

Agora vamos mostrar os clusters finais.

show_clusters(X, final_centroids, final_cluster_labels)

O bloco de código a seguir é usado para exibir a animação de todas as etapas de divisão do conjunto de dados em clusters.

from IPython.display import display, clear_output
def show_animation(X, k):
  fig = plt.figure()
  ax = fig.add_subplot(1, 1, 1)

  # initialization
  num_iterations = 300
  epsilon = 1e-4

  # initial centroid values
  centroids = generate_centroids(X, k)

  # individual iterations
  for iteration in range(0, num_iterations):
    cluster_labels = divide_data(X, centroids, k)

    # display clusters
    ax.cla()
    ax.set_title('Iteration number: ' + str(iteration))

    for i, x in enumerate(X):
      instance_color = cluster_colors[cluster_labels[i]]
      ax.scatter(x[0], x[1], color=instance_color)

    for centroid in centroids:
      ax.scatter(centroid[0], centroid[1], color='black', marker='*')

    display(fig)
    clear_output(wait=True)
    # plt.pause(0.5)

    # calculate centroids for the next iteration
    new_centroids = calculate_new_centroids(X, cluster_labels, k)
    if np.linalg.norm(new_centroids - centroids) < epsilon:
      break

    centroids = new_centroids.copy()

show_animation(X, k)