爬取 Yahoo 股市即時股價 --電機機械 半導體 電子零組件 電子通路 四大類別---Python TKinter
https://tw.stock.yahoo.com/class
上市類股
# 定義要爬取的四個類股及其 URL STOCK_SECTORS = {
"電機機械": "https://tw.stock.yahoo.com/class-quote?sectorId=6&exchange=TAI",
"半導體": "https://tw.stock.yahoo.com/class-quote?sectorId=40&exchange=TAI",
"電子零組件": "https://tw.stock.yahoo.com/class-quote?sectorId=44&exchange=TAI",
"電子通路": "https://tw.stock.yahoo.com/class-quote?sectorId=45&exchange=TAI" }
import tkinter as tk
from tkinter import ttk
import requests
from bs4 import BeautifulSoup
import time
import threading
# 定義要爬取的四個類股及其 URL
STOCK_SECTORS = {
"電機機械": "https://tw.stock.yahoo.com/class-quote?sectorId=6&exchange=TAI",
"半導體": "https://tw.stock.yahoo.com/class-quote?sectorId=40&exchange=TAI",
"電子零組件": "https://tw.stock.yahoo.com/class-quote?sectorId=44&exchange=TAI",
"電子通路": "https://tw.stock.yahoo.com/class-quote?sectorId=45&exchange=TAI"
}
# 爬取資料的函數，現在接受 URL 作為參數
def get_stock_data(url):
"""
從指定的 URL 爬取股票資訊。
"""
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
}
try:
response = requests.get(url, headers=headers)
response.raise_for_status()
soup = BeautifulSoup(response.text, 'html.parser')
table_body = soup.find('div', class_='table-body-wrapper')
if not table_body:
return []
stock_rows = table_body.find_all('li', class_='List(n)')
data = []
for row in stock_rows:
name_and_id_div = row.find('div', class_='Fxs(0)')
# 確保找到所有必要的元素
name_div = name_and_id_div.find('div', class_='Lh(20px)')
id_span = name_and_id_div.find('span', class_='Fz(14px)')
if not name_div or not id_span:
continue
name = name_div.text.strip()
stock_id = id_span.text.strip()
columns = row.find_all('div', class_='Fxg(1)')
if len(columns) >= 4:
price_span = columns[0].find('span')
change_span = columns[1].find('span')
change_percent_span = columns[2].find('span')
if not price_span or not change_span or not change_percent_span:
continue
price = price_span.text.strip()
change = change_span.text.strip()
change_percent = change_percent_span.text.strip()
# 處理漲跌符號，使其更易讀
if 'C($c-trend-up)' in str(change_span):
change = f"▲{change.lstrip('▲')}"
elif 'C($c-trend-down)' in str(change_span):
change = f"▼{change.lstrip('▼')}"
data.append({
"name": name,
"id": stock_id,
"price": price,
"change": change,
"change_percent": change_percent
})
return data
except requests.exceptions.RequestException as e:
print(f"網路連線錯誤: {e}")
return None
except Exception as e:
print(f"解析資料錯誤: {e}")
return None
class StockApp(tk.Tk):
def __init__(self):
super().__init__()
self.title("上市分類行情查詢")
self.geometry("900x700")
self.current_sector = "電機機械" # 預設顯示電機機械
self.create_widgets()
# 啟動首次資料載入
self.update_data()
def create_widgets(self):
"""
建立使用者介面元件，包含下拉選單和表格。
"""
# 建立控制面板框架
control_frame = ttk.Frame(self, padding="10")
control_frame.pack(fill='x')
ttk.Label(control_frame, text="選擇類股:", font=("Helvetica", 12)).pack(side='left', padx=(0, 10))
# 建立下拉式選單
self.sector_combobox = ttk.Combobox(control_frame, values=list(STOCK_SECTORS.keys()))
self.sector_combobox.pack(side='left')
self.sector_combobox.set(self.current_sector) # 設定預設值
self.sector_combobox.bind("<<ComboboxSelected>>", self.on_sector_change)
self.info_label = ttk.Label(control_frame, text="正在載入資料...", font=("Helvetica", 12))
self.info_label.pack(side='right')
# 建立表格框架
table_frame = ttk.Frame(self, padding="10")
table_frame.pack(fill='both', expand=True)
# 建立表格
columns = ("股票名稱", "股票代號", "股價", "漲跌", "漲跌幅(%)")
self.tree = ttk.Treeview(table_frame, columns=columns, show="headings")
self.tree.pack(fill='both', expand=True)
for col in columns:
self.tree.heading(col, text=col, anchor='center')
self.tree.column(col, anchor='center', width=120)
# 添加滾動條
scrollbar = ttk.Scrollbar(table_frame, orient="vertical", command=self.tree.yview)
self.tree.configure(yscrollcommand=scrollbar.set)
scrollbar.pack(side="right", fill="y")
def on_sector_change(self, event):
"""
處理下拉式選單的選擇事件，更新當前類股並重新爬取資料。
"""
self.current_sector = self.sector_combobox.get()
self.update_data()
def update_data(self):
"""
更新表格數據，並每60秒自動呼叫一次。
"""
# 使用多執行緒，避免爬取資料時UI凍結
def fetch_and_update():
url = STOCK_SECTORS.get(self.current_sector)
self.info_label.config(text=f"正在更新【{self.current_sector}】資料...")
if url:
stock_data = get_stock_data(url)
# 清空舊數據
for item in self.tree.get_children():
self.tree.delete(item)
if stock_data:
for stock in stock_data:
self.tree.insert("", "end", values=(
stock['name'],
stock['id'],
stock['price'],
stock['change'],
stock['change_percent']
))
self.info_label.config(text=f"【{self.current_sector}】資料更新時間: {time.strftime('%Y-%m-%d %H:%M:%S')}")
else:
self.info_label.config(text=f"更新【{self.current_sector}】失敗，請檢查網路連線或稍後再試。")
# 每60秒後再次執行
self.after(60000, self.update_data)
# 在一個新的執行緒中執行爬蟲函數
thread = threading.Thread(target=fetch_and_update)
thread.daemon = True
thread.start()
if __name__ == "__main__":
app = StockApp()
app.mainloop()
沒有留言:
張貼留言