今天是第二十三天！昨天（第二十二天）我們幫 Telegram 機器人打造了「動態自選股管理系統（/add, /del, /watchlist）」，讓使用者可以在手機上隨時調整觀察名單。
但在操作個股時，有一句股市名言：「順大勢者昌，逆大勢者亡。」大盤指數代表著整體市場的資金動向與多空氣氛。如果大盤開高走低重挫百點，大部分個股都很難獨善其身；反之若大盤放量大漲，個股開高的勝率就會大幅提升！
今天我們的目標，就是在 Telegram 機器人中導入「台股大盤觀測模組」！不僅要抓取「加權指數（^TWII）」與「櫃買指數（^TWOII）」，還要計算大盤的 5MA/20MA/60MA（生命線/季線）狀態，並產出「全市場多空溫度計」，讓你在每天盤後第一時間先看懂大局，再看個股！
步驟一：先釐清邏輯
大盤診斷與多空溫度計的運作邏輯如下：
抓取大盤數據（market.py）：
多空溫度與位階分析：
判斷漲跌幅與漲跌點數。
產出多空評語：
接下來，知道邏輯後，我們就可以來撰寫程式碼了！
步驟二：完整實作程式碼
我們新建一個 market.py 專門處理大盤資料，並更新 bot.py 註冊指令！
import yfinance as yf
import pandas as pd
import pandas_ta as ta
def get_market_overview():
print("[INFO] 正在抓取加權指數與櫃買指數數據...")
# yfinance 大盤代號：^TWII (加權指數), ^TWOII (櫃買指數)
twii = yf.Ticker("^TWII").history(period="3mo")
twoii = yf.Ticker("^TWOII").history(period="3mo")
if twii.empty:
return None
# 計算加權指數均線
twii['MA5'] = twii['Close'].rolling(window=5).mean()
twii['MA20'] = twii['Close'].rolling(window=20).mean()
twii['MA60'] = twii['Close'].rolling(window=60).mean()
latest = twii.iloc[-1]
prev = twii.iloc[-2]
close = latest['Close']
change = close - prev['Close']
pct_change = (change / prev['Close']) * 100
ma5 = latest['MA5']
ma20 = latest['MA20']
ma60 = latest['MA60']
# 多空溫度計邏輯
score = 0
signals = []
if close > ma5:
score += 1
signals.append("股價站上 5 日線（短線強勢）")
else:
signals.append("股價跌破 5 日线（短線震盪）")
if close > ma20:
score += 1
signals.append("股價站上 20 日月線（月線支撐有效）")
else:
signals.append("股價跌破 20 日月線（短中期轉弱）")
if close > ma60:
score += 1
signals.append("股價站上 60 日季線（中長線多頭趨勢）")
else:
signals.append("股價跌破 60 日季線（中長線趨勢承壓）")
if ma5 > ma20 > ma60:
score += 1
signals.append("均線呈現多頭排列（強勢多頭）")
# 溫度計評級
if score >= 3:
status = "多頭佔優 (熱絡)"
elif score == 2:
status = "多空拉鋸 (觀望)"
else:
status = "空頭佔優 (保守)"
# 櫃買指數數據
twoii_info = ""
if not twoii.empty:
t_latest = twoii.iloc[-1]
t_prev = twoii.iloc[-2]
t_close = t_latest['Close']
t_change = t_close - t_prev['Close']
t_pct = (t_change / t_prev['Close']) * 100
twoii_info = f"櫃買指數 (OTC)：{t_close:.2f} ({t_change:+.2f} / {t_pct:+.2f}%)\n"
report = (
f"【台股大盤與全市場溫度計】\n"
f"日期：{twii.index[-1].strftime('%Y-%m-%d')}\n"
f"-----------------------------------\n"
f"加權指數 (TAIEX)：{close:.2f} ({change:+.2f} / {pct_change:+.2f}%)\n"
f"{twoii_info}"
f"-----------------------------------\n"
f"5MA: {ma5:.2f} | 20MA: {ma20:.2f} | 60MA: {ma60:.2f}\n"
f"市場多空溫度：{status} (得分: {score}/4)\n"
f"-----------------------------------\n"
f"大盤指標診斷：\n" + "\n".join([f"- {s}" for s in signals])
)
return report
import os
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import ContextTypes, CommandHandler, CallbackQueryHandler
from fetcher import get_stock_data
from chart import create_stock_chart
from database import add_to_watchlist, remove_from_watchlist, get_watchlist
from market import get_market_overview
def get_stock_keyboard(symbol):
keyboard = [
[
InlineKeyboardButton("技術線圖", callback_data=f"chart_{symbol}"),
InlineKeyboardButton("籌碼動向", callback_data=f"chip_{symbol}")
],
[
InlineKeyboardButton("營收數據", callback_data=f"rev_{symbol}"),
InlineKeyboardButton("即時新聞", callback_data=f"news_{symbol}")
]
]
return InlineKeyboardMarkup(keyboard)
# 新增 /market 指令
async def market_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text("[INFO] 正在分析台股大盤最新數據，請稍後...")
report = get_market_overview()
if report:
await update.message.reply_text(report)
else:
await update.message.reply_text("[ERROR] 暫時無法獲取大盤指數數據。")
# 新增自選股指令 (/add 2330)
async def add_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
if not context.args:
await update.message.reply_text("請輸入股票代碼，例如：/add 2330")
return
raw_symbol = context.args[0].upper()
chat_id = update.effective_chat.id
if not raw_symbol.endswith('.TW') and not raw_symbol.endswith('.TWO'):
raw_symbol += '.TW'
success = add_to_watchlist(chat_id, raw_symbol)
if success:
await update.message.reply_text(f"[SUCCESS] 已成功將 {raw_symbol} 加入自選股追蹤清單！")
else:
await update.message.reply_text(f"[NOTICE] {raw_symbol} 已經在你的自選股清單中了。")
# 刪除自選股指令 (/del 2330)
async def del_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
if not context.args:
await update.message.reply_text("請輸入股票代碼，例如：/del 2330")
return
raw_symbol = context.args[0].upper()
chat_id = update.effective_chat.id
if not raw_symbol.endswith('.TW') and not raw_symbol.endswith('.TWO'):
raw_symbol += '.TW'
success = remove_from_watchlist(chat_id, raw_symbol)
if success:
await update.message.reply_text(f"[SUCCESS] 已將 {raw_symbol} 從自選股清單中移除。")
else:
await update.message.reply_text(f"[NOTICE] 在自選股清單中找不到 {raw_symbol}。")
# 查看自選股清單 (/watchlist)
async def watchlist_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
chat_id = update.effective_chat.id
stocks = get_watchlist(chat_id)
if not stocks:
await update.message.reply_text("目前自選股清單是空的！請使用 /add <代碼> 新增標的。")
return
text = "【我的口袋自選股清單】\n-----------------------------------\n"
keyboard = []
for idx, sym in enumerate(stocks, 1):
text += f"{idx}. {sym}\n"
keyboard.append([InlineKeyboardButton(f"看盤 {sym}", callback_data=f"view_{sym}")])
text += "-----------------------------------\n提示：可使用 /del <代碼> 移除標的。"
await update.message.reply_text(
text=text,
reply_markup=InlineKeyboardMarkup(keyboard)
)
# /stock 指令主處理邏輯
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"[INFO] 正在獲取 {raw_symbol} 完整資料，請稍後...")
symbol, df, chip_data, revenue_data, news_data = get_stock_data(raw_symbol)
if df is None:
await update.message.reply_text(f"[ERROR] 找不到股票代碼：{raw_symbol}")
return
latest = df.iloc[-1]
summary_text = (
f"【{symbol} 個股儀表板】\n"
f"日期：{df.index[-1].strftime('%Y-%m-%d')}\n"
f"最新收盤價：{latest['Close']:.2f} 元\n\n"
f"請選擇下方按鈕切換詳細看盤維度："
)
await update.message.reply_text(
text=summary_text,
reply_markup=get_stock_keyboard(symbol)
)
# 按鈕點擊事件處理器
async def button_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
query = update.callback_query
await query.answer()
data = query.data
action, symbol = data.split("_", 1)
if action == "view":
_, df, chip_data, revenue_data, news_data = get_stock_data(symbol)
latest = df.iloc[-1]
summary_text = (
f"【{symbol} 個股儀表板】\n"
f"日期：{df.index[-1].strftime('%Y-%m-%d')}\n"
f"最新收盤價：{latest['Close']:.2f} 元\n\n"
f"請選擇下方按鈕切換詳細看盤維度："
)
await query.message.reply_text(
text=summary_text,
reply_markup=get_stock_keyboard(symbol)
)
return
_, df, chip_data, revenue_data, news_data = get_stock_data(symbol)
if action == "chart":
img_path = create_stock_chart(symbol, df)
with open(img_path, 'rb') as photo:
await query.message.reply_photo(
photo=photo,
caption=f"【{symbol} K線與技術指標線圖】",
reply_markup=get_stock_keyboard(symbol)
)
if os.path.exists(img_path):
os.remove(img_path)
elif action == "chip":
text = (
f"【{symbol} 三大法人籌碼動向】(單位: 張)\n"
f"-----------------------------------\n"
f"外資：{chip_data['foreign']:+d}\n"
f"投信：{chip_data['investment']:+d}\n"
f"自營商：{chip_data['dealer']:+d}\n"
f"-----------------------------------\n"
f"法人合計：{chip_data['total']:+d}\n"
)
await query.message.reply_text(text=text, reply_markup=get_stock_keyboard(symbol))
elif action == "rev":
text = (
f"【{symbol} 最新月營收報告】({revenue_data['month']})\n"
f"-----------------------------------\n"
f"單月營收：{revenue_data['revenue'] / 100000:.2f} 億元\n"
f"MoM (月增率)：{revenue_data['mom']:+.1f}%\n"
f"YoY (年增率)：{revenue_data['yoy']:+.1f}%\n"
)
await query.message.reply_text(text=text, reply_markup=get_stock_keyboard(symbol))
elif action == "news":
text = f"【{symbol} 即時消息面分析】({news_data['sentiment']})\n-----------------------------------\n"
for idx, item in enumerate(news_data["list"], 1):
text += f"{idx}. {item['title']}\n"
await query.message.reply_text(text=text, reply_markup=get_stock_keyboard(symbol))
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))
解釋一下程式碼吧！
^TWII 與 ^TWOII：
這是 Yahoo Finance 中代表台灣加權指數與櫃買指數的標準代碼。
多空溫度計（Market Thermometer）：
我們透過計算指數相對於 5MA、20MA、60MA 的位置，量化得出一個 0~4 分的客觀溫度分數，協助交易者第一時間掌握今天是「多頭做多」還是「空頭避險」的主戰場。
步驟三：執行與測試
在終端機內輸入:
python main.py
打開 Telegram，傳送 /market 點播大盤！收到的報告會長這樣：
【台股大盤與全市場溫度計】
日期：2026-10-07
加權指數 (TAIEX)：22850.30 (+150.20 / +0.66%)
櫃買指數 (OTC)：268.45 (+1.80 / +0.67%)
5MA: 22700.10 | 20MA: 22450.00 | 60MA: 21900.50
市場多空溫度：多頭佔優 (熱絡) (得分: 4/4)
大盤指標診斷：
小結一下：
大盤是個股的母體，看個股前先查看 /market 大盤溫度計，能有效降低逆勢操作的風險！
今天我們成功導入了台股加權/櫃買大盤模組與市場多空溫度計（/market）！
iThome鐵人賽
看影片追技術