finlab.dataframe
FinlabDataFrame 是 pandas DataFrame 的擴充類別，專為股票策略開發設計。
使用情境
- 自動對齊不同頻率的資料（日、週、月、季）
- 快速進行選股操作（排名、篩選前 N 名）
- 計算移動平均與技術指標
- 管理進出場訊號（持倉至停損/停利）
- 產業中性化（neutralize）
- 跨股票因子預處理（
df.cs：排名、縮尾、標準化） - 產業內轉換與聚合（
df.sector：產業內排名、平均、標準差） - 投資組合權重建構（
df.weight：反波動率、風險平價、回撤控制）
快速範例
自動對齊與選股
from finlab import data
# data.get() 回傳的都是 FinlabDataFrame
close = data.get('price:收盤價') # 日頻
revenue = data.get('monthly_revenue:當月營收') # 月頻
# 自動對齊日期（revenue 會自動填補到日頻）
position = (close > close.average(20)) & (revenue > 1e8)
排名與篩選
from finlab import data
close = data.get('price:收盤價')
marketcap = data.get('etl:market_value')
# 選擇市值最小的 30 檔
small_cap = marketcap.is_smallest(30)
# 選擇漲幅最大的 20 檔
momentum = (close / close.shift(20)).is_largest(20)
# 組合策略
position = small_cap & momentum
持倉管理
from finlab import data
from finlab.backtest import sim
close = data.get('price:收盤價')
position = close > close.average(20)
# 持倉至停損 10% 或停利 20%
position_with_stops = position.hold_until(stop_loss=0.1, take_profit=0.2)
# 回測
report = sim(position_with_stops, resample='M')
report.display()
FinlabDataFrame vs pandas DataFrame
| 特性 | pandas DataFrame | FinlabDataFrame |
|---|---|---|
| 基本操作 | ✅ 完全支援 | ✅ 完全支援（繼承 pandas） |
| 自動對齊 | ❌ 需手動處理 | ✅ 自動對齊不同頻率資料 |
| 選股方法 | ❌ 需自行撰寫 | ✅ 內建 is_largest(), is_smallest() |
| 技術指標 | ❌ 需額外套件 | ✅ 內建 average(), rise(), fall() |
| 持倉管理 | ❌ 需自行撰寫 | ✅ 內建 hold_until(), exit_when() |
| 產業分析 | ❌ 需自行撰寫 | ✅ 內建 industry_rank(), neutralize_industry() |
| 因子預處理 | ❌ 需自行撰寫 | ✅ 內建 df.cs 跨股票轉換 |
| 產業內分析 | ❌ 需自行撰寫 | ✅ 內建 df.sector 產業內轉換與聚合 |
| 權重建構 | ❌ 需自行撰寫 | ✅ 內建 df.weight 8 種加權方法 |
API Reference
FinlabDataFrame
finlab.dataframe.FinlabDataFrame
Bases: IndexConversionMixin, SignalMixin, NeutralizeMixin, IndustryMixin, DataFrame
回測語法糖
除了使用熟悉的 Pandas 語法外，我們也提供很多語法糖，讓大家開發程式時，可以用簡易的語法完成複雜的功能，讓開發策略更簡潔！
我們將所有的語法糖包裹在 FinlabDataFrame 中，用起來跟 pd.DataFrame 一樣，但是多了很多功能！
只要使用 finlab.data.get() 所獲得的資料，皆為 FinlabDataFrame 格式，
接下來我們就來看看， FinlabDataFrame 有哪些好用的語法糖吧！
當資料日期沒有對齊（例如: 財報 vs 收盤價 vs 月報）時，在使用以下運算符號：
+, -, *, /, >, >=, ==, <, <=, &, |, ~，
不需要先將資料對齊，因為 FinlabDataFrame 會自動幫你處理，以下是示意圖。
以下是範例：cond1 與 cond2 分別為「每天」，和「每季」的資料，假如要取交集的時間，可以用以下語法：
from finlab import data
# 取得 FinlabDataFrame
close = data.get('price:收盤價')
roa = data.get('fundamental_features:ROA稅後息前')
# 運算兩個選股條件交集
cond1 = close > 37
cond2 = roa > 0
cond_1_2 = cond1 & cond2
cond1 跟 cond2 訊號的頻率雖然不相同，
但是由於 cond1 跟 cond2 是 FinlabDataFrame，所以可以直接取交集，而不用處理資料頻率對齊的問題。
總結來說，FinlabDataFrame 與一般 dataframe 唯二不同之處：
1. 多了一些 method，如df.is_largest(), df.sustain()...等。
2. 在做四則運算、不等式運算前，會將 df1、df2 的 index 取聯集，column 取交集。
average
取 n 筆移動平均
若股票在時間窗格內，有 N/2 筆 NaN，則會產生 NaN。 Args: n (positive-int): 設定移動窗格數。 Returns: (pd.DataFrame): data Examples: 股價在均線之上
只需要簡單的語法，就可以將其中一部分的訊號繪製出來檢查：import matplotlib.pyplot as plt
close.loc['2021', '2330'].plot()
sma.loc['2021', '2330'].plot()
cond.loc['2021', '2330'].mul(20).add(500).plot()
plt.legend(['close', 'sma', 'cond'])
fall
Args:
n (positive-int): 設定比較前第n筆低。
Returns:
(pd.DataFrame): data
Examples:
收盤價是否低於10日前股價
is_largest
Args:
n (positive-int): 設定每列前 n 筆大的數值。
Returns:
(pd.DataFrame): data
Examples:
每季 ROA 前 10 名的股票
is_smallest
lazy
Wrap this DataFrame into a lazy computation handle.
Returns a LazyWide proxy that records operations without
executing them. Call .rebalance(freq) or .collect()
on the result to trigger execution at specific dates only.
Example::
close = data.get('price:收盤價')
position = (
close.lazy()
.rolling(60).mean()
.rank(axis=1, pct=True)
.is_largest(10)
.rebalance('M')
)
quantile_row
rank
Cross-sectional or time-series ranking.
| PARAMETER | DESCRIPTION |
|---|---|
valid
|
只有 valid 為 True 的 cell 才參與排名。 valid 為 False/NaN 的 cell 在排名前設為 NaN， 因此不影響 pct=True 的分母。 常見用法：fillna 後傳入原始 notna() mask， 避免新上市或歷史不足的股票污染百分位排名。
TYPE:
|
rebalance
Downsample to keep only the last observation per period.
Equivalent to self.resample(freq).last().
| PARAMETER | DESCRIPTION |
|---|---|
freq
|
Pandas frequency string, e.g.
TYPE:
|
FinlabDataFrame 主要方法：
選股與排名
| 方法 | 說明 | 範例 |
|---|---|---|
is_largest(n) |
選擇最大的 N 個值 | marketcap.is_largest(30) |
is_smallest(n) |
選擇最小的 N 個值 | pe_ratio.is_smallest(20) |
rank() |
計算排名（支援產業排名） | close.rank(pct=True) |
industry_rank() |
產業內排名 | close.industry_rank() |
技術指標
| 方法 | 說明 | 範例 |
|---|---|---|
average(n) |
N 期移動平均 | close.average(20) |
rise(n) |
連續 N 期上漲 | close.rise(3) |
fall(n) |
連續 N 期下跌 | close.fall(2) |
sustain(n) |
條件持續 N 期 | (close > ma20).sustain(5) |
持倉管理
| 方法 | 說明 | 範例 |
|---|---|---|
hold_until() |
持倉至停損/停利 | position.hold_until(stop_loss=0.1) |
exit_when() |
遇到條件出場 | position.exit_when(close < ma20) |
entry_price() |
取得進場價格 | position.entry_price('close') |
產業分析
| 方法 | 說明 | 範例 |
|---|---|---|
neutralize_industry() |
產業中性化 | factor.neutralize_industry() |
groupby_category() |
按產業分組 | close.groupby_category().mean() |
average()
計算移動平均。
使用範例：
from finlab import data
close = data.get('price:收盤價')
# 20 日均線
ma20 = close.average(20)
# 60 日均線
ma60 = close.average(60)
# 均線交叉策略
golden_cross = ma20 > ma60
與 pandas rolling() 的差異
average(20)等同於rolling(20).mean()- 更簡潔易讀
- 回傳值仍為 FinlabDataFrame，可繼續串接其他方法
is_largest() / is_smallest()
選擇排名前/後 N 名。
使用範例：
from finlab import data
close = data.get('price:收盤價')
marketcap = data.get('etl:market_value')
# 選擇市值最小的 30 檔（小型股）
small_cap = marketcap.is_smallest(30)
# 選擇漲幅最大的 20 檔（動能股）
momentum = (close / close.shift(20)).is_largest(20)
# 組合：小型股中的動能股
position = small_cap & momentum
注意事項
is_largest(n)每期（每日/每月）都會重新排名- 若股票數少於 N，則全選
- 回傳 Boolean DataFrame（True 表示選中）
rank()
計算排名（支援百分比排名）。
使用範例：
from finlab import data
close = data.get('price:收盤價')
marketcap = data.get('etl:market_value')
# 計算百分位排名（0-1）
cap_rank = marketcap.rank(pct=True, axis=1)
# 選擇市值前 30%
position = cap_rank > 0.7
# 或使用 is_largest（更直觀）
position = marketcap.is_largest(int(len(marketcap.columns) * 0.3))
industry_rank()
產業內排名。
使用範例：
from finlab import data
close = data.get('price:收盤價')
# 計算產業內排名（0-1）
industry_rank = close.industry_rank()
# 選擇各產業漲幅前 30%
position = industry_rank > 0.7
產業分類來源
- 使用 TEJ 產業分類
- 自動處理產業變更
rise() / fall()
判斷連續上漲/下跌。
使用範例：
from finlab import data
close = data.get('price:收盤價')
# 連續 3 日上漲
rising = close.rise(3)
# 連續 2 日下跌
falling = close.fall(2)
# 策略：連續上漲後反手做空
position = ~rising # 反向訊號
sustain()
判斷條件持續 N 期。
使用範例：
from finlab import data
close = data.get('price:收盤價')
ma20 = close.average(20)
# 連續 5 日站穩 20 日均線
sustained_above_ma = (close > ma20).sustain(5)
# 策略
position = sustained_above_ma
sustain vs rise
rise(n): 連續 n 期數值上漲sustain(n): 連續 n 期條件為 True
hold_until()
持倉至停損/停利觸發。
使用範例：
from finlab import data
from finlab.backtest import sim
close = data.get('price:收盤價')
position = close > close.average(20)
# 基礎用法：停損 10%
position_with_sl = position.hold_until(stop_loss=0.1)
# 停損 10% + 停利 20%
position_with_stops = position.hold_until(stop_loss=0.1, take_profit=0.2)
# 回測
report = sim(position_with_stops, resample='M')
report.display()
# 移動停利（trailing stop）需在 sim() 中設定
report = sim(position, resample='M', trail_stop=0.2)
重要概念
hold_until()會延長持倉至觸發條件- 停損/停利以進場價格為基準計算
- 移動停利需透過
sim(trail_stop=...)設定，從最高價回落指定比例時出場
exit_when()
遇到特定條件出場。
使用範例：
from finlab import data
close = data.get('price:收盤價')
ma20 = close.average(20)
ma60 = close.average(60)
# 進場：短均線向上突破長均線
entry = ma20 > ma60
# 出場：短均線向下跌破長均線
exit_signal = ma20 < ma60
# 持倉至出場訊號
position = entry.exit_when(exit_signal)
exit_when vs hold_until
exit_when(): 自訂出場條件（如技術指標）hold_until(): 固定停損/停利比例
neutralize_industry()
產業中性化，移除產業因子影響。
使用範例：
from finlab import data
# 原始因子（含產業影響）
factor = data.get('fundamental_features:股東權益報酬率')
# 產業中性化（移除產業平均）
neutral_factor = factor.neutralize_industry()
# 使用中性化因子
position = neutral_factor.is_largest(30)
為什麼要中性化？
- 某些產業天生 ROE 高（如金融業）
- 中性化可避免過度集中單一產業
- 提升策略穩定性
投資組合權重建構 df.weight
df.weight accessor 提供一系列權重建構方法，可直接對選股結果（Boolean 或數值 DataFrame）建構與調整投資組合權重。所有方法回傳 FinlabDataFrame，可鏈式串接。
方法一覽
| 方法 | 說明 | 範例 |
|---|---|---|
cap_industry(max_weight) |
限制每個產業的最大權重，超額按比例重新分配 | w.weight.cap_industry(0.3) |
clip_by_volume(total_fund, max_participation_ratio, volume, window) |
依日均成交金額限制個股權重 | w.weight.clip_by_volume(1e7) |
inverse_volatility(window, price) |
反波動率加權，低波動股票獲得較高權重 | w.weight.inverse_volatility(60) |
risk_parity(window, price) |
風險平價加權，使每個持股的風險貢獻相等 | w.weight.risk_parity(60) |
correlation(window, price, diversify) |
依相關性調整權重，分散化或集中化 | w.weight.correlation(diversify=True) |
target_volatility(target, window, price) |
縮放權重以達成目標年化波動率 | w.weight.target_volatility(0.15) |
limit_turnover(max_turnover) |
限制相鄰換倉的雙向換手率 | w.weight.limit_turnover(0.3) |
drawdown_control(max_drawdown, price) |
回撤超過門檻時自動縮減曝險 | w.weight.drawdown_control(0.15) |
使用範例
from finlab import data
from finlab.backtest import sim
close = data.get('price:收盤價')
marketcap = data.get('etl:market_value')
# 選股
position = marketcap.is_smallest(30)
# 鏈式建構權重
weighted = (
position
.weight.inverse_volatility(window=60) # 反波動率加權
.weight.cap_industry(max_weight=0.3) # 產業權重上限 30%
.weight.limit_turnover(max_turnover=0.3) # 換手率上限 30%
.weight.drawdown_control(max_drawdown=0.15) # 回撤控制 15%
)
report = sim(weighted, resample='M')
report.display()
權重方法的先後順序
- 建議先做加權（inverse_volatility / risk_parity），再做約束（cap_industry / clip_by_volume）
limit_turnover和drawdown_control通常放在最後
跨股票資料轉換 df.cs
df.cs（cross-sectional）accessor 提供跨股票的資料轉換方法。所有計算都是在每個日期、跨所有股票進行。
方法一覽
| 方法 | 說明 | 範例 |
|---|---|---|
rank() |
百分位排名（0, 1] | factor.cs.rank() |
winsorize(lower, upper) |
百分位縮尾處理 | factor.cs.winsorize(0.05, 0.95) |
bucket(n) |
等分位分組（1..n） | factor.cs.bucket(5) |
zscore() |
標準化（均值 0，標準差 1） | factor.cs.zscore() |
demean() |
去均值 | factor.cs.demean() |
使用範例
from finlab import data
roe = data.get('fundamental_features:股東權益報酬率')
# 因子預處理：縮尾 + 標準化
clean_roe = roe.cs.winsorize(0.05, 0.95).cs.zscore()
# 分成 5 組回測
group = roe.cs.bucket(5)
long_group = group == 5 # 做多最高組
# 百分位排名選股
position = roe.cs.rank() > 0.8 # 選前 20%
cs vs rank()
df.cs.rank()回傳 (0, 1] 的百分位排名，每個日期跨所有股票計算df.rank(pct=True, axis=1)是 pandas 原生方法，效果類似但cs.rank()更語義化
產業內資料轉換與聚合 df.sector
df.sector accessor 提供兩類功能：
- 產業內轉換：與
df.cs相同的 5 個方法（rank / winsorize / bucket / zscore / demean），但在同產業內計算 - 產業內聚合：mean / std / median / sum / min / max / count，回傳產業內的聚合統計量
轉換方法
| 方法 | 說明 | 範例 |
|---|---|---|
rank() |
產業內百分位排名 | factor.sector.rank() |
winsorize(lower, upper) |
產業內百分位縮尾 | factor.sector.winsorize(0.05, 0.95) |
bucket(n) |
產業內等分位分組 | factor.sector.bucket(3) |
zscore() |
產業內標準化 | factor.sector.zscore() |
demean() |
產業內去均值 | factor.sector.demean() |
聚合方法
| 方法 | 說明 | 範例 |
|---|---|---|
mean() |
產業內平均值 | roe.sector.mean() |
std(ddof=1) |
產業內標準差 | roe.sector.std() |
median() |
產業內中位數 | roe.sector.median() |
sum() |
產業內總和 | revenue.sector.sum() |
min() |
產業內最小值 | close.sector.min() |
max() |
產業內最大值 | close.sector.max() |
count() |
產業內有效值數量 | roe.sector.count() |
使用範例
from finlab import data
roe = data.get('fundamental_features:股東權益報酬率')
close = data.get('price:收盤價')
# 產業內排名選股（各產業內 ROE 前 20%）
position = roe.sector.rank() > 0.8
# 產業內標準化（消除產業差異）
neutralized_roe = roe.sector.zscore()
# 取得產業平均 ROE 做比較
sector_avg_roe = roe.sector.mean()
above_avg = roe > sector_avg_roe # 高於產業平均的股票
sector vs neutralize_industry()
df.sector.zscore()等同於neutralize_industry()的效果（產業中性化）df.sector提供更多彈性：除了中性化，還能做排名、分組、聚合
完整範例：多重篩選策略
from finlab import data
from finlab.backtest import sim
# 步驟 1：載入資料
close = data.get('price:收盤價')
marketcap = data.get('etl:market_value')
revenue = data.get('monthly_revenue:當月營收')
roe = data.get('fundamental_features:股東權益報酬率')
# 步驟 2：定義篩選條件
cond1 = marketcap.is_smallest(100) # 小型股
cond2 = (revenue.average(3) / revenue.average(12)) > 1.2 # 營收成長
cond3 = roe.rank(pct=True, axis=1) > 0.7 # ROE 前 30%
cond4 = (close > close.average(20)).sustain(3) # 連續 3 日站穩 20MA
# 步驟 3：組合條件
position = cond1 & cond2 & cond3 & cond4
# 步驟 4：加入停損停利
position = position.hold_until(stop_loss=0.1, take_profit=0.2)
# 步驟 5：限制持股數
position = position.is_largest(20) # 最多 20 檔
# 步驟 6：回測
report = sim(position, resample='M')
report.display()
# 步驟 7：分析績效
metrics = report.get_metrics()
print(f"年化報酬: {metrics['annual_return']:.2%}")
print(f"夏普率: {metrics['daily_sharpe']:.2f}")
print(f"最大回撤: {metrics['max_drawdown']:.2%}")
常見問題
Q: FinlabDataFrame 如何自動對齊不同頻率的資料？
from finlab import data
# 日頻資料
close = data.get('price:收盤價') # 每天更新
print(close.index[0]) # 2010-01-04
# 月頻資料
revenue = data.get('monthly_revenue:當月營收') # 每月 10 號公告
print(revenue.index[0]) # 2010-01-10
# 直接運算會自動對齊（revenue 向前填補到日頻）
position = (close > 100) & (revenue > 1e8)
# position 是日頻資料，revenue 已自動填補
對齊規則： - 低頻 → 高頻：向前填補（forward fill） - 季報（財報）：自動延遲至財報公告日
Q: 如何串接多個 FinlabDataFrame 方法？
from finlab import data
close = data.get('price:收盤價')
# 串接方法（類似 pandas method chaining）
position = (
close
.average(20) # 計算 20MA
.rank(pct=True, axis=1) # 排名
.is_largest(30) # 取前 30 名
)
# 等同於
ma20 = close.average(20)
ranked = ma20.rank(pct=True, axis=1)
position = ranked.is_largest(30)
Q: is_largest() 和 rank() + 閾值有什麼差別？
from finlab import data
close = data.get('price:收盤價')
# 方法 1：is_largest（固定數量）
position1 = close.is_largest(30) # 永遠選 30 檔
# 方法 2：rank + 閾值（固定比例）
position2 = close.rank(pct=True, axis=1) > 0.8 # 選前 20%（數量會變）
# 差異：
# - 方法 1：持股數固定（適合固定資金配置）
# - 方法 2：持股數浮動（適合市場狀況調整）
Q: 移動停利（trail_stop）如何運作？
移動停利需透過 sim() 的 trail_stop 參數設定，hold_until() 不支援此功能。
from finlab import data
from finlab.backtest import sim
close = data.get('price:收盤價')
position = close > close.average(20)
# 範例：進場價 100 元
# - stop_loss=0.1: 跌破 90 元出場
# - take_profit=0.2: 漲到 120 元出場（固定停利）
# - trail_stop=0.2: 從最高價回落 20% 出場（移動停利）
# 情境：進場 100 → 漲到 150 → 回落到 120
# - take_profit=0.2: 漲到 120 即出場，不會等到 150
# - trail_stop=0.2: 從最高價 150 回落 20% = 120 時出場
# 固定停利
report_fixed = sim(position, resample='M', stop_loss=0.1, take_profit=0.2)
# 移動停利
report_trailing = sim(position, resample='M', stop_loss=0.1, trail_stop=0.2)
Q: 如何在 FinlabDataFrame 上使用 pandas 方法？
from finlab import data
close = data.get('price:收盤價')
# FinlabDataFrame 完全兼容 pandas
close.rolling(20).mean() # pandas 方法
close.fillna(0) # pandas 方法
close.dropna() # pandas 方法
# 所有 pandas 方法都可用
# 運算結果仍為 FinlabDataFrame（保留擴充功能）
參考資源
- 選股條件教學 - FinlabDataFrame 實戰應用
- 歷史回測教學 - 完整回測流程
- 技術指標選股範例 - 實戰策略範例
- pandas 官方文檔 - pandas 基礎知識