教案：使用 Selenium 爬取 Nasdaq 即時價格

課程目標

了解如何使用 Selenium 進行網頁爬取。
學習 lxml 處理 HTML 資料。
熟悉 WebDriver 的基本操作。
掌握動態網頁數據爬取的技巧。

先備知識

Python 基礎語法
HTML / XPath 知識
瀏覽器開發者工具的使用

教學內容

1. 安裝相依套件

本程式使用 Selenium 來控制瀏覽器，lxml 來解析 HTML。請執行以下指令安裝必要的套件：

python -m pip install lxml
python -m pip install webdriver-manager
python -m pip install selenium

2. 環境設定

在 Python 中，我們需要設定 Selenium WebDriver 來開啟瀏覽器，並設置 Chrome 瀏覽器的相關選項。

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager

options = Options()
options.add_argument("--headless")  # 讓瀏覽器在背景執行
options.add_argument("--disable-gpu")
options.add_argument("--window-size=1920x1080")
wd = webdriver.Chrome(service=Service(ChromeDriverManager().install()), options=options)

3. 爬取 Nasdaq 即時價格

我們的目標是從 Nasdaq 網站獲取 Tesla (TSLA) 股票的即時交易數據。

__baseUrl__ = 'https://www.nasdaq.com/market-activity/stocks/tsla/latest-real-time-trades'
wd.get(__baseUrl__)
wd.implicitly_wait(5)  # 設定隱式等待

3.1 解析交易數據

from lxml import etree

def composeTrade(dom):
    trades = []
    rows = dom.xpath('//tr[@class="latest-real-time-trades__row"]')
    for row in rows:
        trades.append({
            "nlsTime": row[0].text,
            "nlsPrice": row[1].text,
            "nlsVolume": row[2].text
        })
    return trades

4. 自動翻頁與資料收集

網站的即時交易數據可能需要翻頁，因此我們使用 Selenium 自動點擊「下一頁」。

from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

dayTrades = []
nextButtonXpath = '//button[@class="pagination__next"]'
while True:
    try:
        wait = WebDriverWait(wd, 10)
        element = wait.until(EC.element_to_be_clickable((By.XPATH, nextButtonXpath)))
        wd.execute_script("arguments[0].scrollIntoView()", element)
        wd.execute_script("arguments[0].click();", element)
    except:
        break

5. 執行程式

完整程式碼如下：

"""
說明：此程式用於爬取Nasdaq即時價格，需在營運時間內執行，部分時段可能無即時交易行情
作者：Jung-Yu Yu jungyuyu@gmail.com 2022-05-02
授權：創用 CC 姓名標示 4.0 國際授權條款
版本：0.2 (優化版)
"""

import time
import logging
from typing import List, Dict
from lxml import etree
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
from selenium.common.exceptions import TimeoutException, StaleElementReferenceException

# 設定 logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

BASE_URL = 'https://www.nasdaq.com/market-activity/stocks/tsla/latest-real-time-trades'
USER_AGENT = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/100.0.4896.127 Safari/537.36"

class NasdaqScraper:
    def __init__(self):
        self.driver = self._initialize_driver()
        self.day_trades = []

    def _initialize_driver(self) -> webdriver.Chrome:
        """初始化並配置Chrome WebDriver"""
        options = Options()
        
        # 基本配置
        options.add_experimental_option("excludeSwitches", ["enable-automation", 'enable-logging'])
        options.add_experimental_option('useAutomationExtension', False)
        options.add_experimental_option("prefs", {
            "profile.password_manager_enabled": False,
            "credentials_enable_service": False
        })
        
        # 效能與隱藏相關設定
        options.add_argument(f"user-agent={USER_AGENT}")
        options.add_argument('--no-sandbox')
        options.add_argument('--disable-dev-shm-usage')
        options.add_argument("--start-maximized")
        
        # 開發時可取消註解觀察畫面
        # options.add_argument('--headless')
        
        return webdriver.Chrome(
            service=Service(ChromeDriverManager().install()),
            options=options
        )

    def _parse_trade_table(self) -> List[Dict]:
        """解析交易表格數據"""
        results = self.driver.find_element(By.XPATH, '//table[@class="latest-real-time-trades__table"]').get_attribute('innerHTML')
        dom = etree.HTML(results)
        rows = dom.xpath('//tr[@class="latest-real-time-trades__row"]')
        
        return [{
            "nlsTime": row[0].text,
            "nlsPrice": row[1].text,
            "nlsVolume": row[2].text
        } for row in rows]

    def scrape(self):
        """主爬取邏輯"""
        try:
            # 開啟網頁並設定初始等待
            self.driver.get(BASE_URL)
            self.driver.implicitly_wait(5)
            self.driver.execute_script("document.body.style.webkitAnimationPlayState='paused'")

            # 獲取交易日期
            trade_date = self.driver.find_element(
                By.XPATH, 
                '//div[@class="symbol-page-header__timestamp"]/span[@class="symbol-page-header__status"]'
            ).text
            logger.info(f"交易日期: {trade_date}")

            # 爬取第一頁
            self.day_trades.extend(self._parse_trade_table())
            
            # 分頁處理
            next_button_xpath = '//button[@class="pagination__next"]'
            while True:
                try:
                    wait = WebDriverWait(self.driver, 10)
                    next_button = wait.until(EC.element_to_be_clickable((By.XPATH, next_button_xpath)))
                    
                    # 捲動並點擊下一頁
                    self.driver.execute_script("arguments[0].scrollIntoView()", next_button)
                    ActionChains(self.driver).move_to_element(next_button).perform()
                    time.sleep(3)  # 調整為適當的等待時間
                    
                    # 獲取當前頁數據
                    self.day_trades.extend(self._parse_trade_table())
                    self.driver.execute_script("arguments[0].click();", next_button)
                    logger.info("下一頁已載入")
                    
                except (TimeoutException, StaleElementReferenceException):
                    logger.info("已達最後一頁或發生錯誤，結束分頁爬取")
                    break
                    
            logger.info(f"總共收集 {len(self.day_trades)} 筆交易數據")
            return self.day_trades
            
        except Exception as e:
            logger.error(f"爬取過程中發生錯誤: {str(e)}")
            return []
            
        finally:
            self.driver.quit()

def main():
    scraper = NasdaqScraper()
    trades = scraper.scrape()
    print(trades)

if __name__ == "__main__":
    main()

作業與討論

嘗試修改程式碼以爬取不同的股票數據（例如 AAPL、GOOGL）。
加入數據儲存功能，將結果寫入 CSV 或 JSON。
研究 Selenium 中的其他等待機制（顯示等待 vs 隱式等待）。

延伸學習

BeautifulSoup 和 Scrapy 的應用
Pandas 進行數據處理與視覺化
Selenium 與 Headless Chrome 的更多自動化應用

使用 Selenium 爬取 Nasdaq 即時價格