unify model names
This commit is contained in:
Dobromir Popov
@@ -6,8 +6,6 @@ Much larger and more sophisticated architecture for better learning
|
||||
|
||||
import os
|
||||
import logging
|
||||
import numpy as np
|
||||
import matplotlib.pyplot as plt
|
||||
from datetime import datetime
|
||||
import math
|
||||
|
||||
@@ -15,10 +13,30 @@ import torch
|
||||
import torch.nn as nn
|
||||
import torch.optim as optim
|
||||
from torch.utils.data import DataLoader, TensorDataset
|
||||
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
|
||||
import torch.nn.functional as F
|
||||
from typing import Dict, Any, Optional, Tuple
|
||||
|
||||
# Try to import optional dependencies
|
||||
try:
|
||||
import numpy as np
|
||||
HAS_NUMPY = True
|
||||
except ImportError:
|
||||
np = None
|
||||
HAS_NUMPY = False
|
||||
|
||||
try:
|
||||
import matplotlib.pyplot as plt
|
||||
HAS_MATPLOTLIB = True
|
||||
except ImportError:
|
||||
plt = None
|
||||
HAS_MATPLOTLIB = False
|
||||
|
||||
try:
|
||||
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
|
||||
HAS_SKLEARN = True
|
||||
except ImportError:
|
||||
HAS_SKLEARN = False
|
||||
|
||||
# Import checkpoint management
|
||||
from NN.training.model_manager import save_checkpoint, load_best_checkpoint
|
||||
from NN.training.model_manager import create_model_manager
|
||||
@@ -423,38 +441,46 @@ class EnhancedCNNModel(nn.Module):
|
||||
'features': self._memory_barrier(processed_features)
|
||||
}
|
||||
|
||||
def predict(self, feature_matrix: np.ndarray) -> Dict[str, Any]:
|
||||
def predict(self, feature_matrix) -> Dict[str, Any]:
|
||||
"""
|
||||
Make predictions on feature matrix
|
||||
Args:
|
||||
feature_matrix: numpy array of shape [sequence_length, features]
|
||||
feature_matrix: tensor or numpy array of shape [sequence_length, features]
|
||||
Returns:
|
||||
Dictionary with prediction results
|
||||
"""
|
||||
self.eval()
|
||||
|
||||
|
||||
with torch.no_grad():
|
||||
# Convert to tensor and add batch dimension
|
||||
if isinstance(feature_matrix, np.ndarray):
|
||||
if HAS_NUMPY and isinstance(feature_matrix, np.ndarray):
|
||||
x = torch.FloatTensor(feature_matrix).unsqueeze(0) # Add batch dim
|
||||
else:
|
||||
elif isinstance(feature_matrix, torch.Tensor):
|
||||
x = feature_matrix.unsqueeze(0)
|
||||
|
||||
else:
|
||||
x = torch.FloatTensor(feature_matrix).unsqueeze(0)
|
||||
|
||||
# Move to device
|
||||
device = next(self.parameters()).device
|
||||
x = x.to(device)
|
||||
|
||||
|
||||
# Forward pass
|
||||
outputs = self.forward(x)
|
||||
|
||||
|
||||
# Extract results with proper shape handling
|
||||
probs = outputs['probabilities'].cpu().numpy()[0]
|
||||
confidence_tensor = outputs['confidence'].cpu().numpy()
|
||||
regime = outputs['regime'].cpu().numpy()[0]
|
||||
volatility = outputs['volatility'].cpu().numpy()
|
||||
|
||||
if HAS_NUMPY:
|
||||
probs = outputs['probabilities'].cpu().numpy()[0]
|
||||
confidence_tensor = outputs['confidence'].cpu().numpy()
|
||||
regime = outputs['regime'].cpu().numpy()[0]
|
||||
volatility = outputs['volatility'].cpu().numpy()
|
||||
else:
|
||||
probs = outputs['probabilities'].cpu().tolist()[0]
|
||||
confidence_tensor = outputs['confidence'].cpu().tolist()
|
||||
regime = outputs['regime'].cpu().tolist()[0]
|
||||
volatility = outputs['volatility'].cpu().tolist()
|
||||
|
||||
# Handle confidence shape properly
|
||||
if isinstance(confidence_tensor, np.ndarray):
|
||||
if HAS_NUMPY and isinstance(confidence_tensor, np.ndarray):
|
||||
if confidence_tensor.ndim == 0:
|
||||
confidence = float(confidence_tensor.item())
|
||||
elif confidence_tensor.size == 1:
|
||||
@@ -465,7 +491,7 @@ class EnhancedCNNModel(nn.Module):
|
||||
confidence = float(confidence_tensor)
|
||||
|
||||
# Handle volatility shape properly
|
||||
if isinstance(volatility, np.ndarray):
|
||||
if HAS_NUMPY and isinstance(volatility, np.ndarray):
|
||||
if volatility.ndim == 0:
|
||||
volatility = float(volatility.item())
|
||||
elif volatility.size == 1:
|
||||
@@ -474,20 +500,29 @@ class EnhancedCNNModel(nn.Module):
|
||||
volatility = float(volatility[0] if len(volatility) > 0 else 0.0)
|
||||
else:
|
||||
volatility = float(volatility)
|
||||
|
||||
|
||||
# Determine action (0=BUY, 1=SELL for 2-action system)
|
||||
action = int(np.argmax(probs))
|
||||
if HAS_NUMPY:
|
||||
action = int(np.argmax(probs))
|
||||
else:
|
||||
action = int(torch.argmax(torch.tensor(probs)).item())
|
||||
action_confidence = float(probs[action])
|
||||
|
||||
|
||||
# Convert logits to list
|
||||
if HAS_NUMPY:
|
||||
raw_logits = outputs['logits'].cpu().numpy()[0].tolist()
|
||||
else:
|
||||
raw_logits = outputs['logits'].cpu().tolist()[0]
|
||||
|
||||
return {
|
||||
'action': action,
|
||||
'action_name': 'BUY' if action == 0 else 'SELL',
|
||||
'confidence': float(confidence),
|
||||
'action_confidence': action_confidence,
|
||||
'probabilities': probs.tolist(),
|
||||
'regime_probabilities': regime.tolist(),
|
||||
'probabilities': probs if isinstance(probs, list) else probs.tolist(),
|
||||
'regime_probabilities': regime if isinstance(regime, list) else regime.tolist(),
|
||||
'volatility_prediction': float(volatility),
|
||||
'raw_logits': outputs['logits'].cpu().numpy()[0].tolist()
|
||||
'raw_logits': raw_logits
|
||||
}
|
||||
|
||||
def get_memory_usage(self) -> Dict[str, Any]:
|
||||
|
||||
@@ -15,11 +15,19 @@ Architecture:
|
||||
import torch
|
||||
import torch.nn as nn
|
||||
import torch.nn.functional as F
|
||||
import numpy as np
|
||||
import logging
|
||||
from typing import Dict, List, Optional, Tuple, Any
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
# Try to import numpy, but provide fallback if not available
|
||||
try:
|
||||
import numpy as np
|
||||
HAS_NUMPY = True
|
||||
except ImportError:
|
||||
np = None
|
||||
HAS_NUMPY = False
|
||||
logging.warning("NumPy not available - COB RL model will have limited functionality")
|
||||
|
||||
from .model_interfaces import ModelInterface
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -164,45 +172,54 @@ class MassiveRLNetwork(nn.Module):
|
||||
'features': x # Hidden features for analysis
|
||||
}
|
||||
|
||||
def predict(self, cob_features: np.ndarray) -> Dict[str, Any]:
|
||||
def predict(self, cob_features) -> Dict[str, Any]:
|
||||
"""
|
||||
High-level prediction method for COB features
|
||||
|
||||
|
||||
Args:
|
||||
cob_features: COB features as numpy array [input_size]
|
||||
|
||||
cob_features: COB features as tensor or numpy array [input_size]
|
||||
|
||||
Returns:
|
||||
Dict containing prediction results
|
||||
"""
|
||||
self.eval()
|
||||
with torch.no_grad():
|
||||
# Convert to tensor and add batch dimension
|
||||
if isinstance(cob_features, np.ndarray):
|
||||
if HAS_NUMPY and isinstance(cob_features, np.ndarray):
|
||||
x = torch.from_numpy(cob_features).float()
|
||||
else:
|
||||
elif isinstance(cob_features, torch.Tensor):
|
||||
x = cob_features.float()
|
||||
|
||||
else:
|
||||
# Try to convert from list or other format
|
||||
x = torch.tensor(cob_features, dtype=torch.float32)
|
||||
|
||||
if x.dim() == 1:
|
||||
x = x.unsqueeze(0) # Add batch dimension
|
||||
|
||||
|
||||
# Move to device
|
||||
device = next(self.parameters()).device
|
||||
x = x.to(device)
|
||||
|
||||
|
||||
# Forward pass
|
||||
outputs = self.forward(x)
|
||||
|
||||
|
||||
# Process outputs
|
||||
price_probs = F.softmax(outputs['price_logits'], dim=1)
|
||||
predicted_direction = torch.argmax(price_probs, dim=1).item()
|
||||
confidence = outputs['confidence'].item()
|
||||
value = outputs['value'].item()
|
||||
|
||||
|
||||
# Convert probabilities to list (works with or without numpy)
|
||||
if HAS_NUMPY:
|
||||
probabilities = price_probs.cpu().numpy()[0].tolist()
|
||||
else:
|
||||
probabilities = price_probs.cpu().tolist()[0]
|
||||
|
||||
return {
|
||||
'predicted_direction': predicted_direction, # 0=DOWN, 1=SIDEWAYS, 2=UP
|
||||
'confidence': confidence,
|
||||
'value': value,
|
||||
'probabilities': price_probs.cpu().numpy()[0],
|
||||
'probabilities': probabilities,
|
||||
'direction_text': ['DOWN', 'SIDEWAYS', 'UP'][predicted_direction]
|
||||
}
|
||||
|
||||
@@ -250,36 +267,45 @@ class COBRLModelInterface(ModelInterface):
|
||||
|
||||
logger.info(f"COB RL Model Interface initialized on {self.device}")
|
||||
|
||||
def predict(self, cob_features: np.ndarray) -> Dict[str, Any]:
|
||||
def predict(self, cob_features) -> Dict[str, Any]:
|
||||
"""Make prediction using the model"""
|
||||
self.model.eval()
|
||||
with torch.no_grad():
|
||||
# Convert to tensor and add batch dimension
|
||||
if isinstance(cob_features, np.ndarray):
|
||||
if HAS_NUMPY and isinstance(cob_features, np.ndarray):
|
||||
x = torch.from_numpy(cob_features).float()
|
||||
else:
|
||||
elif isinstance(cob_features, torch.Tensor):
|
||||
x = cob_features.float()
|
||||
|
||||
else:
|
||||
# Try to convert from list or other format
|
||||
x = torch.tensor(cob_features, dtype=torch.float32)
|
||||
|
||||
if x.dim() == 1:
|
||||
x = x.unsqueeze(0) # Add batch dimension
|
||||
|
||||
|
||||
# Move to device
|
||||
x = x.to(self.device)
|
||||
|
||||
|
||||
# Forward pass
|
||||
outputs = self.model(x)
|
||||
|
||||
|
||||
# Process outputs
|
||||
price_probs = F.softmax(outputs['price_logits'], dim=1)
|
||||
predicted_direction = torch.argmax(price_probs, dim=1).item()
|
||||
confidence = outputs['confidence'].item()
|
||||
value = outputs['value'].item()
|
||||
|
||||
|
||||
# Convert probabilities to list (works with or without numpy)
|
||||
if HAS_NUMPY:
|
||||
probabilities = price_probs.cpu().numpy()[0].tolist()
|
||||
else:
|
||||
probabilities = price_probs.cpu().tolist()[0]
|
||||
|
||||
return {
|
||||
'predicted_direction': predicted_direction, # 0=DOWN, 1=SIDEWAYS, 2=UP
|
||||
'confidence': confidence,
|
||||
'value': value,
|
||||
'probabilities': price_probs.cpu().numpy()[0],
|
||||
'probabilities': probabilities,
|
||||
'direction_text': ['DOWN', 'SIDEWAYS', 'UP'][predicted_direction]
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user