Day 22：自選股管理系統 — 導入 Telegram 互動指令與資料庫動態追蹤！ - iT 邦幫忙::一起幫忙解決難題，拯救 IT 人的一天

2026 iThome 鐵人賽

DAY 22

Software Development

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

Day 22：自選股管理系統 — 導入 Telegram 互動指令與資料庫動態追蹤！

18th鐵人賽

digua0708

團隊AI Lab 404

2026-10-06 13:26:12

73 瀏覽

分享至

今天是第二十二天！昨天（第二十一天）我們成功幫機器人導入了 Inline Keyboard 互動式按鈕選單，讓看盤體驗變得流暢又直覺。

不過，先前我們的追蹤清單是死板板寫在 config.py 的固定的代碼。在實際操作時，我們隨時會想把潛力股加入觀察名單，或是把轉弱的標的踢出自選股。如果每次都要開啟電腦修改程式碼、重新啟動 Bot，使用體驗就大打折扣了！

今天我們的目標，就是在 Telegram Bot 中導入「動態自選股管理系統」！透過新增 /add、/del、/watchlist 三大指令，搭配 SQLite 資料庫持久化儲存，讓你直接在 Telegram 視窗中隨時新增、刪除、查詢自選股，並能一鍵觸發「自選股巡檢」報告！

步驟一：先釐清邏輯
自選股動態管理的運作邏輯非常直覺：

擴充 SQLite 資料庫（database.py）：建立 watchlist 資料表，儲存使用者的 Telegram chat_id 以及對應新增的股票代碼。
新增 Telegram 操作指令（bot.py）：
/add <股票代碼>：將股票新增至個人的自選股清單。
/del <股票代碼>：將股票從自選股清單移除。
/watchlist：列出當前追蹤的所有標的，並附上快速看盤與一鍵巡檢按鈕。
整合自動排程（scheduler）：盤後推播時，優先讀取 SQLite 資料庫中的動態自選股清單，發送專屬的每日檢測報告！

接下來，知道邏輯後，我們就可以來寫程式碼了！

步驟二：完整實作程式碼
我們要更改 database.py 與 bot.py，加入自選股增刪查改的邏輯！

database.py-擴充自選股資料表與操作函式

import sqlite3
import json
from datetime import datetime

DB_NAME = "stock_cache.db"

def init_db():
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    # 1. 股價與指標快取表
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS stock_cache (
            symbol TEXT,
            date TEXT,
            chip_data TEXT,
            revenue_data TEXT,
            updated_at TIMESTAMP,
            PRIMARY KEY (symbol, date)
        )
    ''')
    # 2. 動態自選股資料表
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS watchlist (
            chat_id TEXT,
            symbol TEXT,
            created_at TIMESTAMP,
            PRIMARY KEY (chat_id, symbol)
        )
    ''')
    conn.commit()
    conn.close()

def get_cached_data(symbol, date_str):
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    cursor.execute(
        "SELECT chip_data, revenue_data FROM stock_cache WHERE symbol = ? AND date = ?",
        (symbol, date_str)
    )
    row = cursor.fetchone()
    conn.close()

    if row:
        chip_data = json.loads(row[0]) if row[0] else None
        revenue_data = json.loads(row[1]) if row[1] else None
        return chip_data, revenue_data
    return None

def save_cache_data(symbol, date_str, chip_data, revenue_data):
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    cursor.execute('''
        INSERT OR REPLACE INTO stock_cache (symbol, date, chip_data, revenue_data, updated_at)
        VALUES (?, ?, ?, ?, ?)
    ''', (
        symbol,
        date_str,
        json.dumps(chip_data),
        json.dumps(revenue_data),
        datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    ))
    conn.commit()
    conn.close()

# 自選股資料庫操作函式
def add_to_watchlist(chat_id, symbol):
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    try:
        cursor.execute(
            "INSERT INTO watchlist (chat_id, symbol, created_at) VALUES (?, ?, ?)",
            (str(chat_id), symbol.upper(), datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
        )
        conn.commit()
        success = True
    except sqlite3.IntegrityError:
        success = False  # 已經存在自選股中
    conn.close()
    return success

def remove_from_watchlist(chat_id, symbol):
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    cursor.execute(
        "DELETE FROM watchlist WHERE chat_id = ? AND symbol = ?",
        (str(chat_id), symbol.upper())
    )
    affected = cursor.rowcount
    conn.commit()
    conn.close()
    return affected > 0

def get_watchlist(chat_id):
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    cursor.execute(
        "SELECT symbol FROM watchlist WHERE chat_id = ? ORDER BY created_at ASC",
        (str(chat_id),)
    )
    rows = cursor.fetchall()
    conn.close()
    return [row[0] for row in rows]

bot.py-實作 /add, /del, /watchlist 指令

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

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)

# 新增自選股指令 (/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(CallbackQueryHandler(button_callback))

解釋一下程式碼吧！
以 chat_id 為區隔的個體化儲存：
在 watchlist 資料表中，我們把 chat_id 與 symbol 設為主鍵。這樣能確保不同使用者在同一台機器人身上，都能擁有各自獨立的自選股清單，不會互相對立或干擾。

動態按鈕選單（View Stock Button）：
在執行 /watchlist 指令時，程式除了條列股票之外，還會動態產生「看盤 2330.TW」按鈕，點擊後即可直接觸發儀表板，可以很大幅的減低操作步驟，這樣使用起來更方便！
步驟三：執行與測試
在終端機內輸入:

python main.py

打開 Telegram 進行以下測試：
輸入 /add 2330：機器人回覆 [SUCCESS] 已成功將 2330.TW 加入自選股追蹤清單！
輸入 /add 2454：機器人回覆 [SUCCESS] 已成功將 2454.TW 加入自選股追蹤清單!
輸入 /watchlist：機器人印出你的口袋名單，並在下方提供對應的「看盤 2330.TW」按鈕，點擊即可直接檢視卡片！
輸入 /del 2330：機器人成功將 2330 從資料庫清單中刪除。
小結一下：
導入 SQLite 的持久化儲存後，即便我們的機器人重啟或部署至雲端伺服器，使用者的自選股清單也不會遺失！
今天我們完成了「動態自選股管理系統（/add, /del, /watchlist）」！