雲端運算AI完整實踐指南：從智慧資源調度到邊緣智慧的端到端解決方案

# 雲端負載預測與智慧擴展系統
# 整合時間序列分析、機器學習與Kubernetes自動擴展

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime, timedelta
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
from sklearn.preprocessing import StandardScaler
import warnings
warnings.filterwarnings('ignore')

# 設定繁體中文顯示
plt.rcParams['font.sans-serif'] = ['Microsoft JhengHei']
plt.rcParams['axes.unicode_minus'] = False

class CloudWorkloadPredictor:
    """
    雲端工作負載預測器
    提供時間序列預測、特徵工程與動態擴展決策
    """
    
    def __init__(self, prediction_horizon=24):
        """
        初始化預測器
        
        參數:
            prediction_horizon: 預測時間範圍（小時）
        """
        self.prediction_horizon = prediction_horizon
        self.model = None
        self.scaler = StandardScaler()
        self.feature_columns = []
        
        print(f"雲端工作負載預測器已初始化（預測範圍: {prediction_horizon}小時）")
    
    def generate_sample_data(self, days=90):
        """
        生成範例工作負載資料
        模擬具有趨勢、季節性與隨機波動的時間序列
        
        參數:
            days: 資料天數
            
        回傳:
            workload_df: 工作負載DataFrame
        """
        print(f"\n生成範例工作負載資料（{days}天）...")
        
        np.random.seed(42)
        
        # 生成時間序列
        end_date = datetime.now()
        start_date = end_date - timedelta(days=days)
        timestamps = pd.date_range(start=start_date, end=end_date, freq='H')
        
        n_hours = len(timestamps)
        
        # 基礎負載（線性增長趨勢）
        base_load = 100 + np.linspace(0, 50, n_hours)
        
        # 每日週期性（工作時間高峰）
        daily_pattern = 30 * np.sin(2 * np.pi * np.arange(n_hours) / 24 - np.pi/2)
        
        # 每週週期性（週末較低）
        weekly_pattern = 20 * np.sin(2 * np.pi * np.arange(n_hours) / (24*7))
        
        # 隨機波動
        noise = np.random.normal(0, 10, n_hours)
        
        # 合成工作負載（CPU使用率百分比）
        cpu_usage = base_load + daily_pattern + weekly_pattern + noise
        cpu_usage = np.clip(cpu_usage, 10, 95)  # 限制在合理範圍
        
        # 記憶體使用率（與CPU相關但有獨立變化）
        memory_usage = 0.8 * cpu_usage + np.random.normal(0, 5, n_hours)
        memory_usage = np.clip(memory_usage, 20, 90)
        
        # 網路流量（單位: Mbps）
        network_traffic = 50 + 0.5 * cpu_usage + np.random.normal(0, 10, n_hours)
        network_traffic = np.clip(network_traffic, 10, 200)
        
        # 請求數量（每分鐘）
        request_count = 1000 + 20 * cpu_usage + np.random.normal(0, 100, n_hours)
        request_count = np.clip(request_count, 500, 5000)
        
        # 響應時間（毫秒，與負載正相關）
        response_time = 50 + 2 * cpu_usage + np.random.normal(0, 10, n_hours)
        response_time = np.clip(response_time, 30, 300)
        
        # 建構DataFrame
        workload_df = pd.DataFrame({
            'timestamp': timestamps,
            'cpu_usage': cpu_usage,
            'memory_usage': memory_usage,
            'network_traffic': network_traffic,
            'request_count': request_count,
            'response_time': response_time
        })
        
        print(f"資料生成完成！形狀: {workload_df.shape}")
        print(f"時間範圍: {workload_df['timestamp'].min()} 至 {workload_df['timestamp'].max()}")
        
        return workload_df
    
    def create_features(self, df):
        """
        建立時間序列特徵
        包含時間特徵、滯後特徵、滾動統計
        
        參數:
            df: 原始工作負載DataFrame
            
        回傳:
            features_df: 包含特徵的DataFrame
        """
        print("\n建立時間序列特徵...")
        
        features_df = df.copy()
        
        # 時間特徵
        features_df['hour'] = features_df['timestamp'].dt.hour
        features_df['day_of_week'] = features_df['timestamp'].dt.dayofweek
        features_df['day_of_month'] = features_df['timestamp'].dt.day
        features_df['month'] = features_df['timestamp'].dt.month
        features_df['is_weekend'] = (features_df['day_of_week'] >= 5).astype(int)
        features_df['is_business_hour'] = (
            (features_df['hour'] >= 9) & (features_df['hour'] <= 18)
        ).astype(int)
        
        # 週期性特徵（使用正弦餘弦編碼）
        features_df['hour_sin'] = np.sin(2 * np.pi * features_df['hour'] / 24)
        features_df['hour_cos'] = np.cos(2 * np.pi * features_df['hour'] / 24)
        features_df['day_sin'] = np.sin(2 * np.pi * features_df['day_of_week'] / 7)
        features_df['day_cos'] = np.cos(2 * np.pi * features_df['day_of_week'] / 7)
        
        # 滯後特徵（過去1、3、6、12、24小時的CPU使用率）
        for lag in [1, 3, 6, 12, 24]:
            features_df[f'cpu_lag_{lag}h'] = features_df['cpu_usage'].shift(lag)
        
        # 滾動統計（過去24小時的統計）
        features_df['cpu_rolling_mean_24h'] = (
            features_df['cpu_usage'].rolling(window=24, min_periods=1).mean()
        )
        features_df['cpu_rolling_std_24h'] = (
            features_df['cpu_usage'].rolling(window=24, min_periods=1).std()
        )
        features_df['cpu_rolling_min_24h'] = (
            features_df['cpu_usage'].rolling(window=24, min_periods=1).min()
        )
        features_df['cpu_rolling_max_24h'] = (
            features_df['cpu_usage'].rolling(window=24, min_periods=1).max()
        )
        
        # 變化率特徵
        features_df['cpu_diff_1h'] = features_df['cpu_usage'].diff(1)
        features_df['cpu_diff_24h'] = features_df['cpu_usage'].diff(24)
        
        # 移除包含NaN的列（由於滯後和滾動計算）
        features_df = features_df.dropna()
        
        print(f"特徵建立完成！特徵數量: {len(features_df.columns)}")
        
        return features_df
    
    def prepare_training_data(self, features_df, target_col='cpu_usage'):
        """
        準備訓練資料
        分離特徵與目標變數
        
        參數:
            features_df: 包含特徵的DataFrame
            target_col: 目標變數欄位名稱
            
        回傳:
            X: 特徵矩陣
            y: 目標向量
        """
        print(f"\n準備訓練資料（目標變數: {target_col}）...")
        
        # 排除時間戳記與目標變數
        exclude_cols = ['timestamp', target_col]
        self.feature_columns = [
            col for col in features_df.columns if col not in exclude_cols
        ]
        
        X = features_df[self.feature_columns].values
        y = features_df[target_col].values
        
        print(f"特徵維度: {X.shape}")
        print(f"目標維度: {y.shape}")
        
        return X, y
    
    def train_model(self, X_train, y_train, model_type='rf'):
        """
        訓練預測模型
        
        參數:
            X_train: 訓練特徵
            y_train: 訓練目標
            model_type: 模型類型（'rf': 隨機森林, 'gb': 梯度提升）
        """
        print(f"\n訓練{model_type.upper()}模型...")
        
        # 標準化特徵
        X_train_scaled = self.scaler.fit_transform(X_train)
        
        # 選擇模型
        if model_type == 'rf':
            self.model = RandomForestRegressor(
                n_estimators=100,
                max_depth=15,
                min_samples_split=10,
                random_state=42,
                n_jobs=-1
            )
        elif model_type == 'gb':
            self.model = GradientBoostingRegressor(
                n_estimators=100,
                max_depth=5,
                learning_rate=0.1,
                random_state=42
            )
        else:
            raise ValueError(f"不支援的模型類型: {model_type}")
        
        # 訓練模型
        self.model.fit(X_train_scaled, y_train)
        
        print("模型訓練完成！")
        
        # 輸出特徵重要性
        if hasattr(self.model, 'feature_importances_'):
            feature_importance = pd.DataFrame({
                'feature': self.feature_columns,
                'importance': self.model.feature_importances_
            }).sort_values('importance', ascending=False)
            
            print("\n前10個重要特徵:")
            print(feature_importance.head(10).to_string(index=False))
    
    def evaluate_model(self, X_test, y_test):
        """
        評估模型效能
        
        參數:
            X_test: 測試特徵
            y_test: 測試目標
            
        回傳:
            metrics: 評估指標字典
        """
        print("\n評估模型效能...")
        
        # 標準化測試資料
        X_test_scaled = self.scaler.transform(X_test)
        
        # 預測
        y_pred = self.model.predict(X_test_scaled)
        
        # 計算評估指標
        mae = mean_absolute_error(y_test, y_pred)
        rmse = np.sqrt(mean_squared_error(y_test, y_pred))
        mape = np.mean(np.abs((y_test - y_pred) / y_test)) * 100
        r2 = r2_score(y_test, y_pred)
        
        metrics = {
            'MAE': mae,
            'RMSE': rmse,
            'MAPE': mape,
            'R²': r2
        }
        
        print("\n模型評估結果:")
        for metric_name, value in metrics.items():
            if metric_name == 'MAPE':
                print(f"  {metric_name}: {value:.2f}%")
            else:
                print(f"  {metric_name}: {value:.2f}")
        
        return metrics, y_pred
    
    def predict_future(self, features_df, steps=24):
        """
        預測未來工作負載
        
        參數:
            features_df: 當前特徵DataFrame
            steps: 預測步數（小時）
            
        回傳:
            predictions: 預測結果
        """
        print(f"\n預測未來{steps}小時的工作負載...")
        
        if self.model is None:
            raise ValueError("模型尚未訓練！")
        
        # 取最新的特徵
        latest_features = features_df[self.feature_columns].iloc[-1:].values
        latest_features_scaled = self.scaler.transform(latest_features)
        
        # 預測
        prediction = self.model.predict(latest_features_scaled)[0]
        
        print(f"預測的CPU使用率: {prediction:.2f}%")
        
        return prediction
    
    def make_scaling_decision(self, current_usage, predicted_usage, 
                             scale_up_threshold=70, scale_down_threshold=30):
        """
        做出擴展決策
        
        參數:
            current_usage: 當前使用率
            predicted_usage: 預測使用率
            scale_up_threshold: 擴展上限閾值
            scale_down_threshold: 縮減下限閾值
            
        回傳:
            decision: 擴展決策字典
        """
        print("\n分析擴展決策...")
        
        decision = {
            'action': 'maintain',
            'reason': '',
            'current_usage': current_usage,
            'predicted_usage': predicted_usage
        }
        
        if predicted_usage > scale_up_threshold:
            decision['action'] = 'scale_up'
            decision['reason'] = f'預測使用率({predicted_usage:.1f}%)超過上限閾值({scale_up_threshold}%)'
            decision['recommended_instances'] = int(np.ceil(predicted_usage / scale_up_threshold))
        elif predicted_usage < scale_down_threshold:
            decision['action'] = 'scale_down'
            decision['reason'] = f'預測使用率({predicted_usage:.1f}%)低於下限閾值({scale_down_threshold}%)'
            decision['recommended_instances'] = max(1, int(np.floor(predicted_usage / scale_down_threshold)))
        else:
            decision['reason'] = f'預測使用率({predicted_usage:.1f}%)在正常範圍內'
        
        print(f"決策: {decision['action'].upper()}")
        print(f"原因: {decision['reason']}")
        
        return decision
    
    def visualize_predictions(self, timestamps, actual, predicted):
        """
        視覺化預測結果
        
        參數:
            timestamps: 時間戳記
            actual: 實際值
            predicted: 預測值
        """
        print("\n生成預測視覺化圖表...")
        
        fig, axes = plt.subplots(2, 1, figsize=(14, 10))
        
        # 子圖1: 預測 vs 實際
        ax1 = axes[0]
        ax1.plot(timestamps, actual, label='實際值', 
                color='blue', linewidth=2, alpha=0.7)
        ax1.plot(timestamps, predicted, label='預測值', 
                color='red', linewidth=2, linestyle='--', alpha=0.7)
        ax1.set_xlabel('時間', fontsize=12)
        ax1.set_ylabel('CPU使用率 (%)', fontsize=12)
        ax1.set_title('工作負載預測 vs 實際', fontsize=14, fontweight='bold')
        ax1.legend(fontsize=11)
        ax1.grid(alpha=0.3)
        ax1.tick_params(axis='x', rotation=45)
        
        # 子圖2: 預測誤差
        ax2 = axes[1]
        errors = actual - predicted
        ax2.plot(timestamps, errors, color='purple', linewidth=1.5, alpha=0.7)
        ax2.axhline(y=0, color='red', linestyle='--', linewidth=1)
        ax2.fill_between(timestamps, 0, errors, alpha=0.3, color='purple')
        ax2.set_xlabel('時間', fontsize=12)
        ax2.set_ylabel('預測誤差 (%)', fontsize=12)
        ax2.set_title('預測誤差分析', fontsize=14, fontweight='bold')
        ax2.grid(alpha=0.3)
        ax2.tick_params(axis='x', rotation=45)
        
        plt.tight_layout()
        plt.show()
        
        print("視覺化圖表生成完成！")

# 主程式執行
if __name__ == "__main__":
    print("="*70)
    print("雲端工作負載預測與智慧擴展系統")
    print("="*70)
    
    # 初始化預測器
    predictor = CloudWorkloadPredictor(prediction_horizon=24)
    
    # 生成範例資料
    workload_df = predictor.generate_sample_data(days=90)
    
    # 顯示前幾筆資料
    print("\n前5筆工作負載資料:")
    print(workload_df.head().to_string())
    
    # 建立特徵
    features_df = predictor.create_features(workload_df)
    
    # 準備訓練資料
    X, y = predictor.prepare_training_data(features_df, target_col='cpu_usage')
    
    # 分割訓練集與測試集（時間序列保持順序）
    train_size = int(0.8 * len(X))
    X_train, X_test = X[:train_size], X[train_size:]
    y_train, y_test = y[:train_size], y[train_size:]
    
    print(f"\n訓練集大小: {len(X_train)}")
    print(f"測試集大小: {len(X_test)}")
    
    # 訓練模型
    predictor.train_model(X_train, y_train, model_type='rf')
    
    # 評估模型
    metrics, y_pred = predictor.evaluate_model(X_test, y_test)
    
    # 視覺化預測結果
    test_timestamps = features_df['timestamp'].iloc[-len(y_test):].values
    predictor.visualize_predictions(test_timestamps, y_test, y_pred)
    
    # 做出擴展決策
    current_cpu = y_test[-1]
    predicted_cpu = y_pred[-1]
    decision = predictor.make_scaling_decision(
        current_cpu, predicted_cpu,
        scale_up_threshold=70,
        scale_down_threshold=30
    )
    
    print("\n" + "="*70)
    print("分析完成！")
    print("="*70)