import pandas as pd import numpy as np import pickle import copy import math import torch.nn as nn from torch.utils.data import DataLoader, TensorDataset import torch.optim as optim import torch import time class PositionalEncoding(nn.Module): def __init__(self, d_model, dropout=0.1, max_len=5000): super(PositionalEncoding, self).__init__() self.dropout = nn.Dropout(p=dropout) # Compute the positional encodings pe = torch.zeros(max_len, d_model) position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1) div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model)) pe[:, 0::2] = torch.sin(position * div_term) # Adjust for odd d_model values if d_model % 2 == 0: pe[:, 1::2] = torch.cos(position * div_term) else: pe[:, 1::2] = torch.cos(position * div_term)[:,:-1] pe = pe.unsqueeze(0).transpose(0, 1) self.register_buffer('pe', pe) def forward(self, x): x = x + self.pe[:x.size(0), :] return self.dropout(x) class TransformerPatternRecognition(nn.Module): def __init__(self, d_model, nhead, num_encoder_layers, num_classes, dropout=0.1): super(TransformerPatternRecognition, self).__init__() self.pos_encoder = PositionalEncoding(d_model, dropout) self.transformer = nn.Transformer(d_model, nhead, num_encoder_layers, dropout=dropout) self.bn = nn.BatchNorm1d(d_model) self.fc = nn.Linear(d_model, num_classes) # Initialize weights for p in self.parameters(): if p.dim() > 1: nn.init.kaiming_normal_(p) def forward(self, x): x = self.pos_encoder(x) output = self.transformer(x, x) output = self.bn(output[:, -1, :]) # Batch Normalization after transformer output = self.fc(output) return output def get_dataloader(training_data, batch_size=32): input_tensors = torch.tensor([item[0] for item in training_data], dtype=torch.float32) label_tensors = torch.tensor([item[1] for item in training_data], dtype=torch.long) dataset = TensorDataset(input_tensors, label_tensors) dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True) return dataloader def to_1d_tensor(t): t = t.squeeze() if t.dim() == 0: t = t.unsqueeze(0) return t def calculate_accuracy(outputs, labels, thres = 0.6): # Convert outputs to probabilities using softmax probs = torch.nn.functional.softmax(outputs, dim=1) indices_0 = to_1d_tensor(torch.nonzero(probs[:, 0] > thres).squeeze()) indices_1 = to_1d_tensor(torch.nonzero(probs[:, 1] > thres).squeeze()) # Check if both indices_0 and indices_1 are empty if len(indices_0) == 0 and len(indices_1) == 0: return 0 correct_count = 0 # Check if predictions at indices_0 are correct for index in indices_0: if labels[index] == 0: correct_count += 1 # Check if predictions at indices_1 are correct for index in indices_1: if labels[index] == 1: correct_count += 1 total_indices = len(indices_0) + len(indices_1) return 100 * correct_count / total_indices def evaluate_model(model, dataloader, criterion): total_loss = 0.0 total_correct = 0 total_items = 0 with torch.no_grad(): for inputs, labels in dataloader: inputs, labels = inputs.to(device), labels.to(device) # Move inputs and labels to the model's device outputs = model(inputs) valid_indices = (labels != 2) if not torch.any(valid_indices): continue masked_outputs = outputs[valid_indices] masked_labels = labels[valid_indices] loss = criterion(masked_outputs, masked_labels) total_loss += loss.item() accuracy = calculate_accuracy(masked_outputs, masked_labels, thres=0.6) total_correct += accuracy * len(masked_labels) / 100 # Convert accuracy percentage back to count total_items += len(masked_labels) avg_loss = total_loss / len(dataloader) overall_accuracy = 100 * total_correct / total_items return avg_loss, overall_accuracy # Load the train and test data using pickle with open('train_data.pkl', 'rb') as f: training_data = pickle.load(f) with open('test_data.pkl', 'rb') as f: test_data = pickle.load(f) # Parameters d_model = 7 nhead = 1 num_encoder_layers = 2 num_classes = 3 learning_rate = 0.0001 num_epochs = 100 device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # Model, Optimizer model = TransformerPatternRecognition(d_model, nhead, num_encoder_layers, num_classes).to(device) # Weights for classes weights = torch.tensor([2.0, 2.0, 1.0], device=device) criterion = nn.CrossEntropyLoss(weight=weights) optimizer = optim.Adam(model.parameters(), lr=learning_rate) # Training loop with batching train_dataloader = get_dataloader(training_data) test_dataloader = get_dataloader(test_data) all_outputs = [] all_labels = [] # train for epoch in range(num_epochs): total_loss = 0.0 iter_count = 0 total_iters = len(train_dataloader) for inputs, labels in train_dataloader: inputs, labels = inputs.to(device), labels.to(device) iter_count += 1 optimizer.zero_grad() outputs = model(inputs) loss = criterion(outputs, labels) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() total_loss += loss.item() all_outputs.append(outputs) all_labels.append(labels) # Periodically backup the model's weights if iter_count % 2000 == 0: accumulated_outputs = torch.cat(all_outputs, dim=0) accumulated_labels = torch.cat(all_labels, dim=0) accuracy = calculate_accuracy(accumulated_outputs, accumulated_labels, thres = 0.6) print(f"Epoch {epoch+1}/{num_epochs}, Iteration {iter_count}/{total_iters}, Loss: {total_loss/iter_count:.4f}, Accuracy: {accuracy:.2f}%") # Reset the lists for the next 1000 iterations all_outputs = [] all_labels = [] avg_loss = total_loss / len(train_dataloader) print(f"Epoch {epoch+1}/{num_epochs} - Avg Loss: {avg_loss:.4f}") # Evaluate on test set test_loss, test_accuracy = evaluate_model(model, test_dataloader, criterion) print(f"Epoch {epoch+1}/{num_epochs} - Test Loss: {test_loss:.4f}, Test Accuracy: {test_accuracy:.2f}%")