Detect and Explain Seasonal Anomalies in Production Output with Merlion and Prophet

"""
Production implementation for detecting and explaining seasonal anomalies in production output.
Utilizes Merlion for anomaly detection and Prophet for forecasting.
"""
from typing import Dict, Any, List
import os
import logging
import time
import pandas as pd
from merlion.evaluate import Evaluator
from merlion.models import Prophet
from merlion.utils import TimeSeries
from sqlalchemy import create_engine, text

# Set up logging for monitoring the application
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Config:
    """
    Configuration class to manage environment variables.
    """
    database_url: str = os.getenv('DATABASE_URL')
    retry_attempts: int = int(os.getenv('RETRY_ATTEMPTS', 3))
    retry_delay: float = float(os.getenv('RETRY_DELAY', 1.0))  # seconds

# Function to validate input data structure
async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate input data for processing.
    
    Args:
        data: Input dictionary to validate.
    Returns:
        True if valid.
    Raises:
        ValueError: If validation fails.
    """
    if 'production_output' not in data:
        raise ValueError('Missing production_output key in input data')
    return True

# Function to fetch data from the database
async def fetch_data(engine: str, query: str) -> pd.DataFrame:
    """Fetch data from the database.
    
    Args:
        engine: Database connection string.
        query: SQL query string.
    Returns:
        DataFrame containing fetched data.
    Raises:
        Exception: If fetching data fails.
    """
    try:
        logger.info('Fetching data from the database')  # Log fetching action
        with create_engine(engine).connect() as connection:
            return pd.read_sql(text(query), connection)
    except Exception as e:
        logger.error(f'Error fetching data: {e}')  # Log error
        raise

# Function to save processed results back to the database
async def save_to_db(engine: str, df: pd.DataFrame, table_name: str) -> None:
    """Save DataFrame to the specified database table.
    
    Args:
        engine: Database connection string.
        df: DataFrame to save.
        table_name: Target table name.
    Raises:
        Exception: If saving data fails.
    """
    try:
        logger.info('Saving data to the database')  # Log saving action
        with create_engine(engine).connect() as connection:
            df.to_sql(table_name, connection, if_exists='replace', index=False)
    except Exception as e:
        logger.error(f'Error saving data: {e}')  # Log error
        raise

# Function to transform records into the required format for analysis
def transform_records(data: pd.DataFrame) -> TimeSeries:
    """Transform DataFrame records into a TimeSeries object for analysis.
    
    Args:
        data: Raw DataFrame containing production data.
    Returns:
        TimeSeries object ready for anomaly detection.
    """
    # Assuming the DataFrame has a 'timestamp' column and 'production_output' column
    data['timestamp'] = pd.to_datetime(data['timestamp'])  # Convert timestamps
    series = TimeSeries.from_dataframe(data, time_col='timestamp', value_col='production_output')
    return series

# Function to analyze seasonal anomalies using Merlion and Prophet
async def analyze_seasonal_anomalies(series: TimeSeries) -> List[Dict[str, Any]]:
    """Analyze seasonal anomalies in the TimeSeries data.
    
    Args:
        series: TimeSeries object to analyze.
    Returns:
        List of anomalies detected with explanations.
    """
    model = Prophet()  # Initialize Prophet model
    model.fit(series)  # Fit model to the time series
    forecast = model.predict()  # Predict future values
    evaluator = Evaluator()  # Initialize evaluator for anomaly detection
    anomalies = evaluator.evaluate(series, forecast)  # Evaluate anomalies
    logger.info(f'Detected {len(anomalies)} anomalies')  # Log number of anomalies
    return anomalies

# Main orchestrator class
class AnomalyDetector:
    """Main orchestrator for detecting and explaining seasonal anomalies.
    """
    def __init__(self):
        self.config = Config()  # Load configuration

    async def run_detection(self) -> None:
        """Run the anomaly detection workflow.
        """
        try:
            query = 'SELECT timestamp, production_output FROM production_data;'  # SQL query to fetch data
            data = await fetch_data(self.config.database_url, query)  # Fetch data
            await validate_input(data.to_dict())  # Validate input data

            series = transform_records(data)  # Transform data
            anomalies = await analyze_seasonal_anomalies(series)  # Analyze anomalies

            await save_to_db(self.config.database_url, pd.DataFrame(anomalies), 'detected_anomalies')  # Save results
        except Exception as e:
            logger.error(f'An error occurred during detection: {e}')  # Log error

if __name__ == '__main__':
    # Example usage of the AnomalyDetector class
    detector = AnomalyDetector()  # Initialize detector
    import asyncio
    asyncio.run(detector.run_detection())  # Run detection asynchronously