「一個完整的生產級 AI Agent,不能只管把報告推播出去,更必須將每日產出的市場數據、AI 預判方向與推播狀態進行持久化追蹤,為後續的勝率統計與離線 Evaluation 建立數據基石。」
(配合前幾天介紹的 daily_analysis.py 核心邏輯演進,決定將 Day 19 的主題稍微調整,直接聚焦在當前版本中最硬核的「資料持久化追蹤與 AI 方向標籤提取」實作上!)
在 Day 18 中,我們完成了 daily_analysis.py 透過 Telegram API 推播訊息的 UTF-8 長度切分與語法降級機制。報告成功發送之後,我們隨即面臨另一個營運痛點:如何無痛紀錄並評估 Agent 每日產出的預測品質呢?
如果只把報告留存在 Telegram 或 Local Log 中,要進行長期勝率統計與覆盤會非常困難。因此,我們在 daily_analysis.py 的主流程中整合了 Google Sheets 自動化紀錄機制(record_daily_data),將每日大盤數據、美股指數、AI 預判方向與執行狀態自動寫入雲端試算表。
在 Production 環境中,外部服務(如 Google Cloud API Credentials)可能會因為憑證過期或網路抖動無法連線。為了防止 Google Sheets 紀錄失敗影響到 Telegram 主推播流程,我們在 daily_analysis.py 頂層採用了安全載入機制:
Python
# Google Sheets tracking 容錯載入
try:
from app.sheets_tracker import record_daily_data
SHEETS_ENABLED = True
except ImportError:
SHEETS_ENABLED = False
print("[WARN] sheets_tracker not available")
透過 SHEETS_ENABLED 布林值標記,即便伺服器環境未配置 Google API 金鑰,主流程依然能順利完成盤後分析與 Telegram 推播,不會發生阻斷性崩潰(Non-blocking Architecture)。
當 generate_report() 產出最終報告後,系統會自動掃描報告關鍵字,將非結構化的自然語言報告轉化為結構化的方向標籤:
Python
# Step 6: Record to Google Sheets 中提取 AI 方向標籤
if SHEETS_ENABLED:
print("\n[6/7] Recording to Google Sheets...")
try:
# 提取當日數據
tw_summary = market_data.get('taiwan', {}).get('monthly_summary', {})
us_info = market_data.get('us', {})
# 從 AI 生成報告中自動判斷 AI 方向標籤
ai_direction = "中性"
report_lower = report.lower() if report else ""
if any(w in report_lower for w in ["看多", "偏多", "樂觀", "上漲"]):
ai_direction = "看多"
elif any(w in report_lower for w in ["看空", "偏空", "悲觀", "下跌"]):
ai_direction = "看空"
接著,將當日所有 TWSE 成交金額、美股三大指數與推播執行狀態封裝為標準字典:Python sheets_data = { '日期': datetime.now(TW_TZ).strftime('%Y-%m-%d'), '加權指數': tw_summary.get('收盤指數', 'N/A'), '漲跌幅(%)': tw_summary.get('漲跌百分比(%)', 'N/A'), '成交量(億)': tw_summary.get('成交金額(元)', 'N/A'), 'S&P 500': str(us_info.get('S&P 500', {}).get('price', 'N/A')), 'NASDAQ': str(us_info.get('NASDAQ', {}).get('price', 'N/A')), '道瓊': str(us_info.get('Dow Jones', {}).get('price', 'N/A')), 'AI方向': ai_direction, '推撥狀態': '成功' if success else '失敗', '分析摘要': (report or '').replace('\n', ' ')[:500], } record_daily_data(sheets_data) print(" ✓ Recorded to Google Sheets") except Exception as e: print(f" [ERROR] Sheets recording failed: {e}")
在 main() 函式的末段,當 Telegram 發送成功後,這道資料寫入工序會自動觸發:
```
Plaintext
[1/6] Fetching market data...
[2/6] Querying knowledge base...
[3/6] Generating AI analysis report...
[4/6] Preparing final report...
[5/6] Sending to Telegram...
[6/7] Recording to Google Sheets... <-- 自動持久化寫入
[7/7] Storing analysis in knowledge base...
```
### 營運覆盤效益:
1. 數據零代價累積:每日自動建立結構化歷史資料集,省去手動紀錄的麻煩。
2. 自動勝率對照:將每日提取的 AI方向 與次日 TWSE 實際漲跌比對,即可在 Google Sheets 中公式化計算 Agent 的真實預測勝率。
3. 監控與告警:紀錄 推撥狀態,作為自動化巡檢與維運除錯的依據。
透過 daily_analysis.py 整合 Google Sheets 自動追蹤模組,我們實現了:
1. 營運資料 cycle:數據抓取 $\rightarrow$ AI 三階段推理 $\rightarrow$ Telegram 推播 $\rightarrow$ Google Sheets 自動紀錄。
2. 高容錯降級:採用 Try-Import 模式,確保紀錄模組不會影響推播核心主管線。
明天(Day 20)我們將正式進入 部署架構重大重構:從 Podman 遷移至 Native systemd system service (angelina.service) 的託管實務!
明日預告:【Day 20】部署架構重構:Podman 到 Native systemd 服務遷移 (angelina.service)