ML‑based anomaly detection on Prometheus metrics with Prophet vs Greykite on a 90‑day baseline

import os
import pandas as pd
from prometheus_api_client import PrometheusConnect, MetricRangeDataFrame
from datetime import datetime, timedelta
import logging
from typing import List, Optional

# Configure logging for error tracking
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

def fetch_prometheus_metrics(
    prom_url: str,
    metric_name: str,
    start_time: datetime,
    end_time: datetime,
    step: str = "5m"
) -> Optional[pd.DataFrame]:
    """
    Fetch 90-day Prometheus metric data and preprocess for time series modeling.

    Args:
        prom_url: URL of Prometheus instance (e.g., http://prometheus:9090)
        metric_name: Prometheus metric name (e.g., node_cpu_seconds_total)
        start_time: Start datetime for data fetch
        end_time: End datetime for data fetch
        step: Resolution of scraped data (default 5m)

    Returns:
        Preprocessed DataFrame with columns [ds, y] or None if fetch fails
    """
    try:
        # Initialize Prometheus client with optional auth
        prom = PrometheusConnect(url=prom_url, disable_ssl_verification=True)
        logger.info(f"Fetching metric {metric_name} from {start_time} to {end_time}")

        # Fetch metric range data
        metric_data = prom.get_metric_range_data(
            metric_name=metric_name,
            start_time=start_time,
            end_time=end_time,
            step=step
        )

        if not metric_data:
            logger.error(f"No data returned for metric {metric_name}")
            return None

        # Convert to DataFrame and flatten labels
        df = MetricRangeDataFrame(metric_data)
        df = df.reset_index().rename(columns={"index": "ds", "value": "y"})
        df["ds"] = pd.to_datetime(df["ds"])
        df["y"] = pd.to_numeric(df["y"], errors="coerce")

        # Handle missing values: forward fill up to 2 intervals, then drop remaining
        df = df.sort_values("ds").ffill(limit=2).dropna()

        # Resample to consistent 5m intervals to fill gaps
        df = df.set_index("ds").resample(step).mean().reset_index()

        # Validate minimum data points: need at least 30 days of data (8640 samples at 5m)
        min_samples = 8640  # 30 days * 24h * 12 samples per hour
        if len(df) < min_samples:
            logger.error(f"Insufficient data for {metric_name}: {len(df)} samples, need {min_samples}")
            return None

        logger.info(f"Successfully fetched {len(df)} samples for {metric_name}")
        return df[["ds", "y"]]

    except ConnectionError as e:
        logger.error(f"Failed to connect to Prometheus at {prom_url}: {e}")
        return None
    except ValueError as e:
        logger.error(f"Invalid data format for {metric_name}: {e}")
        return None
    except Exception as e:
        logger.error(f"Unexpected error fetching {metric_name}: {e}", exc_info=True)
        return None

if __name__ == "__main__":
    # Benchmark configuration
    PROM_URL = os.getenv("PROMETHEUS_URL", "http://localhost:9090")
    METRIC_NAME = "node_cpu_seconds_total{mode='idle'}"
    END_TIME = datetime.now()
    START_TIME = END_TIME - timedelta(days=90)

    # Fetch and save data
    cpu_df = fetch_prometheus_metrics(PROM_URL, METRIC_NAME, START_TIME, END_TIME)
    if cpu_df is not None:
        cpu_df.to_csv("cpu_90d_baseline.csv", index=False)
        print(f"Saved baseline data to cpu_90d_baseline.csv")
    else:
        print("Failed to fetch baseline data")

import pandas as pd
import numpy as np
from prophet import Prophet
from prophet.diagnostics import cross_validation, performance_metrics
import joblib
import logging
from typing import Tuple, List
from sklearn.metrics import precision_score, recall_score, f1_score

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

def train_prophet_model(
    baseline_df: pd.DataFrame,
    model_path: str = "prophet_cpu_model.joblib",
    seasonality_mode: str = "additive"
) -> Tuple[Optional[Prophet], Optional[pd.DataFrame]]:
    """
    Train Prophet model on 90-day baseline, run cross-validation, and save model.

    Args:
        baseline_df: Preprocessed DataFrame with [ds, y] columns
        model_path: Path to save trained model
        seasonality_mode: Additive or multiplicative seasonality

    Returns:
        Tuple of (trained Prophet model, cross-validation results) or (None, None) on failure
    """
    try:
        # Validate input DataFrame
        required_cols = ["ds", "y"]
        if not all(col in baseline_df.columns for col in required_cols):
            logger.error(f"Baseline DataFrame missing required columns: {required_cols}")
            return None, None

        # Initialize Prophet with seasonal components matching Prometheus workload patterns
        model = Prophet(
            seasonality_mode=seasonality_mode,
            yearly_seasonality=False,  # 90-day baseline can't capture yearly seasonality
            weekly_seasonality=True,
            daily_seasonality=True,
            changepoint_prior_scale=0.05,  # Default 0.05, tuned for metric stability
            interval_width=0.95  # 95% confidence interval for anomaly threshold
        )

        # Add holiday component for known Kubernetes maintenance windows
        # Maintenance every Tuesday 2-4 AM UTC
        maintenance_df = pd.DataFrame({
            "holiday": "k8s_maintenance",
            "ds": pd.to_datetime([
                "2024-07-02", "2024-07-09", "2024-07-16", "2024-07-23", "2024-07-30",
                "2024-08-06", "2024-08-13", "2024-08-20", "2024-08-27", "2024-09-03",
                "2024-09-10", "2024-09-17", "2024-09-24"
            ]),
            "lower_window": -2,  # Start 2 hours before (midnight)
            "upper_window": 2    # End 2 hours after (4 AM)
        })
        model.add_holidays(maintenance_df)

        logger.info("Training Prophet model on 90-day baseline")
        model.fit(baseline_df)

        # Run 3-fold cross-validation to validate performance
        # Cutoff every 30 days, horizon 7 days
        df_cv = cross_validation(
            model,
            initial="60 days",
            period="30 days",
            horizon="7 days",
            parallel="processes"
        )

        # Calculate performance metrics
        df_metrics = performance_metrics(df_cv)
        avg_mae = df_metrics["mae"].mean()
        logger.info(f"Prophet cross-validation MAE: {avg_mae:.4f}")

        # Save trained model
        joblib.dump(model, model_path)
        logger.info(f"Saved Prophet model to {model_path}")

        return model, df_cv

    except ValueError as e:
        logger.error(f"Invalid baseline data for Prophet training: {e}")
        return None, None
    except RuntimeError as e:
        logger.error(f"Prophet convergence error: {e}")
        return None, None
    except Exception as e:
        logger.error(f"Unexpected error training Prophet: {e}", exc_info=True)
        return None, None

def detect_prophet_anomalies(
    model: Prophet,
    live_df: pd.DataFrame,
    threshold_std: int = 3
) -> Tuple[List[bool], List[float]]:
    """
    Detect anomalies in live Prometheus data using trained Prophet model.

    Args:
        model: Trained Prophet model
        live_df: Live data DataFrame with [ds, y] columns
        threshold_std: Number of standard deviations for anomaly threshold

    Returns:
        Tuple of (anomaly flags list, residual values list)
    """
    try:
        # Generate predictions for live data
        future = model.make_future_dataframe(periods=len(live_df), freq="5min")
        forecast = model.predict(future)

        # Merge live data with forecast
        merged = pd.merge(live_df, forecast[["ds", "yhat", "yhat_lower", "yhat_upper"]], on="ds")

        # Calculate residuals (actual - predicted)
        merged["residual"] = merged["y"] - merged["yhat"]
        residual_std = merged["residual"].std()

        # Flag anomalies: actual outside 95% confidence interval OR residual > threshold_std * std
        merged["is_anomaly"] = (
            (merged["y"] < merged["yhat_lower"]) |
            (merged["y"] > merged["yhat_upper"]) |
            (np.abs(merged["residual"]) > threshold_std * residual_std)
        )

        return merged["is_anomaly"].tolist(), merged["residual"].tolist()

    except Exception as e:
        logger.error(f"Prophet anomaly detection failed: {e}", exc_info=True)
        return [], []

if __name__ == "__main__":
    # Load baseline data
    baseline_df = pd.read_csv("cpu_90d_baseline.csv")
    baseline_df["ds"] = pd.to_datetime(baseline_df["ds"])

    # Train model
    model, cv_results = train_prophet_model(baseline_df)
    if model is not None:
        # Simulate live data (last 24 hours of baseline)
        live_df = baseline_df.tail(288).copy()  # 288 samples = 24h at 5m
        live_df["y"] = live_df["y"] * np.random.uniform(0.5, 1.5, len(live_df))  # Inject noise

        # Detect anomalies
        anomalies, residuals = detect_prophet_anomalies(model, live_df)
        print(f"Detected {sum(anomalies)} anomalies in 24h live data")
        print(f"Precision: {precision_score(live_df['y'] > 0.8, anomalies)}")  # Simplified ground truth

import pandas as pd
import numpy as np
from greykite.framework.templates.autotune import autotune_template
from greykite.framework.templates.forecaster import Forecaster
from greykite.framework.templates.model_templates import ModelTemplateEnum
import joblib
import logging
from typing import Tuple, Optional
from sklearn.metrics import mean_absolute_error

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

def train_greykite_model(
    baseline_df: pd.DataFrame,
    model_path: str = "greykite_cpu_model.joblib"
) -> Optional[object]:
    """
    Train Greykite model on 90-day baseline using autotune, save model.

    Args:
        baseline_df: Preprocessed DataFrame with [ds, y] columns
        model_path: Path to save trained model

    Returns:
        Trained Greykite forecaster or None on failure
    """
    try:
        # Validate input DataFrame
        required_cols = ["ds", "y"]
        if not all(col in baseline_df.columns for col in required_cols):
            logger.error(f"Baseline DataFrame missing required columns: {required_cols}")
            return None

        # Initialize Greykite forecaster with default template
        forecaster = Forecaster()

        # Autotune hyperparameters for 90-day Prometheus metric baseline
        # Template: SILVERKITE (default for general time series)
        autotune_result = autotune_template(
            forecaster=forecaster,
            df=baseline_df,
            template=ModelTemplateEnum.SILVERKITE.name,
            hyperparameter_grid={
                "seasonality": {
                    "yearly_seasonality": [False],  # 90-day baseline can't capture yearly
                    "weekly_seasonality": [True],
                    "daily_seasonality": [True]
                },
                "changepoint": {
                    "changepoint_prior_scale": [0.01, 0.05, 0.1]
                }
            },
            cv_max_splits=3,
            metric="mean_absolute_error"
        )

        logger.info(f"Greykite autotune best MAE: {autotune_result.best_score_:.4f}")

        # Train final model with best hyperparameters
        forecaster.train(
            df=baseline_df,
            template=ModelTemplateEnum.SILVERKITE.name,
            **autotune_result.best_params_
        )

        # Evaluate in-sample performance
        in_sample_pred = forecaster.predict(baseline_df)
        in_sample_mae = mean_absolute_error(baseline_df["y"], in_sample_pred["forecast"])
        logger.info(f"Greykite in-sample MAE: {in_sample_mae:.4f}")

        # Save trained model
        joblib.dump(forecaster, model_path)
        logger.info(f"Saved Greykite model to {model_path}")

        return forecaster

    except ValueError as e:
        logger.error(f"Invalid baseline data for Greykite training: {e}")
        return None
    except ImportError as e:
        logger.error(f"Missing Greykite dependency: {e}")
        return None
    except Exception as e:
        logger.error(f"Unexpected error training Greykite: {e}", exc_info=True)
        return None

def detect_greykite_anomalies(
    forecaster: object,
    live_df: pd.DataFrame,
    confidence_interval: float = 0.95
) -> Tuple[list, list]:
    """
    Detect anomalies in live Prometheus data using trained Greykite model.

    Args:
        forecaster: Trained Greykite forecaster
        live_df: Live data DataFrame with [ds, y] columns
        confidence_interval: Confidence interval for anomaly threshold (0-1)

    Returns:
        Tuple of (anomaly flags list, forecast values list)
    """
    try:
        # Generate predictions for live data
        prediction = forecaster.predict(live_df)

        # Extract forecast and confidence intervals
        forecast = prediction["forecast"]
        lower_col = f"forecast_lower_{int(confidence_interval*100)}"
        upper_col = f"forecast_upper_{int(confidence_interval*100)}"

        if lower_col not in prediction.columns or upper_col not in prediction.columns:
            logger.error(f"Confidence interval {confidence_interval} not available in prediction")
            return [], []

        # Flag anomalies: actual outside confidence interval
        is_anomaly = (
            (live_df["y"].values < prediction[lower_col].values) |
            (live_df["y"].values > prediction[upper_col].values)
        ).tolist()

        return is_anomaly, forecast.tolist()

    except Exception as e:
        logger.error(f"Greykite anomaly detection failed: {e}", exc_info=True)
        return [], []

if __name__ == "__main__":
    # Load baseline data
    baseline_df = pd.read_csv("cpu_90d_baseline.csv")
    baseline_df["ds"] = pd.to_datetime(baseline_df["ds"])

    # Train model
    forecaster = train_greykite_model(baseline_df)
    if forecaster is not None:
        # Simulate live data (last 24 hours of baseline)
        live_df = baseline_df.tail(288).copy()
        live_df["y"] = live_df["y"] * np.random.uniform(0.5, 1.5, len(live_df))  # Inject noise

        # Detect anomalies
        anomalies, forecasts = detect_greykite_anomalies(forecaster, live_df)
        print(f"Detected {sum(anomalies)} anomalies in 24h live data")

        # Calculate latency for 1k samples
        import time
        start = time.time()
        for _ in range(1000):
            detect_greykite_anomalies(forecaster, live_df.tail(1))
        avg_latency = (time.time() - start) / 1000 * 1000  # ms
        print(f"Average inference latency: {avg_latency:.2f}ms per sample")

# Split 90-day baseline into 83-day train, 7-day validation
train_end = baseline_df["ds"].max() - timedelta(days=7)
train_df = baseline_df[baseline_df["ds"] <= train_end]
val_df = baseline_df[baseline_df["ds"] > train_end]
print(f"Train samples: {len(train_df)}, Val samples: {len(val_df)}")

from sklearn.metrics import precision_recall_curve
import numpy as np

# Calculate residuals on validation set
val_residuals = np.abs(val_df["y"] - val_forecast)
# Generate thresholds from 1-5 standard deviations
thresholds = np.arange(1, 5.1, 0.1) * val_residuals.std()
# Calculate precision/recall for each threshold
precisions, recalls, _ = precision_recall_curve(val_df["is_anomaly"], val_residuals.values.reshape(-1,1) > thresholds.reshape(1,-1))
# Find threshold with precision >= 0.95
optimal_idx = np.where(precisions >= 0.95)[0][0]
print(f"Optimal threshold: {thresholds[optimal_idx]:.2f} ({thresholds[optimal_idx]/val_residuals.std():.1f}σ)")

def route_metric_to_model(metric_name: str, prom_url: str) -> str:
    """Route metric to Prophet or Greykite based on cardinality"""
    prom = PrometheusConnect(url=prom_url)
    # Count number of series for metric
    series_count = len(prom.get_current_metric_value(metric_name))
    if series_count > 500:
        return "greykite"
    else:
        return "prophet"
print(f"node_cpu_seconds_total: {route_metric_to_model('node_cpu_seconds_total', 'http://prometheus:9090')}")
print(f"http_request_latency_seconds: {route_metric_to_model('http_request_latency_seconds', 'http://prometheus:9090')}")