今天是第十九天！昨天們成功讓機器人具備了「技術面 + 籌碼面 + 基本面（月營收 YoY/MoM）」的全方位診斷能力，資訊量直接拉滿！

不過隨著功能越來越強大、每次查詢都要同時抓股價、算指標、拿籌碼與營收資料，你可能會發現：每次在 Telegram 輸入 /stock 2330，機器人都要轉圈圈等個幾秒鐘才回覆。如果短時間內多查詢幾次，甚至可能因為頻繁請求 Yahoo Finance 或外部 API 而被暫時封鎖 IP！
今天我們的目標，就是在專案中導入 Python 內建的 SQLite 輕量級資料庫，建立「快取機制」！只要當天已經抓過的資料就直接從資料庫秒讀，大幅提升回應速度，同時保護我們的 API 不被封鎖！
步驟一：先釐清邏輯
快取萬用邏輯非常簡單且直覺，

• 建立 SQLite 資料庫：在本地端建立一個名為 stock_cache.db 的輕量資料庫檔案，用來儲存已抓取的股價與分析資料。
• 查詢前先查快取：當收到 /stock 2330 請求時，先檢查資料庫裡有沒有最新交易日的快取紀錄。
• 快取命中：如果資料庫裡有當天的資料，直接秒讀並產出報告，速度快到飛起！
• 快取未命中：如果資料庫沒有紀錄（或者紀錄不是今天的），才向網路 API 抓取最新數據，並同步寫入 SQLite 資料庫備用。

接下來，知道邏輯後，我們就可以來寫程式碼了！
步驟二：完整實作程式碼
我們新增一個 database.py 模組，並將快取邏輯整合進 fetcher.py 裡面！

1. database.py:我們使用 Python 內建的 sqlite3

import sqlite3
import json
from datetime import datetime

DB_NAME = "stock_cache.db"

def init_db():
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    # 建立快取資料表
    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)
        )
    ''')
    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()

2. fetcher.py:在向網路發出請求前，優先從 SQLite 讀取快取

import pandas as pd
import pandas_ta as ta
import yfinance as yf
from database import get_cached_data, save_cache_data

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}...")

    # 1. 抓取股價歷史數據
    ticker = yf.Ticker(formatted_symbol)
    df = ticker.history(period="6mo")

    if df.empty and formatted_symbol.endswith('.TW'):
        alt_symbol = formatted_symbol.replace('.TW', '.TWO')
        ticker = yf.Ticker(alt_symbol)
        df = ticker.history(period="6mo")
        if not df.empty:
            formatted_symbol = alt_symbol

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

    # 取得最新資料日期
    latest_date = df.index[-1].strftime('%Y-%m-%d')

    # 2. 檢查 SQLite 快取
    cached_chip, cached_rev = get_cached_data(formatted_symbol, latest_date)

    if cached_chip and cached_rev:
        print(f"[CACHE] 命中 SQLite 快取！直接秒讀 {formatted_symbol} ({latest_date}) 數據。")
        return formatted_symbol, df, cached_chip, cached_rev

    # 3. 快取未命中，模擬/爬取外部籌碼與營收數據
    print(f"[NETWORK] 無快取紀錄，向外部 API 請求籌碼與營收資料...")
    chip_data = {
        "foreign": 1250,
        "investment": 800,
        "dealer": -150
    }
    chip_data["total"] = chip_data["foreign"] + chip_data["investment"] + chip_data["dealer"]

    revenue_data = {
        "month": "2026-09",
        "revenue": 75000000,
        "mom": 8.5,
        "yoy": 32.4
    }

    # 4. 寫入 SQLite 快取
    save_cache_data(formatted_symbol, latest_date, chip_data, revenue_data)

    return formatted_symbol, df, chip_data, revenue_data

3. main.py:呼叫 init_db() 初始化 SQLite 資料庫表

from telegram.ext import ApplicationBuilder
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from config import TOKEN, SCHEDULE_HOUR, SCHEDULE_MINUTE
from bot import setup_handlers, scheduled_job
from database import init_db

if __name__ == "__main__":
    print("[START] 機器人與自動排程系統啟動中...")

    # 初始化 SQLite 資料庫
    init_db()
    print("[DB] SQLite 資料庫連線與資料表初始化完成！")

    app = ApplicationBuilder().token(TOKEN).build()
    setup_handlers(app)

    scheduler = AsyncIOScheduler()
    scheduler.add_job(
        scheduled_job,
        trigger=CronTrigger(day_of_week="mon-fri", hour=SCHEDULE_HOUR, minute=SCHEDULE_MINUTE),
        args=[app]
    )
    scheduler.start()

    app.run_polling()

步驟三：執行與測試
最後，請在終端機內輸入:

python main.py

小結一下：
本地專案根目錄會自動生成一個 stock_cache.db 檔案，這就是我們的 SQLite 資料庫！
今天我們成功導入了 SQLite 資料庫快取機制，大幅減少了 API 請求次數、提升了機器人回應速度，並大幅降低封鎖風險！