Predict Demand Spikes with statsforecast and scikit-learn

                      
                     
"""
Production implementation for predicting demand spikes using StatsForecast and Scikit-Learn.
Provides secure, scalable operations.
"""
from typing import Dict, Any, List
import os
import logging
import pandas as pd
import requests
from statsforecast import StatsForecast
from statsforecast.models import ETS, ARIMA
from sklearn.preprocessing import MinMaxScaler
from sqlalchemy import create_engine

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Config:
    """
    Configuration class to hold environment variables.
    """
    database_url: str = os.getenv('DATABASE_URL')
    api_url: str = os.getenv('API_URL')

    def __init__(self):
        if not self.database_url:
            raise ValueError('DATABASE_URL must be set.')
        if not self.api_url:
            raise ValueError('API_URL must be set.')

async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate the input data for prediction.
    
    Args:
        data: Input dictionary containing demand data.
    Returns:
        bool: True if valid.
    Raises:
        ValueError: If validation fails.
    """
    if 'historical_data' not in data:
        raise ValueError('Missing historical_data key in input.')
    if not isinstance(data['historical_data'], list):
        raise ValueError('historical_data must be a list.')
    return True

async def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize the input fields to prevent issues.
    
    Args:
        data: Input dictionary.
    Returns:
        dict: Sanitized input data.
    """
    # Example sanitization logic
    data['historical_data'] = [float(x) for x in data['historical_data']]
    return data

async def fetch_data(url: str) -> pd.DataFrame:
    """Fetch data from a given API endpoint.
    
    Args:
        url: The API URL to fetch data from.
    Returns:
        pd.DataFrame: Data fetched from the API.
    Raises:
        Exception: If the fetch fails.
    """
    try:
        response = requests.get(url)
        response.raise_for_status()
        return pd.DataFrame(response.json())
    except Exception as e:
        logger.error(f'Error fetching data: {e}')
        raise

async def save_to_db(data: pd.DataFrame, table_name: str) -> None:
    """Save data to the specified database table.
    
    Args:
        data: DataFrame to save.
        table_name: Name of the database table.
    Raises:
        Exception: If the save fails.
    """
    try:
        engine = create_engine(Config().database_url)
        data.to_sql(table_name, con=engine, if_exists='replace', index=False)
        logger.info(f'Data saved to {table_name} successfully.')
    except Exception as e:
        logger.error(f'Error saving data to DB: {e}')
        raise

async def normalize_data(data: List[float]) -> List[float]:
    """Normalize data using Min-Max scaling.
    
    Args:
        data: List of historical demand data.
    Returns:
        List[float]: Normalized demand data.
    """
    scaler = MinMaxScaler()
    normalized = scaler.fit_transform(pd.DataFrame(data))
    return normalized.flatten().tolist()

async def process_batch(data: List[float]) -> pd.DataFrame:
    """Process the batch of historical data for prediction.
    
    Args:
        data: List of historical demand data.
    Returns:
        pd.DataFrame: DataFrame with processed data.
    """
    try:
        normalized_data = await normalize_data(data)
        df = pd.DataFrame(normalized_data, columns=['demand'])
        return df
    except Exception as e:
        logger.error(f'Error processing batch data: {e}')
        raise

async def predict_demand(data: pd.DataFrame) -> pd.DataFrame:
    """Predict future demand using StatsForecast.
    
    Args:
        data: DataFrame with processed data.
    Returns:
        pd.DataFrame: Predictions for future demand.
    """
    try:
        model = StatsForecast(df=data, models=[ARIMA(), ETS()])
        predictions = model.forecast(h=5)  # Forecast for next 5 periods
        return predictions
    except Exception as e:
        logger.error(f'Error predicting demand: {e}')
        raise

async def aggregate_metrics(predictions: pd.DataFrame) -> Dict[str, float]:
    """Aggregate metrics from predictions.
    
    Args:
        predictions: DataFrame with predicted values.
    Returns:
        dict: Aggregated metrics.
    """
    metrics = {
        'mean': predictions['demand'].mean(),
        'stddev': predictions['demand'].std()
    }
    return metrics

class DemandPredictor:
    """Main orchestrator for demand prediction workflow.
    """
    def __init__(self, config: Config):
        self.config = config

    async def run(self, input_data: Dict[str, Any]) -> None:
        """Run the demand prediction workflow.
        
        Args:
            input_data: Input data for prediction.
        """
        try:
            await validate_input(input_data)  # Validate input data
            sanitized_data = await sanitize_fields(input_data)
            historical_data = sanitized_data['historical_data']
            processed_data = await process_batch(historical_data)  # Process data
            predictions = await predict_demand(processed_data)  # Predict demand
            await save_to_db(predictions, 'demand_predictions')  # Save to DB
            metrics = await aggregate_metrics(predictions)  # Aggregate metrics
            logger.info(f'Aggregated metrics: {metrics}')
        except Exception as e:
            logger.error(f'Error in workflow: {e}')
            raise

if __name__ == '__main__':
    # Example usage
    config = Config()
    predictor = DemandPredictor(config)
    input_data = {'historical_data': [100, 150, 200, 250, 300]}
    import asyncio
    asyncio.run(predictor.run(input_data))