Hello! Let's develop an advanced AI model for intraday trading of Bitcoin (BTCUSDT) on Binance.

**Goal:**

* **Primary:** Maximize profit from trading BTCUSDT.
* **Secondary:** Achieve an accuracy of predicting the direction of price change (up/down) of at least 60%.

The model should take 100 steps of 5-minute Bitcoin price changes and predict the 101st step.

**Constraints:**

* Maximum allowable drawdown is no more than 5%.
* The maximum number of simultaneously open trades is 1.

**Dataset:**

* Data: 5-minute BTCUSDT price data from Binance for the period from 01.01.2020 to 31.12.2022.
* Format: Groups of 101 steps, where the first 100 steps are input data (price change relative to the previous step, delta), the 101st step is the target variable (direction of price change - up/down).
* Outlier Handling: Exclude steps with a price change greater than 3 standard deviations from the mean.
* Normalization: Apply MinMaxScaler to the input data.

**Dataset Preparation:**

* Split the data into training and test sets*

**Model:**

* I suggest choosing a Transformer architecture with an attention mechanism, so that it focuses on the last steps but does not lose sight of the first ones. It should be like a real trader - analyzing all the steps and looking for patterns.
* Training: Use Reinforcement Learning, where the model will learn from its mistakes and strive to maximize profits while adhering to risk constraints. 
(Apply modern methods to create this advanced model).

**Model Testing:**

* Get steps from the test set.
* Pass 100 steps to the model.
* Get the prediction and compare it with the actual 101st step.
* Figure out the rest yourself.

**Possible problems:**

* Overfitting. The LLM should consider this problem and apply effective methods to prevent it.
* Market volatility. The model should be adaptable to changing market conditions.
* The problem of simply averaging the model's response into a constant.

**Wishes:**

* Evaluate the model's performance using Sharpe Ratio, Sortino Ratio, Maximum Drawdown, and also take into account the percentage of profitable trades.
* Conduct backtesting on historical data for 2023, simulating real trading, taking into account Binance fees.
* Provide a detailed description of the chosen model architecture, training method and justification of the decisions made.

You can apply all your ML specialist skills and create a truly advanced model capable of successfully predicting the price.

Use Pytorch to write advanced model code. The model will need to run on NVIDIA GPUs and CUDA

I made this. This is my project:

traider_ai/
├── models/
│   ├── best_model.pth
│   ├── best_model_OLD.pth
│   └── final_model.pth
├── backtest.py
├── data_download.py
├── data_preparation.py
├── model.py
├── single_prediction.py
├── test.py
├── train.py
└── utils.py


Project files:

backtest.py
```py

# backtest.py

import torch
import numpy as np
import pandas as pd
from model import PriceDirectionTransformer

def backtest(model, sequences, prices, fees=0.001):
    model.eval()
    device = next(model.parameters()).device
    sequences = torch.Tensor(sequences).to(device)
    with torch.no_grad():
        outputs = model(sequences)
        _, predictions = torch.max(outputs, 1)
        predictions = predictions.cpu().numpy()

    profits = []
    for i in range(len(predictions)):
        entry_price = prices[i + 99]
        exit_price = prices[i + 100] if i + 100 < len(prices) else prices[-1]
        if predictions[i] == 1:  # Predict Up
            # Buy at entry_price, sell at exit_price
            effective_entry_price = entry_price * (1 + fees)
            effective_exit_price = exit_price * (1 - fees)
            profit = (effective_exit_price - effective_entry_price) / effective_entry_price
        else:  # Predict Down
            # Short sell at entry_price, buy back at exit_price
            effective_entry_price = entry_price * (1 - fees)
            effective_exit_price = exit_price * (1 + fees)
            profit = (effective_entry_price - effective_exit_price) / effective_entry_price
        profits.append(profit)

    cumulative_returns = np.cumsum(profits)
    return profits, cumulative_returns

if __name__ == "__main__":
    # Load Test Data and Prices
    X_test = np.load('data/X_test.npy')
    y_test = np.load('data/y_test.npy')
    df = pd.read_csv('data/btcusdt_5min.csv', parse_dates=['timestamp'])
    df.sort_values('timestamp', inplace=True)
    df.reset_index(drop=True, inplace=True)
    total_sequences = len(X_test)
    prices = df['close'].values[-(total_sequences + 100):]  # Adjust the slice

    # Load Scaler if needed
    # scaler = joblib.load('data/scaler.save')

    # Initialize Model
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model = PriceDirectionTransformer()
    model.load_state_dict(torch.load('models/best_model.pth', map_location=device))
    model.to(device)

    # Perform Backtesting
    profits, cumulative_returns = backtest(model, X_test, prices)

    # Calculate Metrics
    total_return = cumulative_returns[-1] * 100
    win_rate = (np.array(profits) > 0).sum() / len(profits) * 100
    max_drawdown = np.min(cumulative_returns) * 100
    print(f'Total Return: {total_return:.2f}%')
    print(f'Win Rate: {win_rate:.2f}%')
    print(f'Max Drawdown: {max_drawdown:.2f}%')

    # Plotting Cumulative Returns
    import matplotlib.pyplot as plt
    plt.figure(figsize=(12,6))
    plt.plot(cumulative_returns, label='Cumulative Returns')
    plt.xlabel('Trade Number')
    plt.ylabel('Cumulative Return')
    plt.title('Backtesting Cumulative Returns')
    plt.legend()
    plt.show()
```


model.py
```py

# model.py

import torch
import torch.nn as nn
import numpy as np

class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len=100):
        super().__init__()
        position = torch.arange(0, max_len).unsqueeze(1)
        div_term = torch.exp(
            torch.arange(0, d_model, 2) * (-np.log(10000.0) / d_model)
        )
        pe = torch.zeros(1, max_len, d_model)
        pe[0, :, 0::2] = torch.sin(position * div_term)
        pe[0, :, 1::2] = torch.cos(position * div_term)
        self.register_buffer('pe', pe)

    def forward(self, x):
        return x + self.pe[:, :x.size(1)]

class PriceDirectionTransformer(nn.Module):
    def __init__(self, feature_size=64, num_layers=2, num_heads=4, dropout=0.1):
        super().__init__()
        self.model_type = 'Transformer'
        self.embedding = nn.Linear(1, feature_size)
        self.positional_encoding = PositionalEncoding(feature_size)
        encoder_layers = nn.TransformerEncoderLayer(
            d_model=feature_size,
            nhead=num_heads,
            dropout=dropout,
            dim_feedforward=256,
            batch_first=True,
        )
        self.transformer_encoder = nn.TransformerEncoder(
            encoder_layer=encoder_layers,
            num_layers=num_layers,
        )
        self.fc_out = nn.Linear(feature_size, 2)  # Up or Down
        self.softmax = nn.Softmax(dim=1)

    def forward(self, src):
        src = src.unsqueeze(-1)  # Add feature dimension
        src = self.embedding(src)
        src = self.positional_encoding(src)
        output = self.transformer_encoder(src)
        output = output.mean(dim=1)  # Global Average Pooling
        output = self.fc_out(output)
        output = self.softmax(output)
        return output
```


single_prediction.py
```py
# single_prediction.py

import torch
import numpy as np
import random
from model import PriceDirectionTransformer


def single_prediction(model, X_test, y_test, device):
    # Выбираем случайный индекс
    random_index = random.randint(0, len(X_test) - 1)

    # Извлекаем последовательность и соответствующую метку
    sequence = X_test[random_index]  # Shape: (100,)
    true_label = y_test[random_index]  # 0 или 1

    # Преобразуем последовательность в тензор и отправляем на устройство
    sequence_tensor = torch.Tensor(sequence).unsqueeze(0).to(device)  # Shape: (1, 100)

    # Пропускаем через модель
    model.eval()
    with torch.no_grad():
        output = model(sequence_tensor)  # Shape: (1, 2)
        predicted_probabilities = output.cpu().numpy()[0]  # Вероятности для классов
        predicted_label = np.argmax(predicted_probabilities)

    # Отображаем результаты
    classes = ['Down', 'Up']
    print(f"Sequence index: {random_index}")
    print(f"Predicted probabilities: {predicted_probabilities}")
    print(f"Predicted label: {predicted_label} ({classes[predicted_label]})")
    print(f"True label: {true_label} ({classes[true_label]})")

    # Дополнительно можно посмотреть на сами данные
    # Например, вывести последние значения последовательности
    print(f"Last 5 values in the sequence: {sequence[-5:]}")


if __name__ == "__main__":
    # Устройство для вычислений
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

    # Загрузка тестовых данных
    X_test = np.load('data/X_test.npy')
    y_test = np.load('data/y_test.npy')

    # Загрузка модели
    model = PriceDirectionTransformer()
    model.load_state_dict(torch.load('models/best_model.pth', map_location=device))
    model.to(device)

    # Выполнение предсказания на одной последовательности
    single_prediction(model, X_test, y_test, device)
```


test.py
```py

# test.py

import torch
import torch.nn as nn
from torch.utils.data import DataLoader
from model import PriceDirectionTransformer
from utils import BTCUSDTDataset
import numpy as np

def test_model(model, dataloader, criterion, device):
    model.eval()
    epoch_loss = 0
    correct_predictions = 0
    total_predictions = 0
    all_labels = []
    all_predictions = []
    with torch.no_grad():
        for sequences, labels in dataloader:
            sequences = sequences.to(device)
            labels = labels.to(device)
            outputs = model(sequences)
            loss = criterion(outputs, labels)
            epoch_loss += loss.item()
            _, predicted = torch.max(outputs, 1)
            correct_predictions += (predicted == labels).sum().item()
            total_predictions += labels.size(0)
            all_labels.extend(labels.cpu().numpy())
            all_predictions.extend(predicted.cpu().numpy())
    accuracy = correct_predictions / total_predictions
    return epoch_loss / len(dataloader), accuracy, all_labels, all_predictions

if __name__ == "__main__":
    # Device Configuration
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

    # Load Test Data
    X_test = np.load('data/X_test.npy')
    y_test = np.load('data/y_test.npy')

    test_dataset = BTCUSDTDataset(X_test, y_test)
    test_loader = DataLoader(test_dataset, batch_size=64, shuffle=False)

    # Initialize Model
    model = PriceDirectionTransformer()
    model.load_state_dict(torch.load('models/best_model.pth'))
    model.to(device)

    # Loss Function
    criterion = nn.CrossEntropyLoss()

    # Test Model
    test_loss, test_accuracy, all_labels, all_predictions = test_model(
        model, test_loader, criterion, device
    )

    print(f'Test Loss: {test_loss:.4f}, Test Accuracy: {test_accuracy:.4f}')
    # Additional Metrics can be calculated here
```


train.py
```py

# train.py

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader
from model import PriceDirectionTransformer
from utils import BTCUSDTDataset
import numpy as np

def train_model(model, dataloader, criterion, optimizer, device):
    model.train()
    epoch_loss = 0
    correct_predictions = 0
    total_predictions = 0
    for sequences, labels in dataloader:
        sequences = sequences.to(device)
        labels = labels.to(device)
        optimizer.zero_grad()
        outputs = model(sequences)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()
        epoch_loss += loss.item()
        _, predicted = torch.max(outputs, 1)
        correct_predictions += (predicted == labels).sum().item()
        total_predictions += labels.size(0)
    accuracy = correct_predictions / total_predictions
    return epoch_loss / len(dataloader), accuracy

def validate_model(model, dataloader, criterion, device):
    model.eval()
    epoch_loss = 0
    correct_predictions = 0
    total_predictions = 0
    with torch.no_grad():
        for sequences, labels in dataloader:
            sequences = sequences.to(device)
            labels = labels.to(device)
            outputs = model(sequences)
            loss = criterion(outputs, labels)
            epoch_loss += loss.item()
            _, predicted = torch.max(outputs, 1)
            correct_predictions += (predicted == labels).sum().item()
            total_predictions += labels.size(0)
    accuracy = correct_predictions / total_predictions
    return epoch_loss / len(dataloader), accuracy

if __name__ == "__main__":
    # Hyperparameters
    FEATURE_SIZE = 64
    NUM_LAYERS = 2
    NUM_HEADS = 4
    DROPOUT = 0.1
    BATCH_SIZE = 64
    NUM_EPOCHS = 20
    LEARNING_RATE = 0.001

    # Device Configuration
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

    # Load Data
    X_train = np.load('data/X_train.npy')
    y_train = np.load('data/y_train.npy')
    X_val = np.load('data/X_val.npy')
    y_val = np.load('data/y_val.npy')

    # Create Datasets and Dataloaders
    from o1_made.utils import BTCUSDTDataset
    train_dataset = BTCUSDTDataset(X_train, y_train)
    val_dataset = BTCUSDTDataset(X_val, y_val)
    train_loader = DataLoader(train_dataset, batch_size=BATCH_SIZE, shuffle=True)
    val_loader = DataLoader(val_dataset, batch_size=BATCH_SIZE, shuffle=False)

    # Initialize Model
    model = PriceDirectionTransformer(
        feature_size=FEATURE_SIZE,
        num_layers=NUM_LAYERS,
        num_heads=NUM_HEADS,
        dropout=DROPOUT,
    ).to(device)

    # Loss and Optimizer
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model.parameters(), lr=LEARNING_RATE)

    # Training Loop
    best_val_accuracy = 0
    for epoch in range(NUM_EPOCHS):
        train_loss, train_accuracy = train_model(
            model, train_loader, criterion, optimizer, device
        )
        val_loss, val_accuracy = validate_model(
            model, val_loader, criterion, device
        )
        if val_accuracy > best_val_accuracy:
            best_val_accuracy = val_accuracy
            torch.save(model.state_dict(), 'models/best_model.pth')
        print(
            f'Epoch {epoch+1}/{NUM_EPOCHS}, '
            f'Train Loss: {train_loss:.4f}, Train Acc: {train_accuracy:.4f}, '
            f'Val Loss: {val_loss:.4f}, Val Acc: {val_accuracy:.4f}'
        )

    # Save Final Model
    torch.save(model.state_dict(), 'models/final_model.pth')
    print("Training completed.")
```


utils.py
```py
# utils.py

import torch
from torch.utils.data import Dataset


class BTCUSDTDataset(Dataset):
    def __init__(self, sequences, labels):
        self.sequences = torch.Tensor(sequences)
        self.labels = torch.LongTensor(labels)

    def __len__(self):
        return len(self.sequences)

    def __getitem__(self, idx):
        return self.sequences[idx], self.labels[idx]
```


Can you review my project and make improvements.