popov/gogo2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

trying to fix training

Browse Source
This commit is contained in:
Dobromir Popov
2025-03-29 03:53:38 +02:00
parent 2255a8363a
commit ebbc0ed2d7
7 changed files with 533 additions and 304 deletions
Download Patch File Download Diff File Expand all files Collapse all files

254
NN/models/cnn_model_pytorch.py
View File

@ -178,6 +178,148 @@ class CNNModelPyTorch:
logger.info(f"Model built successfully with {sum(p.numel() for p in self.model.parameters())} parameters")
def train_epoch(self, X_train, y_train, batch_size=32):
"""Train for one epoch and return loss and accuracy"""
# Convert to PyTorch tensors
X_train_tensor = torch.tensor(X_train, dtype=torch.float32).to(self.device)
if self.output_size == 1:
y_train_tensor = torch.tensor(y_train, dtype=torch.float32).to(self.device)
else:
y_train_tensor = torch.tensor(y_train, dtype=torch.long).to(self.device)
# Create DataLoader
train_dataset = TensorDataset(X_train_tensor, y_train_tensor)
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
self.model.train()
running_loss = 0.0
correct = 0
total = 0
for inputs, targets in train_loader:
# Zero gradients
self.optimizer.zero_grad()
# Forward pass
outputs = self.model(inputs)
# Calculate loss
if self.output_size == 1:
loss = self.criterion(outputs, targets.unsqueeze(1))
else:
loss = self.criterion(outputs, targets)
# Backward pass and optimize
loss.backward()
self.optimizer.step()
# Statistics
running_loss += loss.item()
if self.output_size > 1:
_, predicted = torch.max(outputs, 1)
total += targets.size(0)
correct += (predicted == targets).sum().item()
epoch_loss = running_loss / len(train_loader)
epoch_acc = correct / total if total > 0 else 0
return epoch_loss, epoch_acc
def evaluate(self, X_val, y_val):
"""Evaluate on validation data and return loss and accuracy"""
X_val_tensor = torch.tensor(X_val, dtype=torch.float32).to(self.device)
if self.output_size == 1:
y_val_tensor = torch.tensor(y_val, dtype=torch.float32).to(self.device)
else:
y_val_tensor = torch.tensor(y_val, dtype=torch.long).to(self.device)
val_dataset = TensorDataset(X_val_tensor, y_val_tensor)
val_loader = DataLoader(val_dataset, batch_size=32)
self.model.eval()
val_loss = 0.0
correct = 0
total = 0
with torch.no_grad():
for inputs, targets in val_loader:
# Forward pass
outputs = self.model(inputs)
# Calculate loss
if self.output_size == 1:
loss = self.criterion(outputs, targets.unsqueeze(1))
else:
loss = self.criterion(outputs, targets)
val_loss += loss.item()
# Calculate accuracy
if self.output_size > 1:
_, predicted = torch.max(outputs, 1)
total += targets.size(0)
correct += (predicted == targets).sum().item()
return val_loss / len(val_loader), correct / total if total > 0 else 0
def predict(self, X):
"""Make predictions on input data"""
self.model.eval()
X_tensor = torch.tensor(X, dtype=torch.float32).to(self.device)
with torch.no_grad():
outputs = self.model(X_tensor)
if self.output_size > 1:
_, predicted = torch.max(outputs, 1)
return predicted.cpu().numpy()
else:
return outputs.cpu().numpy()
def predict_next_candles(self, X, n_candles=3):
"""
Predict the next n candles for each timeframe.
Args:
X: Input data of shape [batch_size, window_size, features]
n_candles: Number of future candles to predict
Returns:
Dictionary of predictions for each timeframe
"""
self.model.eval()
X_tensor = torch.tensor(X, dtype=torch.float32).to(self.device)
with torch.no_grad():
# Get the last window of data
last_window = X_tensor[-1:] # [1, window_size, features]
# Initialize predictions
predictions = {}
# For each timeframe, predict next n candles
for i, tf in enumerate(self.timeframes):
# Extract features for this timeframe
tf_features = last_window[:, :, i*5:(i+1)*5] # [1, window_size, 5]
# Predict next n candles
tf_predictions = []
current_window = tf_features
for _ in range(n_candles):
# Get prediction for next candle
output = self.model(current_window)
tf_predictions.append(output.cpu().numpy())
# Update window for next prediction
current_window = torch.cat([
current_window[:, 1:, :],
output.unsqueeze(1)
], dim=1)
predictions[tf] = np.concatenate(tf_predictions, axis=0)
return predictions
def train(self, X_train, y_train, X_val=None, y_val=None, batch_size=32, epochs=100):
"""
Train the CNN model.
@ -259,7 +401,7 @@ class CNNModelPyTorch:
# Validation phase
if val_loader is not None:
val_loss, val_acc = self._validate(val_loader)
val_loss, val_acc = self.evaluate(X_val, y_val)
logger.info(f"Epoch {epoch+1}/{epochs} - "
f"loss: {epoch_loss:.4f} - acc: {epoch_acc:.4f} - "
@ -281,51 +423,12 @@ class CNNModelPyTorch:
logger.info("Training completed")
return self.history
def _validate(self, val_loader):
"""Validate the model using the validation set"""
self.model.eval()
val_loss = 0.0
correct = 0
total = 0
with torch.no_grad():
for inputs, targets in val_loader:
# Forward pass
outputs = self.model(inputs)
# Calculate loss
if self.output_size == 1:
loss = self.criterion(outputs, targets.unsqueeze(1))
else:
loss = self.criterion(outputs, targets)
val_loss += loss.item()
# Calculate accuracy
if self.output_size > 1:
_, predicted = torch.max(outputs, 1)
total += targets.size(0)
correct += (predicted == targets).sum().item()
return val_loss / len(val_loader), correct / total if total > 0 else 0
def evaluate(self, X_test, y_test):
def evaluate_metrics(self, X_test, y_test):
"""
Evaluate the model on test data.
Args:
X_test: Test input data
y_test: Test target data
Returns:
dict: Evaluation metrics
Calculate and return comprehensive evaluation metrics as dict
"""
logger.info(f"Evaluating model on {len(X_test)} samples")
# Convert to PyTorch tensors
X_test_tensor = torch.tensor(X_test, dtype=torch.float32).to(self.device)
# Get predictions
self.model.eval()
with torch.no_grad():
y_pred = self.model(X_test_tensor)
@ -336,70 +439,15 @@ class CNNModelPyTorch:
else:
y_pred_class = (y_pred.cpu().numpy() > 0.5).astype(int).flatten()
# Calculate metrics
if self.output_size > 1:
accuracy = accuracy_score(y_test, y_pred_class)
precision = precision_score(y_test, y_pred_class, average='weighted')
recall = recall_score(y_test, y_pred_class, average='weighted')
f1 = f1_score(y_test, y_pred_class, average='weighted')
metrics = {
'accuracy': accuracy,
'precision': precision,
'recall': recall,
'f1_score': f1
}
else:
accuracy = accuracy_score(y_test, y_pred_class)
precision = precision_score(y_test, y_pred_class)
recall = recall_score(y_test, y_pred_class)
f1 = f1_score(y_test, y_pred_class)
metrics = {
'accuracy': accuracy,
'precision': precision,
'recall': recall,
'f1_score': f1
}
metrics = {
'accuracy': accuracy_score(y_test, y_pred_class),
'precision': precision_score(y_test, y_pred_class, average='weighted', zero_division=0),
'recall': recall_score(y_test, y_pred_class, average='weighted', zero_division=0),
'f1_score': f1_score(y_test, y_pred_class, average='weighted', zero_division=0)
}
logger.info(f"Evaluation metrics: {metrics}")
return metrics
def predict(self, X):
"""
Make predictions with the model.
Args:
X: Input data
Returns:
Predictions
"""
# Convert to PyTorch tensor
X_tensor = torch.tensor(X, dtype=torch.float32).to(self.device)
# Get predictions
self.model.eval()
with torch.no_grad():
predictions = self.model(X_tensor)
if self.output_size > 1:
# Multi-class classification
probs = predictions.cpu().numpy()
_, class_preds = torch.max(predictions, 1)
class_preds = class_preds.cpu().numpy()
return class_preds, probs
else:
# Binary classification or regression
preds = predictions.cpu().numpy()
if self.output_size == 1:
# Binary classification
class_preds = (preds > 0.5).astype(int)
return class_preds.flatten(), preds.flatten()
else:
# Regression
return preds.flatten(), None
def save(self, filepath):
"""
Save the model to a file.