---

name: forecast-operational-metrics description: > Forecast infrastructure and application metrics using Prophet or statsmodels for capacity planning, cost optimization, and proactive scaling. Visualize predictions in Grafana and set up alerts for projected resource exhaustion. Use when forecasting infrastructure capacity needs for CPU, memory, or disk, planning hardware procurement for next quarter, predicting cost trends to optimize cloud spending, or setting up proactive scaling policies based on predicted load. license: MIT allowed-tools: Read Write Edit Bash Grep Glob metadata: author: Philipp Thoss version: "1.0" domain: mlops complexity: intermediate language: multi tags: forecasting, prophet, statsmodels, capacity, time-series, grafana locale: zh-CN source_locale: en source_commit: a87e5e03 fence_basis_commit: a87e5e03 translator: claude translation_date: "2026-03-17"

Forecast Operational Metrics

Predict future resource usage and system metrics for capacity planning and cost optimization.

See Extended Examples for complete configuration files and templates.

适用场景

• Need to forecast infrastructure capacity needs (CPU, memory, disk, network)
• Planning hardware/cloud resource procurement for next quarter
• Want to predict cost trends and optimize cloud spending
• Need to set up proactive scaling policies based on predicted load
• Forecasting user traffic for event planning
• Predicting database storage growth for backup planning
• Estimating API usage for rate limiting configuration

输入

• 必需: Historical time series metrics (3-12 months minimum)
• 必需: Metric type (CPU, memory, requests/sec, costs, etc.)
• 必需: Forecast horizon (days, weeks, or months ahead)
• 可选: Known future events (deployments, marketing campaigns, holidays)
• 可选: Seasonality information (daily, weekly, yearly patterns)
• 可选: External regressors (e.g., marketing spend, user signups)

步骤

第 1 步：Set Up Environment and Load Data

Install forecasting libraries and prepare time series data.

# Create virtual environment
python -m venv venv
source venv/bin/activate

# Install forecasting libraries
pip install prophet statsmodels pandas numpy
pip install plotly matplotlib seaborn
pip install prometheus-api-client influxdb-client
pip install grafana-api

Load and prepare data with MetricsLoader:

# forecasting/data_loader.py (abbreviated)
import pandas as pd
from datetime import datetime, timedelta

class MetricsLoader:
    def load_from_prometheus(self, query: str, lookback_days: int = 90, step: str = "1h"):
        """Load historical metrics from Prometheus."""
        # ... implementation (see EXAMPLES.md for complete code)

    def resample_and_aggregate(self, df: pd.DataFrame, freq: str = "1H"):
        """Resample time series to regular intervals."""
        # ... implementation (see EXAMPLES.md)

# Example usage
loader = MetricsLoader(prometheus_url="http://prometheus:9090")
df = loader.load_from_prometheus(
    query='avg(rate(container_cpu_usage_seconds_total[5m]))',
    lookback_days=90,
)
df_daily = loader.resample_and_aggregate(df, freq="1D")

See EXAMPLES.md Step 1 for the complete MetricsLoader implementation.

预期结果： Time series data loaded with regular intervals, missing values filled, ready for forecasting.

失败处理： If data gaps exist, use forward-fill or interpolation, ensure lookback period has sufficient data (90+ days recommended), verify timestamp timezone consistency, check for outliers (>5 sigma) that may skew forecasts.

第 2 步：Implement Prophet Forecasting

Use Facebook Prophet for automatic seasonality detection and forecasting.

# forecasting/prophet_forecaster.py (abbreviated)
from prophet import Prophet

class ProphetForecaster:
    def __init__(self, growth: str = "linear", seasonality_mode: str = "multiplicative"):
        self.growth = growth
        self.prophet_params = {
            "growth": growth,
            "seasonality_mode": seasonality_mode,
            # ... additional parameters (see EXAMPLES.md)
        }

    def fit(self, df: pd.DataFrame, regressors=None, holidays=None):
        """Train Prophet model on historical data."""
        # ... implementation (see EXAMPLES.md)

    def forecast(self, periods: int, freq: str = "D"):
        """Generate forecast for future periods."""
        # ... implementation (see EXAMPLES.md)

# Example usage
forecaster = ProphetForecaster(growth="linear", seasonality_mode="multiplicative")
forecaster.fit(df_daily)
forecast = forecaster.forecast(periods=30, freq="D")
forecaster.plot_forecast(forecast, save_path="results/cpu_forecast.png")

See EXAMPLES.md Step 2 for the complete ProphetForecaster implementation.

预期结果： Forecast generated for 30+ days ahead with confidence intervals, seasonal patterns captured in components plot, cross-validation MAPE < 15%.

失败处理： If forecast looks unrealistic, try different growth model (linear vs logistic), if seasonality missing adjust seasonality_mode, if accuracy poor (<70% MAPE) add more historical data or external regressors, check for data quality issues.

第 3 步：Implement ARIMA/SARIMAX Forecasting (Alternative)

Use statsmodels for traditional time series forecasting.

# forecasting/arima_forecaster.py (abbreviated)
from statsmodels.tsa.statespace.sarimax import SARIMAX

class ARIMAForecaster:
    def __init__(self, order: tuple = (1, 1, 1), seasonal_order: tuple = (1, 1, 1, 7)):
        self.order = order
        self.seasonal_order = seasonal_order

    def fit(self, df: pd.DataFrame, exog=None):
        """Train SARIMAX model."""
        series = df.set_index("timestamp")["value"]
        self.model = SARIMAX(series, exog=exog, order=self.order, seasonal_order=self.seasonal_order)
        self.fitted_model = self.model.fit(disp=False)
        # ... implementation (see EXAMPLES.md)

    def forecast(self, steps: int, exog_future=None):
        """Generate forecast for future periods."""
        # ... implementation (see EXAMPLES.md)

# Auto-select parameters
best_order, best_seasonal = auto_arima(series, seasonal=True)
forecaster = ARIMAForecaster(order=best_order, seasonal_order=best_seasonal)
forecaster.fit(df_hourly)
forecast = forecaster.forecast(steps=168)  # 7 days

See EXAMPLES.md Step 3 for the complete ARIMAForecaster implementation and auto_arima function.

预期结果： ARIMA model fitted with optimal parameters, forecast generated with confidence intervals, diagnostic plots show white noise residuals.

失败处理： If model doesn't converge, simplify parameters (reduce p, q, P, Q), if forecast has wrong trend check differencing order (d, D), if residuals not white noise add more AR/MA terms, ensure series length >2x seasonal period.

第 4 步：Identify Capacity Thresholds and Alerts

Analyze forecast to predict when resources will be exhausted.

# forecasting/capacity_planning.py (abbreviated)
from datetime import datetime

class CapacityPlanner:
    def __init__(self, capacity_limit: float, warning_threshold: float = 0.8):
        self.capacity_limit = capacity_limit
        self.warning_threshold = warning_threshold

    def find_exhaustion_date(self, forecast: pd.DataFrame):
        """Find when forecast exceeds capacity limit."""
        exceeded = forecast[forecast["yhat"] >= self.capacity_limit]
        # ... implementation (see EXAMPLES.md)

    def generate_capacity_report(self, forecast: pd.DataFrame):
        """Generate comprehensive capacity planning report."""
        # ... implementation (see EXAMPLES.md)

# Example usage
planner = CapacityPlanner(capacity_limit=1000, warning_threshold=0.8)
report = planner.generate_capacity_report(forecast)
print(f"Warning Date: {report['warning_date']}")
print(f"Exhaustion Date: {report['exhaustion_date']}")
recommendation = planner.recommend_scaling_action(report)

See EXAMPLES.md Step 4 for the complete CapacityPlanner implementation.

预期结果： Report shows when capacity limits will be reached, recommendations provided with urgency levels, growth rates calculated.

失败处理： If exhaustion date unrealistic, verify capacity_limit is correct, if growth rate too high check for outliers in historical data, consider non-linear growth models for mature systems.

第 5 步：Visualize Forecasts in Grafana

Push forecast data to Grafana for real-time monitoring.

# forecasting/grafana_integration.py (abbreviated)
import requests

class GrafanaForecaster:
    def __init__(self, grafana_url: str, api_key: str, dashboard_uid: str = None):
        self.grafana_url = grafana_url.rstrip("/")
        self.api_key = api_key
        self.dashboard_uid = dashboard_uid

    def create_annotation(self, text: str, tags: list, time: datetime = None):
        """Create annotation in Grafana for forecast events."""
        # ... implementation (see EXAMPLES.md)

    def create_capacity_alert_annotation(self, capacity_report: dict):
        """Create Grafana annotation for capacity warnings."""
        # ... implementation (see EXAMPLES.md)

# Export to CSV for Grafana datasource
def export_forecast_to_csv(forecast: pd.DataFrame, output_path: str):
    """Export forecast in format compatible with Grafana CSV datasource."""
    # ... implementation (see EXAMPLES.md)

# Example usage
grafana = GrafanaForecaster(
    grafana_url="http://grafana:3000",
    api_key="YOUR_API_KEY",
    dashboard_uid="your-dashboard-uid",
)
grafana.create_capacity_alert_annotation(report)
export_forecast_to_csv(forecast, "grafana/forecasts/cpu_forecast.csv")

See EXAMPLES.md Step 5 for the complete GrafanaForecaster implementation.

预期结果： Forecast annotations appear in Grafana dashboards, capacity warnings visible as vertical markers, forecast data accessible via CSV datasource.

失败处理： Verify Grafana API key has correct permissions, check dashboard UID is correct, ensure timestamps in milliseconds for annotations, test API with curl before integrating.

第 6 步：Automate Forecast Generation

Set up scheduled jobs to generate forecasts regularly.

# forecasting/scheduler.py (abbreviated)
import schedule
import time

def generate_daily_forecast():
    """Generate forecast for all monitored metrics."""
    logger.info("Starting daily forecast generation")

    metrics_config = [
        {"name": "cpu_usage", "query": "...", "capacity_limit": 0.8, "forecast_days": 30},
        {"name": "memory_usage", "query": "...", "capacity_limit": 32, "forecast_days": 30},
        {"name": "disk_usage", "query": "...", "capacity_limit": 500, "forecast_days": 90},
    ]

    loader = MetricsLoader(prometheus_url="http://prometheus:9090")

    for metric_config in metrics_config:
        df = loader.load_from_prometheus(query=metric_config["query"], lookback_days=90)
        forecaster = ProphetForecaster()
        forecaster.fit(df)
        forecast = forecaster.forecast(periods=metric_config["forecast_days"])

        planner = CapacityPlanner(capacity_limit=metric_config["capacity_limit"])
        report = planner.generate_capacity_report(forecast)

        export_forecast_to_csv(forecast, f"grafana/forecasts/{metric_config['name']}_forecast.csv")
        # ... (see EXAMPLES.md for complete implementation)

# Schedule daily at 2 AM
schedule.every().day.at("02:00").do(generate_daily_forecast)

while True:
    schedule.run_pending()
    time.sleep(60)

See EXAMPLES.md Step 6 for the complete scheduler implementation.

预期结果： Forecasts generated daily for all metrics, capacity reports logged, CSV files exported for Grafana, alerts sent for critical capacity warnings.

失败处理： Verify scheduler process runs continuously (use systemd/supervisor), check Prometheus connectivity, ensure sufficient disk space for forecast exports, implement retry logic for transient failures, set up monitoring for scheduler itself.

验证清单

• Historical data loaded with 90+ days of continuous metrics
• Prophet forecast captures daily/weekly seasonality in components plot
• Forecast confidence intervals contain 85-95% of actual values in validation
• Capacity exhaustion dates calculated correctly for known scenarios
• ARIMA model residuals appear as white noise in diagnostic plots
• Grafana annotations appear at predicted warning/exhaustion dates
• Automated forecasting runs daily without manual intervention
• Forecast accuracy (MAPE) < 15% on validation set

常见问题

• Insufficient historical data: Need 3-12 months for reliable seasonality detection; avoid forecasting with <60 days
• Ignoring known events: Holidays, deployments, marketing campaigns skew forecasts; add as external regressors or holidays
• Overconfidence in long-term forecasts: Accuracy degrades beyond 30-90 days; use as directional guidance, not exact predictions
• Static capacity limits: Infrastructure changes over time; update capacity_limit when adding resources
• Forecasting anomalies: Outliers in training data propagate to forecast; clean data or use robust methods
• Not updating models: Forecasts stale after system changes; retrain weekly or after significant architecture changes
• Ignoring confidence intervals: Point forecasts misleading; always use lower/upper bounds for planning
• Wrong seasonality period: Daily for hourly data, weekly for daily data; mismatch causes poor forecasts

相关技能

• detect-anomalies-aiops - Anomaly detection complements forecasting for proactive monitoring
• plan-capacity - Infrastructure capacity planning workflows
• build-grafana-dashboards - Visualize forecasts and capacity trends