Long Short-Term Memory (LSTM) neural network in PyTorch

import torch import torch.nn as nn import numpy as np import pandas as pd import matplotlib.pyplot as plt from torch.utils.data import DataLoader, TensorDataset from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler # Set device for computation device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') # Define the LSTM-based Model class LSTMModel(nn.Module): def init(self, input_size, hidden_size1, num_layers, seq_len): super(LSTMModel, self).__init__() self.num_layers = num_layers self.hidden_size1 = hidden_size1 self.seq_len = seq_len # Fully connected layers before LSTM self.fc1 = nn.Linear(input_size, 32) self.fc2 = nn.Linear(32, 64) self.fc3 = nn.Linear(64, 64) # LSTM Layer self.lstm = nn.LSTM(64, hidden_size1, num_layers, batch_first=True, bidirectional=True, dropout=0.5) # Fully connected layers after LSTM self.fc_lstm1 = nn.Linear(2 hidden_size1, 512) self.fc_lstm2 = nn.Linear(512, 256) self.fc_lstm3 = nn.Linear(256, 128) self.fc_lstm4 = nn.Linear(128, 32) self.fc_out = nn.Linear(32, 1) self.relu = nn.ReLU() self.dropout = nn.Dropout(0.5) def forward(self, x): # Input fully connected layers x = self.relu(self.fc1(x)) x = self.relu(self.fc2(x)) x = self.relu(self.fc3(x)) # Pass through LSTM layers h0 = torch.zeros(2 self.num_layers, x.size(0), self.hidden_size1).to(device) c0 = torch.zeros(2 * self.num_layers, x.size(0), self.hidden_size1).to(device) x, = self.lstm(x, (h0, c0)) x = x[:, -1, :] # Take the output of the last LSTM cell # Fully connected layers after LSTM x = self.relu(self.fclstm1(x)) x = self.dropout(self.relu(self.fc_lstm2(x))) x = self.relu(self.fc_lstm3(x)) x = self.relu(self.fc_lstm4(x)) x = self.fc_out(x) return x # Hyperparameters input_size = 30 hidden_size1 = 64 num_layers = 2 seq_len = 50 learning_rate = 0.001 batch_size = 64 num_epochs = 300 # Data loading and preprocessing def load_real_data(file_path): # Load your dataset data = pd.read_csv(file_path) # Assuming that all columns except the last one are features, and the last column is the target (adjust if necessary) X = data.iloc[:, :-1].values # Features y = data.iloc[:, -1].values # Target # Standardize the data scaler = StandardScaler() X = scaler.fit_transform(X) # Reshape to fit the LSTM input X = X.reshape((X.shape[0], seq_len, input_size)) y = y.reshape(-1, 1) return X, y # Replace 'your_data.csv' with the actual path to your dataset X, y = load_real_data('C:/Users/Stephen Mariga/PycharmProjects/pythonProject2/synthetic_lstm_data.csv') # Split data into training and validation sets X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2, random_state=42) # Convert data to tensors and create DataLoader train_dataset = TensorDataset(torch.tensor(X_train, dtype=torch.float32), torch.tensor(y_train, dtype=torch.float32)) val_dataset = TensorDataset(torch.tensor(X_val, dtype=torch.float32), torch.tensor(y_val, dtype=torch.float32)) train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True) val_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False) # Initialize the model, loss function, and optimizer model = LSTMModel(input_size, hidden_size1, num_layers, seq_len).to(device) criterion = nn.MSELoss() optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate) # Training and validation loop train_losses = [] val_losses = [] for epoch in range(num_epochs): model.train() train_loss = 0 for X_batch, y_batch in train_loader: X_batch, y_batch = X_batch.to(device), y_batch.to(device) optimizer.zero_grad() outputs = model(X_batch) loss = criterion(outputs, y_batch) loss.backward() optimizer.step() train_loss += loss.item() train_loss /= len(train_loader) train_losses.append(train_loss) # Validation phase model.eval() val_loss = 0 with torch.no_grad(): for X_batch, y_batch in val_loader: X_batch, y_batch = X_batch.to(device), y_batch.to(device) outputs = model(X_batch) loss = criterion(outputs, y_batch) val_loss += loss.item() val_loss /= len(val_loader) val_losses.append(val_loss) if epoch % 50 == 0: print(f'Epoch {epoch}/{num_epochs}, Train Loss: {train_loss:.4f}, Val Loss: {val_loss:.4f}') # Plot the training and validation losses (MSE) plt.figure(figsize=(8, 6)) plt.plot(train_losses, label='Training error', color='blue') plt.plot(val_losses, label='Validation error', color='orange') plt.title('Training on normalized data') plt.xlabel('Number of epochs') plt.ylabel('Mean square error') plt.legend() plt.show()