""" Regression models for demand forecasting with Optuna hyperparameter tuning. Follows SOLID principles with separation of concerns. """ import numpy as np import pandas as pd from typing import Optional, Dict, Any, Tuple from abc import ABC, abstractmethod import optuna from sklearn.model_selection import cross_val_score, KFold from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor from sklearn.linear_model import Ridge, Lasso, ElasticNet from sklearn.svm import SVR import xgboost as xgb import lightgbm as lgb import pickle class BaseRegressor(ABC): """Abstract base class for regressors (Interface Segregation Principle).""" @abstractmethod def fit(self, X: np.ndarray, y: np.ndarray): """Train the model.""" pass @abstractmethod def predict(self, X: np.ndarray) -> np.ndarray: """Make predictions.""" pass @abstractmethod def get_params(self) -> Dict[str, Any]: """Get model parameters.""" pass class RegressionModelFactory: """Factory for creating regression models (Single Responsibility).""" @staticmethod def create_model(model_type: str, params: Dict[str, Any]): """ Create a regression model based on type. Args: model_type: Type of model ('random_forest', 'xgboost', 'lightgbm', etc.) params: Model parameters Returns: Model instance """ model_type = model_type.lower() if model_type == 'random_forest': return RandomForestRegressor(**params, random_state=42, n_jobs=-1) elif model_type == 'xgboost': return xgb.XGBRegressor(**params, random_state=42, n_jobs=-1) elif model_type == 'lightgbm': return lgb.LGBMRegressor(**params, random_state=42, n_jobs=-1, verbose=-1) elif model_type == 'gradient_boosting': return GradientBoostingRegressor(**params, random_state=42) elif model_type == 'ridge': return Ridge(**params, random_state=42) elif model_type == 'lasso': return Lasso(**params, random_state=42) elif model_type == 'elastic_net': return ElasticNet(**params, random_state=42) elif model_type == 'svr': return SVR(**params) else: raise ValueError(f"Unknown model type: {model_type}") class OptunaRegressor: """Regression model with Optuna hyperparameter optimization.""" def __init__(self, model_type: str = 'random_forest', n_trials: int = 100, cv_folds: int = 5, scoring: str = 'neg_mean_squared_error', random_state: int = 42): """ Initialize Optuna-based regressor. Args: model_type: Type of model to optimize n_trials: Number of Optuna trials cv_folds: Number of cross-validation folds scoring: Scoring metric for optimization random_state: Random seed """ self.model_type = model_type self.n_trials = n_trials self.cv_folds = cv_folds self.scoring = scoring self.random_state = random_state self.best_model = None self.best_params = None self.study = None def _create_trial_params(self, trial: optuna.Trial, model_type: str) -> Dict[str, Any]: """ Create hyperparameter suggestions for a trial. Args: trial: Optuna trial object model_type: Type of model Returns: Dictionary of hyperparameters """ model_type = model_type.lower() if model_type == 'random_forest': return { 'n_estimators': trial.suggest_int('n_estimators', 50, 500), 'max_depth': trial.suggest_int('max_depth', 5, 50), 'min_samples_split': trial.suggest_int('min_samples_split', 2, 20), 'min_samples_leaf': trial.suggest_int('min_samples_leaf', 1, 10), 'max_features': trial.suggest_categorical('max_features', ['sqrt', 'log2', None]) } elif model_type == 'xgboost': return { 'n_estimators': trial.suggest_int('n_estimators', 50, 500), 'max_depth': trial.suggest_int('max_depth', 3, 10), 'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.3, log=True), 'subsample': trial.suggest_float('subsample', 0.6, 1.0), 'colsample_bytree': trial.suggest_float('colsample_bytree', 0.6, 1.0), 'min_child_weight': trial.suggest_int('min_child_weight', 1, 10), 'gamma': trial.suggest_float('gamma', 0, 5), 'reg_alpha': trial.suggest_float('reg_alpha', 0, 10), 'reg_lambda': trial.suggest_float('reg_lambda', 0, 10) } elif model_type == 'lightgbm': return { 'n_estimators': trial.suggest_int('n_estimators', 50, 500), 'max_depth': trial.suggest_int('max_depth', 3, 15), 'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.3, log=True), 'num_leaves': trial.suggest_int('num_leaves', 10, 300), 'subsample': trial.suggest_float('subsample', 0.6, 1.0), 'colsample_bytree': trial.suggest_float('colsample_bytree', 0.6, 1.0), 'min_child_samples': trial.suggest_int('min_child_samples', 5, 100), 'reg_alpha': trial.suggest_float('reg_alpha', 0, 10), 'reg_lambda': trial.suggest_float('reg_lambda', 0, 10) } elif model_type == 'gradient_boosting': return { 'n_estimators': trial.suggest_int('n_estimators', 50, 300), 'max_depth': trial.suggest_int('max_depth', 3, 10), 'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.2, log=True), 'subsample': trial.suggest_float('subsample', 0.6, 1.0), 'min_samples_split': trial.suggest_int('min_samples_split', 2, 20), 'min_samples_leaf': trial.suggest_int('min_samples_leaf', 1, 10) } elif model_type == 'ridge': return { 'alpha': trial.suggest_float('alpha', 0.1, 100, log=True) } elif model_type == 'lasso': return { 'alpha': trial.suggest_float('alpha', 0.1, 100, log=True) } elif model_type == 'elastic_net': return { 'alpha': trial.suggest_float('alpha', 0.1, 100, log=True), 'l1_ratio': trial.suggest_float('l1_ratio', 0.0, 1.0) } elif model_type == 'svr': # Suggest kernel first kernel = trial.suggest_categorical('kernel', ['linear', 'poly', 'rbf', 'sigmoid']) params = { 'kernel': kernel, 'C': trial.suggest_float('C', 0.1, 1000, log=True), 'epsilon': trial.suggest_float('epsilon', 0.01, 1.0, log=True) } # Add gamma for non-linear kernels if kernel in ['rbf', 'poly', 'sigmoid']: params['gamma'] = trial.suggest_categorical('gamma', ['scale', 'auto']) # Add degree for polynomial kernel if kernel == 'poly': params['degree'] = trial.suggest_int('degree', 2, 5) return params else: raise ValueError(f"Unknown model type: {model_type}") def optimize(self, X: np.ndarray, y: np.ndarray) -> optuna.Study: """ Optimize hyperparameters using Optuna. Args: X: Feature matrix y: Target vector Returns: Optuna study object """ def objective(trial): # Get hyperparameters for this trial params = self._create_trial_params(trial, self.model_type) # Create model with trial parameters model = RegressionModelFactory.create_model(self.model_type, params) # Perform cross-validation kfold = KFold(n_splits=self.cv_folds, shuffle=True, random_state=self.random_state) scores = cross_val_score( model, X, y, cv=kfold, scoring=self.scoring, n_jobs=-1 ) # Return mean score (negate if needed) return scores.mean() # Create study and optimize self.study = optuna.create_study( direction='maximize', # We maximize because scoring is neg_mean_squared_error study_name=f'{self.model_type}_optimization', sampler=optuna.samplers.TPESampler(seed=self.random_state) ) print(f"Starting Optuna optimization for {self.model_type} ({self.n_trials} trials)...") self.study.optimize(objective, n_trials=self.n_trials, show_progress_bar=True) # Get best parameters and create best model self.best_params = self.study.best_params self.best_model = RegressionModelFactory.create_model(self.model_type, self.best_params) print(f"\nBest trial: {self.study.best_trial.number}") print(f"Best score: {self.study.best_value:.4f}") print(f"Best params: {self.best_params}") return self.study def fit(self, X: np.ndarray, y: np.ndarray): """ Train the best model found during optimization. Args: X: Feature matrix y: Target vector """ if self.best_model is None: raise ValueError("Model must be optimized before fitting. Call optimize() first.") self.best_model.fit(X, y) def predict(self, X: np.ndarray) -> np.ndarray: """ Make predictions using the best model. Args: X: Feature matrix Returns: Predictions """ if self.best_model is None: raise ValueError("Model must be fitted before prediction. Call fit() first.") return self.best_model.predict(X) def evaluate(self, X: np.ndarray, y: np.ndarray) -> Dict[str, float]: """ Evaluate model performance. Args: X: Feature matrix y: True target values Returns: Dictionary with evaluation metrics """ y_pred = self.predict(X) mse = mean_squared_error(y, y_pred) mae = mean_absolute_error(y, y_pred) rmse = np.sqrt(mse) mape = np.mean(np.abs((y - y_pred) / (y + 1e-8))) * 100 r2 = r2_score(y, y_pred) return { 'MSE': mse, 'MAE': mae, 'RMSE': rmse, 'MAPE': mape, 'R2': r2 } def save_model(self, filepath: str): """Save the trained model to disk as pickle file.""" if self.best_model is None: raise ValueError("No model to save") # Ensure .pkl extension if not filepath.endswith('.pkl'): filepath = filepath.rsplit('.', 1)[0] + '.pkl' model_data = { 'model': self.best_model, 'best_params': self.best_params, 'model_type': self.model_type } with open(filepath, 'wb') as f: pickle.dump(model_data, f) def load_model(self, filepath: str): """Load a trained model from pickle file.""" if not filepath.endswith('.pkl'): filepath = filepath.rsplit('.', 1)[0] + '.pkl' with open(filepath, 'rb') as f: loaded = pickle.load(f) self.best_model = loaded['model'] self.best_params = loaded['best_params'] self.model_type = loaded['model_type']