第7章 時序模型的可視化、異常偵測與監控

# 第7章 時序模型的可視化、異常偵測與監控 本章聚焦於模型產出之後的兩大關鍵環節：**可視化** 與 **異常偵測** 以及**模型監控**。在實務上，無論是部署於生產環境還是對業務決策提供支持，僅僅有一個高性能模型已不足以確保長期價值。必須透過圖形化分析、監測指標與自動化警示，確保模型的預測準確性、可靠性與可解釋性。以下以實際範例說明整體流程，並附上可直接執行的程式碼片段與建議工具。 --- ## 7.1 時序資料可視化 ### 7.1.1 目標與需求 - **趨勢分析**：檢視整體走勢，確認季節性或長期變化。 - **模型預測可視化**：將實際值、訓練期預測、驗證期預測與測試期預測繪製於同一圖表，直觀比較誤差。 - **異常偵測輔助**：快速定位時間點可能的離群現象。 ### 7.1.2 工具選型 | 工具 | 優點 | 缺點 | |------|------|------| | **Plotly Express** | 交互式、簡易 API | 需安裝 `plotly` | | **Matplotlib + Seaborn** | 靜態圖、可嵌入報告 | 互動性較差 | | **Bokeh** | 交互式、支援大資料 | 需要額外前端配置 | 我們選擇 **Plotly Express** 作為示範，因其既能快速生成交互式圖表，也能輕鬆嵌入 Jupyter Notebook 或 Web 服務。 ### 7.1.3 範例程式 python import pandas as pd import plotly.express as px from sklearn.metrics import mean_absolute_error # 假設已經有完整的時序資料 # df 原始資料：datetime, target, feature1, ... # df_pred_train, df_pred_val, df_pred_test 分別為各階段預測結果 # 每個 df 都有列：datetime, pred, true def plot_predictions(df, df_train, df_val, df_test): fig = px.line(df, x='datetime', y='true', labels={'true':'真實值'}, title='時序預測對比') fig.add_scatter(x=df_train['datetime'], y=df_train['pred'], mode='lines', name='訓練預測') fig.add_scatter(x=df_val['datetime'], y=df_val['pred'], mode='lines', name='驗證預測') fig.add_scatter(x=df_test['datetime'], y=df_test['pred'], mode='lines', name='測試預測') fig.update_layout(yaxis_title='目標變數') fig.show() # 範例呼叫 plot_predictions(df, df_pred_train, df_pred_val, df_pred_test) ### 7.1.4 高階可視化技巧 - **雙軸圖**：將預測誤差（MAE、RMSE）作為第二軸，觀察誤差隨時間變化。 - **熱力圖**：展示不同時間窗口內的誤差分佈，便於定位高風險區段。 - **交互式滑桿**：允許使用者在 Web 端選擇顯示時間段，提升探索體驗。 --- ## 7.2 異常偵測 ### 7.2.1 異常定義 - **觀測異常**：單一資料點偏離預期分布。 - **模式異常**：連續多個觀測值偏離長期模式，常見於季節性失調。 - **模型異常**：模型本身的輸出在預測區間內產生極端偏差。 ### 7.2.2 方法選擇 | 方法 | 特色 | 適用場景 | |------|------|----------| | **基於統計的 Z-Score** | 簡單、快速 | 觀測異常，資料量大時適用 | | **Isolation Forest** | 無監督、適合高維 | 多維特徵異常偵測 | | **Prophet + Bayesian Anomaly Detection** | 內建季節性模型 | 時序特殊結構 | | **LSTM Autoencoder** | 捕捉長期依賴 | 高維時序、複雜模式 | 本章以 **Isolation Forest** 與 **LSTM Autoencoder** 為例，說明如何在不同層面進行異常偵測。 ### 7.2.3 Isolation Forest 範例 python from sklearn.ensemble import IsolationForest import numpy as np # 假設 df_test 只包含預測值與實際值 X = df_test[['pred', 'true']].values iso_forest = IsolationForest(contamination=0.01, random_state=42) iso_forest.fit(X) df_test['anomaly_score'] = iso_forest.decision_function(X) df_test['anomaly'] = iso_forest.predict(X) # -1 表示異常 # 取出異常點 anomalies = df_test[df_test['anomaly'] == -1] print('檢測到 {} 個異常點'.format(len(anomalies))) ### 7.2.4 LSTM Autoencoder 範例 python import torch import torch.nn as nn from torch.utils.data import DataLoader, TensorDataset class LSTMAutoencoder(nn.Module): def __init__(self, input_dim, hidden_dim, seq_len): super().__init__() self.encoder = nn.LSTM(input_dim, hidden_dim, batch_first=True) self.decoder = nn.LSTM(hidden_dim, input_dim, batch_first=True) self.seq_len = seq_len def forward(self, x): _, (h, _) = self.encoder(x) repeated = h.repeat(1, self.seq_len, 1) out, _ = self.decoder(repeated) return out # 準備資料：seq_len=30，feature_dim=1 seq_len = 30 X_seq = [] for i in range(len(df_test)-seq_len+1): seq = df_test['true'].values[i:i+seq_len] X_seq.append(seq.reshape(-1,1)) X_seq = torch.tensor(X_seq, dtype=torch.float32) dataset = TensorDataset(X_seq, X_seq) loader = DataLoader(dataset, batch_size=64, shuffle=True) model = LSTMAutoencoder(input_dim=1, hidden_dim=16, seq_len=seq_len) criterion = nn.MSELoss() optimizer = torch.optim.Adam(model.parameters(), lr=1e-3) # 只示範 5 次 epoch for epoch in range(5): for xb, yb in loader: pred = model(xb) loss = criterion(pred, yb) optimizer.zero_grad(); loss.backward(); optimizer.step() 訓練完畢後，對測試序列做重構，計算重構誤差。將誤差與門檻做比對，即可得到異常點。門檻可用 **百分位法** 或 **統計閾值** 決定。 ### 7.2.5 實務建議 1. **多層級偵測**：觀測層級與模型層級分別設置門檻，避免單一方法誤報。 2. **門檻自動化**：利用 **Percentile(95)** 或 **RobustScaler** 讓門檻隨時間自適應。 3. **可視化整合**：將異常點在預測圖上標註，方便直覺判斷。 --- ## 7.3 模型監控與持續交付 ### 7.3.1 監控指標 | 指標 | 目的 | |------|------| | **Data Drift** | 識別輸入特徵分布變化 | | **Concept Drift** | 識別目標變數與特徵關係變化 | | **Prediction Drift** | 觀測預測值分布變化 | | **Latency** | 推論時間 | | **Error Rates** | MAE, RMSE, MAPE 之時間序列 | | **System Metrics** | CPU, GPU 使用率、記憶體 | ### 7.3.2 工具鏈 - **MLflow**：模型管理、實驗記錄。 - **Prometheus + Grafana**：時序資料監控、告警。 - **Seldon Core / KFServing**：Kubernetes 上的模型服務。 - **KubePrometheus**：整合監控、告警、日誌。 - **TensorBoard**：可視化訓練與推論指標。 ### 7.3.3 建立監控工作流程 1. **模型日誌**：每次推論回傳 `prediction`, `true_value`（若可得）與 `timestamp`。 2. **資料上傳**：將日誌寫入 Kafka / Cloud Storage，或直接寫入 Prometheus Pushgateway。 3. **Prometheus 采集**：設置 exporter 讀取模型日誌並產生 metric。 4. **Grafana Dashboard**：以 **時間滑桿** 顯示 MAE、RMSE、Prediction Drift。 5. **告警規則**：當 MAE 超過 3σ 或 Data Drift 超過 0.2 時發送 Slack / Email 通知。 6. **自動回測**：定期載入最新一週資料，使用 **A/B test** 或 **Offline Validation** 檢驗模型是否仍符合基準。 #### 範例：Prometheus Metric python from prometheus_client import Gauge, start_http_server import time # 設定指標 mae_gauge = Gauge('model_mae', 'Mean Absolute Error of the model') latency_gauge = Gauge('model_latency_seconds', 'Inference latency') start_http_server(8000) while True: # 假設我們已經有測試批次的 predictions & truths mae = mean_absolute_error(truths, preds) latency = compute_latency() mae_gauge.set(mae) latency_gauge.set(latency) time.sleep(60) ### 7.3.4 CI/CD 與版本管理 - **模型訓練 CI**：使用 GitHub Actions 或 GitLab CI 執行 `train.py`，並將模型訓練成果儲存至 S3/Google Cloud Storage，並註冊至 MLflow Model Registry。 - **部署 CD**：當 MLflow 觸發 `new_staging_version` 時，Seldon Core 透過 Helm 升級容器，並自動替換 Kubernetes Deployment。 - **Rollbacks**：若監控告警頻繁，Kubernetes 可回滾至 `previous_version`，並透過 `kubectl rollout undo` 或 Helm 的 `history` 指令快速回退。 --- ## 7.4 小結 本章從 **可視化**、**異常偵測** 到 **模型監控** 全面覆蓋了時序機器學習模型的產出評估與運營流程。關鍵點如下： 1. **結合觀測層級與模型層級** 的異常偵測，提升偵測準確度。 2. 利用 **Prometheus + Grafana** 建立實時監控面板，並設置告警觸發。 3. 以 **MLflow + Helm** 搭建 CI/CD 流程，確保模型從實驗到生產的完整鏈路。 4. **自動化門檻** 與 **自適應 drift 檢測**，確保模型在變動環境下仍具可預測性。 > **實務小提醒**：雖然時序模型難以直接得到 `true_value`，但若能透過 **監控系統** 或 **實時評估** 取得真實標籤，能大幅提升監控精度與異常修正速度。 --- ## 7.5 附錄：完整的監控 & 報告示例 以下提供一個簡易的 **Streamlit** 應用，將上述可視化、異常偵測與監控統一展示： python import streamlit as st import pandas as pd import matplotlib.pyplot as plt from streamlit_echarts import st_echarts # 讀取日誌 log_df = pd.read_csv('model_logs.csv') # 1. 預測曲線 st.subheader('預測 vs 實際') fig, ax = plt.subplots(figsize=(10,4)) ax.plot(log_df['timestamp'], log_df['true_value'], label='True') ax.plot(log_df['timestamp'], log_df['prediction'], label='Pred') ax.set_xlabel('時間') ax.set_ylabel('目標') ax.legend() st.pyplot(fig) # 2. MAE 時序 st.subheader('MAE 隨時間變化') mae_series = log_df.groupby('date').apply(lambda g: g['true_value'].sub(g['prediction']).abs().mean()) st.line_chart(mae_series) # 3. 異常標記 st.subheader('異常偵測結果') st.dataframe(log_df[log_df['anomaly']==-1]) --- ### 結語 透過上述可視化、異常偵測與監控流程，您將能在實際業務中快速定位模型問題，並以自動化方式進行模型回測、部署與回滾，確保模型始終保持最佳效能。 --- **後續學習**：接下來的章節將深入 **可解釋 AI (XAI)** 與 **邊緣推論優化**，進一步提升時序模型在不同環境下的可用性與效能。