""" Time Series Forecasting Service Uses Prophet and statistical models for business forecasting """ import logging from datetime import datetime, timedelta from typing import List, Dict, Any, Optional import numpy as np import pandas as pd from scipy import stats logger = logging.getLogger(__name__) class ForecastService: """Business forecasting using time series analysis""" def __init__(self): self.prophet_available = False try: from prophet import Prophet self.Prophet = Prophet self.prophet_available = True logger.info("Prophet is available for forecasting") except ImportError: logger.warning("Prophet not available, using fallback methods") async def forecast_messages( self, historical_data: List[Dict[str, Any]], periods: int = 30 ) -> Dict[str, Any]: """ Forecast message volume Args: historical_data: List of dicts with 'date' and 'count' periods: Number of days to forecast Returns: Forecast results """ if not historical_data: return self._empty_forecast() # Convert to DataFrame df = pd.DataFrame(historical_data) df['ds'] = pd.to_datetime(df['date']) df['y'] = df['count'] if self.prophet_available and len(df) >= 14: return await self._forecast_with_prophet(df, periods) else: return await self._forecast_with_statsmodels(df, periods) async def _forecast_with_prophet( self, df: pd.DataFrame, periods: int ) -> Dict[str, Any]: """Use Prophet for forecasting""" try: model = self.Prophet( yearly_seasonality=True, weekly_seasonality=True, daily_seasonality=False, changepoint_prior_scale=0.05 ) model.fit(df[['ds', 'y']]) # Make future dataframe future = model.make_future_dataframe(periods=periods) forecast = model.predict(future) # Extract predictions predictions = [] for _, row in forecast.tail(periods).iterrows(): predictions.append({ "date": row['ds'].strftime('%Y-%m-%d'), "predicted": max(0, round(row['yhat'])), "lower_bound": max(0, round(row['yhat_lower'])), "upper_bound": max(0, round(row['yhat_upper'])) }) return { "method": "prophet", "forecast": predictions, "total_predicted": sum(p['predicted'] for p in predictions), "confidence_interval": 0.95 } except Exception as e: logger.error(f"Prophet forecast failed: {e}") return await self._forecast_with_statsmodels(df, periods) async def _forecast_with_statsmodels( self, df: pd.DataFrame, periods: int ) -> Dict[str, Any]: """Fallback forecasting using simple methods""" values = df['y'].values # Calculate trend if len(values) >= 7: # Use last 7 days average as base recent_avg = np.mean(values[-7:]) overall_avg = np.mean(values) # Calculate simple growth rate if len(values) >= 14: first_half = np.mean(values[:len(values)//2]) second_half = np.mean(values[len(values)//2:]) daily_growth = (second_half - first_half) / (len(values) / 2) / first_half if first_half > 0 else 0 else: daily_growth = 0 # Calculate standard deviation for confidence interval std_dev = np.std(values) else: recent_avg = np.mean(values) if len(values) > 0 else 100 daily_growth = 0 std_dev = recent_avg * 0.2 # Generate predictions predictions = [] last_date = df['ds'].max() for i in range(1, periods + 1): pred_date = last_date + timedelta(days=i) predicted = recent_avg * (1 + daily_growth * i) # Add weekly seasonality (rough approximation) day_of_week = pred_date.weekday() if day_of_week >= 5: # Weekend predicted *= 0.7 elif day_of_week == 0: # Monday predicted *= 1.1 predicted = max(0, round(predicted)) predictions.append({ "date": pred_date.strftime('%Y-%m-%d'), "predicted": predicted, "lower_bound": max(0, round(predicted - 1.96 * std_dev)), "upper_bound": round(predicted + 1.96 * std_dev) }) return { "method": "statistical", "forecast": predictions, "total_predicted": sum(p['predicted'] for p in predictions), "confidence_interval": 0.95, "daily_growth_rate": daily_growth } async def forecast_revenue( self, historical_data: List[Dict[str, Any]], periods: int = 30 ) -> Dict[str, Any]: """ Forecast revenue Args: historical_data: List of dicts with 'date' and 'revenue' periods: Number of days to forecast Returns: Revenue forecast """ # Similar to message forecast but for revenue modified_data = [ {"date": d["date"], "count": d.get("revenue", 0)} for d in historical_data ] result = await self.forecast_messages(modified_data, periods) # Rename fields for revenue for pred in result.get("forecast", []): pred["revenue"] = pred.pop("predicted") pred["revenue_lower"] = pred.pop("lower_bound") pred["revenue_upper"] = pred.pop("upper_bound") result["total_revenue_predicted"] = result.pop("total_predicted", 0) return result async def detect_anomalies( self, data: List[Dict[str, Any]], threshold: float = 2.0 ) -> Dict[str, Any]: """ Detect anomalies in time series data Args: data: List of dicts with 'date' and 'value' threshold: Z-score threshold for anomaly detection Returns: Anomaly detection results """ if len(data) < 7: return {"anomalies": [], "message": "Insufficient data"} df = pd.DataFrame(data) values = df['value'].values # Calculate rolling statistics window = min(7, len(values) // 2) rolling_mean = pd.Series(values).rolling(window=window, center=True).mean() rolling_std = pd.Series(values).rolling(window=window, center=True).std() # Calculate z-scores z_scores = np.abs((values - rolling_mean) / rolling_std) # Find anomalies anomalies = [] for i, (z, val) in enumerate(zip(z_scores, values)): if pd.notna(z) and z > threshold: anomalies.append({ "date": df.iloc[i]['date'], "value": float(val), "expected": float(rolling_mean.iloc[i]), "z_score": float(z), "severity": "high" if z > 3 else "medium" }) return { "anomalies": anomalies, "total_anomalies": len(anomalies), "threshold_used": threshold, "data_points_analyzed": len(data) } def _empty_forecast(self) -> Dict[str, Any]: return { "method": "none", "forecast": [], "total_predicted": 0, "message": "Insufficient historical data" } # Singleton instance forecast_service = ForecastService()