iT邦幫忙

2026 iThome 鐵人賽

DAY 29
0
Software Development

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

Day 29:日誌管理與健康監控 — 打造企業級的日誌追蹤系統!

  • 分享至 

  • xImage
  •  

今天是第二十九天!昨天我們幫機器人加入了強固的防禦性編程與全域例外處理,確保 Telegram 機器人在面對網路瞬斷或異常輸入時,不會發生當機閃退的狀況。

當機器人正式上線到雲端(例如 Render 或 Linux VPS)24 小時運作時,另一個非常重要的課題就是「日誌管理(Logging Management)」。過去我們習慣用 print() 將訊息印在終端機上,但當程式跑了好幾天、甚至累積了數萬行訊息時:

  1. print() 無法自動記錄發生時間與錯誤等級(INFO, WARNING, ERROR)。
  2. 當雲端容器重新啟動或崩潰時,終端機上的紀錄就會全部消失,事後根本無從查證是哪一個環節出了問題。

今天我們的目標,就是要用 Python 內建的 logging 模組 取代 print(),打造具備時間戳記、日誌等級與檔案輪替(Rotating File Handler)的企業級日誌管理系統!

步驟一:先釐清邏輯
專業日誌系統的運作邏輯非常清晰:

  1. 分級記錄(Log Levels):
  • INFO:記錄正常運作事件(如「成功抓取 2330.TW 數據」)。
  • WARNING:記錄非致命但需注意的狀況(如「快取未命中,改走外部 API」)。
  • ERROR:記錄重大錯誤或例外(如「API 連線逾時」)。
  1. 輸出至檔案與終端機雙軌並行:同時將日誌印在終端機(方便即時觀看)以及寫入本地的 bot.log 檔案中(方便長期追蹤)。
  2. 日誌輪替(Rotating File) :限制日誌檔案的大小,避免日誌檔無限膨脹吃光伺服器硬碟空間。

步驟二:完整實作程式碼
我們建立一個統一的日誌管理設定,並將全專案的 print() 全面替換為 logging!

  1. logger.py(建立全域日誌設定模組)
import logging
import os
from logging.handlers import RotatingFileHandler

def setup_logger():
    # 建立日誌格式
    log_format = logging.Formatter(
        '[%(asctime)s] [%(levelname)s] [%(filename)s:%(lineno)d]: %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )

    # 取得根日誌器
    logger = logging.getLogger("StockBot")
    logger.setLevel(logging.INFO)

    # 避免重複註冊 Handler
    if not logger.handlers:
        # 1. 終端機輸出 Handler
        console_handler = logging.StreamHandler()
        console_handler.setFormatter(log_format)
        logger.addHandler(console_handler)

        # 2. 檔案輸出 Handler (限制單檔最大 5MB,最多保留 3 個備份檔)
        file_handler = RotatingFileHandler(
            "bot.log", maxBytes=5*1024*1024, backupCount=3, encoding="utf-8"
        )
        file_handler.setFormatter(log_format)
        logger.addHandler(file_handler)

    return logger

# 全域共用 logger 實例
logger = setup_logger()
  1. 在各模組中導入 logger 取代 print
    我們以 fetcher.py 為例,將原本的 print() 改為 logger.info() 或 logger.error():
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
from logger import logger

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'

        logger.info(f"正在安全取得數據: {formatted_symbol}...")

        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:
            logger.warning(f"找不到代碼 {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')

        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

        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:
        logger.error(f"get_stock_data 發生未預期錯誤: {e}", exc_info=True)
        return None, None, None, None, None

(同理,你可以將專案中其他檔案如 main.py、bot.py、market.py 裡的 print 替換成 logger.info() 或 logger.error()。)

解釋一下程式碼吧!

  • RotatingFileHandler(日誌自動輪替):
    當 bot.log 檔案大小達到 5MB 時,系統會自動將舊檔案備份並建立新檔案,最多保留 3 個備份,完美防止日誌檔塞爆伺服器硬碟。

  • exc_info=True:
    在 logger.error() 中加上這個參數,日誌系統會自動把完整的錯誤追溯堆疊記錄進檔案中,這對後續除錯非常有幫助!
    步驟三:執行與驗證
    在終端機內輸入:

python main.py

你會發現終端機輸出的訊息變得非常整齊且帶有精確的時間戳記:

同時,專案根目錄下會自動生成一個 bot.log 檔案,用來完整記錄所有的運行足跡!
小結一下:
透過 logging 模組,我們告別了陽春的 print(),將機器人升級為具備結構化紀錄、等級區分與檔案備份的企業級架構!
今天我們完成了日誌管理與健康監控模組的建置!


上一篇
Day 28:防禦性編程與異常處理 — 讓 Telegram 機器人 24 小時不當機!
下一篇
Day 30:30 天鐵人賽完賽總結 — 從零打造全自動、專業、模組化的台股 Telegram 機器人!
系列文
30 天打造專屬盯盤機器人:資料爬取、策略指標到 Telegram/LINE 推播 共 30 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言