預運指

用 Prophet 或 statsmodels 預基設與應指以謀容、優費、主擴。

詳見 Extended Examples 全配置與模板。

用時

• 須預基設容需（CPU、記憶、盤、網）
• 為下季謀硬/雲資購
• 欲預費勢以優雲支
• 須按預載設主擴策
• 為事預用戶流
• 為備謀預庫儲長
• 為速限配估 API 用

入

• 必要：歷時序指（最少三至十二月）
• 必要：指類（CPU、記、請/秒、費等）
• 必要：預程（日、週、或月前）
• 可選：知未來事（部署、市宣、假）
• 可選：季信（日、週、年模）
• 可選：外回歸（如市支、用戶註）

法

第一步：設環並載數

裝預庫並備時序數。

# 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

以 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")

見 EXAMPLES.md Step 1 全 MetricsLoader。

得： 時序數以正隔載，缺值填，可預。

敗則： 若數有隙，用前填或插值；確回看期有足數（薦 90+ 日）；驗時戳時區一致；察可歪預之異（>5 sigma）。

第二步：施 Prophet 預

用 Facebook Prophet 作自動季察與預。

# 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")

見 EXAMPLES.md Step 2 全 ProphetForecaster。

得： 三十日以上預附可信區，季模於部圖捕，交叉驗證 MAPE < 15%。

敗則： 若預不實，試異長模（線對邏）；若缺季，調 seasonality_mode；若準差（<70% MAPE），加更多史數或外回歸；察數質。

第三步：施 ARIMA/SARIMAX 預（替）

用 statsmodels 作傳時序預。

# 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

見 EXAMPLES.md Step 3 全 ARIMAForecaster 與 auto_arima。

得： ARIMA 模以最佳參擬，預生附可信區，診圖示殘為白噪。

敗則： 若模不收斂，簡參（減 p、q、P、Q）；若預勢誤，察差分階（d、D）；若殘非白噪，加更多 AR/MA 項；確序長 >2x 季期。

第四步：識容閾與警

析預以測資何時竭。

# 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)

見 EXAMPLES.md Step 4 全 CapacityPlanner。

得： 報示容限何時達，薦附急級，長率算。

敗則： 若竭日不實，驗 capacity_limit 正；若長率過高，察史數異；於熟系考非線長模。

第五步：於 Grafana 視預

推預數至 Grafana 供實時監。

# 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")

見 EXAMPLES.md Step 5 全 GrafanaForecaster。

得： 預注現於 Grafana 面板，容警為豎標，預數可經 CSV 源達。

敗則： 驗 Grafana API 鑰有正權；察面板 UID 正；確時戳於注為毫秒；接前以 curl 試 API。

第六步：自動預生

設定時業常規生預。

# 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)

見 EXAMPLES.md Step 6 全調度。

得： 諸指預日生，容報記，CSV 出供 Grafana，重容警發。

敗則： 驗調度程續運（用 systemd/supervisor）；察 Prometheus 通；確預出盤空足；於暫敗施重試邏；於調度本設監。

驗

• 史數載附 90+ 日連指
• Prophet 預於部圖捕日/週季
• 預可信區驗時含 85-95% 實值
• 於知景容竭日算正
• ARIMA 殘於診圖現如白噪
• Grafana 注現於預警/竭日
• 自動預日運無手涉
• 驗集預準（MAPE）< 15%

陷

• 史數不足：可靠季察需三至十二月；勿以 <60 日預
• 略知事：假、部署、市宣歪預；加為外回歸或假
• 長程預過信：30-90 日後準衰；為向導，非確預
• 靜容限：基設隨時易；加資時更 capacity_limit
• 異值預：訓數之異傳入預；清數或用穩法
• 不更模：系改後預陳；週訓或於大架改後重訓
• 略可信區：點預誤導；恆用下/上界供謀
• 誤季期：時數用日、日數用週；失配致差預

參

• detect-anomalies-aiops - 異察補預以主監
• plan-capacity - 基設容謀流
• build-grafana-dashboards - 視預與容勢