架構與設計

資料流程

本頁內容8

呼叫 data.get() 之後，資料從哪裡來、什麼時候更新、怎麼存在你的電腦？這一頁說明整個流程，以及讀資料最省時間的寫法。

資料怎麼到你手上

第一次讀一個資料集時從雲端下載，存進本機快取；之後直接讀快取，雲端有新版本才再下載。

資料來源與更新時間

類別	資料來源	更新頻率	資料範圍
股價	證交所／櫃買中心	每日 18:00	2007 年至今
財報	公開資訊觀測站	每季財報公布後	2000 年至今
月營收	公開資訊觀測站	每月 10 日後	2000 年至今
籌碼	證交所	每日 18:00	2010 年至今
技術指標	FinLab 計算	即時計算	依股價資料

每天的更新時程（台灣時間）：

類別	資料來源	更新頻率	資料範圍
股價（`us_price`）	NYSE、NASDAQ 與 NYSE American 上市股票	每日美股收盤後	2016 年至今
ETF 與基金價格（`us_fund_price`）	美國上市的 ETF 與共同基金	每日美股收盤後	2016 年至今
財報（`us_income_statement`、`us_balance_sheet`、`us_cash_flow`）	SEC 10-Q／10-K 申報	每次申報後，日期對齊申報日	2015 年至今
技術指標	FinLab 計算	即時計算	依股價資料

美股股價與財報每天在美股收盤後更新一次。

本機快取：下載一次，之後讀硬碟

快取存在哪裡

下載的資料快取在 ~/finlab_db（Google Colab 掛載雲端硬碟時為 My Drive/Colab_tmp/finlab_db）。要改用其他目錄：

from finlab import data

data.set_storage(data.FileStorage('/path/to/custom/cache'))

資料更新後怎麼拿到新版本

每次呼叫 data.get()，FinLab 會比對本機快取和雲端公告的版本，決定直接用快取、只補最近的資料，還是完整重新下載：

公告版本來自 fdata 發布的共用索引，最多每 5 分鐘確認一次；確認結果存在本機的預設資料目錄（~/finlab_db），同一台電腦 5 分鐘內新開的行程直接沿用，不再連線。索引取不到時改問簽名服務。

資料更新或修正後，約 5 分鐘內重新呼叫 data.get 即可取得新版本（finlab 2.2.5 起）。

免費帳號的快取

免費帳號拿到的資料與付費帳號不同（台股只到免費期限），所以存在預設資料目錄下的 free 子資料夾，同一台電腦的每個行程共用，付費帳號不會讀到。

免費帳號不看公告版本索引，每次讀取都向簽名服務報上快取的時間：雲端沒有更新的檔案時直接用本機快取，不下載，也不計入每日下載額度；有更新才重新下載。

大型資料表的增量更新

大型完整表（例如 broker_transactions、important_info_announcement）另外發布增量檔 <資料集>.delta：快取時間不早於它的 full_refresh_before 時，只下載最近有變動的日子，替換快取中這幾天的列，再以整表雜湊驗證；更舊的快取先下載完整表再套用增量。

雜湊不符就改為完整下載，不會回傳不一致的資料。

釋放記憶體或清除快取

from finlab import data

# 釋放記憶體中的資料（硬碟上的快取保留）
data.free_memory()

# 刪除本機快取並恢復預設設定（之後每個資料集都要重新完整下載，快取損壞時才需要）
# data.clear()

讀資料更快的寫法

一次載入，重複使用

不好的寫法：每次回測都呼叫 data.get()。

# 執行 100 次回測，每次都載入資料（慢！）
for param in range(100):
    close = data.get('price:收盤價')  # 重複載入
    position = close > close.average(param)
    report = sim(position)

好的寫法：只載入一次，重複使用。

from finlab import data
from finlab.backtest import sim

# 只載入一次
close = data.get('price:收盤價')

# 執行 100 次回測
for param in range(100):
    position = close > close.average(param)
    report = sim(position)

只取需要的日期

# 只取 2020 年後的資料
close = data.get('price:收盤價')
close_recent = close[close.index >= '2020-01-01']

# 或使用 loc
close_recent = close.loc['2020-01-01':]

只取需要的股票

# 只取特定股票
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]

一次算完所有股票

不好的寫法：一檔一檔重新取資料、重新計算。

for stock in ['2330', '2317', '2454']:
    close = data.get('price:收盤價')[stock]
    ma20 = close.rolling(20).mean()  # 每次都重新計算

好的寫法：

close = data.get('price:收盤價')[['2330', '2317', '2454']]
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='all')

異常值：漲停鎖死

# 移除漲跌停鎖死的資料
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]

常用的資料表

from finlab import data

# 股價
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:ROE稅後')  # 股東權益報酬率
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:融資使用率')  # 融資使用率

from finlab import data

data.set_market('us')

# 股價
adj_close = data.get('price:adj_close')  # 還原股價（考慮股利與股票分割）
raw_close = data.get('price:close')  # 原始股價
volume = data.get('price:volume')  # 成交量（股）
market_value = data.get('etl:us_market_value')  # 市值（美元）

# 財報：每季財報對齊 SEC 申報日
eps = data.get('us_income_statement:eps_diluted')  # 稀釋每股盈餘

# 股東權益報酬率
net_income = data.get('us_income_statement:net_income')
equity = data.get('us_balance_sheet:total_stockholders_equity')
roe = net_income / equity

# 營業利益率
revenue = data.get('us_income_statement:revenue')
operating_margin = data.get('us_income_statement:operating_income') / revenue

# 營收年增率
rev_yoy = revenue.pct_change(4)

月營收、法人買賣超與融資融券只有台股有。

存成 CSV、再讀回來

close = data.get('price:收盤價')

# 儲存整個 DataFrame
close.to_csv('close_prices.csv')

# 儲存特定股票
close[['2330', '2317']].to_csv('selected_stocks.csv')

讀回來後轉成 FinlabDataFrame，就能繼續用 FinLab 的方法：

import pandas as pd
from finlab.dataframe import FinlabDataFrame

# 載入 CSV
close = pd.read_csv('close_prices.csv', index_col=0, parse_dates=True)

# 轉換為 FinlabDataFrame
close = FinlabDataFrame(close)

# 現在可以使用 FinLab 的方法
ma20 = close.average(20)

下一步

下載資料 data.get() 的完整用法，以及怎麼找資料表名稱。
finlab.data 資料下載與快取設定的 API 參考。
FinlabDataFrame 讀進來的表能用哪些方法。
模組架構資料交給回測、分析與下單模組之後的流程。

這一頁有幫到你嗎？