finlab.tools
finlab.tools.event_study
create_factor_data
create factor data, which contains future return
PARAMETER | DESCRIPTION |
---|---|
factor |
factor data where index is datetime and columns is asset id
TYPE:
|
adj_close |
adj close where index is datetime and columns is asset id
TYPE:
|
days |
future return considered
TYPE:
|
Return
Analytic plots and tables
Warning
This function is not identical to finlab.ml.alphalens.create_factor_data
Examples:
from finlab.tools.event_study import create_factor_data
from finlab.tools.event_study import event_study
factor = data.get('price_earning_ratio:股價淨值比')
adj_close = data.get('etl:adj_close')
benchmark = data.get('benchmark_return:發行量加權股價報酬指數')
# create event dataframe
dividend_info = data.get('dividend_announcement')
v = dividend_info[['stock_id', '除權交易日']].set_index(['stock_id', '除權交易日'])
v['value'] = 1
event = v[~v.index.duplicated()].reset_index().drop_duplicates(
subset=['stock_id', '除權交易日']
).pivot(index='除權交易日', columns='stock_id', values='value').notna()
# calculate factor_data
factor_data = create_factor_data({'pb':factor}, adj_close, event=event)
r = event_study(factor_data, benchmark, adj_close)
plt.bar(r.columns, r.mean().values)
plt.plot(r.columns, r.mean().cumsum().values)
event_study
event_study(factor_data, benchmark_adj_close, stock_adj_close, sample_period=(-45, -20), estimation_period=(-5, 20), plot=True)
Run event study and returns the abnormal returns of each stock on each day.
PARAMETER | DESCRIPTION |
---|---|
factor_data |
factor data where index is datetime and columns is asset id
TYPE:
|
benchmark_adj_close |
benchmark for CAPM
TYPE:
|
stock_adj_close |
stock price for CAPM
TYPE:
|
sample_period |
period for fitting CAPM
TYPE:
|
estimation_period |
period for calculating alpha (abnormal return)
TYPE:
|
plot |
plot the result
TYPE:
|
Return
Abnormal returns of each stock on each day.
Examples:
from finlab.tools.event_study import create_factor_data
from finlab.tools.event_study import event_study
factor = data.get('price_earning_ratio:股價淨值比')
adj_close = data.get('etl:adj_close')
benchmark = data.get('benchmark_return:發行量加權股價報酬指數')
# create event dataframe
dividend_info = data.get('dividend_announcement')
v = dividend_info[['stock_id', '除權交易日']].set_index(['stock_id', '除權交易日'])
v['value'] = 1
event = v[~v.index.duplicated()].reset_index().drop_duplicates(
subset=['stock_id', '除權交易日']
).pivot(index='除權交易日', columns='stock_id', values='value').notna()
# calculate factor_data
factor_data = create_factor_data({'pb':factor}, adj_close, event=event)
r = event_study(factor_data, benchmark, adj_close)
plt.bar(r.columns, r.mean().values)
plt.plot(r.columns, r.mean().cumsum().values)
finlab.tools.factor_analysis
calc_centrality
對指定的時間序列數據計算滾動資產集中度。
此函式為通用函式,可應用於任何以時間為索引、資產為欄位的 DataFrame(例如因子報酬)。
它現在是頻率無關的,滾動窗口由整數 window_periods
指定。
PARAMETER | DESCRIPTION |
---|---|
return_df |
一個時間序列 DataFrame,索引為日期,欄位為資產(例如因子名稱)。
雖然參數名稱為
TYPE:
|
window_periods |
滾動窗口的長度,以數據點的「數量」計。例如,如果
TYPE:
|
n_components |
用於計算的 PCA 主成分數量。
TYPE:
|
RETURNS | DESCRIPTION |
---|---|
DataFrame
|
pd.DataFrame: 包含滾動集中度分數的 DataFrame。 |
Example
date | FactorA | FactorB |
---|---|---|
2025-01-01 | 0.1 | 0.2 |
2025-01-02 | 0.1 | 0.2 |
2025-01-03 | 0.1 | 0.2 |
2025-01-04 | 0.1 | 0.2 |
2025-01-05 | 0.1 | 0.2 |
calc_factor_return
計算基於特徵和標籤的等權重投資組合週期表現。
此函式是因子績效計算的核心引擎,接受預先準備好的特徵和標籤, 然後計算每個因子的投資組合報酬。
函式會自動處理以下流程: 1. 驗證所有特徵都是布林值 2. 對每個因子計算等權重投資組合的週期報酬 3. 自動裁剪掉第一個非空行之前的數據
PARAMETER | DESCRIPTION |
---|---|
features |
特徵 DataFrame,索引為日期,欄位為因子名稱,值應為布林值。
TYPE:
|
labels |
標籤 Series,索引為日期,值為超額報酬。
TYPE:
|
RETURNS | DESCRIPTION |
---|---|
DataFrame
|
pd.DataFrame: 一個索引為日期、欄位為各因子名稱的 DataFrame,其值為每個週期的等權重投組表現。 輸出會自動從第一個非空行開始,確保數據完整性。 |
RAISES | DESCRIPTION |
---|---|
ValueError
|
如果特徵不是布林值。 |
Example
from finlab import data from finlab.tools.factor_analysis import calc_factor_return, generate_features_and_labels
price = data.get('etl:adj_close') marketcap = data.get('etl:market_value') revenue = data.get('monthly_revenue:當月營收')
先生成特徵和標籤
features, labels = generate_features_and_labels({ ... 'marketcap': marketcap.rank(pct=True, axis=1) < 0.3, ... 'revenue': (revenue.average(3) / revenue.average(12)).rank(pct=True, axis=1) < 0.3, ... 'momentum': price / price.shift(20) - 1 > 0 ... }, resample=revenue.index)
計算因子報酬
factor_return = calc_factor_return(features, labels)
輸出範例
print(factor_return.head()) | datetime | marketcap | revenue | momentum | |:--------------------|------------:|------------:|------------:| | 2013-04-30 00:00:00 | 0.018 | -0.005 | 0.009 | | 2013-05-31 00:00:00 | 0.004 | -0.003 | -0.001 | | 2013-06-30 00:00:00 | -0.013 | -0.006 | 0.023 | | 2013-07-31 00:00:00 | 0.007 | -0.007 | 0.001 | | 2013-08-31 00:00:00 | 0.014 | -0.003 | -0.005 |
calc_ic
計算特徵與標籤之間的相關係數(IC),可選擇是否對特徵進行排名。
PARAMETER | DESCRIPTION |
---|---|
features |
特徵資料,索引為MultiIndex(日期, 股票代碼),欄位為因子名稱。
TYPE:
|
labels |
標籤資料,索引為MultiIndex(日期, 股票代碼)。
TYPE:
|
rank |
是否對特徵進行排名。預設為False。
TYPE:
|
RETURNS | DESCRIPTION |
---|---|
pd.DataFrame: 每個日期、每個因子的IC值。 |
calc_metric
計算因子
PARAMETER | DESCRIPTION |
---|---|
factor |
因子
TYPE:
|
adj_close |
股價
TYPE:
|
days |
預測天數. Defaults to [10, 20, 60, 120].
TYPE:
|
func |
計算函數. Defaults to corr.
TYPE:
|
RETURNS | DESCRIPTION |
---|---|
pd.DataFrame: 因子計算結果 |
Example
factor = data.indicator('RSI') adj_close = data.get('etl:adj_close') calc_metric(factor, adj_close)
date | factor_10 | factor_20 | factor_60 | factor_120 |
---|---|---|---|---|
2010-01-01 | 0.1 | 0.2 | 0.3 | 0.4 |
2010-01-02 | 0.1 | 0.2 | 0.3 | 0.4 |
2010-01-03 | 0.1 | 0.2 | 0.3 | 0.4 |
2010-01-04 | 0.1 | 0.2 | 0.3 | 0.4 |
2010-01-05 | 0.1 | 0.2 | 0.3 | 0.4 |
calc_regression_stats
對 DataFrame 中的每個時間序列進行線性回歸,並回傳原始統計數據。
此函式使用純 pandas 操作進行向量化計算,不依賴 scipy。
PARAMETER | DESCRIPTION |
---|---|
df |
時間序列 DataFrame,索引為 DatetimeIndex,欄位為不同的指標序列。
TYPE:
|
p_value_threshold |
p 值閾值,用於判斷趨勢的統計顯著性。預設為 0.05。
TYPE:
|
r_squared_threshold |
R² 閾值,用於判斷趨勢的解釋力。預設為 0.1。
TYPE:
|
RETURNS | DESCRIPTION |
---|---|
DataFrame
|
pd.DataFrame: 線性回歸的統計結果,包含以下欄位: - slope: 線性回歸斜率 - p_value: 斜率的 p 值 - r_squared: 決定係數 (R²) - tail_estimate: 時間序列尾部的估計值 - trend: 趨勢分類 ("up", "down", "flat") |
Example
假設 ic_df 是一個包含 IC 時間序列的 DataFrame
ic_df = calc_ic(features, labels)
1. 計算回歸統計數據
trend_stats = calc_regression_stats(ic_df) print(trend_stats)
2. 基於回傳結果進行客製化分析
範例:找出統計顯著 (p-value < 0.05) 且趨勢向上 (slope > 0) 的因子
significant_up_trend = trend_stats[ ... (trend_stats['p_value'] < 0.05) & (trend_stats['slope'] > 0) ... ] print(significant_up_trend)
3. 查看趨勢分類結果
up_trends = trend_stats[trend_stats['trend'] == 'up'] down_trends = trend_stats[trend_stats['trend'] == 'down'] flat_trends = trend_stats[trend_stats['trend'] == 'flat']
calc_shapley_values
計算因子的 Shapley 值,用於評估每個因子對投資組合表現的邊際貢獻。
Shapley 值是一種合作博弈論中的概念,用於公平分配聯盟的總收益給各個參與者。 在因子分析中,我們將每個因子視為一個"參與者",投資組合的報酬視為"聯盟的總收益"。
計算過程: 1. 對所有可能的因子組合(從單一因子到全部因子) 2. 計算每個組合的投資組合報酬 3. 根據 Shapley 值公式計算每個因子的邊際貢獻
PARAMETER | DESCRIPTION |
---|---|
features |
特徵 DataFrame,索引為日期,欄位為因子名稱,值應為布林值。 每個因子代表一個投資策略(True 表示選中該股票)。
TYPE:
|
labels |
標籤 Series,索引為日期,值為超額報酬。 應為 MultiIndex,包含 'datetime' 和 'stock_id' 層級。
TYPE:
|
RETURNS | DESCRIPTION |
---|---|
DataFrame
|
pd.DataFrame: 包含每個因子 Shapley 值的 DataFrame。 索引為日期,欄位為因子名稱,值為該因子的 Shapley 值。 |
RAISES | DESCRIPTION |
---|---|
ValueError
|
|
Example
from finlab import data from finlab.tools.factor_analysis import calc_shapley_values, generate_features_and_labels
price = data.get('etl:adj_close') marketcap = data.get('etl:market_value') revenue = data.get('monthly_revenue:當月營收')
生成特徵和標籤
features, labels = generate_features_and_labels({ ... 'marketcap': marketcap.rank(pct=True, axis=1) < 0.3, ... 'revenue': (revenue.average(3) / revenue.average(12)).rank(pct=True, axis=1) < 0.3, ... 'momentum': price / price.shift(20) - 1 > 0 ... }, resample=revenue.index)
計算 Shapley 值
shapley_df = calc_shapley_values(features, labels)
print(shapley_df.head()) | datetime | marketcap | revenue | momentum | |:--------------------|------------:|------------:|------------:| | 2013-04-30 00:00:00 | 0.012 | -0.003 | 0.006 | | 2013-05-31 00:00:00 | 0.002 | -0.001 | -0.001 | | 2013-06-30 00:00:00 | -0.008 | -0.004 | 0.015 | | 2013-07-31 00:00:00 | 0.004 | -0.004 | 0.001 | | 2013-08-31 00:00:00 | 0.009 | -0.002 | -0.003 |
Note
- Shapley 值的計算複雜度為 O(2^n),其中 n 為因子數量
- 對於大量因子,計算時間可能較長
- 建議因子數量不超過 10 個以確保合理的計算時間
generate_features_and_labels
生成因子特徵和標籤,這是因子分析的核心步驟。
此函式封裝了因子分析中特徵和標籤生成的標準流程:
1. 使用 finlab.ml.feature.combine
將因子字典轉換為特徵 DataFrame
2. 使用 finlab.ml.label.excess_over_mean
生成超額報酬標籤
PARAMETER | DESCRIPTION |
---|---|
dfs |
因子字典,包含因子名稱和對應的因子數據或計算函式。
此為
TYPE:
|
resample |
重採樣頻率,用於特徵和標籤的生成。 例如: 'M' (月度), 'Q' (季度), 'Y' (年度)
TYPE:
|
RETURNS | DESCRIPTION |
---|---|
tuple[DataFrame, Series]
|
tuple[pd.DataFrame, pd.Series]: - pd.DataFrame: 特徵 DataFrame,索引為日期,欄位為因子名稱 - pd.Series: 標籤 Series,索引為日期,值為超額報酬 |
RAISES | DESCRIPTION |
---|---|
ValueError
|
如果輸入的因子字典為空或無效。 |
Example
from finlab import data from finlab.tools.factor_analysis import generate_features_and_labels
price = data.get('etl:adj_close') marketcap = data.get('etl:market_value') revenue = data.get('monthly_revenue:當月營收')
features, labels = generate_features_and_labels({ ... 'marketcap': marketcap.rank(pct=True, axis=1) < 0.3, ... 'revenue': (revenue.average(3) / revenue.average(12)).rank(pct=True, axis=1) < 0.3, ... 'momentum': price / price.shift(20) - 1 > 0 ... }, resample=revenue.index)
print(f"Features shape: {features.shape}") print(f"Labels shape: {labels.shape}") Features shape: (120, 3) Labels shape: (120,)
ic
計算因子的IC
PARAMETER | DESCRIPTION |
---|---|
factor |
因子
TYPE:
|
adj_close |
股價
TYPE:
|
days |
預測天數. Defaults to [10, 20, 60, 120].
TYPE:
|
RETURNS | DESCRIPTION |
---|---|
pd.DataFrame: 因子計算結果 |
Example
factor = data.indicator('RSI') adj_close = data.get('etl:adj_close') calc_metric(factor, adj_close)
date | factor_10 | factor_20 | factor_60 | factor_120 |
---|---|---|---|---|
2010-01-01 | 0.1 | 0.2 | 0.3 | 0.4 |
2010-01-02 | 0.1 | 0.2 | 0.3 | 0.4 |
2010-01-03 | 0.1 | 0.2 | 0.3 | 0.4 |
2010-01-04 | 0.1 | 0.2 | 0.3 | 0.4 |
2010-01-05 | 0.1 | 0.2 | 0.3 | 0.4 |
is_boolean_series
Check if a pandas Series contains boolean values, handling NaN values.
In older pandas versions, boolean Series with NaN values get converted to float dtype. This function detects such cases by checking if the non-NaN values are boolean.
PARAMETER | DESCRIPTION |
---|---|
series |
The Series to check
TYPE:
|
RETURNS | DESCRIPTION |
---|---|
bool
|
True if the Series contains boolean values (including NaN), False otherwise
TYPE:
|
Example
import pandas as pd import numpy as np
Pure boolean
s1 = pd.Series([True, False, True]) is_boolean_series(s1) # True
Boolean with NaN (becomes float in old pandas)
s2 = pd.Series([True, False, np.nan]) is_boolean_series(s2) # True
Non-boolean
s3 = pd.Series([1, 2, 3]) is_boolean_series(s3) # False