「一個架構良善的 AI Agent 系統，其檢驗在於 Pipeline End-to-End 的實際執行狀況。從市場數據抓取、知識庫查詢、Gemini 三階段推理、Telegram 推播、Google Sheets 紀錄到當日報告回寫知識庫，7 個步驟必須像精密齒輪般順暢運轉。」

先前，我們陸續重構並驗證了 Angelina Agent 的各大獨立模組（包含三階段 Prompt、Telegram Byte-Safe 切分、Google Sheets 追蹤與 systemd 定時器排程...,etc）。今天我們將把視角拉回最核心的主流程。
我們將進行 端到端 整合測試，完整驗證當我們在執行時，系統是如何依序觸發 pipeline 步驟，完成資料處理、AI 分析報告生成與狀態紀錄。

本篇重點摘要

1. 剖析 7 大執行步驟的架構。
2. 觀察實際執行 Console Log 輸出與各模組溝通邊界。
3. 驗證從數據擷取到 Telegram 發送、Google Sheets 自動紀錄與知識庫寫回的完整 cycle 。

一、daily_analysis.py 7 大步驟執行脈絡剖析

在 daily_analysis.py 的 main() 函式中，管線採取嚴密的單向流動（Dataflow Pipeline）設計，每個步驟在 Terminal 都有清晰的階段 Log 標記：

# daily_analysis.py 主流程 7 大步驟程式碼架構對應
async def main():
    # [1/6] Fetching market data... (TWSE, US Market, Yahoo Stock News)
    # [2/6] Querying knowledge base... (/chat /learning-stats)
    # [3/6] Generating AI analysis report... (Stage 1-3 Gemini Pipeline)
    # [4/6] Preparing final report... (Adding Title Header)
    # [5/6] Sending to Telegram... (Byte-Safe Splitting & Rate-Limit Sleep)
    # [6/7] Recording to Google Sheets... (if SHEETS_ENABLED)
    # [7/7] Storing analysis in knowledge base... (/chat /learn endpoint)

二、E2E 整合測試與 Console 日誌驗證

當我們透過命令列手動啟動整合測試：

/opt/angelina/venv/bin/python /opt/angelina/daily_analysis.py
系統輸出的真實 Console 畫面與步驟剖析：
[1/6] Fetching market data...
台灣市場 (fetch_twse_daily)：自動呼叫證交所 API，抓取每月每日彙整 (FMTQIK)、漲幅前十 (MI_INDEX20) 以及三大法人買賣超 (BFI82U)。

美國市場 (fetch_us_market)：呼叫 Yahoo Finance 抓取 S&P 500、NASDAQ 與道瓊指數最新點位與漲跌幅。

盤後新聞 (fetch_news)：爬取 Yahoo 股市頭條新聞，提取當日熱門話題。

[2/6] Querying knowledge base...
知識庫查詢 (get_knowledge_context)：對 FastAPI 主服務（http://localhost:8080/chat）發送 /learning-stats 特殊指令，取得當前 Agent 累積的投資框架與統計資訊。

[3/6] Generating AI analysis report...
Gemini 三階段推理 (generate_report)：

Stage 1：輸入 Raw Data 產出精簡 JSON 盤態方向（bullish/bearish/neutral）。

Stage 2：依據 Stage 1 方向至在地 ChromaDB 向量庫提取針對性防禦/攻擊策略。

Stage 3：結合 7 條 Core Execution Rules 與方向硬性約束（Mandatory Line），生成高質量分析報告。

[4/6] Preparing final report...
報告格式前處理：加入當日日期標題頭，並觸發 _fmt() 前處理正則，自動將 9 位數以上的金額轉換為「億/兆元」單位。

[5/6] Sending to Telegram...
send_telegram：檢測訊息長度，若超過 4000 字則依據換行符安全切割，發送時使用 Hash 去重避免重複推播，且每段訊息之間加入 await asyncio.sleep(1) 防範 Rate Limit。

[6/7] Recording to Google Sheets...
record_daily_data：掃描報告提取 ai_direction（看多/看空/中性）標籤，連同指數點位與推播狀態寫入雲端 Google Sheets 試算表。

[7/7] Storing analysis in knowledge base...
store_to_knowledge：將產出的當日分析報告透過 Telegram/Chat 管道呼叫 /chat 的 /learn 指令，讓 Agent 自動切碎 Chunk 寫回 ChromaDB，實現自我歷史記憶 cycle ！

三、E2E 測試中的容錯與防禦邊界驗證

在整合測試過程中，我們也驗證了兩個關鍵的降級防禦機制：
1. 非阻斷式 Google Sheets 寫入 (Non-blocking Design)
測試情境：故意不安裝 Google Sheets 相關套件（觸發 ImportError）。
測試結果：系統觸發 SHEETS_ENABLED = False，輸出 [WARN] sheets_tracker not available，[6/7] 步驟自動顯示為 skipping，後續的 [7/7] 知識庫回寫依然順利完成，主管線完全不崩潰。

  2. 知識庫寫回容錯 (Non-critical Error Handling)
     測試情境：模擬 FastAPI 主服務暫時不穩定或回應失敗。
     測試結果：Step 7 捕捉 Exception 並輸出 ⚠ Could not store in knowledge base (non-critical)，測試主流程依然劃下 Daily analysis complete! 的圓滿句點。

四、今日總結

透過 daily_analysis.py 的端到端整合測試，我們確認了：
1. 7 步管線連貫性：從 Raw Data 抓取到最終知識庫回寫，數據傳遞順暢。
2. 高韌性邊界：Try-Except 與 Non-critical 設計確保了非核心模組失敗時，主推播管線依然 100% 安定執行。

明日預告：敬請期待~