2025 iThome 鐵人賽

DAY 16

AI & Data

雲端情人 - AI 愛系列第 16 篇

Day 16｜讓 LINE 金融助理系統優化：更抗風險：重試、退避、快取、斷路器與可觀測性——包含設計思路、精簡可用的程式片段與參考資料

17th鐵人賽

Max Cheng

2025-09-09 08:11:21

96 瀏覽

分享至

下面是我整理好的。

⸻

Day 16｜讓 LINE 金融助理更抗風險

主題重點：當你的 Bot 依賴第三方資料源（Yahoo、匯率 API、新聞 RSS…），偶發連線錯誤、流量管制、格式變動都會讓服務「有時好、有時壞」。今天把幾個實務耐用的工程手法一次補齊：重試＋退避（含抖動）、斷路器、結果快取、尊重 Retry-After、可觀測性（Log / Trace / Error）。

⸻

為什麼一定要「重試＋退避＋抖動（jitter）」？

只要是雲端 API，就可能回 429/5xx 或短暫超時。業界最佳實務是指數退避（exponential backoff），而且加上抖動避免很多客戶端同時在固定節奏撞回來，造成「同步雪崩」。AWS 架構文章展示了不同退避策略的比較，“Full Jitter” 通常表現最佳；Google 也在官方文件建議採用指數退避。

快速可用的 Python 寫法（Tenacity）


from tenacity import retry, wait_exponential_jitter, stop_after_attempt, retry_if_exception_type
import httpx

@retry(
    wait=wait_exponential_jitter(initial=0.5, max=8.0),  # 指數退避 + 抖動
    stop=stop_after_attempt(6),
    retry=retry_if_exception_type((httpx.ReadTimeout, httpx.ConnectError)),
    reraise=True,
)
async def fetch_json(url: str, headers: dict | None = None, timeout=5.0):
    async with httpx.AsyncClient(timeout=timeout) as c:
        r = await c.get(url, headers=headers)
        r.raise_for_status()
        return r.json()

Tenacity 是 Python 常用的重試工具，README 與文件都有多種等待策略（包含 jitter）。

⸻

加上「斷路器」避免雪崩

當下游持續故障時，斷路器（Circuit Breaker）可以在短期內「快速失敗」，不再浪費連線等待，同時保護整體服務；一段時間後再嘗試「半開」恢復。這是經典可靠性模式。可用 aiobreaker（async）或 pybreaker（sync）。

from aiobreaker import CircuitBreaker

breaker = CircuitBreaker(fail_max=5, reset_timeout=30)

@breaker
async def safe_fetch(url: str):
    return await fetch_json(url)  # 這裡可串前面的帶重試的 fetch

⸻

結果快取：減負載、加穩定

財經「快照」類查詢（例如報價、ETF 概覽）通常秒級即可。善用HTTP 快取語義（Cache-Control, ETag, Last-Modified）與應用層快取（如 requests-cache）能顯著降壓，同時在下游短暫抖動時，用「最近一次成功結果」救場。HTTP 的快取語義在 IETF RFC 9111 有標準定義；requests-cache 可將 requests/httpx 回應透明地落地（檔案、SQLite、Redis）。

同步用法示例（httpx 亦有相似外掛生態）

import requests
from requests_cache import CachedSession

session = CachedSession(
    'cache.sqlite', expire_after=30, allowable_codes=(200,), allowable_methods=('GET',)
)
r = session.get("https://api.example.com/snapshot")
data = r.json()

⸻

尊重伺服端的 Retry-After 提示

當你收到 429 Too Many Requests 或 503 Service Unavailable，若回應標頭包含 Retry-After（日期或秒數），客戶端應依建議等待再重試，否則只會加重對方壓力。該標頭的語義與格式定義於 HTTP 規範（歷史上載於 RFC2616，也被後續 HTTP Semantics 覆蓋）。OpenAI 等提供者也在官方文件建議以指數退避處理 429。

import httpx, asyncio
from email.utils import parsedate_to_datetime

async def respectful_get(url):
async with httpx.AsyncClient() as c:
r = await c.get(url)
if r.status_code in (429, 503):
ra = r.headers.get("Retry-After")
if ra:
try:
delay = int(ra)
except ValueError:
delay = max((parsedate_to_datetime(ra) - parsedate_to_datetime(r.headers["Date"])).total_seconds(), 1)
await asyncio.sleep(delay)
r = await c.get(url)
r.raise_for_status()
return r

⸻

可觀測性：Log、Trace、Error 監看要一次到位
• 結構化日誌：用 uvicorn.error、logging 加上 request-id，把錯誤與關鍵欄位打全。
• 分散式追蹤（Tracing）：FastAPI 可用 OpenTelemetry 的 opentelemetry-instrumentation-fastapi 快速掛上自動追蹤，配合你愛用的後端（Jaeger、Tempo、SigNoz）。
• 例外通報：Sentry 的 Python SDK/ASGI/FastAPI 整合能自動捕捉未處理例外與效能資訊。

最快測法：以 opentelemetry 自動掛鉤（示意）

opentelemetry-instrument \
  --traces_exporter console \
  --metrics_exporter none \
  uvicorn app_fastapi:app

⸻

把能力「疊」在一起：通用耐用的抓取層

下方是一個把重試＋退避+jitter、斷路器、尊重 Retry-After、快取疊起來的「資料介面層」。Webhook handler 只依賴這一層，其他換源或改 API 也只動這裡。

resilience.py

import asyncio, json, time
import httpx
from email.utils import parsedate_to_datetime
from tenacity import retry, wait_exponential_jitter, stop_after_attempt, retry_if_exception_type
from aiobreaker import CircuitBreaker

breaker = CircuitBreaker(fail_max=5, reset_timeout=30)

def _respect_retry_after(resp: httpx.Response) -> float | None:
if resp.status_code in (429, 503):
ra = resp.headers.get("Retry-After")
if not ra: return None
try:
return float(ra)
except ValueError:
try:
server_date = parsedate_to_datetime(resp.headers.get("Date"))
retry_at = parsedate_to_datetime(ra)
return max((retry_at - server_date).total_seconds(), 0.0)
except Exception:
return None
return None

@retry(
wait=wait_exponential_jitter(initial=0.5, max=8.0),
stop=stop_after_attempt(6),
retry=retry_if_exception_type((httpx.ReadTimeout, httpx.ConnectError)),
reraise=True,
)
@breaker
async def robust_get_json(url: str, headers: dict | None = None, timeout=5.0) -> dict:
async with httpx.AsyncClient(timeout=timeout) as c:
r = await c.get(url, headers=headers)
if (delay := _respect_retry_after(r)) is not None:
await asyncio.sleep(delay)
r = await c.get(url, headers=headers)
r.raise_for_status()
# 小型應用可加一層 in-memory LRU / TTL 快取（略）
return r.json()

⸻

你的 LINE 金融助理可以這樣用

7.1 例：抓 ETF 快照（含英文尾碼與台股代碼）

import re

def normalize_symbol(s: str) -> str:
    s = s.strip().upper()

# 台股：4~6碼 + 可選尾碼英文，例如 00937B

  if re.fullmatch(r"\d{4,6}[A-Z]?", s):
       return f"{s}.TW"
   return s

async def yahoo_quote(symbol: str) -> dict:
    sym = normalize_symbol(symbol)
    url = f"https://query1.finance.yahoo.com/v7/finance/quote?symbols={sym}"
    data = await robust_get_json(url, headers={"User-Agent":"Mozilla/5.0"})
    result = (data or {}).get("quoteResponse", {}).get("result", [])
    if not result:
        raise RuntimeError(f"No quote for {sym}")
    q = result[0]
    return {
        "symbol": sym,
        "name": q.get("longName") or q.get("shortName") or sym,
        "price": q.get("regularMarketPrice"),
        "change": q.get("regularMarketChange"),
        "change_percent": q.get("regularMarketChangePercent"),
        "currency": q.get("currency"),
        "ts": q.get("regularMarketTime"),
    }

7.2 例：黃金價格多來源備援
• 來源 A（官方網站 HTML）→ 以解析器抓「賣出價」欄
• 來源 B（金融報價）→ XAUUSD 類型匯總
• 來源 C（最後保底）→ 回傳最近快取的成功結果 + 顯示時間戳記
搭配上面的 robust_get_json，只要任何一路成功就回覆使用者；若都失敗，回「暫時取不到、已切換到最近一次成功值（時間）」。

退避／斷路器／快取／尊重 Retry-After 的組合，通常能把「偶發 429/5xx」帶來的失敗率壓到使用者幾乎無感。設計思路見上文引用。

⸻

上線前檢核清單（超實用）
1. 逾時：Client/Server 逾時全都設好（外呼 API 建議 3～8s）。
2. 重試策略：指數退避 + 抖動，最多 5～7 次。
3. 尊重 Retry-After：處理 429/503 時間或秒數。
4. 斷路器：連續錯誤快速失敗，定時半開。
5. 快取：把秒級變動資料做 TTL；了解 Cache-Control/ETag。
6. 可觀測性：
  • 例外通報（Sentry 或同類）
  • 追蹤（OpenTelemetry FastAPI）
  • 結構化日誌 + request-id。
7. 降級回應：拿不到即時值時，回前次成功值＋時間戳，並標示「非即時」。

⸻

今日結語

Day 16 把可靠性「底座」補齊：把抓資料這一層武裝起來，前台（對話、人設、Flex UI）不需要知道下游有多不穩。明天你要再接新來源、或對方短暫雪崩，系統也能穩穩撐住。這些做法不是只為了「不卡」，更是為了讓使用者信任你的服務。

參考資料（延伸閱讀）
• Exponential Backoff and Jitter（AWS 架構部落格）
• Google：建議使用指數退避重試（官方文件）
• Tenacity（Python 重試）文件/README
• Circuit Breaker 模式（Fowler）與 aiobreaker/pybreaker 專案
• HTTP Caching Semantics（RFC 9111）
• requests-cache 文件（將回應透明化快取）
• Retry-After 標頭（HTTP 規範 / OpenAI 429 建議）
• OpenTelemetry FastAPI Instrumentation 文件

⸻

需要把上面的程式片段整合進你現有的 app_fastapi.py 嗎？告訴我你目前的檔案結構，我直接給你「可貼上」的 resilience.py 與引用示例版面。