Python 抓台股資料教學：用 API 取得 OHLCV、成交量與個股行情

為什麼用 API 抓台股資料？

很多台股資料分析專案一開始都會從爬蟲開始。爬蟲適合 prototype，但如果要建立長期可維護的量化研究、資料 dashboard 或 AI agent workflow，API 會更穩定。

API 的價值不只是取得資料，而是提供一套可被程式長期消費的資料介面：

• endpoint 穩定
• 欄位名稱一致
• 日期格式一致
• 錯誤碼清楚
• 支援批次查詢
• 有 rate limit 說明
• 可以接進 backtest、ETL、dashboard 或 AI agent tool calling

如果你還不熟悉台股 API 的資料類型，可以先看 台股 API 完整指南。

如果你要專注在回測資料品質、K 線與除權息處理，可以再看 台股歷史股價 API 設計。

如果你下一步要把資料串到策略回測與風控流程，建議接著看 台股量化交易入門。

Python 環境準備

這篇文章使用 Python、requests 和 pandas 示範。你可以用 venv、pyenv、conda 或任何你習慣的 Python 環境。

python -m venv .venv
source .venv/bin/activate

pip install requests pandas python-dotenv

建議不要把 API key 寫死在程式裡。可以用環境變數或 .env 管理。

TW_MARKET_DATA_API_KEY=your_api_key_here

import os
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("TW_MARKET_DATA_API_KEY")

if not API_KEY:
    raise RuntimeError("Missing TW_MARKET_DATA_API_KEY")

用 requests 呼叫台股 API

以下示範如何用 Python 呼叫台股 OHLCV endpoint。OHLCV 代表 open、high、low、close、volume，是歷史股價與回測最常用的基礎資料。

import requests

BASE_URL = "https://api.example.com"

headers = {
    "Authorization": f"Bearer {API_KEY}"
}

params = {
    "symbol": "2330",
    "from": "2025-01-01",
    "to": "2025-12-31"
}

response = requests.get(
    f"{BASE_URL}/v1/tw/stocks/2330/ohlcv",
    headers=headers,
    params=params,
    timeout=20
)

response.raise_for_status()

payload = response.json()

print(payload.keys())

上方 endpoint 是示意。實際路徑請以 TW Market Data docs 為準。

把 API response 轉成 pandas DataFrame

API response 通常是 JSON。對資料分析與量化研究來說，最常見的下一步是轉成 pandas DataFrame。

import pandas as pd

data = payload["data"]

df = pd.DataFrame(data)

df["date"] = pd.to_datetime(df["date"])
df = df.sort_values("date").reset_index(drop=True)

print(df.head())

轉成 DataFrame 後，就可以進一步做：

• 畫價格走勢
• 計算報酬率
• 計算技術指標
• 建立量化 signal
• 匯出成 parquet / csv
• 接進 backtesting engine

df["daily_return"] = df["close"].pct_change()

print(df[["date", "symbol", "close", "daily_return"]].tail())

OHLCV 欄位怎麼設計？

穩定的欄位設計比單次抓到資料更重要。對台股資料 API 來說，OHLCV schema 至少應該包含以下欄位。

對回測來說，date、symbol、close 和 volume 是最低限度。若要做更完整的策略研究，market、turnover、adjusted price、交易日曆與除權息資料也會很重要。

一次查詢多檔股票

量化研究通常不會只看一檔股票，而是會建立一個 universe。例如大型股、ETF 成分股、特定產業或自訂股票池。

如果 API 支援批次查詢，可以用 batch endpoint。若目前只支援單檔查詢，也可以在 client 端迴圈呼叫，但要注意 rate limit。

symbols = ["2330", "2317", "2454"]

frames = []

for symbol in symbols:
    response = requests.get(
        f"{BASE_URL}/v1/tw/stocks/{symbol}/ohlcv",
        headers=headers,
        params={
            "from": "2025-01-01",
            "to": "2025-12-31"
        },
        timeout=20
    )
    response.raise_for_status()

    data = response.json()["data"]
    frame = pd.DataFrame(data)
    frames.append(frame)

prices = pd.concat(frames, ignore_index=True)
prices["date"] = pd.to_datetime(prices["date"])

print(prices.head())

上方 endpoint 是示意。實際路徑請以 TW Market Data docs 為準。

如果後續要做 portfolio-level backtest，建議把資料整理成長表格式：

長表格式比寬表格式更適合 API、資料庫、ETL 和多商品分析。

加入錯誤處理、retry 與 timeout

Production workflow 不能假設每次 request 都會成功。你至少需要處理：

• timeout
• 429 rate limit
• 401 authentication error
• 404 symbol not found
• 5xx server error
• response schema 不符合預期

import time
import requests

def fetch_ohlcv(symbol: str, start: str, end: str, max_retries: int = 3) -> pd.DataFrame:
    url = f"{BASE_URL}/v1/tw/stocks/{symbol}/ohlcv"

    for attempt in range(max_retries):
        try:
            response = requests.get(
                url,
                headers=headers,
                params={
                    "from": start,
                    "to": end
                },
                timeout=20
            )

            if response.status_code == 429:
                time.sleep(2 ** attempt)
                continue

            response.raise_for_status()

            payload = response.json()
            data = payload.get("data", [])

            frame = pd.DataFrame(data)

            if frame.empty:
                return frame

            frame["date"] = pd.to_datetime(frame["date"])
            frame = frame.sort_values("date").reset_index(drop=True)

            return frame

        except requests.RequestException:
            if attempt == max_retries - 1:
                raise

            time.sleep(2 ** attempt)

    return pd.DataFrame()

實務上也可以用 tenacity 這類套件管理 retry，但在教學文章中先用簡單函式比較容易理解。

用 cache 降低 API request

歷史 OHLCV 不需要每次都重新抓。對研究與回測來說，cache 可以降低 API request、加速 notebook，也避免重複消耗 rate limit。

from pathlib import Path

CACHE_DIR = Path(".cache/ohlcv")
CACHE_DIR.mkdir(parents=True, exist_ok=True)

def get_cache_path(symbol: str, start: str, end: str) -> Path:
    return CACHE_DIR / f"{symbol}_{start}_{end}.parquet"

def load_or_fetch_ohlcv(symbol: str, start: str, end: str) -> pd.DataFrame:
    cache_path = get_cache_path(symbol, start, end)

    if cache_path.exists():
        return pd.read_parquet(cache_path)

    frame = fetch_ohlcv(symbol, start, end)

    if not frame.empty:
        frame.to_parquet(cache_path, index=False)

    return frame

如果要用 parquet，請安裝 pyarrow：

pip install pyarrow

接到台股回測或 AI agent workflow

抓資料只是第一步。對 TW Market Data 的使用者來說，更重要的是資料可以接進後續 workflow。

df = load_or_fetch_ohlcv("2330", "2025-01-01", "2025-12-31")

df["ma20"] = df["close"].rolling(20).mean()
df["ma60"] = df["close"].rolling(60).mean()

df["signal"] = 0
df.loc[df["ma20"] > df["ma60"], "signal"] = 1
df.loc[df["ma20"] <= df["ma60"], "signal"] = 0

print(df[["date", "close", "ma20", "ma60", "signal"]].tail())

def get_daily_ohlcv_tool(symbol: str, start: str, end: str) -> dict:
    frame = load_or_fetch_ohlcv(symbol, start, end)

    return {
        "symbol": symbol,
        "start": start,
        "end": end,
        "rows": len(frame),
        "data": frame.tail(20).to_dict(orient="records"),
        "not_investment_advice": True
    }

完整範例程式

以下是一個簡化版完整範例，示範從 API request、DataFrame 整理到 moving average signal。

import os
import time
from pathlib import Path

import pandas as pd
import requests
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("TW_MARKET_DATA_API_KEY")
BASE_URL = "https://api.example.com"

if not API_KEY:
    raise RuntimeError("Missing TW_MARKET_DATA_API_KEY")

headers = {
    "Authorization": f"Bearer {API_KEY}"
}

CACHE_DIR = Path(".cache/ohlcv")
CACHE_DIR.mkdir(parents=True, exist_ok=True)

def fetch_ohlcv(symbol: str, start: str, end: str, max_retries: int = 3) -> pd.DataFrame:
    url = f"{BASE_URL}/v1/tw/stocks/{symbol}/ohlcv"

    for attempt in range(max_retries):
        try:
            response = requests.get(
                url,
                headers=headers,
                params={
                    "from": start,
                    "to": end
                },
                timeout=20
            )

            if response.status_code == 429:
                time.sleep(2 ** attempt)
                continue

            response.raise_for_status()

            payload = response.json()
            frame = pd.DataFrame(payload.get("data", []))

            if frame.empty:
                return frame

            frame["date"] = pd.to_datetime(frame["date"])
            frame = frame.sort_values("date").reset_index(drop=True)

            return frame

        except requests.RequestException:
            if attempt == max_retries - 1:
                raise

            time.sleep(2 ** attempt)

    return pd.DataFrame()

def load_or_fetch_ohlcv(symbol: str, start: str, end: str) -> pd.DataFrame:
    cache_path = CACHE_DIR / f"{symbol}_{start}_{end}.parquet"

    if cache_path.exists():
        return pd.read_parquet(cache_path)

    frame = fetch_ohlcv(symbol, start, end)

    if not frame.empty:
        frame.to_parquet(cache_path, index=False)

    return frame

df = load_or_fetch_ohlcv("2330", "2025-01-01", "2025-12-31")

df["daily_return"] = df["close"].pct_change()
df["ma20"] = df["close"].rolling(20).mean()
df["ma60"] = df["close"].rolling(60).mean()
df["signal"] = (df["ma20"] > df["ma60"]).astype(int)

print(df.tail())

上方 endpoint 是示意。實際路徑請以 TW Market Data docs 為準。

這是教學用範例，不構成投資建議。實際交易前需要完整回測、風險控管、交易成本與合規檢查。