第 9 章 監控與持續改進

# 第 9 章 監控與持續改進 > **核心訊息**：在量化投資的生命週期中，*監控* 與 *迴圈改進* 是將歷史優勢轉化為長期可持續績效的關鍵橋樑。\n本章將帶你從概念到實作，打造一套完整、可擴充的監控體系，並示範如何利用數據驅動的迴圈來自動化策略優化。 --- ## 9.1 監控的目的與範疇 | 監控類別 | 主要目的 | 典型指標 | |----------|----------|-----------| | **績效監控** | 跟踪策略收益與風險 | 夏普比率、Sortino、年化報酬率、最大回撤 | | **風險監控** | 確保風險在可接受範圍內 | VaR、CVaR、日均損失、最大連續虧損 | | **資料監控** | 保證資料完整與正確 | 缺失率、重複率、異常值比例 | | **執行監控** | 檢測交易執行問題 | 滑點、延遲、失敗率、執行頻率 | | **模型監控** | 追蹤模型表現衰退 | 預測偏差、分類準確率、AUC | > **思考框架**：一個成熟的監控系統不僅僅是「發警報」——它應該提供可量化、可行動的洞察，並驅動迴圈改進。 --- ## 9.2 監控指標設計 ### 9.2.1 風險指標 | 指標 | 定義 | 觸發條件 | |------|------|----------| | **VaR** | 95% 信心水平下的風險邊際 | 日 VaR 超過 1% 時停止交易 | | **CVaR** | 在 VaR 以上的平均損失 | CVaR 超過 1.5% | | **最大連續虧損** | 連續虧損期間的累計損失 | 超過 3% 時重新評估風險參數 | ### 9.2.2 資料完整性 - **缺失率**：若日常資料缺失率 > 5%，觸發數據補償流程。 - **異常分布**：利用 z‑score 或 IQR 檢測異常點；若異常比例 > 1%，則自動排除或手動審查。 ### 9.2.3 執行指標 - **滑點**：實際成交價與委託價之差；若滑點 > 0.3% ，重新評估執行策略。 - **延遲**：從下單到成交的時間；若延遲 > 500ms，考慮優化 API 或更換交易所。 > **實務提醒**：指標設計要與風險偏好、資金規模、投資時段同步調整，避免「硬編」指標造成誤判。 --- ## 9.3 監控基礎架構 | 元件 | 功能 | 推薦技術 | |------|------|-----------| | **數據收集** | 交易日誌、行情、風險數據 | Kafka、RabbitMQ | | **儲存** | 時間序列數據、指標數據 | TimescaleDB、InfluxDB | | **可視化** | 實時儀表板 | Grafana、Plotly Dash | | **告警** | 觸發訊息、通知 | Alertmanager、Slack、Email | | **自動化腳本** | 指標計算、模型漂移偵測 | Python（pandas、scikit‑learn） | > **組態範例**：Grafana 連接 InfluxDB，使用 Prometheus exporter 監控交易延遲；Alertmanager 透過 Slack webhook 傳送即時告警。 --- ## 9.4 模型漂移偵測 ### 9.4.1 漂移類型 | 漂移 | 說明 | |------|------| | **概念漂移** | 數據分佈（X）變化 | | **標籤漂移** | 真實目標變化 | | **模型漂移** | 模型輸出與預期偏差 | ### 9.4.2 檢測方法 | 方法 | 目的 | 優缺點 | |------|------|-------| | **Kolmogorov–Smirnov (KS) Test** | 比較兩段樣本分佈 | 易用、統計嚴謹；對於高維資料不適用 | | **Population Stability Index (PSI)** | 量化分佈變化 | 直觀；需設置閾值 | | **Rolling Window Summary** | 監測均值/方差變化 | 簡單；可能誤判短期波動 | | **Autoencoder Reconstruction Error** | 高維資料自動偵測 | 高效；需要訓練 | ### 9.4.3 典型程式範例（Python） python import pandas as pd import numpy as np from scipy.stats import ks_2samp # 1️⃣ 讀取最近 30 天與 60 天歷史數據 recent = pd.read_csv('recent_features.csv') # 30 天 historical = pd.read_csv('historical_features.csv') # 60 天 # 2️⃣ KS Test 針對單一特徵 def ks_test(col): return ks_2samp(recent[col], historical[col]).pvalue # 3️⃣ PSI 計算 def psi(target, ref, bins=10): target_counts = np.histogram(target, bins=bins, density=True)[0] ref_counts = np.histogram(ref, bins=bins, density=True)[0] psi_value = np.sum((target_counts - ref_counts) * np.log(target_counts / ref_counts)) return psi_value # 4️⃣ 漂移報告 report = [] for col in recent.columns: pval = ks_test(col) psi_val = psi(recent[col], historical[col]) drift = { 'feature': col, 'ks_pval': pval, 'psi': psi_val, 'drift_flag': psi_val > 0.1 } report.append(drift) report_df = pd.DataFrame(report) print(report_df) > **實務建議**：對於多模型、不同時間尺度的策略，應將漂移檢測納入 *每天 4 次* 的自動化任務，並將結果寫入 TimescaleDB，方便歷史追蹤。 --- ## 9.5 迴圈改進流程（CRAB） | 步驟 | 描述 | 工具 | |------|------|------| | **Capture** | 透過監控儀表板收集回饋 | Grafana、Alertmanager | | **Review** | 人工或自動分析回饋 | Jupyter、GitHub Issues | | **Adjust** | 調整模型參數或交易規則 | Python / Docker | | **Build** | 重新訓練或加入新特徵 | scikit‑learn、XGBoost | | **Test** | A/B 或模擬驗證 | Backtest engine、Paper Trading | | **Deploy** | 將新版本上線 | Docker、Kubernetes、MLflow | | **Benchmark** | 與舊版本比較 | 版本化指標、AUC、KS | > **實作提示**：利用 *MLflow* 追蹤實驗參數、模型指標與版本，並在 Kubernetes 中透過 `kubectl rollout` 控制滾動更新。 --- ## 9.6 案例：因子策略的完整監控流程 | 步驟 | 內容 | |------|------| | **1️⃣ 設定儀表板** | 顯示日報酬、夏普、VaR、滑點、延遲 | | **2️⃣ 風險告警** | 95% VaR 超過 1% → 通知並暫停所有委託 | | **3️⃣ 模型漂移檢測** | 以 PSI < 0.02 為閾值；若超過，觸發模型重訓 | | **4️⃣ 重訓流程** | 以 30 天過去資料重新訓練；使用 `scikit‑learn` 的 `RandomForestRegressor` | | **5️⃣ A/B 測試** | 同時執行舊版與新版；收集 10 天回報差異 | | **6️⃣ 決策** | 若新版夏普提升 > 0.2，切換至新版；否則回退 | ### 9.6.1 監控腳本範例（簡化版） python import pandas as pd import numpy as np from datetime import datetime from sklearn.ensemble import RandomForestRegressor # 讀取策略日誌 log = pd.read_csv('strategy_log.csv', parse_dates=['date']) # 1️⃣ 指標計算 log['ret'] = log['position'] * log['price_change'] log['cum_ret'] = log['ret'].cumsum() log['drawdown'] = log['cum_ret'].cummax() - log['cum_ret'] max_dd = log['drawdown'].max() # 2️⃣ 風險告警 if max_dd > 0.03: print('⚠️ Max DD > 3% -> 觸發重估') # 這裡可呼叫 Slack API 或寫入 Alertmanager # 3️⃣ 模型漂移檢測（PSI） from sklearn.preprocessing import StandardScaler scaler = StandardScaler() X_recent = scaler.fit_transform(log[['feature1', 'feature2']].tail(30)) X_historical = scaler.fit_transform(log[['feature1', 'feature2']].head(60)) # PSI 計算（兩段平均值與標準差） psi_val = np.mean((np.mean(X_recent, axis=0) - np.mean(X_historical, axis=0))**2) if psi_val > 0.05: print('⚠️ PSI > 0.05 -> 進行重訓') # 重新訓練模型 model = RandomForestRegressor(n_estimators=200) model.fit(X_recent, log['target'].tail(30)) # 保存新模型 import joblib joblib.dump(model, 'models/model_v2.pkl') --- ## 9.7 綜合建議與最佳實踐 1. **指標分層設計**：每個監控層級應有至少兩個「閾值」— *警告* 與 *嚴重*，以避免過度觸發。 2. **告警頻率控制**：採用 *co‑alescing* 或 *sliding‑window suppression*，確保訊息不被「告警噪音」淹沒。 3. **自動化程度**：從 10% 自動化（手動審查）向 90% 自動化（自動重訓、A/B 測試）逐步提升。 4. **版本管理**：將所有模型、腳本、資料管線皆放入 Git 或 DVC，並在每次部署時標註版本號。 5. **資料治理**：實時驗證資料流完整性；若發現重複或缺失，立即觸發補償機制，避免漂移偵測誤判。 6. **合規與可追蹤性**：所有告警與操作都應可回溯至原始交易日誌，符合監管需求。 --- > **結語**：監控不是終點，而是持續改進的起點。隨著市場環境、交易所規則與模型演進的變化，定期回顧與優化監控指標，才能確保策略始終保持「最優」狀態。