第二章：市場數據與資料倉儲

# 第二章：市場數據與資料倉儲 在量化投資中，**資料是最核心的資產**。沒有可靠的數據來源、乾淨的數據結構，任何模型再先進也無法發揮實際效益。本章將帶領讀者從資料來源、品質、處理，到如何建構高效、可維護的資料倉儲與 API 獲取機制，為後續的因子挖掘與機器學習打下堅實基礎。 ## 2.1 市場資料來源 | 類型 | 典型來源 | 主要特點 | 適用場景 | |------|----------|----------|----------| | 股價/交易資料 | Yahoo Finance、Alpha Vantage、Tiingo、Polygon | 無需付費、頻率高（日、5min、1min） | 基礎行情、回測、低頻策略 | | 基本面資料 | Quandl、SEC EDGAR、Compustat | 週期性、結構化 | 基本面因子、價值投資 | | 替代資料 | FRED、Quandl 期貨、社交媒體 API、衛星影像 | 非傳統、稀疏、噪聲大 | 替代因子、情緒分析 | | 期貨/衍生品 | CME、ICE、IBKR | 高杠杆、波動性大 | 套利、風險管理 | | 匯率/商品 | OANDA、FXCM | 24/7、流動性高 | 全球配置、對沖 | > **實務提醒**：不同資料來源的時間戳格式、交易時段、缺失值處理方式都有差異，需在資料聚合前先統一時間基準與頻率。 ## 2.2 資料品質與預處理 ### 2.2.1 資料完整性 - **缺失值**：使用線性插值、時間窗口平均或向前/向後填充。對於開盤價與收盤價，常見做法是「向前填充」以保留交易訊號。 - **異常值**：利用 z‑score、IQR 或者布丁距離檢測離群點，決定是否修正或刪除。 ### 2.2.2 交易日曆與時間戳 | 步驟 | 目的 | |------|------| | 轉換時區 | 以 UTC 或本地時間為基準，避免跨時區套利錯誤 | | 交易時段統一 | 例如 09:30‑16:00 美股，確保所有時間序列對齊 | | 交易日曆生成 | 以 `pandas.tseries.offsets` 產生交易日索引，填補非交易日空白 | ### 2.2.3 資料格式化 - **欄位命名**：統一大小寫、下劃線命名，例如 `adj_close`, `volume`。 - **資料型別**：確保數值型別為 `float64`，避免因 `int64` 轉換造成精度損失。 - **時間序列索引**：設定為 `DatetimeIndex`，並設 `freq='B'`（交易日）或 `freq='1min'`。 ## 2.3 時間序列處理 ### 2.3.1 重新取樣 (Resampling) ```python import pandas as pd # 假設 df 為日頻資料 # 轉成 5 分鐘頻率 df_5min = df.resample('5T').agg({ 'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum', }) ``` ### 2.3.2 滑動窗口統計 ```python # 20 日移動平均 df['ma20'] = df['adj_close'].rolling(window=20).mean() # 20 日波動率 df['vol20'] = df['adj_close'].pct_change().rolling(window=20).std() ``` ### 2.3.3 滑動窗口交叉驗證 ```python from sklearn.model_selection import TimeSeriesSplit tsc = TimeSeriesSplit(n_splits=5) for train_index, test_index in tsc.split(df): X_train, X_test = df.iloc[train_index], df.iloc[test_index] # 進行模型訓練與評估 ``` ## 2.4 資料倉儲設計 ### 2.4.1 架構圖（Mermaid） ```mermaid graph TD A[資料來源] --> B[ETL 作業] B --> C[資料清洗] C --> D[資料倉儲] D --> E[資料服務層] E --> F[分析 & 回測] ``` ### 2.4.2 資料模型 | 表格 | 主鍵 | 主要欄位 | |------|------|----------| | `prices` | `ticker`, `datetime` | `open`, `high`, `low`, `close`, `adj_close`, `volume` | | `factors` | `ticker`, `datetime` | `momentum`, `volatility`, `sentiment_score` | | `meta` | `ticker` | `sector`, `market_cap` | > **最佳實踐**：採用 Columnar 格式（Parquet、Feather）可顯著降低 IO 時間，尤其在大規模歷史資料回測時。 ### 2.4.3 版本控制 - **DVC (Data Version Control)**：將資料檔案納入 Git 管理，保持版本可追蹤。 - **Delta Lake**：支援 ACID 交易、時間旅行，適合線上實盤資料。 ## 2.5 API 獲取機制 ### 2.5.1 使用 `yfinance` 取得歷史行情 ```python import yfinance as yf def fetch_historical(ticker: str, start: str, end: str, interval: str = '1d'): df = yf.download(ticker, start=start, end=end, interval=interval) df.reset_index(inplace=True) df['ticker'] = ticker return df # 範例：下載 AAPL 2018-2023 aapl = fetch_historical('AAPL', '2018-01-01', '2023-12-31') ``` ### 2.5.2 速率限制與重試機制 ```python import time import requests class RateLimitedClient: def __init__(self, limit: int, per: int): self.limit = limit self.per = per self.tokens = limit self.last = time.time() def acquire(self): now = time.time() elapsed = now - self.last self.tokens += elapsed * self.limit / self.per if self.tokens > self.limit: self.tokens = self.limit if self.tokens < 1: time.sleep((1 - self.tokens) * self.per / self.limit) self.tokens -= 1 self.last = time.time() client = RateLimitedClient(limit=5, per=60) # 每分鐘 5 次 client.acquire() resp = requests.get('https://api.example.com/data') ``` ### 2.5.3 資料快取 - **Redis**：用於快取近期行情，減少 API 呼叫。\n- **SQLite**：對於單機環境，適合小型資料集快取。\n ## 2.6 實務範例：建立「量化資料倉儲」 以下為完整流程範例，示範如何從資料下載、清洗、儲存，到查詢。 ```python import pandas as pd import yfinance as yf import pyarrow.parquet as pq import os # 1. 下載資料 symbols = ['AAPL', 'MSFT', 'GOOG'] all_data = [] for s in symbols: df = yf.download(s, start='2015-01-01', end='2023-12-31', interval='1d') df.reset_index(inplace=True) df['ticker'] = s all_data.append(df) # 2. 合併並清洗 prices = pd.concat(all_data, ignore_index=True) prices.dropna(subset=['Adj Close'], inplace=True) prices['datetime'] = pd.to_datetime(prices['Date']) prices.drop(['Date', 'Open', 'High', 'Low', 'Close', 'Adj Close', 'Volume'], axis=1, inplace=True) # 3. 儲存為 Parquet output_dir = 'data/warehouse/prices' os.makedirs(output_dir, exist_ok=True) prices.to_parquet(os.path.join(output_dir, 'prices.parquet'), index=False, compression='snappy') # 4. 查詢範例 df = pd.read_parquet(os.path.join(output_dir, 'prices.parquet')) # 取得 AAPL 2020 年 6 月的資料 mask = (df['ticker'] == 'AAPL') & (df['datetime'] >= '2020-06-01') & (df['datetime'] < '2020-07-01') print(df.loc[mask]) ``` > **實務提醒**：在實盤環境中，需考慮資料同步頻率與交易時段。建議使用事件驅動型資料管道（如 Kafka）搭配即時快取。 ## 2.7 小結 本章闡述了量化投資資料的全流程：從多元資料來源、嚴謹的品質管控，到高效的時間序列處理與資料倉儲設計，並提供了實際可執行的 API 獲取與快取機制。透過這些基礎，讀者可自行構建可維護、可擴充的資料基礎設施，為後續因子挖掘與機器學習打下堅實基礎。