"""
Cerebrum Forex - RandomForest Model
Ensemble tree-based model for signal prediction.
Replaces CatBoost for sklearn 1.8+ compatibility.
"""
import logging
import pickle
from pathlib import Path
from typing import Tuple
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import balanced_accuracy_score
from config.settings import default_settings, IS_FROZEN
from .base_model import BaseModel
logger = logging.getLogger(__name__)
[docs]
class RandomForestModel(BaseModel):
"""RandomForest model for forex signal prediction"""
def __init__(self, timeframe: str, model_dir: Path):
super().__init__(timeframe, model_dir)
self.params = {
'n_estimators': 500, # Number of trees
'max_depth': 10, # Max tree depth
'min_samples_split': 10, # Min samples to split
'min_samples_leaf': 5, # Min samples per leaf
'max_features': 'sqrt', # Features per split
'bootstrap': True, # Use bootstrap sampling
'max_samples': 0.8, # Use 80% of samples per tree (simulated regularization/early stopping)
'class_weight': 'balanced', # Handle imbalanced classes
'random_state': 42,
'n_jobs': 1 if IS_FROZEN else 2, # Throttled to prevent CPU contention
'verbose': 0,
}
@property
def name(self) -> str:
return "randomforest"
[docs]
def train(self, X: np.ndarray, y: np.ndarray,
X_val: np.ndarray = None, y_val: np.ndarray = None,
class_weights: dict = None) -> float:
"""Train RandomForest model with optional external validation set and class weights"""
try:
# Ensure y is 1D array
y = np.asarray(y).ravel()
# Detect number of classes dynamically
unique_classes = np.unique(y)
num_classes = len(unique_classes)
logger.info(f"[RandomForest {self.timeframe}] Detected {num_classes} classes: {unique_classes}")
logger.info(f"[RandomForest {self.timeframe}] Training with {len(X)} samples, {X.shape[1]} features")
# Use external val set if provided, else split internally
if X_val is None or y_val is None:
logger.info(f"[RandomForest {self.timeframe}] No external val set, doing internal split")
try:
X_train, X_val, y_train, y_val = train_test_split(
X, y, test_size=default_settings.validation_ratio, random_state=42, stratify=y
)
logger.info(f"[RandomForest {self.timeframe}] Stratified split OK")
except ValueError as e:
logger.warning(f"[RandomForest {self.timeframe}] Stratified failed: {e}, using regular split")
X_train, X_val, y_train, y_val = train_test_split(
X, y, test_size=default_settings.validation_ratio, random_state=42
)
else:
X_train, y_train = X, y
y_val = np.asarray(y_val).ravel()
# Log shapes after split
logger.info(f"[RandomForest {self.timeframe}] After split: X_train={X_train.shape}, X_val={X_val.shape}")
# Apply class weights if provided (override balanced)
if class_weights is not None and isinstance(class_weights, dict):
self.params['class_weight'] = class_weights
# Create and train model
self.model = RandomForestClassifier(**self.params)
logger.info(f"[RandomForest {self.timeframe}] Calling model.fit()...")
self.model.fit(X_train, y_train)
# Evaluate
y_pred = self.model.predict(X_val)
self.accuracy = balanced_accuracy_score(y_val.ravel(), y_pred.ravel())
self.is_trained = True
logger.info(f"[RandomForest {self.timeframe}] ✓ Balanced Accuracy: {self.accuracy:.2%}")
# Save model
self.save()
return self.accuracy
except Exception as e:
logger.error(f"RandomForest training failed: {e}", exc_info=True)
return 0.0
[docs]
def predict(self, X: np.ndarray) -> Tuple[str, float]:
"""Make prediction with RandomForest"""
if not self.is_trained and not self.load():
logger.warning("RandomForest model not trained")
return "NEUTRAL", 0.0
try:
# Handle Feature Mismatch (Train vs Predict)
# 1. New System
expected_features = self.feature_names
# 2. Legacy Fallback
if not expected_features and hasattr(self.model, 'feature_names_in_'):
expected_features = self.model.feature_names_in_
if hasattr(X, 'columns') and expected_features:
# Check if we have all required features
missing = [f for f in expected_features if f not in X.columns]
if missing:
logger.warning(f"RandomForest mismatch: Missing {len(missing)} features ({missing[:3]}...). Returning NEUTRAL.")
return "NEUTRAL", 0.0
# Check for "Unnamed" features (Legacy)
if len(expected_features) > 0 and str(expected_features[0]).startswith("Column_") and not str(X.columns[0]).startswith("Column_"):
logger.warning(f"RandomForest schema mismatch: Model expects raw features (Column_X) but got named features. Returning NEUTRAL.")
return "NEUTRAL", 0.0
# Select only expected columns in correct order
X = X[expected_features]
# Ensure 2D input
if hasattr(X, 'values'):
X = X.values
if X.ndim == 1:
X = X.reshape(1, -1)
# Get probabilities
proba = self.model.predict_proba(X)
pred_class = np.argmax(proba, axis=1)[0]
confidence = proba[0][pred_class]
signal = self.signal_from_prediction(pred_class)
return signal, float(confidence)
except Exception as e:
logger.error(f"RandomForest prediction failed: {e}")
return "NEUTRAL", 0.0
[docs]
def load(self) -> bool:
"""Load model from pickle file"""
if not self.model_path.exists():
return False
try:
with open(self.model_path, 'rb') as f:
data = pickle.load(f)
self.model = data['model']
self.accuracy = data.get('accuracy', 0.0)
self.params = data.get('params', self.params)
self.feature_names = data.get('feature_names', [])
self.is_trained = True
logger.info(f"RandomForest model loaded from {self.model_path} ({len(self.feature_names)} features)")
return True
except Exception as e:
logger.error(f"Failed to load RandomForest model: {e}")
return False