FinLab 資料流程詳解
本文詳細說明 FinLab 的資料架構、來源、更新頻率、快取機制與查詢最佳實踐。
資料架構概覽
graph TB
A[資料來源] --> B[GCS Bucket]
B --> C[BigQuery]
C --> D[finlab.data.get]
D --> E{本地快取?}
E -->|有| F[返回快取資料]
E -->|無| G[從 BigQuery 下載]
G --> H[儲存至快取]
H --> F
F --> I[FinlabDataFrame]
I --> J[策略開發]
style A fill:#e1f5ff
style D fill:#fff4e1
style F fill:#e8f5e9資料來源
台股資料
| 類別 | 資料來源 | 更新頻率 | 資料範圍 |
|---|---|---|---|
| 股價 | 證交所/櫃買中心 | 每日 18:00 | 2007年至今 |
| 財報 | 公開資訊觀測站 | 每季財報公布後 | 2000年至今 |
| 月營收 | 公開資訊觀測站 | 每月 10 日後 | 2000年至今 |
| 籌碼 | 證交所 | 每日 18:00 | 2010年至今 |
| 技術指標 | FinLab 計算 | 即時計算 | 依股價資料 |
美股資料
| 類別 | 資料來源 | 更新頻率 | 資料範圍 |
|---|---|---|---|
| 股價 | Yahoo Finance | 每日 | 2010年至今 |
| 財報 | SEC EDGAR | 每季 | 2010年至今 |
資料更新時程
gantt
title 台股資料每日更新時程
dateFormat HH:mm
axisFormat %H:%M
section 股價資料
收盤 :done, 13:30, 1m
爬蟲開始 :done, 18:00, 10m
資料上傳 :done, 18:10, 20m
可下載 :crit, 18:30, 1m
section 籌碼資料
爬蟲開始 :done, 18:00, 30m
資料上傳 :done, 18:30, 30m
可下載 :crit, 19:00, 1m
section 財報資料
公布 :done, 08:00, 1m
爬蟲開始 :done, 09:00, 60m
資料上傳 :done, 10:00, 30m
可下載 :crit, 10:30, 1m資料快取機制
本地快取目錄
import finlab
# 預設快取位置
print(finlab.get_data_dir())
# 輸出: /Users/username/.finlab/data
# 自訂快取位置
finlab.set_data_dir('/path/to/custom/cache')
快取更新邏輯
flowchart TD
A[呼叫 data.get] --> B{本地有快取?}
B -->|否| C[從 BigQuery 下載]
B -->|是| D{快取是否過期?}
D -->|是| C
D -->|否| E[使用快取資料]
C --> F[儲存至本地]
F --> E
E --> G[返回 DataFrame]快取管理
from finlab import data
# 清除所有快取
data.clear_cache()
# 清除特定資料快取
data.clear_cache('price:收盤價')
# 強制重新下載(忽略快取)
close = data.get('price:收盤價', force_download=True)
資料查詢最佳實踐
1. 批次載入資料
❌ 不好的做法:每次都呼叫 data.get()
# 執行 100 次回測,每次都載入資料(慢!)
for param in range(100):
close = data.get('price:收盤價') # 重複載入
position = close > close.average(param)
report = sim(position)
✅ 好的做法:一次載入,重複使用
# 只載入一次
close = data.get('price:收盤價')
# 執行 100 次回測
for param in range(100):
position = close > close.average(param)
report = sim(position)
2. 使用日期篩選
# 只載入 2020 年後的資料
close = data.get('price:收盤價')
close_recent = close[close.index >= '2020-01-01']
# 或使用 loc
close_recent = close.loc['2020-01-01':]
3. 使用股票篩選
# 只載入特定股票
close = data.get('price:收盤價')
close_subset = close[['2330', '2317', '2454']]
# 或使用篩選條件
market_cap = data.get('etl:market_value')
large_cap = market_cap > 100_000_000_000 # 市值 > 1000 億
close_large = close[large_cap]
4. 避免重複計算
❌ 不好的做法
for stock in ['2330', '2317', '2454']:
close = data.get('price:收盤價')[stock]
ma20 = close.rolling(20).mean() # 每次都重新計算
✅ 好的做法
資料品質與處理
處理缺失值
from finlab import data
close = data.get('price:收盤價')
# 檢查缺失值
missing_ratio = close.isna().sum() / len(close)
print(f"缺失值比例:\n{missing_ratio[missing_ratio > 0.1]}")
# 方法 1: Forward fill (推薦)
close_filled = close.ffill()
# 方法 2: 移除有缺失值的股票
close_clean = close.dropna(axis=1, how='any')
# 方法 3: 移除有缺失值的日期
close_clean = close.dropna(axis=0, how='any')
處理異常值
# 移除漲跌停鎖死的資料
close = data.get('price:收盤價')
open_price = data.get('price:開盤價')
# 檢查是否漲停鎖死(開盤 = 收盤 = 漲停)
limit_up = close / close.shift() - 1 > 0.095
locked = (close == open_price) & limit_up
# 移除鎖死日期的資料
close_filtered = close[~locked]
常見資料表說明
股價資料
# 還原股價(考慮除權息)
adj_close = data.get('etl:adj_close')
# 原始股價
raw_close = data.get('price:收盤價')
# 成交量
volume = data.get('price:成交股數')
財報資料
# 每股盈餘
eps = data.get('financial_statement:每股盈餘')
# 股東權益報酬率
roe = data.get('fundamental_features:股東權益報酬率')
# 營業利益率
operating_margin = data.get('fundamental_features:營業利益率')
月營收資料
# 當月營收
rev = data.get('monthly_revenue:當月營收')
# 營收年增率
rev_yoy = data.get('monthly_revenue:去年同月增減(%)')
籌碼資料
# 投信買賣超
trust = data.get('institutional_investors_trading_summary:投信買賣超股數')
# 外資買賣超
foreign = data.get('institutional_investors_trading_summary:外資買賣超股數')
# 融資使用率
margin_ratio = data.get('margin_transactions:融資使用率')
資料儲存與分享
儲存資料至 CSV
close = data.get('price:收盤價')
# 儲存整個 DataFrame
close.to_csv('close_prices.csv')
# 儲存特定股票
close[['2330', '2317']].to_csv('selected_stocks.csv')
從 CSV 載入資料
import pandas as pd
# 載入 CSV
close = pd.read_csv('close_prices.csv', index_col=0, parse_dates=True)
# 轉換為 FinlabDataFrame
from finlab.dataframe import FinlabDataFrame
close = FinlabDataFrame(close)
# 現在可以使用 FinLab 的方法
ma20 = close.average(20)