iT邦幫忙

2026 iThome 鐵人賽

DAY 10
0
Build on Google AI

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

【Day 10】數據追蹤與績效評估:Google Sheets 數據寫入與預測準確率追蹤 (sheets_tracker.py)

  • 分享至 

  • xImage
  •  

「沒有數據記錄的分析只是猜測,能將每一次 AI 預測落盤並進行長效覆盤與準確率追蹤,才能打造真正具備商業價值的金融 Agent。」

在 Day 09 中,我們介紹了每日自動化盤後分析與 Telegram 推播腳本 (daily_analysis.py)。今天我們將進入自動化 pipeline 的最後一個核心模組——sheets_tracker.py Google Sheets 數據追蹤與績效評估模組。

透過 Google Sheets API 與 gspread 套件,系統能在每日盤後分析完成後,自動將加權指數、成交量、美股三大指數價位、AI 看法方向與報告摘要結構化寫入雲端試算表,並提供 update_prediction_accuracy 預測準確率回填機制,為未來評估 AI 分析準確度提供長期數據積累。

本篇重點摘要

  1. 為什麼選擇 Google Sheets 作為輕量化雲端報表與績效資料庫呢?
  2. 拆解 sheets_tracker.py 11 大標準欄位(HEADERS)設計與服務帳號認證。
  3. 實作工作表動態初始化機制 (init_sheet):自動建立 Daily Tracker 分頁與標題列校正。
  4. 剖析每日數據追加 (record_daily_data) 與歷史預測結果回填 (update_prediction_accuracy) 邏輯。

一、報表選型與憑證架構

在 Self-hosted 且 $0 元營運的前提下,使用 Google Sheets 作為數據終端具有顯著優勢:
1. 無需自建 UI 報表:利用 Google Sheets 原生的圖表與篩選功能,維運者可隨時透過手機或瀏覽器查看每日市場指標與 AI 方向。
2. 無縫整合 Service Account:使用與 drive_sync.py 相同的服務帳號金鑰文件 (/opt/angelina/config/service-account.json),透過 spreadsheets 與 drive 作用域 (Scopes) 進行授權,免去使用者登入驗證。
3. 異質系統橋樑:作為 daily_analysis.py 與外部數據分析工具(如 Python Pandas 或 Looker Studio)之間的數據橋樑。

二、sheets_tracker.py 欄位規格與初始化機制

  1. 11 大標準欄位(HEADERS)規格
    為了完整記錄台股與美股的盤後數據,sheets_tracker.py 定義了標準欄位結構:
Python
HEADERS = [
    '日期', '加權指數', '漲跌幅(%)', '成交量(億)',
    'S&P 500', 'NASDAQ', '道瓊',
    '三大法人淨買賣(億)', 'AI方向', '推撥狀態', '分析摘要'
]
2. 工作表自動初始化與修復 (init_sheet)
在腳本首次執行或試算表尚未建立分頁時,init_sheet() 具備自我修復與建置能力:

自動尋找或新增名為 Daily Tracker 的工作表。

若試算表為空,自動寫入標準 HEADERS。

若第一列為數據而非標題,自動於第一列插入 HEADERS,確保欄位對齊。

Python
def init_sheet():
    """初始化 gspread 客戶端與工作表,自動建構 Daily Tracker 與標題列"""
    try:
        credentials = Credentials.from_service_account_file(
            SERVICE_ACCOUNT_PATH, scopes=SCOPES
        )
        client = gspread.authorize(credentials)
        spreadsheet = client.open_by_key(SPREADSHEET_ID)

        # 尋找或自動建立 Daily Tracker 工作表
        try:
            worksheet = spreadsheet.worksheet(WORKSHEET_NAME)
        except gspread.exceptions.WorksheetNotFound:
            worksheet = spreadsheet.add_worksheet(
                title=WORKSHEET_NAME, rows=1000, cols=len(HEADERS)
            )

        # 確保第 1 列永遠包含正確的標題列
        all_vals = worksheet.get_all_values()
        if not all_vals:
            worksheet.append_row(HEADERS)
        elif all_vals[0] != HEADERS:
            worksheet.insert_row(HEADERS, 1)

        return worksheet
    except Exception as e:
        print(f"[SheetsTracker] Error initializing sheet: {e}")
        return None

三、數據寫入與預測準確率回填實務

  1. 每日盤後數據追加 (record_daily_data)
    當 daily_analysis.py 完成分析後,會建構包含市場數據與 AI 方向(看多/看空/中性)的字典,呼叫 record_daily_data 追加新的資料列:
Python
def record_daily_data(data: dict):
    """追加今日數據至 Daily Tracker 試算表末端"""
    try:
        sheet = init_sheet()
        if sheet is None:
            return

        # 依據 HEADERS 順序提取資料,缺漏欄位自動補空字串
        row = [str(data.get(header, '')) for header in HEADERS]
        sheet.append_row(row)
        print(f"[SheetsTracker] Daily data recorded for {data.get('日期', 'unknown date')}.")
    except Exception as e:
        print(f"[SheetsTracker] Error recording daily data: {e}")
  1. 歷史預測準確率更新 (update_prediction_accuracy)
    為了在未來進行 AI 預測能力的評估與校正(Backtesting & Evaluation),模組提供了 update_prediction_accuracy 函式。
    它能依據指定日期搜尋特定資料列,並自動在試算表末端建立或更新 實際結果 欄位:
Python
def update_prediction_accuracy(date_str: str, actual_result: str):
    """依據指定日期更新『實際結果』欄位,用於未來追蹤預測準確率"""
    try:
        sheet = init_sheet()
        if sheet is None:
            return

        all_values = sheet.get_all_values()
        headers_row = all_values[0] if all_values else []

        # 動態檢查或建立 '實際結果' 欄位
        if '實際結果' in headers_row:
            result_col = headers_row.index('實際結果') + 1  # gspread 採用 1-based 索引
        else:
            result_col = len(headers_row) + 1
            sheet.update_cell(1, result_col, '實際結果')

        # 搜尋日期欄位並更新特定儲存格
        date_col_values = sheet.col_values(1)
        if date_str in date_col_values:
            row_index = date_col_values.index(date_str) + 1
            sheet.update_cell(row_index, result_col, actual_result)
            print(f"[SheetsTracker] Updated prediction accuracy for {date_str}: {actual_result}")
    except Exception as e:
        print(f"[SheetsTracker] Error updating prediction accuracy: {e}")

四、章節總結與自動化管線回顧

到這裡,Week 3 的「數據 pipeline 自動化與周邊服務整合」已全部實作完畢!
我們建立了一套完整的資料進出閉環:
1. 資料輸入管線 (drive_sync.py):自動將 Google Drive 上的理財筆記與報告,透過 /learn 指令注入 RAG 知識庫。
2. 資料處理與推播 (daily_analysis.py):自動爬取台股/美股數據,呼叫 Gemini 進行推理,經由 Telegram 推播並自動存回知識庫。
3. 資料持久化追蹤 (sheets_tracker.py):結構化記錄每日數據與 AI 方向,並支援預測準確率的回填與覆盤。

後續我們將進入前端介面、系統全監控與部署總結——實作零框架 Web Chat UI、FastAPI 健康檢查與系統維運指標 API!

明日預告:【Day 11】前端互動介面:極簡 RHEL 風格 Web Chat UI 與靜態檔案掛載 (static/index.html)


上一篇
【Day 09】每日自動化盤後分析:TWSE/美股數據爬取、Gemini 分析與 Telegram 機器人推播 (daily_analysis.py)
下一篇
【Day 11】前端互動介面:極簡前端與獨立單元測試驗證模組 (static/validation.js)
系列文
打造零成本企業級 AI Agent:以 Gemini 2.5 Flash 構建金融分析助手與維運實戰11
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言