NeuralNetworks

1
import numpy as np
2
import pandas as pd
3
import matplotlib.pyplot as plt
4
from sklearn.model_selection import train_test_split
5
from sklearn.preprocessing import StandardScaler
6
from sklearn.metrics import mean_squared_error
7
from statistics import mean
8
from typing import Dict, List, Tuple
9

10
np.random.seed(42)
11

12

13
class Neural:
14

15
    def __init__(self, layers: List[int], epochs: int, 
16
                 learning_rate: float = 0.001, batch_size: int=32,
17
                 validation_split: float = 0.2, verbose: int=1):
18
        self._layer_structure: List[int] = layers
19
        self._batch_size: int = batch_size
20
        self._epochs: int = epochs
21
        self._learning_rate: float = learning_rate
22
        self._validation_split: float = validation_split
23
        self._verbose: int = verbose
24
        self._losses: Dict[str, float] = {"train": [], "validation": []}
25
        self._is_fit: bool = False
26
        self.__layers = None
27

28
    def fit(self, X: np.ndarray, y: np.ndarray) -> None:
29
        # validation split
30
        X, X_val, y, y_val = train_test_split(X, y, test_size=self._validation_split, random_state=42)
31
        # initialization of layers
32
        self.__layers = self.__init_layers()
33
        for epoch in range(self._epochs):
34
            epoch_losses = []
35
            for i in range(1, len(self.__layers)):
36
                # forward pass
37
                x_batch = X[i:(i+self._batch_size)]
38
                y_batch = y[i:(i+self._batch_size)]
39
                pred, hidden = self.__forward(x_batch)
40
                # calculate loss
41
                loss = self.__calculate_loss(y_batch, pred)
42
                epoch_losses.append(np.mean(loss ** 2))
43
                #backward
44
                self.__backward(hidden, loss)
45
            valid_preds, _ = self.__forward(X_val)
46
            train_loss = mean(epoch_losses)
47
            valid_loss = np.mean(self.__calculate_mse(valid_preds,y_val))
48
            self._losses["train"].append(train_loss)
49
            self._losses["validation"].append(valid_loss)
50
            if self._verbose:
51
                print(f"Epoch: {epoch} Train MSE: {train_loss} Valid MSE: {valid_loss}")
52
        self._is_fit = True
53
        return
54

55
    def predict(self, X: np.ndarray) -> np.ndarray:
56
        if self._is_fit == False:
57
            raise Exception("Model has not been trained yet.")
58
        pred, hidden = self.__forward(X)
59
        return pred
60

61
    def plot_learning(self) -> None:
62
        plt.plot(self._losses["train"],label="loss")
63
        plt.plot(self._losses["validation"],label="validation")
64
        plt.legend()
65

66
    def __init_layers(self) -> List[np.ndarray]:
67
        layers = []
68
        for i in range(1, len(self._layer_structure)):
69
            layers.append([
70
                np.random.rand(self._layer_structure[i-1], self._layer_structure[i]) / 5 - .1,
71
                np.ones((1,self._layer_structure[i]))
72
            ])
73
        return layers
74

75
    # Прямой проход
76
    def __forward(self, batch: np.ndarray) -> Tuple[np.ndarray, List[np.ndarray]]:
77
        hidden = [batch.copy()]
78
        for i in range(len(self.__layers)):
79
            batch = np.matmul(batch, self.__layers[i][0]) + self.__layers[i][1]
80
            if i < len(self.__layers) - 1:
81
                batch = np.maximum(batch, 0)
82
            # Store the forward pass hidden values for use in backprop
83
            hidden.append(batch.copy())
84
        return batch, hidden
85

86
    def __calculate_loss(self, actual: np.ndarray, predicted: np.ndarray) -> np.ndarray:
87
        "mse"
88
        return predicted - actual
89

90
    # Функция ошибки
91
    def __calculate_mse(self, actual: np.ndarray, predicted: np.ndarray) -> np.ndarray:
92
        return (actual - predicted) ** 2
93

94
    # Обратный проход
95
    def __backward(self, hidden: List[np.ndarray], grad: np.ndarray) -> None:
96
        for i in range(len(self.__layers)-1, -1, -1):
97
            if i != len(self.__layers) - 1:
98
                grad = np.multiply(grad, np.heaviside(hidden[i+1], 0))
99

100
            w_grad = hidden[i].T @ grad
101
            b_grad = np.mean(grad, axis=0)
102

103
            self.__layers[i][0] -= w_grad * self._learning_rate
104
            self.__layers[i][1] -= b_grad * self._learning_rate
105

106
            grad = grad @ self.__layers[i][0].T
107
        return
108

109

110
#  Фиктивные данные для тестирования класса Neural.
111
def generate_data():
112
    # Define correlation values
113
    corr_a = 0.8
114
    corr_b = 0.4
115
    corr_c = -0.2
116

117
    # Generate independent features
118
    a = np.random.normal(0, 1, size=100000)
119
    b = np.random.normal(0, 1, size=100000)
120
    c = np.random.normal(0, 1, size=100000)
121
    d = np.random.randint(0, 4, size=100000)
122
    e = np.random.binomial(1, 0.5, size=100000)
123

124
    # Generate target feature based on independent features
125
    target = 50 + corr_a*a + corr_b*b + corr_c*c + d*10 + 20*e + np.random.normal(0, 10, size=100000)
126

127
    # Create DataFrame with all features
128
    df = pd.DataFrame({'a': a, 'b': b, 'c': c, 'd': d, 'e': e, 'target': target})
129
    return df
130

131

132
# Клиентский код
133
df = generate_data()
134

135
# Separate the features and target
136
X = df.drop('target', axis=1)
137
y = df['target']
138

139
scaler = StandardScaler()
140
X = scaler.fit_transform(X)
141

142
# Split the dataset into training and testing sets
143
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
144
y_train = y_train.to_numpy().reshape(-1,1)
145
y_test = y_test.to_numpy().reshape(-1,1)
146

147
layer_structure = [X_train.shape[1],10,10,1]
148
nn = Neural(layer_structure, 20, 1e-5, 64, 0.2, 1)
149

150
nn.fit(X_train, y_train)
151

152
y_pred = nn.predict(X_test)
153
nn.plot_learning()
154

155
print("Test error: ",mean_squared_error(y_test, y_pred))
156