Symulowanie-wizualne/sw_lab1.ipynb

from load_data import get_dataset
import numpy as np
from collections import Counter
from tabulate import tabulate
from statistics import mean

Zadanie 1 (4 pkt)

Napisz kod klasy KNearestNeighbor implementującej klasyfikator knn. Należy zimplementować następujące metody:

• konstruktor pobierający listę obrazów treningowych (zgodną zw składową 'values' wczytanego słownika) oraz listę ich etykiet
• metoda l_p_metric(image1, image2, p): zwracająca wartość odległości pomiędzy dwoma obrazami, mierzoną normą typu l_p - parametr p określa 'potęgę' normy
• metoda predict(test_images, k,p): zwracająca listę prognozowanych etykiet dla obrazów testowych (parametr test_images). Paramter drugi określa liczbę przeszukiwanych sąsiadów, natomiast paramter trzeci określa potęgę wybranej metryki.
• metoda accuracy(test_images, k,p) zwracająca dokładność klasyfikatora na zbiorze testowym. Parametr drugi i trzeci są jak w metodzie predict()

class KNearestNeighbor():
    def __init__(self, values, labels):
        self.values = values
        self.labels = labels

    def l_p_metric(self, image1, image2, p):
        return np.sum(np.abs(image1 - image2) ** p) ** (1/p)

    def predict(self, X, K=1, P=1):
        predicted = []

        for image in X:
            metrics = []

            for value in self.values:
                metrics.append(self.l_p_metric(image, value, P))

            mins = sorted(range(len(metrics)), key = lambda sub: metrics[sub])[:K]
            nearest = [self.labels[x] for x in mins]
            pred = max(nearest, key=Counter(nearest).get)

            predicted.append(pred)

        return predicted

    def accuracy(self, expected, predicted):
        return sum(1 for x, y in zip(expected, predicted) if x == y) / len(expected)

Zadanie 2 (2 pkt)

Napisz kod funkcji crossValidation(X, y, n = 10, k=1, p=1): obliczającą algorytm kNN z n-krotną walidacją krzyżową.

def crossValidation(X_train, y_train, X_test, y_test, n=10, k=1, p=1):
    tab = []
    X_folds = []
    y_folds = []
        
    f_size = len(X_train)//n
    index = 0
    
    for i in range(n):
        if i == n-1:
                X_folds.append(X_train[index:])
                y_folds.append(y_train[index:])
                continue
        X_folds.append(X_train[index : index+f_size])
        y_folds.append(y_train[index : index+f_size])
        index = index + f_size
    
    for i in range(n):
        X_train_temp = X_folds[:i] + X_folds[i+1:]
        X_train = np.concatenate((X_train_temp))

        y_train_temp = y_folds[:i] + y_folds[i+1:]
        y_train = np.concatenate((y_train_temp))
        
        X_test = X_folds[i]
        y_test = y_folds[i]

        Knn = KNearestNeighbor(X_train, y_train)
        
        pred = Knn.predict(X_test, k, p)
        a = Knn.accuracy(y_test, pred)
        tab.append(a)
        
        result = mean(tab)
            
    return result

X_train, y_train, X_test, y_test = get_dataset(new_size=64)

kNN  = KNearestNeighbor(X_train, y_train)

Ks = [1, 5, 10]
Ps = [1, 2]

accuracy = [ [
    k, p, 
    kNN.accuracy(y_test, kNN.predict(X_test, K=k, P=p)),
    crossValidation(X_train, y_train, X_test, y_test, n=len(X_train), k=k, p=p)] for k in Ks for p in Ps ]

print(tabulate(accuracy, headers=['K', 'P', 'kNN accuracy', 'CrossValidation accuracy']))

  K    P    kNN accuracy    CrossValidation accuracy
---  ---  --------------  --------------------------
  1    1        0.583012                    0.658228
  1    2        0.552124                    0.617332
  5    1        0.555985                    0.574489
  5    2        0.544402                    0.565725
 10    1        0.501931                    0.523856
 10    2        0.501931                    0.534567

Zadanie 3 (4 pkt)

Napisz kod klasy LogisticRegression implementującej klasyfikator wieloklasowej regresji logistycznej z funkcją softmax() (ze standardowymi nazwami dwóch kluczowych funkcji: fit(), predict()). Zastosuj ten kod do pobranych danych (zbiór walidacyjny losujemy ze zbioru treningowego) - oblicz następujące charakterystyki modelu dla danych walidacyjnych oraz treningowych: dokładność (accuracy), precyzję (precision), czułość(recall) oraz F1 - dla poszczególnych klas oraz globalnie (zob. np. tu).

class LogisticRegression():
    def mapY(self, y, cls):
        m = len(y)
        yBi = np.matrix(np.zeros(m)).reshape(m, 1)
        yBi[y == cls] = 1.
        return yBi

    def indicatorMatrix(self, y):
        classes = np.unique(y.tolist())
        m = len(y)
        k = len(classes)
        Y = np.matrix(np.zeros((m, k)))
        for i, cls in enumerate(classes):
            Y[:, i] = self.mapY(y, cls)
        return Y
    
    # Zapis macierzowy funkcji softmax
    def softmax(self, X):
        return np.exp(X) / np.sum(np.exp(X))
    
    # Funkcja regresji logistcznej
    def h(self, theta, X):
        return 1.0/(1.0 + np.exp(-X * theta))
    
    # Funkcja kosztu dla regresji logistycznej
    def J(self, h, theta, X, y):
        m = len(y)
        h_val = h(theta, X)
        s1 = np.multiply(y, np.log(h_val))
        s2 = np.multiply((1 - y), np.log(1 - h_val))
        return -np.sum(s1 + s2, axis=0) / m

    # Gradient dla regresji logistycznej
    def dJ(self, h, theta, X, y):
        return 1.0 / len(y) * (X.T * (h(theta, X) - y))

    # Metoda gradientu prostego dla regresji logistycznej
    def GD(self, h, fJ, fdJ, theta, X, y, alpha=0.01, eps=10**-3, maxSteps=10000):
        errorCurr = fJ(h, theta, X, y) # fJ -> J, fdJ -> dJ
        errors = [[errorCurr, theta]]
        while True:
            # oblicz nowe theta
            theta = theta - alpha * fdJ(h, theta, X, y)
            # raportuj poziom błędu
            errorCurr, errorPrev = fJ(h, theta, X, y), errorCurr
            # kryteria stopu
            if abs(errorPrev - errorCurr) <= eps:
                break
            if len(errors) > maxSteps:
                break
            errors.append([errorCurr, theta]) 
        return theta, errors

    def trainMaxEnt(self, X, Y):
        n = X.shape[1]
        thetas = []
        for c in range(Y.shape[1]):
            YBi = Y[:,c]
            theta = np.matrix(np.random.random(n)).reshape(n,1)
            # Macierz parametrów theta obliczona dla każdej klasy osobno.
            thetaBest, errors = self.GD(self.h, self.J, self.dJ, theta, 
                                X, YBi, alpha=0.1, eps=10**-4)
            thetas.append(thetaBest)
        return thetas

    def classify(self, thetas, X):
        regs = np.array([(X*theta).item() for theta in thetas])
        probs = self.softmax(regs)
        result = np.argmax(probs)
        return result

    def class_score(self, expected, predicted):
        # accuracy = TP + TN / FP + FN + TP + TN
        accuracy = sum(1 for exp, pred in zip(expected, predicted) if exp == pred) / len(expected)
        # precision = TP / FP + TP
        precision = sum(
            1 for exp, pred in zip(expected, predicted) if exp == 1.0 and pred == 1.0) / sum(
                1 for exp, pred in zip(expected, predicted) if exp == 1.0)
        # recall = TP / FN + TP
        recall = sum(
            1 for exp, pred in zip(expected, predicted) if exp == 1.0 and pred == 1.0) / sum(
                1 for exp, pred in zip(expected, predicted) if pred == 1.0)
        f1 = (2 * precision * recall) / (precision + recall)
        return accuracy, precision, recall, f1

    def fit(self, X_train, y_train):
        Y = self.indicatorMatrix(y_train)
        self.thetas = self.trainMaxEnt(X_train, Y)

    def predict(self, X_test):
        return np.array([self.classify(self.thetas, x) for x in X_test])
    
    def score(self, expected, predicted):
        score = {
            'Class' : [], 
            'Accuracy': [],
            'Precision': [],
            'Recall': [],
            'F1': []}

        oh_expected = self.indicatorMatrix(expected).T.tolist()
        oh_predicted = self.indicatorMatrix(predicted).T.tolist()
        n_classes = len(oh_expected)

        for i in range(n_classes):
            e = oh_expected[i]
            p = oh_predicted[i]
            a, p, r, f1 = self.class_score(e, p)
            score['Class'].append(i)
            score['Accuracy'].append(a)
            score['Precision'].append(p)
            score['Recall'].append(r)
            score['F1'].append(f1)

        score['Class'].append('Global')
        score['Accuracy'].append(sum(1 for exp, pred in zip(expected, predicted) if exp == pred) / len(expected))
        score['Precision'].append(np.mean(score['Precision']))
        score['Recall'].append(np.mean(score['Recall']))
        score['F1'].append(np.mean(score['F1']))

        return score

X_train, y_train, X_test, y_test = get_dataset(new_size=32) 

logreg = LogisticRegression()
logreg.fit(X_train, y_train)

predicted = logreg.predict(X_test)
score = logreg.score(y_test, predicted)

print(tabulate(score, headers='keys'))

/var/folders/7c/v61kq2b95dzbt7s47fxy0grm0000gn/T/ipykernel_2525/2624688826.py:30: RuntimeWarning: divide by zero encountered in log
  s2 = np.multiply((1 - y), np.log(1 - h_val))
/var/folders/7c/v61kq2b95dzbt7s47fxy0grm0000gn/T/ipykernel_2525/2624688826.py:30: RuntimeWarning: invalid value encountered in multiply
  s2 = np.multiply((1 - y), np.log(1 - h_val))
/var/folders/7c/v61kq2b95dzbt7s47fxy0grm0000gn/T/ipykernel_2525/2624688826.py:47: RuntimeWarning: invalid value encountered in subtract
  if abs(errorPrev - errorCurr) <= eps:

Class      Accuracy    Precision    Recall        F1
-------  ----------  -----------  --------  --------
0          0.96139      0.823529  0.976744  0.893617
1          0.857143     0.557692  0.674419  0.610526
2          0.872587     0.788462  0.650794  0.713043
3          0.861004     0.596154  0.673913  0.632653
4          0.776062     0.557692  0.453125  0.5
Global     0.664093     0.664706  0.685799  0.669968

Zadanie 4 (1 pkt)

Oblicz ile danych z poszczególnych klas znajduje się po dodatniej/ujemnej stronie hiperpłaszczyzny klasyfikacyjnej dla danej klasy.

one_hot = logreg.indicatorMatrix(predicted)
length = len(one_hot)
one_hot = one_hot.sum(axis=0).tolist()[0]

hyperplane = [
    [i for i in np.unique(predicted)], 
    [int(x) for x in one_hot],
    [length - int(x) for x in one_hot]]
    

print(tabulate(np.array(hyperplane).T, headers=['Klasa', 'Dodatnia strona', 'Ujemna strona']))

  Klasa    Dodatnia strona    Ujemna strona
-------  -----------------  ---------------
      0                 48              211
      1                 52              207
      2                 65              194
      3                 47              212
      4                 47              212