iT邦幫忙

2026 iThome 鐵人賽

DAY 12
0
Software Development

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

Day 12:籌碼動能大解密 — 導入成交量能與法人籌碼監控!

  • 分享至 

  • xImage
  •  

今天是第十二天!昨天我們成功賦予了機器人「技術面雙指標(RSI + MACD)」的戰術大腦。
有了 K 線與技術指標,我們已經能看出「價格」的趨勢。但在台股市場裡,有一句老話說得好:「價格是人用錢堆出來的,量才是靈魂!」
除了看價格線圖(技術面),觀察成交量變化與三大法人(外資、投信、自營商)的籌碼動向,更能幫我們抓出主力到底是在偷偷佈局蓄勢待發,還是在拉高出貨誘多!
今天我們的目標,就是要把「量能與籌碼面診斷」無縫整合進我們的 Telegram 看盤機器人,讓它一秒進化成技術+籌碼雙修的全方位分析師!

為什麼要看成交量與籌碼?

成交量(Volume):
爆量突破:價格突破壓力點且成交量大增(高於 5 日均量),通常代表主力真金白銀進場,突破的可信度高!
量縮觀望:價格波動但成交量極低,代表市場觀望氣氛濃,暫時缺乏推升動能。
三大法人動向:
外資:資金體量龐大,偏好大型權值股(如台積電),影響大盤趨勢最深。
投信:國內基金代表,偏好中小型飆股,操作節奏快,常有「季末作帳/結帳」行情。
自營商:券商自有資金,偏向極短線避險與套利。
步驟一:更新 main.py 腳本
打開你的 main.py,把內容替換成這份整合了技術面(MA/RSI/MACD)+ 籌碼量能診斷的最新程式碼:

import os
import pandas as pd
import pandas_ta as ta
import yfinance as yf
import mplfinance as mpf
import urllib.request
import time

# 1. 通訊金鑰與設定
TOKEN = "你的_BOT_TOKEN"
CHAT_ID = "你的_CHAT_ID"

# 想要追蹤的自選股清單
STOCK_LIST = [
    {"symbol": "2330.TW", "name": "台積電"},
    {"symbol": "2317.TW", "name": "鴻海"},
    {"symbol": "2454.TW", "name": "聯發科"},
    {"symbol": "0050.TW", "name": "元大台灣50"}
]

# 發送圖片到 Telegram 函式 
def send_telegram_photo(token, chat_id, photo_path, caption_text):
    url = f"https://api.telegram.org/bot{token}/sendPhoto"
    boundary = '----WebKitFormBoundary7MA4YWxkTrZu0gW'

    if not os.path.exists(photo_path):
        print(f"[ERROR] 找不到圖片檔案: {photo_path}")
        return False

    with open(photo_path, 'rb') as f:
        photo_data = f.read()

    # 關鍵:所有字串全都強制編碼為 utf-8 的 bytes 處理
    parts = [
        f'--{boundary}\r\nContent-Disposition: form-data; name="chat_id"\r\n\r\n{chat_id}\r\n'.encode('utf-8'),
        f'--{boundary}\r\nContent-Disposition: form-data; name="caption"\r\n\r\n{caption_text}\r\n'.encode('utf-8'),
        f'--{boundary}\r\nContent-Disposition: form-data; name="photo"; filename="{os.path.basename(photo_path)}"\r\nContent-Type: image/png\r\n\r\n'.encode('utf-8') + photo_data + b'\r\n',
        f'--{boundary}--\r\n'.encode('utf-8')
    ]

    payload = b''.join(parts)
    headers = {'Content-Type': f'multipart/form-data; boundary={boundary}'}

    try:
        req = urllib.request.Request(url, data=payload, headers=headers)
        with urllib.request.urlopen(req) as response:
            return True
    except Exception as e:
        print(f"[ERROR] Telegram 發送失敗:{e}")
        return False

# 分析單檔股票
def process_stock(symbol, name):
    print(f"\n[INFO] 正在深入分析:{name} ({symbol})...")
    
    # 抓取 6 個月歷史資料
    df = yf.download(symbol, period="6mo", progress=False)

    if df.empty:
        print(f"[WARNING] 無法取得 {symbol} 資料,略過。")
        return

    # 清理 yfinance 多層索引
    if isinstance(df.columns, pd.MultiIndex):
        df.columns = df.columns.get_level_values(0)

    # 1. 計算技術指標 (MA, RSI, MACD)
    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. 量能與籌碼動向分析 (計算 5 日均量)
    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"{symbol.replace('.', '_')}_chart.png"

    # 設定三個圖層(主圖: K線+MA,副圖1: RSI,副圖2: MACD 柱狀圖)
    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"{name} ({symbol}) Tech & Volume 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"【{name} ({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(死亡交叉)")

    # RSI 判斷
    if rsi_val >= 70:
        signals.append("[RSI] 過熱區 (>70),留意回檔風險")
    elif rsi_val <= 30:
        signals.append("[RSI] 超賣區 (<30),關注反彈買點")

    # MACD 判斷
    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)

    # 發送訊息至 Telegram
    success = send_telegram_photo(TOKEN, CHAT_ID, img_filename, caption)
    if success:
        print(f"[SUCCESS] {name} 全方位報告已成功發送至 Telegram!")
    
    # 刪除暫存檔
    if os.path.exists(img_filename):
        os.remove(img_filename)

# 主流程
if __name__ == "__main__":
    print("[START] 啟動第 12 天:技術面 + 籌碼量能全方位盤後掃描...")
    for item in STOCK_LIST:
        process_stock(item["symbol"], item["name"])
        time.sleep(1)
    print("\n[FINISH] 今日全方位巡檢完畢!")

步驟二:執行與驗證

python main.py

最後結果會長這樣:

小結一下:
到了第 12 天,我們的盤後小幫手已經完成了:
自動巡檢多股(2330, 2317, 2454, 0050 ... 自選清單隨意擴充)
三圖層視覺化(K線/均線 + 成交量 + RSI + MACD)
技術與籌碼雙診斷(均線交叉 + RSI過熱過冷 + MACD動能轉折 + 量能爆發監控)
編碼與穩定發送(徹底擺脫 Unicode ASCII 轉碼地獄)

我在驗證的時候最容易犯的錯是在終端機打反指令,一定要記得是python main.py ,不是 main python.py!


上一篇
Day 11:升級雙重技術指標(RSI + MACD)與指標背離提醒
下一篇
Day 13:讓機器人聽懂人話 — 打造 Telegram 互動式點播功能!
系列文
30 天打造專屬盯盤機器人:資料爬取、策略指標到 Telegram/LINE 推播 共 17 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言