Project structure:

o1_made/
├── models/
│   ├── best_model.pth
│   └── final_model.pth
├── backtest.py
├── data_download.py
├── data_preparation.py
├── model.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
```


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


Вот атк выглядит обучение модели:
```
Epoch 1/20, Train Loss: 0.5551, Train Acc: 0.7365, Val Loss: 0.5531, Val Acc: 0.7430
Epoch 2/20, Train Loss: 0.5406, Train Acc: 0.7559, Val Loss: 0.5454, Val Acc: 0.7492
Epoch 3/20, Train Loss: 0.5389, Train Acc: 0.7578, Val Loss: 0.5497, Val Acc: 0.7447
Epoch 4/20, Train Loss: 0.5384, Train Acc: 0.7579, Val Loss: 0.5442, Val Acc: 0.7538
Epoch 5/20, Train Loss: 0.5380, Train Acc: 0.7590, Val Loss: 0.5435, Val Acc: 0.7543
Epoch 6/20, Train Loss: 0.5396, Train Acc: 0.7579, Val Loss: 0.5431, Val Acc: 0.7550
Epoch 7/20, Train Loss: 0.5384, Train Acc: 0.7578, Val Loss: 0.5427, Val Acc: 0.7550
Epoch 8/20, Train Loss: 0.5379, Train Acc: 0.7586, Val Loss: 0.5481, Val Acc: 0.7488
Epoch 9/20, Train Loss: 0.5377, Train Acc: 0.7588, Val Loss: 0.5426, Val Acc: 0.7546
Epoch 10/20, Train Loss: 0.5377, Train Acc: 0.7586, Val Loss: 0.5422, Val Acc: 0.7542
Epoch 11/20, Train Loss: 0.5372, Train Acc: 0.7592, Val Loss: 0.5477, Val Acc: 0.7478
Epoch 12/20, Train Loss: 0.5375, Train Acc: 0.7584, Val Loss: 0.5480, Val Acc: 0.7495
Epoch 13/20, Train Loss: 0.5375, Train Acc: 0.7594, Val Loss: 0.5444, Val Acc: 0.7518
Epoch 14/20, Train Loss: 0.5372, Train Acc: 0.7595, Val Loss: 0.5441, Val Acc: 0.7515
Epoch 15/20, Train Loss: 0.5373, Train Acc: 0.7590, Val Loss: 0.5429, Val Acc: 0.7539
Epoch 16/20, Train Loss: 0.5370, Train Acc: 0.7596, Val Loss: 0.5438, Val Acc: 0.7539
Epoch 17/20, Train Loss: 0.5369, Train Acc: 0.7595, Val Loss: 0.5424, Val Acc: 0.7541
Epoch 18/20, Train Loss: 0.5367, Train Acc: 0.7593, Val Loss: 0.5425, Val Acc: 0.7537
Epoch 19/20, Train Loss: 0.5368, Train Acc: 0.7597, Val Loss: 0.5430, Val Acc: 0.7541
Epoch 20/20, Train Loss: 0.5368, Train Acc: 0.7603, Val Loss: 0.5421, Val Acc: 0.7541
Training completed.
```
Будем честны, это очень плохо.

Вот так выглядит тестирование backtest:
```
Total Return: -9313.39%
Win Rate: 5.77%
Max Drawdown: -9313.39%
```
Там линейный график, чем больше `Trade Number`, тем меньше `Cumulative Return`