今天是第二十八天!昨天我們順利解決了後來發現的第十三天程式碼中的變數型態小陷阱(將 Pandas Series 的 .str 修正為純字串操作),讓整個 Telegram 機器人能夠順暢執行。
在前面二十幾天的旅程中,我們從單純的 K 線圖繪製、技術指標計算、三大法人籌碼、月營收基本面、新聞情緒分析,一路升級到 SQLite 快取、按鈕選單、自選股管理、大盤溫度計以及雲端 Docker 部署。
但是,當機器人上線到雲端 24 小時運作時,往往會面臨各種不可控的突發狀況:
網路瞬斷或 API 逾時:Yahoo Finance 或 Telegram 伺服器偶爾會沒回應。
查無代碼或異常輸入:使用者不小心輸入了錯誤的股票代碼(例如 /stock 99999)。
資料庫連線鎖定:多個請求同時讀寫 SQLite 時發生衝突。
如果沒有做好完善的例外處理(Exception Handling),機器人只要遇到一次錯誤就會直接崩潰(Crash),甚至停止運行。今天我們的目標,就是為整個專案導入強固的防禦性編程機制!
步驟一:先釐清邏輯
打造高可用性機器人的防禦邏輯非常嚴謹:
步驟二:完整實作程式碼
我們來為 fetcher.py 與 bot.py 加上強固的防禦防線!
import pandas as pd
import pandas_ta as ta
import yfinance as yf
from database import get_cached_data, save_cache_data
from news import fetch_stock_news
def get_stock_data(symbol):
try:
formatted_symbol = str(symbol).strip().upper()
if not formatted_symbol.endswith('.TW') and not formatted_symbol.endswith('.TWO'):
formatted_symbol += '.TW'
print(f"[INFO] 正在安全取得數據: {formatted_symbol}...")
# 1. 嘗試抓取股價歷史數據
ticker = yf.Ticker(formatted_symbol)
df = ticker.history(period="6mo")
if df.empty and formatted_symbol.endswith('.TW'):
alt_symbol = formatted_symbol.replace('.TW', '.TWO')
ticker = yf.Ticker(alt_symbol)
df = ticker.history(period="6mo")
if not df.empty:
formatted_symbol = alt_symbol
if df.empty:
print(f"[WARNING] 找不到代碼 {formatted_symbol} 的有效歷史數據。")
return None, None, None, None, None
if isinstance(df.columns, pd.MultiIndex):
df.columns = df.columns.get_level_values(0)
# 計算技術指標
df['MA5'] = df['Close'].rolling(window=5).mean()
df['MA20'] = df['Close'].rolling(window=20).mean()
df['RSI'] = df.ta.rsi(length=14)
macd_df = df.ta.macd(fast=12, slow=26, signal=9)
df['MACD'] = macd_df['MACD_12_26_9']
df['MACD_signal'] = macd_df['MACDs_12_26_9']
df['MACD_hist'] = macd_df['MACDh_12_26_9']
df['Vol_MA5'] = df['Volume'].rolling(window=5).mean()
latest_date = df.index[-1].strftime('%Y-%m-%d')
# 2. 檢查 SQLite 快取
cached_chip, cached_rev = get_cached_data(formatted_symbol, latest_date)
if not (cached_chip and cached_rev):
chip_data = {"foreign": 1250, "investment": 800, "dealer": -150}
chip_data["total"] = chip_data["foreign"] + chip_data["investment"] + chip_data["dealer"]
revenue_data = {"month": "2026-09", "revenue": 75000000, "mom": 8.5, "yoy": 32.4}
save_cache_data(formatted_symbol, latest_date, chip_data, revenue_data)
else:
chip_data, revenue_data = cached_chip, cached_rev
# 3. 安全抓取新聞
news_list, sentiment, bull_cnt, bear_cnt = fetch_stock_news(formatted_symbol)
news_data = {
"list": news_list,
"sentiment": sentiment,
"bullish": bull_cnt,
"bearish": bear_cnt
}
return formatted_symbol, df, chip_data, revenue_data, news_data
except Exception as e:
print(f"[ERROR] get_stock_data 發生未預期錯誤: {e}")
return None, None, None, None, None
async def error_handler(update: object, context: ContextTypes.DEFAULT_TYPE):
print(f"[ERROR] 捕捉到未處理的例外狀況: {context.error}")
if isinstance(update, Update) and update.effective_message:
await update.effective_message.reply_text(
"[NOTICE] 系統剛剛處理您的請求時發生了一點小狀況,可能為外部數據源暫時無回應,請稍後再試一次!"
)
def setup_handlers(app):
app.add_handler(CommandHandler("start", lambda u, c: u.message.reply_text("歡迎使用台股看盤小幫手!")))
app.add_handler(CommandHandler("stock", stock_command))
app.add_handler(CommandHandler("add", add_command))
app.add_handler(CommandHandler("del", del_command))
app.add_handler(CommandHandler("watchlist", watchlist_command))
app.add_handler(CommandHandler("market", market_command))
app.add_handler(CallbackQueryHandler(button_callback))
# 註冊全域錯誤處理器
app.add_error_handler(error_handler)
*解釋一下程式碼吧!
防禦性包覆(Defensive Try-Except):
在對外的 API 呼叫與資料處理核心加入 try-except,即使外部數據源(Yahoo Finance)發生瞬斷或回傳空資料,程式也不會直接噴錯終止,而是安全地回傳 None。
app.add_error_handler(error_handler):
這是 Telegram Bot 專案能夠穩定運行 24 小時的靈魂!它可以攔截所有漏網之魚的例外,並在對話框中給使用者溫馨的提示,而不是讓機器人默默掛掉。
步驟三:測試與驗證
在終端機內打:
python main.py
小結一下:
透過嚴謹的異常處理與全域錯誤攔截,我們大幅提升了台股機器人的強韌度,確保它在雲端能 24 小時不間斷穩定運航!
今天我們完成了防禦性編程與異常處理升級!