""" Prophet Service Module Handles Facebook Prophet time series forecasting operations """ import os import json import logging import threading from typing import Dict, List, Optional, Any, Tuple, Union from datetime import datetime, timedelta import pandas as pd import numpy as np from prophet import Prophet import base64 import io import joblib import matplotlib.pyplot as plt from app.utils.helpers import get_logger from app.models.mysql_model import MySQLModel # pandas 2.2+ renamed period-end frequency aliases _FREQ_ALIASES: Dict[str, str] = {'M': 'ME', 'Q': 'QE', 'Y': 'YE', 'A': 'YE'} class ProphetService: """Service for handling Facebook Prophet forecasting operations""" _cache: Dict[str, Any] = {} # shared across all instances in same process _cache_lock = threading.Lock() # thread-safety for _cache _table_ensured = False # DDL guard — run CREATE TABLE only once per process _table_lock = threading.Lock() def __init__(self): self.logger = get_logger("prophet_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["PROPHET_MODEL_DIR"] def _ensure_model_dir(self) -> None: os.makedirs(self._get_model_dir(), exist_ok=True) def _ensure_table(self) -> None: if ProphetService._table_ensured: return with ProphetService._table_lock: if ProphetService._table_ensured: return ddl = """ CREATE TABLE IF NOT EXISTS `ml_prophet_models` ( `id` INT UNSIGNED NOT NULL AUTO_INCREMENT, `model_id` VARCHAR(120) NOT NULL, `file_path` VARCHAR(512) NOT NULL, `config` JSON NOT NULL, `data_points` INT UNSIGNED NOT NULL DEFAULT 0, `train_start` DATE NOT NULL, `train_end` DATE NOT NULL, `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_created_at` (`created_at`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """ self._db.execute_query(ddl) ProphetService._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, file_path, config, data_points, train_start, train_end): sql = """ INSERT INTO ml_prophet_models (model_id, file_path, config, data_points, train_start, train_end) VALUES (%s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE file_path=VALUES(file_path), config=VALUES(config), data_points=VALUES(data_points), train_start=VALUES(train_start), train_end=VALUES(train_end), updated_at=CURRENT_TIMESTAMP """ self._db.execute_query(sql, (model_id, file_path, json.dumps(config), data_points, train_start, train_end)) def _fetch_metadata(self, model_id: str) -> Optional[Dict]: rows = self._db.execute_query( "SELECT * FROM ml_prophet_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"]) return row def _resolve_model(self, model_id: str) -> Dict: """Three-layer fetch: process cache → DB metadata + disk.""" # Layer 1: process cache (zero I/O) with ProphetService._cache_lock: if model_id in ProphetService._cache: return ProphetService._cache[model_id] # Layer 2: DB metadata + disk 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-heal: orphaned DB record — remove it self._db.execute_query("DELETE FROM ml_prophet_models WHERE model_id = %s", (model_id,)) raise RuntimeError( f"Model '{model_id}' file is missing. Orphaned record removed. Please re-train." ) # Warm cache for subsequent requests in this worker with ProphetService._cache_lock: ProphetService._cache[model_id] = payload return payload # ------------------------------------------------------------------ # Data preparation # ------------------------------------------------------------------ def _prepare_dataframe(self, data: List[Dict], date_column: str = 'ds', value_column: str = 'y') -> pd.DataFrame: """ Prepare data for Prophet model Args: data: List of dictionaries with date and value columns date_column: Name of the date column value_column: Name of the value column Returns: Pandas DataFrame formatted for Prophet """ try: df = pd.DataFrame(data) # Ensure date column is datetime if date_column in df.columns: df[date_column] = pd.to_datetime(df[date_column]) df = df.sort_values(date_column) # Ensure value column is numeric if value_column in df.columns: df[value_column] = pd.to_numeric(df[value_column], errors='coerce') # Remove rows with NaN values df = df.dropna() self.logger.info(f"Prepared DataFrame with {len(df)} rows for Prophet") return df except Exception as e: self.logger.error(f"Error preparing DataFrame: {str(e)}") raise ValueError(f"Invalid data format: {str(e)}") # ------------------------------------------------------------------ # Public methods # ------------------------------------------------------------------ def train_model( self, data: List[Dict], model_id: Optional[str] = None, config: Optional[Dict] = None ) -> Dict[str, Any]: """ Train a Prophet model Args: data: Time series data model_id: Optional model identifier for storage config: Prophet configuration parameters Returns: Training results with model info """ try: # Prepare data df = self._prepare_dataframe(data) if len(df) < 2: raise ValueError("Insufficient data for training (minimum 2 data points required)") # Default configuration prophet_config = { 'seasonality_mode': 'additive', 'yearly_seasonality': True, 'weekly_seasonality': True, 'daily_seasonality': False, 'changepoint_prior_scale': 0.05, 'seasonality_prior_scale': 10.0, 'holidays_prior_scale': 10.0, 'changepoint_range': 0.8, 'interval_width': 0.80, } # Override with user config if config: prophet_config.update(config) # Initialize and fit model model = Prophet(**prophet_config) model.fit(df) # Generate model ID if not provided if not model_id: model_id = f"prophet_{datetime.now().strftime('%Y%m%d_%H%M%S')}" train_start = df['ds'].min().strftime('%Y-%m-%d') train_end = df['ds'].max().strftime('%Y-%m-%d') payload = { 'model': model, 'config': prophet_config, 'training_data': df, 'created_at': datetime.now(), 'data_points': len(df), 'train_start': train_start, 'train_end': train_end, } with ProphetService._cache_lock: file_path = self._save_to_disk(model_id, payload) self._upsert_metadata(model_id, file_path, prophet_config, len(df), train_start, train_end) ProphetService._cache[model_id] = payload self.logger.info(f"Trained Prophet model {model_id} with {len(df)} data points") return { 'model_id': model_id, 'status': 'trained', 'data_points': len(df), 'config': prophet_config, 'training_date_range': { 'start': df['ds'].min().isoformat(), 'end': df['ds'].max().isoformat() } } except Exception as e: self.logger.error(f"Error training Prophet model: {str(e)}") raise def generate_forecast( self, model_id: str, periods: int = 30, freq: str = 'D', include_history: bool = True ) -> Dict[str, Any]: """ Generate forecast using trained model Args: model_id: ID of the trained model periods: Number of periods to forecast freq: Frequency of forecast ('D' for daily, 'H' for hourly, etc.) include_history: Whether to include historical data in response Returns: Forecast results """ try: model_info = self._resolve_model(model_id) model = model_info['model'] # Normalize legacy pandas freq alias (e.g. 'M' → 'ME' for pandas 2.2+) freq = _FREQ_ALIASES.get(freq, freq) # Create future dataframe future = model.make_future_dataframe(periods=periods, freq=freq) # Generate forecast forecast = model.predict(future) # Prepare response — row-oriented list so PHP can iterate directly. # Prophet's make_future_dataframe produces len(df) history rows + `periods` future rows. # When include_history=False, slice to future rows only. start_idx = len(model_info['training_data']) if not include_history else 0 forecast_rows = [] for i in range(start_idx, len(forecast)): forecast_rows.append({ 'ds': forecast['ds'].iloc[i].strftime('%Y-%m-%d'), 'yhat': max(0.0, round(float(forecast['yhat'].iloc[i]), 4)), 'yhat_lower': max(0.0, round(float(forecast['yhat_lower'].iloc[i]), 4)), 'yhat_upper': max(0.0, round(float(forecast['yhat_upper'].iloc[i]), 4)), }) result = { 'model_id': model_id, 'forecast_periods': periods, 'frequency': freq, 'forecast': forecast_rows, 'generated_at': datetime.now().isoformat() } if include_history: training_data = model_info['training_data'] result['historical_data'] = { 'ds': training_data['ds'].dt.strftime('%Y-%m-%d').tolist(), 'y': training_data['y'].round(4).tolist() } self.logger.info(f"Generated forecast for model {model_id}: {periods} periods") return result except Exception as e: self.logger.error(f"Error generating forecast: {str(e)}") raise def generate_plot( self, model_id: str, periods: int = 30, freq: str = 'D', width: int = 800, height: int = 600 ) -> str: """ Generate forecast plot as base64 encoded image Args: model_id: ID of the trained model periods: Number of periods to forecast freq: Frequency of forecast width: Plot width in pixels height: Plot height in pixels Returns: Base64 encoded PNG image """ try: model_info = self._resolve_model(model_id) model = model_info['model'] # Normalize legacy pandas freq alias (e.g. 'M' → 'ME' for pandas 2.2+) freq = _FREQ_ALIASES.get(freq, freq) # Create future dataframe and forecast future = model.make_future_dataframe(periods=periods, freq=freq) forecast = model.predict(future) # Create plot fig = model.plot(forecast, figsize=(width/100, height/100)) # Convert to base64 buffer = io.BytesIO() fig.savefig(buffer, format='png', dpi=100, bbox_inches='tight') buffer.seek(0) image_base64 = base64.b64encode(buffer.getvalue()).decode('utf-8') buffer.close() plt.close(fig) self.logger.info(f"Generated plot for model {model_id}") return f"data:image/png;base64,{image_base64}" except Exception as e: self.logger.error(f"Error generating plot: {str(e)}") raise def get_model_info(self, model_id: str) -> Optional[Dict[str, Any]]: """ Get information about a trained model Args: model_id: ID of the model Returns: Model information or None if not found """ with ProphetService._cache_lock: if model_id in ProphetService._cache: c = ProphetService._cache[model_id] return { 'data_points': c['data_points'], 'config': c['config'], 'created_at': c['created_at'].isoformat(), 'train_start': c.get('train_start'), 'train_end': c.get('train_end'), } meta = self._fetch_metadata(model_id) if meta is None: return None return { 'data_points': meta['data_points'], 'config': meta['config'], 'created_at': str(meta['created_at']), 'train_start': str(meta['train_start']), 'train_end': str(meta['train_end']), } def list_models(self) -> List[Dict[str, Any]]: """ List all trained models Returns: List of model summaries """ rows = self._db.execute_query( "SELECT model_id, data_points, config, train_start, train_end, created_at " "FROM ml_prophet_models ORDER BY created_at DESC" ) result = [] for row in rows: config = row['config'] if isinstance(row['config'], dict) else json.loads(row['config']) result.append({ 'model_id': row['model_id'], 'data_points': row['data_points'], 'config': config, 'train_start': str(row['train_start']), 'train_end': str(row['train_end']), 'created_at': str(row['created_at']), }) return result def delete_model(self, model_id: str) -> bool: """ Delete a trained model Args: model_id: ID of the model to delete Returns: True if deleted, False if not found """ meta = self._fetch_metadata(model_id) with ProphetService._cache_lock: in_cache = model_id in ProphetService._cache if meta is None and not in_cache: return False with ProphetService._cache_lock: ProphetService._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_prophet_models WHERE model_id = %s", (model_id,)) return True def validate_data_format(self, data: List[Dict]) -> Tuple[bool, str]: """ Validate input data format for Prophet Args: data: Input data to validate Returns: Tuple of (is_valid, error_message) """ if not isinstance(data, list) or len(data) < 2: return False, "Data must be a list with at least 2 data points" required_keys = {'ds', 'y'} for i, item in enumerate(data): if not isinstance(item, dict): return False, f"Data point {i} must be a dictionary" if not required_keys.issubset(item.keys()): return False, f"Data point {i} must contain 'ds' and 'y' keys" # Validate date format try: pd.to_datetime(item['ds']) except: return False, f"Invalid date format in data point {i}" # Validate numeric value try: float(item['y']) except: return False, f"Invalid numeric value in data point {i}" return True, ""