USER
Моя модель для предсказания 101 шага от приведенных 100 шагов цены Bitcoin.
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
```
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]
```
data_preparation.py:
```py
# data_preparation.py
import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler
def load_data(file_path):
"""
Load BTCUSDT data from a CSV file.
Expected columns: ['timestamp', 'close']
"""
df = pd.read_csv(file_path, parse_dates=['timestamp'])
df.sort_values('timestamp', inplace=True)
return df
def preprocess_data(df):
"""
Preprocess the data:
- Calculate price change deltas.
- Remove outliers beyond 3 standard deviations.
- Normalize using MinMaxScaler.
- Prepare sequences and labels.
"""
# Calculate price change deltas
df['delta'] = df['close'].pct_change().fillna(0)
# Remove outliers
mean = df['delta'].mean()
std = df['delta'].std()
threshold = 3 * std
df = df[np.abs(df['delta'] - mean) <= threshold]
# Normalize deltas
scaler = MinMaxScaler()
df['delta_scaled'] = scaler.fit_transform(df[['delta']])
# Prepare sequences of 100 steps and labels
sequences = []
labels = []
data = df['delta_scaled'].values
for i in range(len(data) - 101): # Changed from 100 to 101 to prevent IndexError
seq = data[i:i+100]
delta_next = data[i+100] - data[i+99]
label = 1 if delta_next > 0 else 0 # Up or Down
sequences.append(seq)
labels.append(label)
sequences = np.array(sequences)
labels = np.array(labels)
return sequences, labels, scaler
def split_data(sequences, labels):
"""
Split data into training, validation, and test sets.
"""
num_samples = sequences.shape[0]
train_end = int(num_samples * 0.7)
val_end = int(num_samples * 0.85)
X_train = sequences[:train_end]
y_train = labels[:train_end]
X_val = sequences[train_end:val_end]
y_val = labels[train_end:val_end]
X_test = sequences[val_end:]
y_test = labels[val_end:]
return X_train, y_train, X_val, y_val, X_test, y_test
def save_data(X_train, y_train, X_val, y_val, X_test, y_test, scaler):
"""
Save processed data and scaler for future use.
"""
np.save('data/X_train.npy', X_train)
np.save('data/y_train.npy', y_train)
np.save('data/X_val.npy', X_val)
np.save('data/y_val.npy', y_val)
np.save('data/X_test.npy', X_test)
np.save('data/y_test.npy', y_test)
# Save scaler
import joblib
joblib.dump(scaler, 'data/scaler.save')
if __name__ == "__main__":
# Ensure you have the data file in the specified path
df = load_data('data/btcusdt_5min.csv')
sequences, labels, scaler = preprocess_data(df)
X_train, y_train, X_val, y_val, X_test, y_test = split_data(sequences, labels)
save_data(X_train, y_train, X_val, y_val, X_test, y_test, scaler)
print("Data preparation completed.")
```
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.")
```
backtest.py:
```py
# backtest.py
import torch
from torch.utils.data import DataLoader
from model import PriceDirectionTransformer
from utils import BTCUSDTDataset
import numpy as np
import pandas as pd
import joblib
def backtest(model, sequences, prices, fees=0.001):
"""
Perform backtesting on the given sequences and price data.
"""
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()
# Simulate trades
positions = []
profits = []
for i in range(len(predictions)):
# The price at which we enter the trade
entry_price = prices[i + 99] # Adjusted index to align with sequence
# The price at which we exit the trade
exit_price = prices[i + 100] if i + 100 < len(prices) else prices[-1]
if predictions[i] == 1: # Predict Up
entry_price *= (1 + fees)
exit_price = (1 - fees)
profit = (exit_price - entry_price) / entry_price
else: # Predict Down
entry_price = (1 - fees)
exit_price = (1 + fees)
profit = (entry_price - exit_price) / entry_price
positions.append(predictions[i])
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 = df.iloc[-(len(X_test)+101):] # Corresponding to X_test sequences and next prices
prices = df['close'].values # Use 'close' prices
# Load Scaler if needed
# scaler = joblib.load('data/scaler.save')
# Initialize Model
model = PriceDirectionTransformer()
model.load_state_dict(torch.load('models/best_model.pth'))
model.to('cuda' if torch.cuda.is_available() else 'cpu')
# Perform Backtesting
profits, cumulative_returns = backtest(model, X_test, prices)
# Calculate Metrics
total_return = cumulative_returns[-1]
win_rate = (np.array(profits) > 0).sum() / len(profits)
max_drawdown = np.min(cumulative_returns)
print(f'Total Return: {total_return/100:.2f}%')
print(f'Win Rate: {win_rate/100:.2f}%')
print(f'Max Drawdown: {max_drawdown/100:.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()
```
Вот что получилось в backtest.py:
```
Total Return: -223.45%
Win Rate: 0.00%
Max Drawdown: -223.45%
```
Что в модели не так?
Можешь провести анализ и сказать что не так, а также предоставить улучшениеASSISTANT
The poor performance of your model and backtest results stems from several issues in your code, particularly in the data preparation and backtesting stages. Below, I'll analyze the problems and provide suggested improvements to help you get better results.
## Issues in Your Code
### 1. Incorrect Label Calculation in Data Preparation
**Problem:**
In your `data_preparation.py`, the labels are calculated based on the difference between two scaled deltas:
```python
delta_next = data[i+100] - data[i+99]
label = 1 if delta_next > 0 else 0
```
Since `data[i]` represents the scaled percentage change (delta) at time `i`, subtracting one scaled delta from another doesn't provide meaningful information about future price direction. This approach results in labels that don't accurately represent whether the price is moving up or down.
**Solution:**
Labels should be calculated based on the actual price movement between two time points. Specifically, you should compare the prices at `i+99` and `i+100` to determine if the price went up or down. Here's how you can fix it:
```python
prices = df['close'].values
for i in range(len(data) - 100):
seq = data[i:i + 100]
# The label is whether the price increased from time i+99 to i+100
price_last = prices[i + 99]
price_next = prices[i + 100]
label = 1 if price_next > price_last else 0
sequences.append(seq)
labels.append(label)
```
### 2. Faulty Profit Calculation in Backtesting
**Problem:**
In your `backtest.py`, the profit calculation is incorrect. You're setting `exit_price` to constants `(1 - fees)` or `(1 + fees)`, which doesn't make sense in the context of actual price movements:
```python
if predictions[i] == 1: # Predict Up
entry_price *= (1 + fees)
exit_price = (1 - fees)
profit = (exit_price - entry_price) / entry_price
else: # Predict Down
entry_price = (1 - fees)
exit_price = (1 + fees)
profit = (entry_price - exit_price) / entry_price
```
This approach ignores the actual exit prices and miscalculates the profits, leading to erroneous backtest results.
**Solution:**
Revise the profit calculation to use the actual `entry_price` and `exit_price`, adjusting for fees appropriately. Here's the corrected code:
```python
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)
```
### 3. Misalignment of Data Indices
**Problem:**
The indices used when aligning sequences with prices may be off by one due to zero-based indexing, potentially causing misalignment between your sequences and corresponding prices.
**Solution:**
Ensure that your indices align correctly by carefully adjusting them based on the context. For example, when preparing sequences and labels, if you use `data[i:i+100]`, then `prices[i + 99]` corresponds to the last price in the sequence, and `prices[i + 100]` is the next price.
### 4. Potential Data Leakage
**Problem:**
If data is not properly shuffled or split, there could be data leakage between training, validation, and test sets, leading to overfitting and poor generalization.
**Solution:**
Ensure that your data splitting is done correctly, and avoid any overlap between the datasets. Use proper shuffling and consider time-based split if applicable.
### 5. Model Architecture and Hyperparameters
**Problem:**
The model may not be sufficiently complex, or the hyperparameters may not be optimal for capturing the patterns in the data.
**Solution:**
Experiment with different model architectures and hyperparameters. Consider increasing the number of layers, using different activation functions, or trying other model types like LSTM or GRU that are often used for time series data.
## Corrected and Improved Code
### Updated `data_preparation.py`
```python
# data_preparation.py
import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler
def load_data(file_path):
df = pd.read_csv(file_path, parse_dates=['timestamp'])
df.sort_values('timestamp', inplace=True)
df.reset_index(drop=True, inplace=True)
return df
def preprocess_data(df):
# Calculate price change deltas
df['delta'] = df['close'].pct_change().fillna(0)
# Remove outliers
mean = df['delta'].mean()
std = df['delta'].std()
threshold = 3 * std
df = df[np.abs(df['delta'] - mean) <= threshold]
df.reset_index(drop=True, inplace=True)
# Normalize deltas
scaler = MinMaxScaler()
df['delta_scaled'] = scaler.fit_transform(df[['delta']])
# Prepare sequences of 100 steps and labels
sequences = []
labels = []
data = df['delta_scaled'].values
prices = df['close'].values
for i in range(len(data) - 100):
seq = data[i:i + 100]
# The label is whether the price increased from time i+99 to i+100
price_last = prices[i + 99]
price_next = prices[i + 100]
label = 1 if price_next > price_last else 0
sequences.append(seq)
labels.append(label)
sequences = np.array(sequences)
labels = np.array(labels)
return sequences, labels, scaler
def split_data(sequences, labels):
num_samples = sequences.shape[0]
train_end = int(num_samples * 0.7)
val_end = int(num_samples * 0.85)
X_train = sequences[:train_end]
y_train = labels[:train_end]
X_val = sequences[train_end:val_end]
y_val = labels[train_end:val_end]
X_test = sequences[val_end:]
y_test = labels[val_end:]
return X_train, y_train, X_val, y_val, X_test, y_test
def save_data(X_train, y_train, X_val, y_val, X_test, y_test, scaler):
np.save('data/X_train.npy', X_train)
np.save('data/y_train.npy', y_train)
np.save('data/X_val.npy', X_val)
np.save('data/y_val.npy', y_val)
np.save('data/X_test.npy', X_test)
np.save('data/y_test.npy', y_test)
import joblib
joblib.dump(scaler, 'data/scaler.save')
if __name__ == "__main__":
df = load_data('data/btcusdt_5min.csv')
sequences, labels, scaler = preprocess_data(df)
X_train, y_train, X_val, y_val, X_test, y_test = split_data(sequences, labels)
save_data(X_train, y_train, X_val, y_val, X_test, y_test, scaler)
print("Data preparation completed.")
```
### Updated `backtest.py`
```python
# 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()
```
**Note:** Adjust the path `'models/best_model.pth'` if necessary, and ensure that the model's `.to(device)` and `map_location` match your device configuration.
## Additional Recommendations
- **Model Architecture:** Consider experimenting with different model architectures, such as Long Short-Term Memory (LSTM) networks, which are well-suited for time-series data.
- **Hyperparameter Tuning:** Adjust hyperparameters like learning rate, number of layers, and batch size. Using techniques like grid search or Bayesian optimization can help find optimal values.
- **Feature Engineering:** Incorporate additional features that might help the model, such as technical indicators (e.g., moving averages, RSI, MACD).
- **Data Augmentation:** Use techniques to augment your dataset if it's not sufficiently large, such as adding noise or simulating additional data.
- **Validation Strategy:** Since financial data is time-dependent, consider using time-based cross-validation techniques, like walk-forward validation, to better assess model performance.
- **Regularization:** Apply regularization methods like dropout or L2 regularization to prevent overfitting.
- **Early Stopping:** Monitor validation loss and implement early stopping during training to prevent overfitting.
- **Normalization Consistency:** Ensure that the same scaler used during training is applied during backtesting and inference to maintain consistency.
- **Logging and Monitoring:** Use logging libraries or tensorboard to keep track of your training metrics over epochs.
By addressing the issues in your code and implementing these improvements, you should see better model performance and more realistic backtesting results. Remember that predicting financial markets is inherently challenging due to their complexity and the presence of random noise, so continuous experimentation and refinement are key.