「沒有數據記錄的分析只是猜測,能將每一次 AI 預測落盤並進行長效覆盤與準確率追蹤,才能打造真正具備商業價值的金融 Agent。」
在 Day 09 中,我們介紹了每日自動化盤後分析與 Telegram 推播腳本 (daily_analysis.py)。今天我們將進入自動化 pipeline 的最後一個核心模組——sheets_tracker.py Google Sheets 數據追蹤與績效評估模組。
透過 Google Sheets API 與 gspread 套件,系統能在每日盤後分析完成後,自動將加權指數、成交量、美股三大指數價位、AI 看法方向與報告摘要結構化寫入雲端試算表,並提供 update_prediction_accuracy 預測準確率回填機制,為未來評估 AI 分析準確度提供長期數據積累。
在 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)之間的數據橋樑。
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
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}")
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)