iT邦幫忙

2026 iThome 鐵人賽

DAY 15
0
Build on Google AI

打造零成本企業級 AI Agent:以 Gemini 2.5 Flash 構建金融分析助手與維運實戰系列 第 19

【Day 19】數據追蹤與自動覆盤:Google Sheets 紀錄整合與 AI 方向預判紀錄

  • 分享至 

  • xImage
  •  

「一個完整的生產級 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 預判方向與執行狀態自動寫入雲端試算表。

本篇重點摘要

  1. 剖析 daily_analysis.py 整合 sheets_tracker.py 的容錯載入(Try-Import)設計。
  2. 解析如何從 AI 報告中自動提取「看多/看空/中性」方向標籤。
  3. 實作 Google Sheets 每日維護資料集(包含指數、成交量、推播狀態與分析摘要)。

一、模組容錯載入與非阻斷式設計

在 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)。

二、AI 預判方向動態提取與結構化封裝

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


上一篇
【Day 18】推播優化:Telegram 訊息 UTF-8 Byte-Safe 精確切分演算法
系列文
打造零成本企業級 AI Agent:以 Gemini 2.5 Flash 構建金融分析助手與維運實戰19
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言