今天是第二十天!昨天我們幫機器人裝上了 SQLite 資料庫快取機制,大幅提升了查詢速度,讓 Telegram 回覆變得秒傳秒回!
不過在台股市場裡,除了技術面、籌碼面與基本面之外,還有一個對短期股價波動極具影響力的因素——「消息面」!當個股突然出現法說會利多、擴廠消息、或是產品遭客戶砍單等新聞時,股價往往會隨之劇烈波動。如果機器人在提供技術與籌碼分析的同時,還能順便幫我們整理當天的「最新重大新聞與利多利空評價」,看盤就會更加全面!
今天我們的目標,就是在我們的專案架構中,加入「個股新聞爬蟲與關鍵字情緒分析」模組,讓我們的 Telegram 機器人邁向「技術 + 籌碼 + 基本 + 消息」更加全面的看盤神器!
步驟一:先釐清邏輯
新聞爬蟲與情緒分析的運作邏輯非常直覺:
抓取個股最新新聞(news.py):利用 BeautifulSoup 或 yfinance 的新聞 API 抓取個股最新 3~5 則相關新聞標題與連結。
關鍵字情緒比對(Sentiment Scoring):
接下來,知道邏輯後,我們就可以來撰寫程式碼了!
步驟二:完整實作程式碼
我們新建一個 news.py 專門處理新聞爬蟲與情緒分析,並整合至 fetcher.py 與 bot.py。
1.news.py
import yfinance as yf
# 利多與利空關鍵字字典
BULLISH_KEYWORDS = ["創新高", "大增", "擴廠", "轉盈", "接單", "看好", "買超", "漲停", "成長", "優於預期"]
BEARISH_KEYWORDS = ["創新低", "大跌", "虧損", "砍單", "裁員", "看淡", "賣超", "跌停", "衰退", "低於預期"]
def fetch_stock_news(symbol, max_news=3):
print(f"[INFO] 正在爬取 {symbol} 的即時個股新聞...")
ticker = yf.Ticker(symbol)
# 取得 yfinance 提供的新聞資料
raw_news = ticker.news
news_list = []
if not raw_news:
return news_list, "中性平穩", 0, 0
bullish_count = 0
bearish_count = 0
for item in raw_news[:max_news]:
# yfinance 新聞結構維護
content = item.get('content', {}) if isinstance(item.get('content'), dict) else item
title = content.get('title', item.get('title', '無標題'))
link = content.get('clickThroughUrl', {}).get('url') if isinstance(content.get('clickThroughUrl'), dict) else item.get('link', '')
# 簡單情緒比對
for kw in BULLISH_KEYWORDS:
if kw in title:
bullish_count += 1
for kw in BEARISH_KEYWORDS:
if kw in title:
bearish_count += 1
news_list.append({
"title": title,
"link": link
})
# 判斷整體消息面情緒
if bullish_count > bearish_count:
sentiment = "消息面偏多"
elif bearish_count > bullish_count:
sentiment = "消息面偏空"
else:
sentiment = "消息面中性"
return news_list, sentiment, bullish_count, bearish_count
import pandas as pd
import pandas_ta as ta
import yfinance as yf
from database import get_cached_data, save_cache_data
from news import fetch_stock_news
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, 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 not (cached_chip and cached_rev):
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
}
save_cache_data(formatted_symbol, latest_date, chip_data, revenue_data)
else:
chip_data, revenue_data = cached_chip, cached_rev
# 3. 抓取最新個股新聞與情緒分析
news_list, sentiment, bull_cnt, bear_cnt = fetch_stock_news(formatted_symbol)
news_data = {
"list": news_list,
"sentiment": sentiment,
"bullish": bull_cnt,
"bearish": bear_cnt
}
return formatted_symbol, df, chip_data, revenue_data, news_data
def generate_report_text(symbol, df, chip_data=None, revenue_data=None, news_data=None):
latest = df.iloc[-1]
prev = df.iloc[-2]
rsi_val = latest['RSI']
latest_vol = latest['Volume']
vol_ma5_val = latest['Vol_MA5']
caption = f"【{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"
# 籌碼面區塊
if chip_data:
caption += f"-----------------------------------\n"
caption += f"【三大法人籌碼動向】(單位: 張)\n"
caption += f"外資:{chip_data['foreign']:+d} | 投信:{chip_data['investment']:+d}\n"
caption += f"法人合計:{chip_data['total']:+d}\n"
# 基本面營收區塊
if revenue_data:
caption += f"-----------------------------------\n"
caption += f"【最新月營收報告】({revenue_data['month']})\n"
caption += f"單月營收:{revenue_data['revenue'] / 100000:.2f} 億元 | YoY:{revenue_data['yoy']:+.1f}%\n"
# 消息面新聞區塊
if news_data and news_data["list"]:
caption += f"-----------------------------------\n"
caption += f"【即時消息面分析】({news_data['sentiment']})\n"
for idx, item in enumerate(news_data["list"], 1):
caption += f"{idx}. {item['title']}\n"
caption += f"-----------------------------------\n"
signals = []
# 消息面訊號
if news_data:
if news_data["sentiment"] == "消息面偏多":
signals.append("[消息面] 近期媒體新聞多提及利多關鍵字,關注市場熱度!")
elif news_data["sentiment"] == "消息面偏空":
signals.append("[消息面] 近期新聞出現較多利空關鍵字,留意短線避險情緒!")
# 基本面訊號
if revenue_data and revenue_data['yoy'] >= 30:
signals.append("[基本面] 月營收年增率 (YoY) 突破 30%,營收強勁成長!")
# 籌碼面訊號
if chip_data and chip_data['foreign'] > 0 and chip_data['investment'] > 0:
signals.append("[籌碼] 外資與投信同買,法人作多力道強勁!")
# 技術面訊號
if prev['MA5'] <= prev['MA20'] and latest['MA5'] > latest['MA20']:
signals.append("[MA] 5MA 向上突破 20MA(黃金交叉)")
if not signals:
caption += "評語:技術、籌碼、基本與消息面處於平穩狀態,建議持續觀察。"
else:
caption += "關鍵訊號彙整:\n" + "\n".join(signals)
return caption
解釋一下程式碼吧!
關鍵字匹配演算法(Keyword Matching):
我們設定了常用的台股新聞利多與利空關鍵字,自動掃描標題文字並計算比例。這種做法雖然簡單,但在新聞題材熱絡時能非常迅速地對市場風向做出大致分類!
技術 + 籌碼 + 基本 + 消息 整合:
報告內容現在涵蓋了四大看盤面向,讓你的 Telegram 看盤報告不再只是一堆枯燥的數字,而是像一份由專業研究員產出的綜合日報。
步驟三:執行與測試
在終端機內輸入:
python main.py
小結一下:
消息面適合用來輔助觀察市場情緒,但仍建議搭配籌碼與技術面一起交叉印證,避免追高受傷。
今天我們成功導入了新聞爬蟲與關鍵字情緒分析,完成了「技術面 + 籌碼面 + 基本面 + 消息面」機器人的最後一塊拼圖!