""" Regression/Classification Service Handles XGBoost model training, persistence, and prediction. Mirrors ProphetService pattern: 3-layer cache, .joblib disk, ml_regression_models DB table. """ import os import json import threading from typing import Dict, List, Optional, Any, Tuple from datetime import datetime import pandas as pd import numpy as np import joblib from app.utils.helpers import get_logger from app.models.mysql_model import MySQLModel class RegressionService: """Unified service for XGBoost classifier and regressor models.""" _cache: Dict[str, Any] = {} _cache_lock = threading.Lock() _table_ensured = False _table_lock = threading.Lock() CLASSIFIER_DEFAULTS = { 'n_estimators': 200, 'max_depth': 4, 'learning_rate': 0.05, 'eval_metric': 'logloss', 'random_state': 42, } REGRESSOR_DEFAULTS = { 'n_estimators': 200, 'max_depth': 4, 'learning_rate': 0.05, 'objective': 'reg:squarederror', 'random_state': 42, } RISK_THRESHOLDS_COLLECTION = {'HIGH': 0.60, 'MEDIUM': 0.35} RISK_THRESHOLDS_SALES = {'LOW': 75.0, 'MEDIUM': 90.0} def __init__(self): self.logger = get_logger("regression_service") self._db = MySQLModel() self._ensure_table() self._ensure_model_dir() # ------------------------------------------------------------------ # Private helpers # ------------------------------------------------------------------ def _get_model_dir(self) -> str: from flask import current_app return current_app.config.get("REGRESSION_MODEL_DIR", os.path.join(current_app.config.get("PROPHET_MODEL_DIR", "storage/app"), "regression")) def _ensure_model_dir(self) -> None: os.makedirs(self._get_model_dir(), exist_ok=True) def _ensure_table(self) -> None: if RegressionService._table_ensured: return with RegressionService._table_lock: if RegressionService._table_ensured: return ddl = """ CREATE TABLE IF NOT EXISTS `ml_regression_models` ( `id` INT UNSIGNED NOT NULL AUTO_INCREMENT, `model_id` VARCHAR(120) NOT NULL, `file_path` VARCHAR(512) NOT NULL, `model_type` ENUM('classifier','regressor') NOT NULL, `config` JSON NOT NULL, `features` JSON NOT NULL, `data_points` INT UNSIGNED NOT NULL DEFAULT 0, `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (`id`), UNIQUE KEY `uq_model_id` (`model_id`), KEY `idx_model_type` (`model_type`), KEY `idx_created_at` (`created_at`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """ self._db.execute_query(ddl) RegressionService._table_ensured = True def _model_file_path(self, model_id: str) -> str: safe_id = "".join(c if c.isalnum() or c in ("-", "_") else "_" for c in model_id) return os.path.join(self._get_model_dir(), f"{safe_id}.joblib") def _save_to_disk(self, model_id: str, payload: Dict) -> str: file_path = self._model_file_path(model_id) joblib.dump(payload, file_path, compress=3) return file_path def _load_from_disk(self, model_id: str, file_path: str) -> Dict: if not os.path.isfile(file_path): raise FileNotFoundError(f"Model file for '{model_id}' not found at '{file_path}'") return joblib.load(file_path) def _upsert_metadata(self, model_id: str, file_path: str, model_type: str, config: dict, features: List[str], data_points: int) -> None: sql = """ INSERT INTO ml_regression_models (model_id, file_path, model_type, config, features, data_points) VALUES (%s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE file_path=VALUES(file_path), model_type=VALUES(model_type), config=VALUES(config), features=VALUES(features), data_points=VALUES(data_points), updated_at=CURRENT_TIMESTAMP """ self._db.execute_query(sql, ( model_id, file_path, model_type, json.dumps(config), json.dumps(features), data_points )) def _fetch_metadata(self, model_id: str) -> Optional[Dict]: rows = self._db.execute_query( "SELECT * FROM ml_regression_models WHERE model_id = %s LIMIT 1", (model_id,) ) if not rows: return None row = rows[0] if isinstance(row.get("config"), str): row["config"] = json.loads(row["config"]) if isinstance(row.get("features"), str): row["features"] = json.loads(row["features"]) return row def _resolve_model(self, model_id: str) -> Dict: """Three-layer fetch: process cache → DB metadata + disk.""" with RegressionService._cache_lock: if model_id in RegressionService._cache: return RegressionService._cache[model_id] meta = self._fetch_metadata(model_id) if meta is None: raise ValueError(f"Model '{model_id}' not found") try: # Derive path from model_id against the current model dir (portable across # host/container moves); stored file_path is audit-only, not load-bearing. payload = self._load_from_disk(model_id, self._model_file_path(model_id)) except FileNotFoundError: self._db.execute_query( "DELETE FROM ml_regression_models WHERE model_id = %s", (model_id,) ) raise RuntimeError( f"Model '{model_id}' file missing. Orphaned record removed. Re-train required." ) with RegressionService._cache_lock: RegressionService._cache[model_id] = payload return payload # ------------------------------------------------------------------ # Public methods # ------------------------------------------------------------------ def train( self, model_id: str, X: pd.DataFrame, y: pd.Series, model_type: str = 'classifier', config: Optional[Dict] = None ) -> Dict[str, Any]: """ Train an XGBoost model. Args: model_id: Unique identifier for this model X: Feature DataFrame y: Target Series model_type: 'classifier' or 'regressor' config: Override default XGBoost hyperparameters Returns: {model_id, status, data_points, features, model_type, metrics} """ try: from xgboost import XGBClassifier, XGBRegressor from sklearn.preprocessing import StandardScaler from sklearn.model_selection import train_test_split from sklearn.metrics import (classification_report, mean_absolute_error, r2_score, accuracy_score) if len(X) < 5: raise ValueError(f"Insufficient data ({len(X)} rows, minimum 5 required)") features = list(X.columns) if model_type == 'classifier': defaults = dict(self.CLASSIFIER_DEFAULTS) defaults.update(config or {}) pos = int(y.sum()) neg = len(y) - pos if pos > 0 and neg > 0: defaults['scale_pos_weight'] = neg / pos model = XGBClassifier(**{k: v for k, v in defaults.items() if k != 'random_state'}) else: defaults = dict(self.REGRESSOR_DEFAULTS) defaults.update(config or {}) model = XGBRegressor(**{k: v for k, v in defaults.items() if k != 'random_state'}) scaler = StandardScaler() X_scaled = scaler.fit_transform(X.fillna(0)) # Hold-out metrics when enough data metrics = {} if len(X) >= 10: X_tr, X_te, y_tr, y_te = train_test_split( X_scaled, y, test_size=0.2, random_state=42 ) model.fit(X_tr, y_tr) if model_type == 'classifier': preds = model.predict(X_te) metrics['accuracy'] = round(float(accuracy_score(y_te, preds)), 4) else: preds = model.predict(X_te) metrics['mae'] = round(float(mean_absolute_error(y_te, preds)), 4) metrics['r2'] = round(float(r2_score(y_te, preds)), 4) # Refit on full data model.fit(X_scaled, y) else: model.fit(X_scaled, y) payload = { 'model': model, 'scaler': scaler, 'features': features, 'model_type': model_type, 'config': defaults, 'data_points': len(X), 'created_at': datetime.now(), } with RegressionService._cache_lock: file_path = self._save_to_disk(model_id, payload) self._upsert_metadata(model_id, file_path, model_type, defaults, features, len(X)) RegressionService._cache[model_id] = payload self.logger.info(f"Trained {model_type} model '{model_id}' on {len(X)} rows") return { 'model_id': model_id, 'status': 'trained', 'model_type': model_type, 'data_points': len(X), 'features': features, 'metrics': metrics, } except Exception as e: self.logger.error(f"Error training model '{model_id}': {str(e)}") raise def predict_proba(self, model_id: str, X: pd.DataFrame) -> List[float]: """Return class-1 probabilities for a classifier model.""" payload = self._resolve_model(model_id) if payload['model_type'] != 'classifier': raise ValueError(f"Model '{model_id}' is not a classifier") X_scaled = payload['scaler'].transform(X[payload['features']].fillna(0)) return payload['model'].predict_proba(X_scaled)[:, 1].tolist() def predict(self, model_id: str, X: pd.DataFrame) -> List[float]: """Return regression predictions.""" payload = self._resolve_model(model_id) if payload['model_type'] != 'regressor': raise ValueError(f"Model '{model_id}' is not a regressor") X_scaled = payload['scaler'].transform(X[payload['features']].fillna(0)) return payload['model'].predict(X_scaled).tolist() def get_model_info(self, model_id: str) -> Optional[Dict]: with RegressionService._cache_lock: if model_id in RegressionService._cache: c = RegressionService._cache[model_id] return { 'model_id': model_id, 'model_type': c['model_type'], 'data_points': c['data_points'], 'features': c['features'], 'config': c['config'], 'created_at': c['created_at'].isoformat(), } meta = self._fetch_metadata(model_id) if meta is None: return None return { 'model_id': meta['model_id'], 'model_type': meta['model_type'], 'data_points': meta['data_points'], 'features': meta['features'], 'config': meta['config'], 'created_at': str(meta['created_at']), 'updated_at': str(meta['updated_at']), } def list_models(self) -> List[Dict]: rows = self._db.execute_query( "SELECT model_id, model_type, data_points, features, created_at, updated_at " "FROM ml_regression_models ORDER BY created_at DESC" ) result = [] for row in rows: features = row['features'] if isinstance(row['features'], list) else json.loads(row['features']) result.append({ 'model_id': row['model_id'], 'model_type': row['model_type'], 'data_points': row['data_points'], 'features': features, 'created_at': str(row['created_at']), 'updated_at': str(row['updated_at']), }) return result def delete_model(self, model_id: str) -> bool: meta = self._fetch_metadata(model_id) with RegressionService._cache_lock: in_cache = model_id in RegressionService._cache if meta is None and not in_cache: return False with RegressionService._cache_lock: RegressionService._cache.pop(model_id, None) if meta: try: file_path = self._model_file_path(model_id) if os.path.isfile(file_path): os.remove(file_path) except OSError as e: self.logger.warning(f"Could not delete model file: {e}") self._db.execute_query( "DELETE FROM ml_regression_models WHERE model_id = %s", (model_id,) ) return True # ------------------------------------------------------------------ # Static risk classification helpers # ------------------------------------------------------------------ @staticmethod def collection_risk_level(probability: float) -> str: if probability >= RegressionService.RISK_THRESHOLDS_COLLECTION['HIGH']: return 'HIGH' if probability >= RegressionService.RISK_THRESHOLDS_COLLECTION['MEDIUM']: return 'MEDIUM' return 'LOW' @staticmethod def sales_target_risk_level(attainment_pct: float) -> str: if attainment_pct <= RegressionService.RISK_THRESHOLDS_SALES['LOW']: return 'LOW' if attainment_pct <= RegressionService.RISK_THRESHOLDS_SALES['MEDIUM']: return 'MEDIUM' return 'HIGH'