Capacity Planning

# capacity_planning/forecasting.py
import pandas as pd
from prophet import Prophet
from dataclasses import dataclass
from typing import List, Dict, Tuple
import numpy as np

@dataclass
class ForecastResult:
    timestamp: pd.Timestamp
    predicted_load: float
    lower_bound: float
    upper_bound: float
    confidence: float

class DemandForecaster:
    """
    ML-powered demand forecasting using
    Prophet with custom seasonality.
    """
    
    def __init__(self):
        self.model = None
        self.seasonality_config = {
            "weekly": True,
            "daily": True,
            "yearly": True,
            "custom": [
                {"name": "monthly_sales", "period": 30.5, "fourier_order": 5},
                {"name": "end_of_quarter", "period": 91.25, "fourier_order": 3}
            ]
        }
    
    def train(
        self,
        historical_data: pd.DataFrame,
        metric: str = "request_rate"
    ) -> None:
        """
        Train forecasting model on
        historical load data.
        """
        
        # Prepare data for Prophet
        df = historical_data[["timestamp", metric]].copy()
        df.columns = ["ds", "y"]
        
        # Initialize Prophet with tuned hyperparameters
        self.model = Prophet(
            changepoint_prior_scale=0.05,
            seasonality_prior_scale=10,
            holidays_prior_scale=10,
            seasonality_mode="multiplicative",
            interval_width=0.95
        )
        
        # Add custom seasonality
        for custom in self.seasonality_config["custom"]:
            self.model.add_seasonality(
                name=custom["name"],
                period=custom["period"],
                fourier_order=custom["fourier_order"]
            )
        
        # Add special events (sales, launches)
        self.model.add_country_holidays(country_name="US")
        
        # Fit model
        self.model.fit(df)
    
    def forecast(
        self,
        horizon_hours: int = 168,  # 1 week
        granularity_minutes: int = 15
    ) -> List[ForecastResult]:
        """
        Generate demand forecast for
        specified time horizon.
        """
        
        if not self.model:
            raise ValueError("Model not trained. Call train() first.")
        
        # Create future dataframe
        periods = (horizon_hours * 60) // granularity_minutes
        future = self.model.make_future_dataframe(
            periods=periods,
            freq=f"{granularity_minutes}min"
        )
        
        # Generate predictions
        forecast = self.model.predict(future)
        
        # Extract results for future dates
        results = []
        for _, row in forecast.tail(periods).iterrows():
            results.append(ForecastResult(
                timestamp=row["ds"],
                predicted_load=max(0, row["yhat"]),
                lower_bound=max(0, row["yhat_lower"]),
                upper_bound=row["yhat_upper"],
                confidence=0.95
            ))
        
        return results
    
    def detect_anomalies(
        self,
        actual_data: pd.DataFrame,
        threshold_std: float = 3.0
    ) -> List[Dict]:
        """
        Detect load anomalies by comparing
        actual vs predicted values.
        """
        
        predictions = self.model.predict(actual_data[["ds"]])
        
        residuals = actual_data["y"] - predictions["yhat"]
        std = residuals.std()
        
        anomalies = []
        for i, (_, row) in enumerate(actual_data.iterrows()):
            if abs(residuals.iloc[i]) > threshold_std * std:
                anomalies.append({
                    "timestamp": row["ds"],
                    "actual": row["y"],
                    "predicted": predictions["yhat"].iloc[i],
                    "deviation_std": residuals.iloc[i] / std
                })
        
        return anomalies