iT邦幫忙

2026 iThome 鐵人賽

DAY 14
0
Software Development

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

Day 14:邁向專業工程師 — 專案模組化重構與架構優化!

  • 分享至 

  • xImage
  •  

今天是第十四天!前幾天我們做了好多功能,到了昨天(第十三天)我們成功讓機器人聽懂 Telegram 的 /stock 指令,實現了隨叫隨到的「互動點播」!
還記得我們在第九天畫 K 線圖、第十一天算 RSI/MACD 技術指標、第十二天加入成交量能判斷,以及第十三天建立 Telegram 指令監聽。但你有沒有發現,現在的 main.py 已經塞滿了幾百行程式碼?未來如果要再加功能,程式碼會變得又臭又難維護。
今天我們的目標,就是要進行一場「程式碼模組化重構」!把原本擠在一起的程式碼拆解成獨立的模組,做出一套結構清晰、好維護的專業專案架構!
步驟一:先釐清邏輯
模組化的核心觀念就是「讓專門的人做專門的事」。我們把原本龐大的 main.py 拆成 5 個檔案,流程劃分非常直覺:

  1. config.py(設定檔):專門存放 Telegram Token 與預設的股票清單。
  2. fetcher.py(資料模組):專門向 yfinance 抓取數據,並計算 MA、RSI 與 MACD 指標。
  3. plotter.py(繪圖模組):專門使用 mplfinance 畫出三層 K 線技術圖表,並存成圖片。
  4. bot.py(機器人模組):專門負責組裝中文診斷文字,以及處理 Telegram 的 /start 與 /stock 指令。
  5. main.py(主程式入口):只留最精簡的啟動邏輯,負責把機器人跑起來!

接下來,知道邏輯後,我們就可以來建立檔案與撰寫程式碼了!
步驟二:完整實作程式碼
請在你的 stock-bot 資料夾下,依序建立以下檔案。還是要先確認終端機前面有沒有 (venv) 喔!

  1. config.py

# 1. Telegram 設定
TOKEN = "你的_BOT_TOKEN"

# 2. 預設自選股清單
DEFAULT_STOCKS = [
    {"symbol": "2330.TW", "name": "台積電"},
    {"symbol": "2317.TW", "name": "鴻海"},
    {"symbol": "2454.TW", "name": "聯發科"},
    {"symbol": "0050.TW", "name": "元大台灣50"}
]

2.fetcher.py

# fetcher.py
import pandas as pd
import pandas_ta as ta
import yfinance as yf

def get_stock_data(symbol):
    formatted_symbol = symbol.strip().upper()
    if not formatted_symbol.endswith('.TW') and not formatted_symbol.endswith('.TWO'):
        formatted_symbol += '.TW'

    print(f"[INFO] 正在抓取數據: {formatted_symbol}...")

    df = yf.download(formatted_symbol, period="6mo", progress=False)

    if df.empty and formatted_symbol.endswith('.TW'):
        alt_symbol = formatted_symbol.replace('.TW', '.TWO')
        df = yf.download(alt_symbol, period="6mo", progress=False)
        if not df.empty:
            formatted_symbol = alt_symbol

    if df.empty:
        return 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()

    return formatted_symbol, df

3.plotter.py

# plotter.py
import os
import mplfinance as mpf

def create_stock_chart(symbol, df):
    df_plot = df.tail(60).copy()
    img_filename = f"{symbol.replace('.', '_')}_chart.png"

    ap = [
        mpf.make_addplot(df_plot['MA5'], color='orange', width=1.2),
        mpf.make_addplot(df_plot['MA20'], color='blue', width=1.2),
        mpf.make_addplot(df_plot['RSI'], panel=2, color='purple', ylabel='RSI(14)'),
        mpf.make_addplot(df_plot['MACD_hist'], type='bar', panel=3, color='dimgray', ylabel='MACD Hist')
    ]

    mpf.plot(
        df_plot,
        type='candle',
        volume=True,
        addplot=ap,
        title=f"{symbol} Technical Analysis",
        savefig=img_filename,
        panel_ratios=(4, 1, 1, 1)
    )

    return img_filename

4.bot.py

# bot.py
import os
from telegram import Update
from telegram.ext import CommandHandler, ContextTypes
from fetcher import get_stock_data
from plotter import create_stock_chart

def generate_report_text(symbol, df):
    latest = df.iloc[-1]
    prev = df.iloc[-2]

    rsi_val = latest['RSI']
    macd_hist_latest = latest['MACD_hist']
    macd_hist_prev = prev['MACD_hist']
    latest_vol = latest['Volume']
    vol_ma5_val = latest['Vol_MA5']

    caption = f"【{symbol} 即時點播分析報告】\n"
    caption += f"日期:{df.index[-1].strftime('%Y-%m-%d')}\n"
    caption += f"收盤價:{latest['Close']:.2f}\n"
    caption += f"5MA: {latest['MA5']:.2f} | 20MA: {latest['MA20']:.2f}\n"
    caption += f"RSI(14):{rsi_val:.1f}\n"
    caption += f"-----------------------------------\n"

    signals = []
    if prev['MA5'] <= prev['MA20'] and latest['MA5'] > latest['MA20']:
        signals.append("[MA] 5MA 向上突破 20MA(黃金交叉)")
    elif prev['MA5'] >= prev['MA20'] and latest['MA5'] < latest['MA20']:
        signals.append("[MA] 5MA 向下跌破 20MA(死亡交叉)")

    if rsi_val >= 70:
        signals.append("[RSI] 過熱區 (>70),留意回檔風險")
    elif rsi_val <= 30:
        signals.append("[RSI] 超賣區 (<30),關注反彈買點")

    if macd_hist_prev < 0 and macd_hist_latest > 0:
        signals.append("[MACD] 柱狀圖由負轉正,多頭動能轉強")
    elif macd_hist_prev > 0 and macd_hist_latest < 0:
        signals.append("[MACD] 柱狀圖由正轉負,空頭動能增強")

    if latest_vol > vol_ma5_val * 1.5:
        signals.append("[量能] 今日成交量爆發(高於 5 日均量 1.5 倍)!")
    elif latest_vol < vol_ma5_val * 0.6:
        signals.append("[量能] 成交量明顯萎縮,市場觀望氣氛濃")

    if not signals:
        caption += "評語:指標處於區間震盪,建議持續觀察。"
    else:
        caption += "關鍵訊號彙整:\n" + "\n".join(signals)

    return caption

async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    welcome_text = (
        "歡迎使用盤後看盤小幫手!\n\n"
        "你可以使用以下指令點播個股分析:\n"
        "格式:/stock <股票代號>\n"
        "範例:/stock 2330 或 /stock 2317"
    )
    await update.message.reply_text(welcome_text)

async def stock_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if not context.args:
        await update.message.reply_text("[提示] 請輸入股票代號!範例:/stock 2330")
        return

    raw_symbol = context.args[0]
    await update.message.reply_text(f"[系統] 收到!正在為您繪製 {raw_symbol} 的分析圖表,請稍候...")

    symbol, df = get_stock_data(raw_symbol)

    if df is None:
        await update.message.reply_text(f"[ERROR] 找不到股票代號 {raw_symbol} 的歷史資料,請確認代號是否正確。")
        return

    img_path = create_stock_chart(symbol, df)
    caption_text = generate_report_text(symbol, df)

    with open(img_path, 'rb') as photo:
        await update.message.reply_photo(photo=photo, caption=caption_text)

    if os.path.exists(img_path):
        os.remove(img_path)

def setup_handlers(app):
    app.add_handler(CommandHandler("start", start_command))
    app.add_handler(CommandHandler("stock", stock_command))
  1. main.py
# main.py
from telegram.ext import ApplicationBuilder
from config import TOKEN
from bot import setup_handlers

if __name__ == "__main__":
    print("[START] 機器人模組化架構啟動中,等待使用者指令...")
    
    app = ApplicationBuilder().token(TOKEN).build()
    setup_handlers(app)
    app.run_polling()

解釋一下程式碼吧!
跨檔案模組引用 (Import):
在 bot.py 的開頭,我們使用 from fetcher import get_stock_data 與 from plotter import create_stock_chart,直接把別的檔案寫好的函式拿過來用!這樣每個檔案只需要專注處理自己的工作。
超乾淨的 main.py:
重構後,主程式入口只剩下不到 15 行,它只負責讀取 config.py 的 Token,並呼叫 bot.py 註冊指令後啟動監聽,整體視覺乾淨又好看。
步驟三:執行程式
在終端機內輸入:

python main.py

打開 Telegram,傳送 /stock 2330 給機器人測試看看,收到回傳圖文就代表模組化重構大成功囉!
最後會長下面這樣:

小結一下:
記得把 Token 填在 config.py 裡面,主程式才會順利運作!
今天我們成功把原本擠在單一檔案的龐大程式碼,拆解成符合低耦合標準的 5 個模組。
經過模組化調整後,架構變得非常乾淨!明天(第十五天)是鐵人賽過半的里程碑,我們要在這套乾淨的架構上,加上「APScheduler 自動排程器」,讓機器人可以在每天盤後自動觸發推播!


上一篇
Day 13:讓機器人聽懂人話 — 打造 Telegram 互動式點播功能!
下一篇
Day 15:邁向全自動化 — 導入 APScheduler 打造盤後定時自動推播!
系列文
30 天打造專屬盯盤機器人:資料爬取、策略指標到 Telegram/LINE 推播 共 17 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言