今天是第十四天!前幾天我們做了好多功能,到了昨天(第十三天)我們成功讓機器人聽懂 Telegram 的 /stock 指令,實現了隨叫隨到的「互動點播」!
還記得我們在第九天畫 K 線圖、第十一天算 RSI/MACD 技術指標、第十二天加入成交量能判斷,以及第十三天建立 Telegram 指令監聽。但你有沒有發現,現在的 main.py 已經塞滿了幾百行程式碼?未來如果要再加功能,程式碼會變得又臭又難維護。
今天我們的目標,就是要進行一場「程式碼模組化重構」!把原本擠在一起的程式碼拆解成獨立的模組,做出一套結構清晰、好維護的專業專案架構!
步驟一:先釐清邏輯
模組化的核心觀念就是「讓專門的人做專門的事」。我們把原本龐大的 main.py 拆成 5 個檔案,流程劃分非常直覺:
接下來,知道邏輯後,我們就可以來建立檔案與撰寫程式碼了!
步驟二:完整實作程式碼
請在你的 stock-bot 資料夾下,依序建立以下檔案。還是要先確認終端機前面有沒有 (venv) 喔!
# 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))
# 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 自動排程器」,讓機器人可以在每天盤後自動觸發推播!