今天是第十三天！過去這幾天，我們的 Telegram 看盤機器人已經學會了多股巡檢、技術面雙指標（RSI + MACD）以及籌碼量能診斷，甚至還能透過排程器在每天盤後自動推播，真的越來越像個專業分析師！
但你有沒有遇過這種情境？
「突然想看某檔股票現在的線圖，但排程時間還沒到……」或者「想臨時查一下不在預設清單裡的股票……」
沒錯！單向的「自動推播」雖然方便，但「隨叫隨到的互動點播」才是機器人的靈魂啊！
今天我們的目標，就是要幫 Telegram 機器人掛上 Bot Commands（互動指令）！讓你在手機對話框裡輸入 /stock 2330 或 /stock 2317，機器人就會立刻收到指令，即時畫圖、分析並回傳熱騰騰的報告給你喔~
接下來我們，先釐清邏輯，
核心原理：什麼是 Long Polling？
要讓機器人「隨時聽候差遣」，我們的 Python 程式就必須持續跟 Telegram 的伺服器保持連線，詢問：「請問現在有使用者傳訊息給我嗎？」
這種機制就叫做 Polling（輪詢）。今天我們會利用 Python 最強大且好用的 Telegram 套件 python-telegram-bot，來幫我們更輕鬆的搞定監聽與回應！
步驟一：安裝 python-telegram-bot 套件
打開你的終端機（記得確認前面有 (venv) 標籤喔！），輸入以下命令：
pip install python-telegram-bot
步驟二：撰寫 main.py 互動回應腳本
這次我們要重構程式碼，把分析邏輯模組化，並加入 Telegram 的指令監聽器！
打開 main.py，把這份支援 /stock <代號> 點播的最新程式碼寫上去：
import os
import io
import pandas as pd
import pandas_ta as ta
import yfinance as yf
import mplfinance as mpf
import time
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes
# 1. 通訊金鑰設定
TOKEN = "你的_BOT_TOKEN"
# 分析單檔股票並產生圖表與報告文字
def analyze_stock(symbol):
# 處理台股代號格式 (如果使用者沒輸入 .TW，自動補上)
formatted_symbol = symbol.str.strip().upper()
if not formatted_symbol.endswith('.TW') and not formatted_symbol.endswith('.TWO'):
formatted_symbol += '.TW'
print(f"[INFO] 收到點播，正在分析: {formatted_symbol}...")
# 抓取 6 個月歷史資料
df = yf.download(formatted_symbol, period="6mo", progress=False)
if df.empty:
# 如果上市找不到，嘗試上櫃 (.TWO)
if 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, f"[ERROR] 找不到股票代號 {symbol} 的歷史資料，請確認代號是否正確。"
# 清理 yfinance 多層索引
if isinstance(df.columns, pd.MultiIndex):
df.columns = df.columns.get_level_values(0)
# 1. 計算技術指標
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']
# 2. 量能分析
df['Vol_MA5'] = df['Volume'].rolling(window=5).mean()
latest_vol = df['Volume'].iloc[-1]
vol_ma5_val = df['Vol_MA5'].iloc[-1]
# 取最近 60 天資料繪圖
df_plot = df.tail(60).copy()
img_filename = f"{formatted_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"{formatted_symbol} Technical Analysis",
savefig=img_filename,
panel_ratios=(4, 1, 1, 1)
)
# 3. 撰寫診斷文字
latest = df.iloc[-1]
prev = df.iloc[-2]
rsi_val = latest['RSI']
macd_hist_latest = latest['MACD_hist']
macd_hist_prev = prev['MACD_hist']
caption = f"【{formatted_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 img_filename, caption
# /start 指令回應
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)
# /stock 指令回應
async def stock_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
# 檢查使用者是否有輸入股票代號
if not context.args:
await update.message.reply_text("[提示] 請輸入股票代號！範例：/stock 2330")
return
stock_id = context.args[0]
await update.message.reply_text(f"[系統] 收到！正在為您繪製 {stock_id} 的分析圖表，請稍候...")
# 進行股票分析
img_path, caption_text = analyze_stock(stock_id)
if img_path is None:
await update.message.reply_text(caption_text)
else:
# 發送圖文訊息回給使用者
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)
# 主流程：啟動 Bot 監聽
if __name__ == "__main__":
print("[START] 機器人啟動中，等待使用者指令...")
app = ApplicationBuilder().token(TOKEN).build()
# 註冊指令處理器
app.add_handler(CommandHandler("start", start_command))
app.add_handler(CommandHandler("stock", stock_command))
# 開始持續監聽 (Polling)
app.run_polling()
步驟三：執行與互動測試
在終端機輸入我們最熟悉的黃金指令（Python 在前，檔名在後！）：
python main.py
接下來，終端機會顯示:
[START] 機器人啟動中，等待使用者指令...
打開 Telegram 找到你的機器人：
最後畫面長這樣!
小結一下：
今天我們成功把看盤機器人升級成了「主動推播 + 被動點播」！
現在我們的機器人具備了：
即時互動功能：隨時在 Telegram 輸入 /stock 代號 就能點播任何台股個股。
自動補全邏輯：輸入 2330 會自動幫你補上 .TW，上櫃股票還會智慧試錯 .TWO。
異步訊息處理：使用 async/await 架構，回應流暢不卡頓！
iThome鐵人賽
看影片追技術