predictive_analytics.py

import numpy as np
import pandas as pd
from statsmodels.tsa.arima.model import ARIMA
from statsmodels.tsa.holtwinters import ExponentialSmoothing
import logging
from typing import Dict, List, Tuple, Optional
import json

log = logging.getLogger("predictive_analytics")

class TimeSeriesForecaster:
    def __init__(self, method="arima", seasonal_periods=24):
        """
        Initialize time series forecaster.

        Args:
            method: Forecasting method ('arima', 'exponential_smoothing')
            seasonal_periods: Number of periods in a season (e.g., 24 for hourly data)
        """
        self.method = method
        self.seasonal_periods = seasonal_periods
        self.model = None

    def fit(self, data: List[float]) -> None:
        """Fit the forecasting model on historical data."""
        if len(data) < 10:
            log.warning("Insufficient data for forecasting")
            return

        data_array = np.array(data)

        if self.method == "arima":
            try:
                # Simple ARIMA(1,1,1) - can be made more sophisticated
                self.model = ARIMA(data_array, order=(1, 1, 1))
                self.model = self.model.fit()
            except Exception as e:
                log.warning(f"ARIMA fitting failed: {e}")
                self.model = None

        elif self.method == "exponential_smoothing":
            try:
                self.model = ExponentialSmoothing(data_array, seasonal_periods=self.seasonal_periods, trend='add', seasonal='add')
                self.model = self.model.fit()
            except Exception as e:
                log.warning(f"Exponential smoothing fitting failed: {e}")
                self.model = None

        log.info(f"Time series forecaster fitted with method: {self.method}")

    def forecast(self, steps: int = 24) -> Tuple[List[float], List[float], List[float]]:
        """
        Generate forecast.

        Args:
            steps: Number of steps to forecast

        Returns:
            Tuple of (forecast_values, lower_bounds, upper_bounds)
        """
        if self.model is None:
            log.warning("Model not fitted, cannot forecast")
            return [], [], []

        try:
            if self.method == "arima":
                forecast_result = self.model.forecast(steps=steps)
                forecast_values = forecast_result.tolist()

                # Simple confidence intervals (can be improved)
                std_dev = np.std(self.model.resid) if hasattr(self.model, 'resid') else 1.0
                lower_bounds = (np.array(forecast_values) - 1.96 * std_dev).tolist()
                upper_bounds = (np.array(forecast_values) + 1.96 * std_dev).tolist()

            elif self.method == "exponential_smoothing":
                forecast_result = self.model.forecast(steps)
                forecast_values = forecast_result.tolist()

                # Placeholder confidence intervals
                std_dev = np.std(self.model.resid) if hasattr(self.model, 'resid') else 1.0
                lower_bounds = (np.array(forecast_values) - 1.96 * std_dev).tolist()
                upper_bounds = (np.array(forecast_values) + 1.96 * std_dev).tolist()

            return forecast_values, lower_bounds, upper_bounds

        except Exception as e:
            log.warning(f"Forecasting failed: {e}")
            return [], [], []

def forecast_timeseries(filepath: str, hours_ahead: int = 24, method: str = "arima") -> Dict[str, Dict]:
    """
    Forecast timeseries data from JSONL file.

    Args:
        filepath: Path to timeseries JSONL file
        hours_ahead: Hours to forecast ahead
        method: Forecasting method

    Returns:
        Dictionary with forecast data by entity
    """
    try:
        # Load timeseries data
        data = []
        with open(filepath, 'r', encoding='utf-8') as f:
            for line in f:
                data.append(json.loads(line.strip()))

        if not data:
            return {}

        # Group by entity and metric
        grouped_data = {}
        for record in data:
            key = f"{record['entity_id']}_{record['metric_name']}"
            if key not in grouped_data:
                grouped_data[key] = []
            grouped_data[key].append(record)

        forecast_results = {}

        for key, records in grouped_data.items():
            # Sort by timeframe
            records.sort(key=lambda x: x['timeframe']['from'])

            # Extract values
            values = []
            timestamps = []
            for record in records:
                if record['points']:
                    for point in record['points']:
                        values.append(point['value'])
                        timestamps.append(point['timestamp'])

            if len(values) < 10:
                log.warning(f"Insufficient data for {key}")
                continue

            # Fit forecaster
            forecaster = TimeSeriesForecaster(method=method)
            forecaster.fit(values)

            # Generate forecast
            forecast_values, lower_bounds, upper_bounds = forecaster.forecast(steps=hours_ahead)

            if forecast_values:
                # Generate future timestamps (assuming hourly data)
                last_timestamp = timestamps[-1] if timestamps else int(pd.Timestamp.now().timestamp() * 1000)
                future_timestamps = [last_timestamp + (i + 1) * 3600 * 1000 for i in range(hours_ahead)]

                forecast_results[key] = {
                    'historical': {
                        'timestamps': timestamps,
                        'values': values
                    },
                    'forecast': {
                        'timestamps': future_timestamps,
                        'values': forecast_values,
                        'lower_bound': lower_bounds,
                        'upper_bound': upper_bounds
                    }
                }

        return forecast_results

    except FileNotFoundError:
        log.warning(f"Timeseries file not found: {filepath}")
        return {}
    except Exception as e:
        log.warning(f"Forecasting failed: {e}")
        return {}

if __name__ == "__main__":
    # Example usage
    import random

    # Generate sample data
    historical_data = [100 + random.gauss(0, 5) for _ in range(50)]

    forecaster = TimeSeriesForecaster(method="arima")
    forecaster.fit(historical_data)

    forecast_values, lower_bounds, upper_bounds = forecaster.forecast(steps=10)

    print(f"Generated {len(forecast_values)} forecast points")
    if forecast_values:
        print(f"First forecast value: {forecast_values[0]:.2f}")
        print(f"Confidence interval: [{lower_bounds[0]:.2f}, {upper_bounds[0]:.2f}]")