ML000/networks.py

import numpy as np
import matplotlib.pyplot as plt

# input_size = 5 * 3
# hidden_size = 64
# output_size = 10
# model = networks.ImprovedNeuralNetwork(input_size, hidden_size1, hidden_size2, output_size)
class SimpleNeuralNetwork:
    def __init__(self, input_size, hidden_size, output_size):
        self.input_size = input_size
        self.hidden_size = hidden_size
        self.output_size = output_size

        # # Инициализация весов
        # self.W1 = np.random.randn(input_size, hidden_size) * 0.01
        # self.b1 = np.zeros((1, hidden_size))
        # self.W2 = np.random.randn(hidden_size, output_size) * 0.01
        # self.b2 = np.zeros((1, output_size))

        # Инициализация весов (Xavier initialization)
        self.W1 = np.random.randn(input_size, hidden_size) * np.sqrt(2.0 / input_size)
        self.b1 = np.zeros((1, hidden_size))
        self.W2 = np.random.randn(hidden_size, output_size) * np.sqrt(2.0 / hidden_size)
        self.b2 = np.zeros((1, output_size))

    def relu(self, Z):
        return np.maximum(0, Z)

    def relu_derivative(self, Z):
        return Z > 0

    def forward(self, X):
        # Входной слой к скрытому слою
        # Входные данные умножаются на матрицу весов (м/у входными и скрытым) и добавляется вектор смещения скрытого слоя
        # \[
        # \mathbf{Z}_1 = \mathbf{X} \mathbf{W}_1 + \mathbf{b}_1
        # \]
        # numpy.dot(a, b, out=None) # Скалярное произведение
        # a: Первый входной массив.
        # b: Второй входной массив.
        # out: Выходной массив, в который будет записан результат. Если не указан, результат будет возвращен как новый массив.
        self.Z1 = np.dot(X, self.W1) + self.b1
        # применяется функция активации ReLU
        # \[
        # \mathbf{A}_1 = \text{ReLU}(\mathbf{Z}_1) = \max(0, \mathbf{Z}_1)
        # \]
        self.A1 = self.relu(self.Z1) # self.A1 = np.tanh(self.Z1)
        # Скрытый слой к выходному слою
        # активация (предыдущий этап) умножается на матрицу весов (м/у скрытым и выходным) и добавляется вектор смещения выходного слоя
        self.Z2 = np.dot(self.A1, self.W2) + self.b2
        # применяется функция активации softmax
        # \[
        # \mathbf{A}_2 = \text{softmax}(\mathbf{Z}_2) = \frac{\exp(\mathbf{Z}_2)}{\sum \exp(\mathbf{Z}_2)}
        # \]
        # numpy.sum(a, axis=None, dtype=None, out=None, keepdims=<no value>, initial=<no value>, where=<no value>)
        #     a: Входной массив или объект, который может быть преобразован в массив.
        #     axis: Ось или оси по которым вычисляется сумма. Если axis равно None (по умолчанию), сумма вычисляется по всем элементам массива.
        #     dtype: Тип данных результата. Если не указан, тип данных результата будет таким же, как и тип данных входного массива.
        #     out: Выходной массив, в который будет записан результат. Если не указан, результат будет возвращен как новый массив.
        #     keepdims: Если True, размерность результата будет такой же, как и размерность входного массива, но с размером 1 по указанным осям. По умолчанию False.
        #     initial: Начальное значение для суммирования. Если не указано, начальное значение будет 0.
        #     where: Маска, определяющая, какие элементы массива будут включены в сумму. Если не указано, все элементы массива будут включены.

        self.A2 = np.exp(self.Z2) / np.sum(np.exp(self.Z2), axis=1, keepdims=True)
        return self.A2

    # Кросс-энтропийная функция потерь для задачи классификации определяется следующим образом:
    #
    # \[ L(\mathbf{Y}, \mathbf{\hat{Y}}) = -\frac{1}{N} \sum_{i=1}^{N} \sum_{c=1}^{C} y_{i,c} \log(\hat{y}_{i,c}) \]
    #
    # где:
    # - \( N \) — количество примеров в наборе данных.
    # - \( C \) — количество классов.
    # - \( \mathbf{Y} \) — матрица истинных меток размером \( N \times C \), где \( y_{i,c} \) равно 1, если пример \( i \) принадлежит классу \( c \), и 0 в противном случае.
    # - \( \mathbf{\hat{Y}} \) — матрица предсказанных вероятностей размером \( N \times C \), где \( \hat{y}_{i,c} \) — предсказанная вероятность того, что пример \( i \) принадлежит классу \( c \).
    #
    # Кросс-энтропийная функция потерь основана на теории информации и измеряет количество информации, необходимой для передачи сообщения. В контексте классификации, она измеряет разницу между предсказанными вероятностями и истинными метками.
    #
    # ### Преимущества
    #
    # 1. **Интерпретируемость**: Кросс-энтропийная функция потерь имеет четкую интерпретацию в терминах теории информации.
    # 2. **Дифференцируемость**: Она является дифференцируемой функцией, что позволяет использовать градиентный спуск для оптимизации.
    # 3. **Эффективность**: Она эффективно работает с вероятностными предсказаниями, что делает её подходящей для задач классификации.
    #
    # ### Недостатки
    #
    # 1. **Чувствительность к плохим предсказаниям**: Кросс-энтропийная функция потерь может быть чувствительна к плохим предсказаниям, особенно если предсказанная вероятность близка к 0 или 1.
    # 2. **Необходимость нормализации**: Предсказанные вероятности должны быть нормализованы, чтобы их сумма была равна 1. Это обычно достигается с помощью функции активации softmax.
    def compute_loss(self, Y, Y_hat):
        # Используется кросс-энтропийная функция потерь
        # \[
        # L(\mathbf{Y}, \mathbf{Y}_{\text{hat}}) = -\frac{1}{m} \sum_{i=1}^{m} \log(\mathbf{Y}_{\text{hat}}[i, \mathbf{Y}[i]])
        # \]
        m = Y.shape[0]
        logprobs = np.log(Y_hat[range(m), Y])
        loss = -np.sum(logprobs) / m
        return loss

    def backward(self, X, Y, Y_hat):
        m = X.shape[0]
        # Выходной слой
        # Градиенты функции потерь
        # \[
        # \mathbf{dZ}_2 = \mathbf{Y}_{\text{hat}} - \mathbf{Y}
        # \]
        dZ2 = Y_hat - np.eye(self.output_size)[Y]
        # Градиенты весов
        # \[
        # \mathbf{dW}_2 = \frac{1}{m} \mathbf{A}_1^T \mathbf{dZ}_2
        # \]
        dW2 = (1 / m) * np.dot(self.A1.T, dZ2)
        # Градиенты смещений
        # \[
        # \mathbf{db}_2 = \frac{1}{m} \sum \mathbf{dZ}_2
        # \]
        db2 = (1 / m) * np.sum(dZ2, axis=0, keepdims=True)

        # Скрытый слой
        # Градиенты функции потерь
        # \[
        # \mathbf{dA}_1 = \mathbf{dZ}_2 \mathbf{W}_2^T
        # \]
        dA1 = np.dot(dZ2, self.W2.T)
        # Градиенты функции потерь
        # \[
        # \mathbf{dZ}_1 = \mathbf{dA}_1 \cdot \text{ReLU}'(\mathbf{Z}_1)
        # \]
        dZ1 = dA1 * self.relu_derivative(self.Z1) # dZ1 = dA1 * (1 - np.power(self.A1, 2))
        # Градиенты весов
        # \[
        # \mathbf{dW}_1 = \frac{1}{m} \mathbf{X}^T \mathbf{dZ}_1
        # \]
        dW1 = (1 / m) * np.dot(X.T, dZ1)
        # Градиенты смещений
        # \[
        # \mathbf{db}_1 = \frac{1}{m} \sum \mathbf{dZ}_1
        # \]
        db1 = (1 / m) * np.sum(dZ1, axis=0, keepdims=True)

        return dW1, db1, dW2, db2

    def update_parameters(self, dW1, db1, dW2, db2, learning_rate):
        # \[
        # \mathbf{W}_1 := \mathbf{W}_1 - \alpha \mathbf{dW}_1
        # \]
        self.W1 -= learning_rate * dW1
        # \[
        # \mathbf{b}_1 := \mathbf{b}_1 - \alpha \mathbf{db}_1
        # \]
        self.b1 -= learning_rate * db1
        # \[
        # \mathbf{W}_2 := \mathbf{W}_2 - \alpha \mathbf{dW}_2
        # \]
        self.W2 -= learning_rate * dW2
        # \[
        # \mathbf{b}_2 := \mathbf{b}_2 - \alpha \mathbf{db}_2
        # \]
        self.b2 -= learning_rate * db2

    def train(self, X, Y, learning_rate=0.01, epochs=1000):
        self.losses = []
        self.accuracies = []
        for epoch in range(epochs):
            Y_hat = self.forward(X)
            loss = self.compute_loss(Y, Y_hat)
            dW1, db1, dW2, db2 = self.backward(X, Y, Y_hat)
            self.update_parameters(dW1, db1, dW2, db2, learning_rate)
            self.losses.append(loss)
            self.accuracies.append(self.accuracy(Y, np.argmax(Y_hat, axis=1)))
            if epoch % 100 == 0:
                print(f'Epoch {epoch}, Loss: {loss}, Accuracy: {self.accuracies[-1]}')

    def predict(self, X):
        Y_hat = self.forward(X)
        return np.argmax(Y_hat, axis=1)

    def accuracy(self, Y_true, Y_pred):
        return np.mean(Y_true == Y_pred)

    def plot_metrics(self):
        plt.figure(figsize=(12, 5))

        plt.subplot(1, 2, 1)
        plt.plot(self.losses, label='Loss')
        plt.xlabel('Epoch')
        plt.ylabel('Loss')
        plt.title('Loss over Epochs')
        plt.legend()

        plt.subplot(1, 2, 2)
        plt.plot(self.accuracies, label='Accuracy')
        plt.xlabel('Epoch')
        plt.ylabel('Accuracy')
        plt.title('Accuracy over Epochs')
        plt.legend()

        plt.tight_layout()
        plt.show()

    def evaluate(self, X_test, Y_test):
        Y_pred = self.predict(X_test)
        accuracy = self.accuracy(Y_test, Y_pred)
        print("Accuracy:", accuracy)

# input_size = 5 * 3
# hidden_size1 = 128
# hidden_size2 = 64
# output_size = 10
# ТОДО: Многослойная нейронная сеть
class ImprovedNeuralNetwork:
    def __init__(self, input_size, hidden_size1, hidden_size2, output_size):
        self.input_size = input_size
        self.hidden_size1 = hidden_size1
        self.hidden_size2 = hidden_size2
        self.output_size = output_size

        # Инициализация весов
        self.W1 = np.random.randn(input_size, hidden_size1) * 0.01
        self.b1 = np.zeros((1, hidden_size1))
        self.W2 = np.random.randn(hidden_size1, hidden_size2) * 0.01
        self.b2 = np.zeros((1, hidden_size2))
        self.W3 = np.random.randn(hidden_size2, output_size) * 0.01
        self.b3 = np.zeros((1, output_size))

    def relu(self, Z):
        return np.maximum(0, Z)

    def relu_derivative(self, Z):
        return Z > 0

    def forward(self, X):
        self.Z1 = np.dot(X, self.W1) + self.b1
        self.A1 = np.tanh(self.Z1) # self.relu(self.Z1)
        self.Z2 = np.dot(self.A1, self.W2) + self.b2
        self.A2 = np.tanh(self.Z2) # self.relu(self.Z2)
        self.Z3 = np.dot(self.A2, self.W3) + self.b3
        self.A3 = np.exp(self.Z3) / np.sum(np.exp(self.Z3), axis=1, keepdims=True)
        return self.A3

    def compute_loss(self, Y, Y_hat):
        m = Y.shape[0]
        logprobs = np.log(Y_hat[range(m), Y])
        loss = -np.sum(logprobs) / m
        return loss

    def backward(self, X, Y, Y_hat):
        m = X.shape[0]

        dZ3 = Y_hat - np.eye(self.output_size)[Y]
        dW3 = (1 / m) * np.dot(self.A2.T, dZ3)
        db3 = (1 / m) * np.sum(dZ3, axis=0, keepdims=True)

        dA2 = np.dot(dZ3, self.W3.T)
        dZ2 = dA2 * (1 - np.power(self.A2, 2)) # dA2 * self.relu_derivative(self.Z2)
        dW2 = (1 / m) * np.dot(self.A1.T, dZ2)
        db2 = (1 / m) * np.sum(dZ2, axis=0, keepdims=True)

        dA1 = np.dot(dZ2, self.W2.T)
        dZ1 = dA1 * (1 - np.power(self.A1, 2)) # dA1 * self.relu_derivative(self.Z1)
        dW1 = (1 / m) * np.dot(X.T, dZ1)
        db1 = (1 / m) * np.sum(dZ1, axis=0, keepdims=True)

        return dW1, db1, dW2, db2, dW3, db3

    def update_parameters(self, dW1, db1, dW2, db2, dW3, db3, learning_rate):
        self.W1 -= learning_rate * dW1
        self.b1 -= learning_rate * db1
        self.W2 -= learning_rate * dW2
        self.b2 -= learning_rate * db2
        self.W3 -= learning_rate * dW3
        self.b3 -= learning_rate * db3

    def train(self, X, Y, learning_rate=0.01, epochs=1000):
        for epoch in range(epochs):
            Y_hat = self.forward(X)
            loss = self.compute_loss(Y, Y_hat)
            dW1, db1, dW2, db2, dW3, db3 = self.backward(X, Y, Y_hat)
            self.update_parameters(dW1, db1, dW2, db2, dW3, db3, learning_rate)
            if epoch % 100 == 0:
                print(f'Epoch {epoch}, Loss: {loss}')

    def predict(self, X):
        Y_hat = self.forward(X)
        return np.argmax(Y_hat, axis=1)

# ТОДО: CNN
class SimpleCNeuralNetwork:
    def __init__(self, input_size, hidden_size, output_size, filter_size=3, num_filters=32):
        self.input_size = input_size
        self.hidden_size = hidden_size
        self.output_size = output_size
        self.filter_size = filter_size
        self.num_filters = num_filters

        # Инициализация весов для сверточного слоя
        self.filters = np.random.randn(num_filters, filter_size, filter_size) * np.sqrt(2.0 / (filter_size * filter_size))
        self.conv_bias = np.zeros((num_filters,))

        # Инициализация весов для полносвязного слоя
        self.W1 = np.random.randn(input_size * input_size * num_filters, hidden_size) * np.sqrt(2.0 / (input_size * input_size * num_filters))
        self.b1 = np.zeros((1, hidden_size))
        self.W2 = np.random.randn(hidden_size, output_size) * np.sqrt(2.0 / hidden_size)
        self.b2 = np.zeros((1, output_size))

    def relu(self, Z):
        return np.maximum(0, Z)

    def relu_derivative(self, Z):
        return Z > 0

    def conv2d(self, X, filters, bias, stride=1):
        batch_size, height, width = X.shape # 1000, 5, 3
        filter_size = filters.shape[1] # 3
        num_filters = filters.shape[0] # 32

        # Размер выходного изображения
        output_height = (height - filter_size) // stride + 1 # 3
        output_width = (width - filter_size) // stride + 1 # 1

        # Инициализация выходного массива
        output = np.zeros((batch_size, output_height, output_width, num_filters)) # (1000, 3, 1, 32)

        # Применение свертки
        for i in range(output_height):
            for j in range(output_width):
                for f in range(num_filters):
                    region = X[:, i * stride:i * stride + filter_size, j * stride:j * stride + filter_size]
                    output[:, i, j, f] = np.sum(region * filters[f], axis=(1, 2)) + bias[f]

        return output

    def forward(self, X):
        # Сверточный слой
        self.Z1 = self.conv2d(X, self.filters, self.conv_bias)
        self.A1 = self.relu(self.Z1)

        # Сплющивание
        self.A1_flat = self.A1.reshape(self.A1.shape[0], -1)

        # Полносвязный слой
        # ТОДО: Разобраться с размерностями матриц
        self.Z2 = np.dot(self.A1_flat, self.W1) + self.b1
        self.A2 = self.relu(self.Z2)

        # Выходной слой
        self.Z3 = np.dot(self.A2, self.W2) + self.b2
        self.A3 = np.exp(self.Z3) / np.sum(np.exp(self.Z3), axis=1, keepdims=True)
        return self.A3

    def compute_loss(self, Y, Y_hat):
        m = Y.shape[0]
        logprobs = np.log(Y_hat[range(m), Y])
        loss = -np.sum(logprobs) / m
        return loss

    def backward(self, X, Y, Y_hat):
        m = X.shape[0]

        # Градиенты выходного слоя
        dZ3 = Y_hat - np.eye(self.output_size)[Y]
        dW2 = (1 / m) * np.dot(self.A2.T, dZ3)
        db2 = (1 / m) * np.sum(dZ3, axis=0, keepdims=True)

        # Градиенты полносвязного слоя
        dA2 = np.dot(dZ3, self.W2.T)
        dZ2 = dA2 * self.relu_derivative(self.Z2)
        dW1 = (1 / m) * np.dot(self.A1_flat.T, dZ2)
        db1 = (1 / m) * np.sum(dZ2, axis=0, keepdims=True)

        # Градиенты сверточного слоя
        dA1 = dZ2.reshape(self.A1.shape) * self.relu_derivative(self.Z1)
        dZ1 = np.zeros_like(self.Z1)
        for i in range(m):
            for f in range(self.num_filters):
                for h in range(self.Z1.shape[1]):
                    for w in range(self.Z1.shape[2]):
                        dZ1[i, h, w, f] = np.sum(dA1[i, h, w, f] * self.filters[f])

        dfilters = np.zeros_like(self.filters)
        for i in range(m):
            for f in range(self.num_filters):
                for h in range(self.Z1.shape[1]):
                    for w in range(self.Z1.shape[2]):
                        region = X[i, h:h+self.filter_size, w:w+self.filter_size, :]
                        dfilters[f] += dA1[i, h, w, f] * region

        dconv_bias = np.sum(dA1, axis=(0, 1, 2))

        return dW1, db1, dW2, db2, dfilters, dconv_bias

    def update_parameters(self, dW1, db1, dW2, db2, dfilters, dconv_bias, learning_rate):
        self.W1 -= learning_rate * dW1
        self.b1 -= learning_rate * db1
        self.W2 -= learning_rate * dW2
        self.b2 -= learning_rate * db2
        self.filters -= learning_rate * dfilters
        self.conv_bias -= learning_rate * dconv_bias

    def train(self, X, Y, learning_rate=0.01, epochs=1000):
        self.losses = []
        self.accuracies = []
        for epoch in range(epochs):
            Y_hat = self.forward(X)
            loss = self.compute_loss(Y, Y_hat)
            dW1, db1, dW2, db2, dfilters, dconv_bias = self.backward(X, Y, Y_hat)
            self.update_parameters(dW1, db1, dW2, db2, dfilters, dconv_bias, learning_rate)
            self.losses.append(loss)
            self.accuracies.append(self.accuracy(Y, np.argmax(Y_hat, axis=1)))
            if epoch % 100 == 0:
                print(f'Epoch {epoch}, Loss: {loss}, Accuracy: {self.accuracies[-1]}')

    def predict(self, X):
        Y_hat = self.forward(X)
        return np.argmax(Y_hat, axis=1)

    def accuracy(self, Y_true, Y_pred):
        return np.mean(Y_true == Y_pred)

    def confusion_matrix(self, Y_true, Y_pred, num_classes):
        cm = np.zeros((num_classes, num_classes), dtype=int)
        for true, pred in zip(Y_true, Y_pred):
            cm[true, pred] += 1
        return cm

    def plot_metrics(self):
        plt.figure(figsize=(12, 5))

        plt.subplot(1, 2, 1)
        plt.plot(self.losses, label='Loss')
        plt.xlabel('Epoch')
        plt.ylabel('Loss')
        plt.title('Loss over Epochs')
        plt.legend()

        plt.subplot(1, 2, 2)
        plt.plot(self.accuracies, label='Accuracy')
        plt.xlabel('Epoch')
        plt.ylabel('Accuracy')
        plt.title('Accuracy over Epochs')
        plt.legend()

        plt.tight_layout()
        plt.show()

    def evaluate(self, X_test, Y_test):
        Y_pred = self.predict(X_test)
        accuracy = self.accuracy(Y_test, Y_pred)
        cm = self.confusion_matrix(Y_test, Y_pred, self.output_size)
        print("Accuracy:", accuracy)
        print("Confusion Matrix:\n", cm)