import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import mean_squared_error
from statistics import mean
from typing import Dict, List, Tuple
np.random.seed(42)
class Neural:
def __init__(self, layers: List[int], epochs: int,
learning_rate: float = 0.001, batch_size: int=32,
validation_split: float = 0.2, verbose: int=1)
:
self._layer_structure: List[int] = layers
self._batch_size: int = batch_size
self._epochs: int = epochs
self._learning_rate: float = learning_rate
self._validation_split: float = validation_split
self._verbose: int = verbose
self._losses: Dict[str, float] = {"train": [], "validation": []}
self._is_fit: bool = False
self.__layers = None
def fit(self, X: np.ndarray, y: np.ndarray) -> None:
X, X_val, y, y_val = train_test_split(X, y, test_size=self._validation_split, random_state=42)
self.__layers = self.__init_layers()
for epoch in range(self._epochs):
epoch_losses = []
for i in range(1, len(self.__layers)):
x_batch = X[i:(i+self._batch_size)]
y_batch = y[i:(i+self._batch_size)]
pred, hidden = self.__forward(x_batch)
loss = self.__calculate_loss(y_batch, pred)
epoch_losses.append(np.mean(loss ** 2))
self.__backward(hidden, loss)
valid_preds, _ = self.__forward(X_val)
train_loss = mean(epoch_losses)
valid_loss = np.mean(self.__calculate_mse(valid_preds,y_val))
self._losses["train"].append(train_loss)
self._losses["validation"].append(valid_loss)
if self._verbose:
print(f"Epoch: {epoch} Train MSE: {train_loss} Valid MSE: {valid_loss}")
self._is_fit = True
return
def predict(self, X: np.ndarray) -> np.ndarray:
if self._is_fit == False:
raise Exception("Model has not been trained yet.")
pred, hidden = self.__forward(X)
return pred
def plot_learning
(self) -> None:
plt.plot(self._losses["train"],label="loss")
plt.plot(self._losses["validation"],label="validation")
plt.legend()
def __init_layers(self) -> List[np.ndarray]:
layers = []
for i in range(1, len(self._layer_structure)):
layers.append([
np.random.rand(self._layer_structure[i-1], self._layer_structure[i]) / 5 - .1,
np.ones((1,self._layer_structure[i]))
])
return layers
def __forward(self, batch: np.ndarray) -> Tuple[np.ndarray, List[np.ndarray]]:
hidden = [batch.copy()]
for i in range(len(self.__layers)):
batch = np.matmul(batch, self.__layers[i][0]) + self.__layers[i][1]
if i < len(self.__layers) - 1:
batch = np.maximum(batch, 0)
hidden.append(batch.copy())
return batch, hidden
def __calculate_loss(self,actual: np.ndarray, predicted: np.ndarray) -> np.ndarray:
"mse"
return predicted - actual
def __calculate_mse(self, actual: np.ndarray, predicted: np.ndarray) -> np.ndarray:
return (actual - predicted) ** 2
def __backward(self, hidden: List[np.ndarray], grad: np.ndarray) -> None:
for i in range(len(self.__layers)-1, -1, -1):
if i != len(self.__layers) - 1:
grad = np.multiply(grad, np.heaviside(hidden[i+1], 0))
w_grad = hidden[i].T @ grad
b_grad = np.mean(grad, axis=0)
self.__layers[i][0] -= w_grad * self._learning_rate
self.__layers[i][1] -= b_grad * self._learning_rate
grad = grad @ self.__layers[i][0].T
return
def generate_data():
corr_a = 0.8
corr_b = 0.4
corr_c = -0.2
a = np.random.normal(0, 1, size=100000)
b = np.random.normal(0, 1, size=100000)
c = np.random.normal(0, 1, size=100000)
d = np.random.randint(0, 4, size=100000)
e = np.random.binomial(1, 0.5, size=100000)
target = 50 + corr_a*a + corr_b*b + corr_c*c + d*10 + 20*e + np.random.normal(0, 10, size=100000)
df = pd.DataFrame({'a': a, 'b': b, 'c': c, 'd': d, 'e': e, 'target': target})
return df
df = generate_data()
X = df.drop('target', axis=1)
y = df['target']
scaler = StandardScaler()
X = scaler.fit_transform(X)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
y_train = y_train.to_numpy().reshape(-1,1)
y_test = y_test.to_numpy().reshape(-1,1)
layer_structure = [X_train.shape[1],10,10,1]
nn = Neural(layer_structure, 20, 1e-5, 64, 0.2, 1)
nn.fit(X_train, y_train)
y_pred = nn.predict(X_test)
nn.plot_learning()
print("Test error: ",mean_squared_error(y_test, y_pred))
"""
输出结果:
Epoch: 0 Train MSE: 6066.584227303851 Valid MSE: 5630.972575612107
Epoch: 1 Train MSE: 5870.45827241517 Valid MSE: 5384.024826048717
Epoch: 2 Train MSE: 5584.489636840577 Valid MSE: 4993.952466830458
Epoch: 3 Train MSE: 5127.64238543267 Valid MSE: 4376.563641292963
Epoch: 4 Train MSE: 4408.550555767417 Valid MSE: 3470.967255214888
Epoch: 5 Train MSE: 3370.6165240733935 Valid MSE: 2333.4365011529103
Epoch: 6 Train MSE: 2112.702666917853 Valid MSE: 1245.1547720938968
Epoch: 7 Train MSE: 1001.3618108374816 Valid MSE: 565.5834115291266
Epoch: 8 Train MSE: 396.9514096548994 Valid MSE: 298.31216370120575
Epoch: 9 Train MSE: 198.29006090703072 Valid MSE: 204.83294115572235
Epoch: 10 Train MSE: 139.2931182121901 Valid MSE: 162.0341771457693
Epoch: 11 Train MSE: 113.971621253487 Valid MSE: 138.35491897074462
Epoch: 12 Train MSE: 100.19734344395454 Valid MSE: 124.60170156400542
Epoch: 13 Train MSE: 92.35069581444299 Valid MSE: 116.55999261926036
Epoch: 14 Train MSE: 87.88890529435344 Valid MSE: 111.85169154584908
Epoch: 15 Train MSE: 85.37162170152865 Valid MSE: 109.08001681897412
Epoch: 16 Train MSE: 83.96135084225956 Valid MSE: 107.42929147368837
Epoch: 17 Train MSE: 83.17564386183105 Valid MSE: 106.42800615549532
Epoch: 18 Train MSE: 82.73977092210092 Valid MSE: 105.80581167903857
Epoch: 19 Train MSE: 82.49876360284046 Valid MSE: 105.40815905002043
Test error: 105.40244085384184
"""