Project structure:

o1_made/
├── 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=5000):
        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(max_len, d_model)
        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)
        pe = pe.unsqueeze(0)  # [1, max_len, d_model]
        self.register_buffer('pe', pe)

    def forward(self, x):
        # x shape: [batch_size, seq_len, d_model]
        x = x + self.pe[:, :x.size(1), :].to(x.device)
        return x

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_layer = nn.TransformerEncoderLayer(
            d_model=feature_size,
            nhead=num_heads,
            dim_feedforward=256,
            dropout=dropout,
            activation='relu',
            batch_first=True,
        )
        self.transformer_encoder = nn.TransformerEncoder(
            encoder_layer=encoder_layer, num_layers=num_layers
        )
        self.dropout = nn.Dropout(dropout)
        self.fc_out = nn.Linear(feature_size, 2)  # Up or Down

    def forward(self, src):
        # src shape: [batch_size, seq_len]
        src = src.unsqueeze(-1)  # [batch_size, seq_len, 1]
        src = self.embedding(src)  # [batch_size, seq_len, feature_size]
        src = self.positional_encoding(src)  # [batch_size, seq_len, feature_size]
        output = self.transformer_encoder(src)  # [batch_size, seq_len, feature_size]
        output = output.mean(dim=1)  # Global Average Pooling over sequence length
        output = self.dropout(output)
        output = self.fc_out(output)  # [batch_size, num_classes]
        return output  # Без применения Softmax
```


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 = 128
    NUM_LAYERS = 3
    NUM_HEADS = 8
    DROPOUT = 0.2
    BATCH_SIZE = 256
    NUM_EPOCHS = 50
    LEARNING_RATE = 0.001
    PATIENCE = 5  # Для ранней остановки

    # 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')

    # Class weights for handling class imbalance
    class_counts = np.bincount(y_train)
    class_weights = torch.tensor([1.0 / class_counts[0], 1.0 / class_counts[1]], dtype=torch.float).to(device)
    print(f"Class counts: {class_counts}, Class weights: {class_weights.cpu().numpy()}")

    # Create Datasets and Dataloaders
    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(weight=class_weights)
    optimizer = optim.Adam(model.parameters(), lr=LEARNING_RATE)
    scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, mode='max', factor=0.5, patience=2, verbose=True)

    # Training Loop
    best_val_accuracy = 0
    epochs_no_improve = 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
        )

        scheduler.step(val_accuracy)

        if val_accuracy > best_val_accuracy:
            best_val_accuracy = val_accuracy
            torch.save(model.state_dict(), 'models/best_model.pth')
            epochs_no_improve = 0
        else:
            epochs_no_improve += 1

        print(
            f'Epoch {epoch+1}/{NUM_EPOCHS}, '
            f'LR: {optimizer.param_groups[0]["lr"]:.6f}, '
            f'Train Loss: {train_loss:.4f}, Train Acc: {train_accuracy:.4f}, '
            f'Val Loss: {val_loss:.4f}, Val Acc: {val_accuracy:.4f}'
        )

        # Early stopping
        if epochs_no_improve >= PATIENCE:
            print(f'Early stopping on epoch {epoch+1}')
            break

    # 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]
```


Вот так выглядит обучение:
```
Epoch 1/50, LR: 0.001000, Train Loss: 0.6975, Train Acc: 0.4995, Val Loss: 0.6933, Val Acc: 0.5012
Epoch 2/50, LR: 0.001000, Train Loss: 0.6939, Train Acc: 0.4994, Val Loss: 0.6932, Val Acc: 0.5012
Epoch 3/50, LR: 0.001000, Train Loss: 0.6934, Train Acc: 0.5009, Val Loss: 0.6932, Val Acc: 0.4988
Epoch 4/50, LR: 0.000500, Train Loss: 0.6933, Train Acc: 0.4997, Val Loss: 0.6936, Val Acc: 0.4988
```
Исправь это.
Предложи улучшения модели