This is a training script for a model making predictions on a 1 min usd/euro chart. It predicts if the price will be lower, much lower, higher, much higher, or the same price, 30 minutes into the future. The trained model will be used in a meratrader5 environment as part of an ea making trades, which is why it is saved as an onnx model. Improve its speed, effencienccy, and predictive ability without adding features. Make the loss calculation more appropriate for the task at hand.

import os
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
import pandas as pd
import numpy as np
from sklearn.preprocessing import MinMaxScaler
from sklearn.model_selection import train_test_split
import onnx
import onnxruntime
from tqdm import tqdm

# Configuration
class Config:
    def __init__(self):
        self.task_name = 'short_term_forecast'
        self.seq_len = 180  # Input sequence length
        self.pred_len = 30  # Prediction length (30 minutes into the future)
        self.d_model = 1024  # Model dimension
        self.enc_in = 4  # Number of input features (Open, High, Low, Close)
        self.e_layers = 6  # Number of encoder layers
        self.dropout = 0.4
        self.batch_size = 128
        self.learning_rate = 0.00001
        self.epochs = 100
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        self.data_path = '/mnt/d/FX/fx_csv/22_23_1M.csv'
        self.model_save_path = '/mnt/d/FX/Mixer_Models'
        self.results_save_path = '/mnt/d/FX/Mixer_Results'

configs = Config()

class ScalerModule(nn.Module):
    def __init__(self, scaler):
        super(ScalerModule, self).__init__()
        self.min_ = nn.Parameter(torch.FloatTensor(scaler.min_), requires_grad=False)
        self.scale_ = nn.Parameter(torch.FloatTensor(scaler.scale_), requires_grad=False)

    def forward(self, x):
        return (x - self.min_) / self.scale_

# Model definition
class ResBlock(nn.Module):
    def __init__(self, configs):
        super(ResBlock, self).__init__()
        self.temporal = nn.Sequential(
            nn.Linear(configs.seq_len, configs.d_model),
            nn.ReLU(),
            nn.Linear(configs.d_model, configs.seq_len),
            nn.Dropout(configs.dropout)
        )
        self.channel = nn.Sequential(
            nn.Linear(configs.enc_in, configs.d_model),
            nn.ReLU(),
            nn.Linear(configs.d_model, configs.enc_in),
            nn.Dropout(configs.dropout)
        )

    def forward(self, x):
        # x: [B, L, D]
        x = x + self.temporal(x.transpose(1, 2)).transpose(1, 2)
        x = x + self.channel(x)
        return x

class Model(nn.Module):
    def __init__(self, configs, scaler):
        super(Model, self).__init__()
        self.scaler = ScalerModule(scaler)
        self.task_name = configs.task_name
        self.layer = configs.e_layers
        self.model = nn.ModuleList([ResBlock(configs) for _ in range(configs.e_layers)])
        self.pred_len = configs.pred_len
        self.projection = nn.Linear(configs.seq_len, configs.pred_len)
        self.final_layer = nn.Linear(configs.enc_in, 1)  # Output a single value
        self.classifier = nn.Linear(configs.pred_len, 5)  # 5 classes for prediction

    def forecast(self, x_enc):
        # x: [B, L, D]
        x_enc = self.scaler(x_enc)
        for i in range(self.layer):
            x_enc = self.model[i](x_enc)
        enc_out = self.projection(x_enc.transpose(1, 2)).transpose(1, 2)
        enc_out = self.final_layer(enc_out).squeeze(-1)
        return self.classifier(enc_out)

    def forward(self, x_enc):
        if self.task_name == 'short_term_forecast':
            return self.forecast(x_enc)
        else:
            raise ValueError('Only short_term_forecast task implemented')

# Dataset
class ForexDataset(Dataset):
    def __init__(self, data, seq_len, pred_len):
        self.data = data
        self.seq_len = seq_len
        self.pred_len = pred_len

    def __len__(self):
        return len(self.data) - self.seq_len - self.pred_len + 1

    def __getitem__(self, idx):
        x = self.data[idx:idx+self.seq_len]
        y = self.data[idx+self.seq_len+self.pred_len-1, 3]  # Close price after pred_len
        
        # Calculate the price change
        current_price = self.data[idx+self.seq_len-1, 3]
        price_change = (y - current_price) / current_price
        
        # Classify the price change
        if price_change < -0.01:
            label = 0  # much lower
        elif -0.01 <= price_change < -0.001:
            label = 1  # lower
        elif -0.001 <= price_change <= 0.001:
            label = 2  # same
        elif 0.001 < price_change <= 0.01:
            label = 3  # higher
        else:
            label = 4  # much higher
        
        return torch.FloatTensor(x), torch.LongTensor([label])

# Data preprocessing
def preprocess_data(configs):
    print("Loading and preprocessing data...")
    df = pd.read_csv(configs.data_path)
    df['DateTime'] = pd.to_datetime(df['DateTime'])
    df = df.set_index('DateTime')
    
    scaler = MinMaxScaler()
    data = scaler.fit_transform(df[['Open', 'High', 'Low', 'Close']])
    
    dataset = ForexDataset(data, configs.seq_len, configs.pred_len)
    train_size = int(0.8 * len(dataset))
    test_size = len(dataset) - train_size
    train_dataset, test_dataset = torch.utils.data.random_split(dataset, [train_size, test_size])
    
    train_loader = DataLoader(train_dataset, batch_size=configs.batch_size, shuffle=True)
    test_loader = DataLoader(test_dataset, batch_size=configs.batch_size, shuffle=False)
    
    return train_loader, test_loader, scaler

def calculate_metrics(output, target):
    _, predicted = torch.max(output.data, 1)
    correct = (predicted == target).sum().item()
    
    # Convert output probabilities to continuous predictions
    continuous_pred = torch.sum(output * torch.tensor([-2, -1, 0, 1, 2]).to(output.device), dim=1)
    continuous_target = torch.tensor([-2, -1, 0, 1, 2]).to(target.device)[target]
    
    mae = torch.mean(torch.abs(continuous_pred - continuous_target)).item()
    mse = torch.mean((continuous_pred - continuous_target) ** 2).item()
    rmse = torch.sqrt(torch.tensor(mse)).item()
    
    return correct, mae, mse, rmse

# Training function
def train(model, train_loader, optimizer, criterion, device, epoch):
    model.train()
    total_loss = 0
    correct = 0
    total = 0
    total_mae = 0
    total_mse = 0
    
    progress_bar = tqdm(train_loader, desc=f"Epoch {epoch+1}")
    for batch_x, batch_y in progress_bar:
        optimizer.zero_grad()
        batch_x, batch_y = batch_x.to(device), batch_y.to(device)
        output = model(batch_x)
        loss = criterion(output, batch_y.squeeze())
        loss.backward()
        optimizer.step()
        
        total_loss += loss.item()
        batch_correct, batch_mae, batch_mse, _ = calculate_metrics(output, batch_y.squeeze())
        correct += batch_correct
        total += batch_y.size(0)
        total_mae += batch_mae * batch_y.size(0)
        total_mse += batch_mse * batch_y.size(0)
        
        # Update progress bar
        avg_loss = total_loss / (progress_bar.n + 1)
        accuracy = 100 * correct / total
        avg_mae = total_mae / total
        avg_rmse = torch.sqrt(torch.tensor(total_mse / total)).item()
        progress_bar.set_postfix({
            'loss': f'{avg_loss:.4f}',
            'acc': f'{accuracy:.2f}%',
            'MAE': f'{avg_mae:.4f}',
            'RMSE': f'{avg_rmse:.4f}'
        })
    
    return (total_loss / len(train_loader), 100 * correct / total, 
            total_mae / total, torch.sqrt(torch.tensor(total_mse / total)).item())

def evaluate(model, test_loader, criterion, device):
    model.eval()
    total_loss = 0
    correct = 0
    total = 0
    total_mae = 0
    total_mse = 0
    
    with torch.no_grad():
        for batch_x, batch_y in tqdm(test_loader, desc="Evaluating"):
            batch_x, batch_y = batch_x.to(device), batch_y.to(device)
            output = model(batch_x)
            loss = criterion(output, batch_y.squeeze())
            total_loss += loss.item()
            
            batch_correct, batch_mae, batch_mse, _ = calculate_metrics(output, batch_y.squeeze())
            correct += batch_correct
            total += batch_y.size(0)
            total_mae += batch_mae * batch_y.size(0)
            total_mse += batch_mse * batch_y.size(0)
    
    avg_loss = total_loss / len(test_loader)
    accuracy = 100 * correct / total
    avg_mae = total_mae / total
    avg_rmse = torch.sqrt(torch.tensor(total_mse / total)).item()
    return avg_loss, accuracy, avg_mae, avg_rmse

# Save model function
def save_model(model, path, epoch, test_loss):
    if not os.path.exists(path):
        os.makedirs(path)
    models = sorted([f for f in os.listdir(path) if f.endswith('.onnx')])
    if len(models) >= 3:
        os.remove(os.path.join(path, models[0]))
    
    model_path = os.path.join(path, f'model_epoch_{epoch}_loss_{test_loss:.4f}.onnx')
    dummy_input = torch.randn(1, configs.seq_len, configs.enc_in).to(configs.device)
    torch.onnx.export(model, dummy_input, model_path, export_params=True, opset_version=11,
                      input_names=['input'], output_names=['output'],
                      dynamic_axes={'input': {0: 'batch_size'}, 'output': {0: 'batch_size'}})
    print(f"Model saved: {model_path}")

# Main training loop
def main():
    print(f"Using device: {configs.device}")
    train_loader, test_loader, scaler = preprocess_data(configs)
    
    model = Model(configs, scaler).to(configs.device)
    optimizer = optim.Adam(model.parameters(), lr=configs.learning_rate)
    criterion = nn.CrossEntropyLoss()
    
    best_test_loss = float('inf')
    
    for epoch in range(configs.epochs):
        train_loss, train_acc, train_mae, train_rmse = train(model, train_loader, optimizer, criterion, configs.device, epoch)
        test_loss, test_acc, test_mae, test_rmse = evaluate(model, test_loader, criterion, configs.device)
        
        print(f"\nEpoch {epoch+1}/{configs.epochs}")
        print(f"Train Loss: {train_loss:.4f}, Train Accuracy: {train_acc:.2f}%")
        print(f"Train MAE: {train_mae:.4f}, Train RMSE: {train_rmse:.4f}")
        print(f"Test Loss: {test_loss:.4f}, Test Accuracy: {test_acc:.2f}%")
        print(f"Test MAE: {test_mae:.4f}, Test RMSE: {test_rmse:.4f}")
        
        if test_loss < best_test_loss:
            best_test_loss = test_loss
            save_model(model, configs.model_save_path, epoch+1, test_loss)
    
    # Save final results
    with open(os.path.join(configs.results_save_path, 'training_results.txt'), 'w') as f:
        f.write(f"Final Train Loss: {train_loss:.4f}\n")
        f.write(f"Final Train Accuracy: {train_acc:.2f}%\n")
        f.write(f"Final Train MAE: {train_mae:.4f}\n")
        f.write(f"Final Train RMSE: {train_rmse:.4f}\n")
        f.write(f"Final Test Loss: {test_loss:.4f}\n")
        f.write(f"Final Test Accuracy: {test_acc:.2f}%\n")
        f.write(f"Final Test MAE: {test_mae:.4f}\n")
        f.write(f"Final Test RMSE: {test_rmse:.4f}\n")
        f.write(f"Best Test Loss: {best_test_loss:.4f}\n")
    
    print("Training completed. Results saved.")

if __name__ == "__main__":
    main()