iT邦幫忙

2026 iThome 鐵人賽

DAY 28
0
Software Development

30 天打造專屬盯盤機器人:資料爬取、策略指標到 Telegram/LINE 推播系列 第 28 篇

Day 28:防禦性編程與異常處理 — 讓 Telegram 機器人 24 小時不當機!

  • 分享至 

  • xImage
  •  

今天是第二十八天!昨天我們順利解決了後來發現的第十三天程式碼中的變數型態小陷阱(將 Pandas Series 的 .str 修正為純字串操作),讓整個 Telegram 機器人能夠順暢執行。

在前面二十幾天的旅程中,我們從單純的 K 線圖繪製、技術指標計算、三大法人籌碼、月營收基本面、新聞情緒分析,一路升級到 SQLite 快取、按鈕選單、自選股管理、大盤溫度計以及雲端 Docker 部署。

但是,當機器人上線到雲端 24 小時運作時,往往會面臨各種不可控的突發狀況:

網路瞬斷或 API 逾時:Yahoo Finance 或 Telegram 伺服器偶爾會沒回應。

查無代碼或異常輸入:使用者不小心輸入了錯誤的股票代碼(例如 /stock 99999)。

資料庫連線鎖定:多個請求同時讀寫 SQLite 時發生衝突。

如果沒有做好完善的例外處理(Exception Handling),機器人只要遇到一次錯誤就會直接崩潰(Crash),甚至停止運行。今天我們的目標,就是為整個專案導入強固的防禦性編程機制!

步驟一:先釐清邏輯
打造高可用性機器人的防禦邏輯非常嚴謹:

  1. 核心函式加上 try-except 包覆:在 fetcher.py 抓取資料、chart.py 繪圖以及 bot.py 的指令回覆中,全面加入防禦性攔截。
  2. 友善的錯誤提示訊息:當發生例外或找不到資料時,不噴出看不懂的 Traceback,而是回傳像 [ERROR] 暫時無法取得該標的數據,請檢查代碼或稍後再試! 的友善提示。
  3. 全域錯誤處理器(Error Handler):在 telegram.ext 中註冊 add_error_handler,捕捉所有未預期的非同步例外,防止應用程式閃退。

步驟二:完整實作程式碼
我們來為 fetcher.py 與 bot.py 加上強固的防禦防線!

  1. fetcher.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
  1. bot.py(註冊全域 Error Handler)
    在 bot.py 中新增一個全域錯誤處理函式,並在 setup_handlers 中註冊:
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 小時不間斷穩定運航!
今天我們完成了防禦性編程與異常處理升級!


上一篇
Day 27:實戰演練與自動化排程 — 將台股機器人與 Google Tasks 任務清單完美串接!
下一篇
Day 29:日誌管理與健康監控 — 打造企業級的日誌追蹤系統!
系列文
30 天打造專屬盯盤機器人:資料爬取、策略指標到 Telegram/LINE 推播 共 30 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言