Bases: 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
擷取 1101 台泥 的訊號如下圖,可以看到
cond1
跟
cond2
訊號的頻率雖然不相同,
但是由於
cond1
跟
cond2
是
FinlabDataFrame
,所以可以直接取交集,而不用處理資料頻率對齊的問題。
總結來說,FinlabDataFrame 與一般 dataframe 唯二不同之處:
1. 多了一些 method,如df.is_largest()
, df.sustain()
...等。
2. 在做四則運算、不等式運算前,會將 df1、df2 的 index 取聯集,column 取交集。
Source code in finlab/dataframe.py
| def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.id = uuid.uuid4().int
|
average
取 n 筆移動平均
若股票在時間窗格內,有 N/2 筆 NaN,則會產生 NaN。
Args:
n (positive-int): 設定移動窗格數。
Returns:
(pd.DataFrame): data
Examples:
股價在均線之上
from finlab import data
close = data.get('price:收盤價')
sma = close.average(10)
cond = close > sma
只需要簡單的語法,就可以將其中一部分的訊號繪製出來檢查:
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'])
Source code in finlab/dataframe.py
| def average(self, n):
"""取 n 筆移動平均
若股票在時間窗格內,有 N/2 筆 NaN,則會產生 NaN。
Args:
n (positive-int): 設定移動窗格數。
Returns:
(pd.DataFrame): data
Examples:
股價在均線之上
```py
from finlab import data
close = data.get('price:收盤價')
sma = close.average(10)
cond = close > sma
```
只需要簡單的語法,就可以將其中一部分的訊號繪製出來檢查:
```py
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'])
```
<img src="https://i.ibb.co/Mg1P85y/sma.png" alt="sma">
"""
return self.rolling(n, min_periods=int(n/2)).mean()
|
deadline
財務索引轉換成公告截止日
將財務季報 (ex:2022Q1) 從文字格式轉為公告截止日的datetime格式,
通常使用情境為對不同週期的dataframe做reindex,常用於以公告截止日作為訊號產生日。
Returns:
(pd.DataFrame): data
Examples:
data.get('financial_statement:現金及約當現金').deadline()
data.get('monthly_revenue:當月營收').deadline()
Source code in finlab/dataframe.py
| def deadline(self):
"""財務索引轉換成公告截止日
將財務季報 (ex:2022Q1) 從文字格式轉為公告截止日的datetime格式,
通常使用情境為對不同週期的dataframe做reindex,常用於以公告截止日作為訊號產生日。
Returns:
(pd.DataFrame): data
Examples:
```py
data.get('financial_statement:現金及約當現金').deadline()
data.get('monthly_revenue:當月營收').deadline()
```
"""
if len(self.index) == 0 or not isinstance(self.index[0], str):
return self
if self.index[0].find('M') != -1:
return self._index_str_to_date_month()
elif self.index[0].find('Q') != -1:
return self._index_str_to_date_season(detail=False)
raise Exception("Cannot apply deadline to dataframe. "
"Index is not compatable."
"Index should be 2013-Q1 or 2013-M1."
)
|
fall
數值下降中
取是否比前第n筆低,若符合條件的值則為True,反之為False。
Args:
n (positive-int): 設定比較前第n筆低。
Returns:
(pd.DataFrame): data
Examples:
收盤價是否低於10日前股價
from finlab import data
data.get('price:收盤價').fall(10)
Source code in finlab/dataframe.py
| def fall(self, n=1):
"""數值下降中
取是否比前第n筆低,若符合條件的值則為True,反之為False。
<img src="https://i.ibb.co/Y72bN5v/Screen-Shot-2021-10-26-at-6-43-41-AM.png" alt="Screen-Shot-2021-10-26-at-6-43-41-AM">
Args:
n (positive-int): 設定比較前第n筆低。
Returns:
(pd.DataFrame): data
Examples:
收盤價是否低於10日前股價
```py
from finlab import data
data.get('price:收盤價').fall(10)
```
"""
return self < self.shift(n)
|
groupby_category
資料按產業分群
類似 pd.DataFrame.groupby()
的處理效果。
Returns:
(pd.DataFrame): data
Examples:
半導體平均股價淨值比時間序列
from finlab import data
pe = data.get('price_earning_ratio:股價淨值比')
pe.groupby_category().mean()['半導體'].plot()
全球 2020 量化寬鬆加上晶片短缺,使得半導體股價淨值比衝高。
Source code in finlab/dataframe.py
| def groupby_category(self):
"""資料按產業分群
類似 `pd.DataFrame.groupby()`的處理效果。
Returns:
(pd.DataFrame): data
Examples:
半導體平均股價淨值比時間序列
```py
from finlab import data
pe = data.get('price_earning_ratio:股價淨值比')
pe.groupby_category().mean()['半導體'].plot()
```
<img src="https://i.ibb.co/Tq2fKBp/pbmean.png" alt="pbmean">
全球 2020 量化寬鬆加上晶片短缺,使得半導體股價淨值比衝高。
"""
from finlab import data
categories = data.get('security_categories')
cat = categories.set_index('stock_id').category.to_dict()
org_set = set(cat.values())
set_remove_illegal = set(
o for o in org_set if isinstance(o, str) and o != 'nan')
set_remove_illegal
refine_cat = {}
for s, c in cat.items():
if c == None or c == 'nan':
refine_cat[s] = '其他'
continue
if c == '電腦及週邊':
refine_cat[s] = '電腦及週邊設備業'
continue
if c[-1] == '業' and c[:-1] in set_remove_illegal:
refine_cat[s] = c[:-1]
else:
refine_cat[s] = c
col_categories = pd.Series(self.columns.map(
lambda s: refine_cat[s] if s in cat else '其他'))
return self.groupby(col_categories.values, axis=1)
|
hold_until
hold_until(exit, nstocks_limit=None, stop_loss=-np.inf, take_profit=np.inf, trade_at='close', rank=None, market='AUTO')
訊號進出場
這大概是所有策略撰寫中,最重要的語法糖,上述語法中 entries
為進場訊號,而 exits
是出場訊號。所以 entries.hold_until(exits)
,就是進場訊號為 True
時,買入並持有該檔股票,直到出場訊號為 True
則賣出。
此函式有很多細部設定,可以讓你最多選擇 N 檔股票做輪動。另外,當超過 N 檔進場訊號發生,也可以按照客制化的排序,選擇優先選入的股票。最後,可以設定價格波動當輪動訊號,來增加出場的時機點。
PARAMETER |
DESCRIPTION |
exit |
TYPE:
Dataframe
|
nstocks_limit |
TYPE:
int)`
DEFAULT:
None
|
stop_loss |
價格波動輪動訊號,預設為None,不生成輪動訊號。範例:0.1,代表成本價下跌 10% 時產生出場訊號。
TYPE:
float
DEFAULT:
-inf
|
take_profit |
價格波動輪動訊號,預設為None,不生成輪動訊號。範例:0.1,代表成本價上漲 10% 時產生出場訊號。
TYPE:
float
DEFAULT:
inf
|
trade_at |
價格波動輪動訊號參考價,預設為'close'。可選 close 或 open 。
TYPE:
str
DEFAULT:
'close'
|
rank |
當天進場訊號數量超過 nstocks_limit 時,以 rank 數值越大的股票優先進場。
TYPE:
Dataframe
DEFAULT:
None
|
RETURNS |
DESCRIPTION |
DataFrame
|
|
Examples:
價格 > 20 日均線入場, 價格 < 60 日均線出場,最多持有10檔,超過 10 個進場訊號,則以股價淨值比小的股票優先選入。
from finlab import data
from finlab.backtest import sim
close = data.get('price:收盤價')
pb = data.get('price_earning_ratio:股價淨值比')
sma20 = close.average(20)
sma60 = close.average(60)
entries = close > sma20
exits = close < sma60
#pb前10小的標的做輪動
position = entries.hold_until(exits, nstocks_limit=10, rank=-pb)
sim(position)
Source code in finlab/dataframe.py
| def hold_until(self, exit, nstocks_limit=None, stop_loss=-np.inf, take_profit=np.inf, trade_at='close', rank=None, market='AUTO'):
"""訊號進出場
這大概是所有策略撰寫中,最重要的語法糖,上述語法中 `entries` 為進場訊號,而 `exits` 是出場訊號。所以 `entries.hold_until(exits)` ,就是進場訊號為 `True` 時,買入並持有該檔股票,直到出場訊號為 `True ` 則賣出。
<img src="https://i.ibb.co/PCt4hPd/Screen-Shot-2021-10-26-at-6-35-05-AM.png" alt="Screen-Shot-2021-10-26-at-6-35-05-AM">
此函式有很多細部設定,可以讓你最多選擇 N 檔股票做輪動。另外,當超過 N 檔進場訊號發生,也可以按照客制化的排序,選擇優先選入的股票。最後,可以設定價格波動當輪動訊號,來增加出場的時機點。
Args:
exit (pd.Dataframe): 出場訊號。
nstocks_limit (int)`: 輪動檔數上限,預設為None。
stop_loss (float): 價格波動輪動訊號,預設為None,不生成輪動訊號。範例:0.1,代表成本價下跌 10% 時產生出場訊號。
take_profit (float): 價格波動輪動訊號,預設為None,不生成輪動訊號。範例:0.1,代表成本價上漲 10% 時產生出場訊號。
trade_at (str): 價格波動輪動訊號參考價,預設為'close'。可選 `close` 或 `open`。
rank (pd.Dataframe): 當天進場訊號數量超過 nstocks_limit 時,以 rank 數值越大的股票優先進場。
Returns:
(pd.DataFrame): data
Examples:
價格 > 20 日均線入場, 價格 < 60 日均線出場,最多持有10檔,超過 10 個進場訊號,則以股價淨值比小的股票優先選入。
```py
from finlab import data
from finlab.backtest import sim
close = data.get('price:收盤價')
pb = data.get('price_earning_ratio:股價淨值比')
sma20 = close.average(20)
sma60 = close.average(60)
entries = close > sma20
exits = close < sma60
#pb前10小的標的做輪動
position = entries.hold_until(exits, nstocks_limit=10, rank=-pb)
sim(position)
```
"""
if nstocks_limit is None:
nstocks_limit = len(self.columns)
self_reindex = self.index_str_to_date()
exit_reindex = exit.index_str_to_date()
rank_reindex = rank.index_str_to_date() if rank is not None else None
union_index = self_reindex.index.union(exit_reindex.index)
intersect_col = self_reindex.columns.intersection(exit_reindex.columns)
if stop_loss != -np.inf or take_profit != np.inf:
market = finlab.market_info.get_market_info(
self_reindex, user_market_info=market)
if not isinstance(market, finlab.market_info.MarketInfo):
raise Exception("It seems like the market has"
"not been specified well when using the hold_until"
" function. Please provide the appropriate"
" market parameter to the hold_until function "
"to ensure it can determine the correct market"
" for the transaction.")
price = market.get_price(trade_at, adj=True)
union_index = union_index.union(
price.loc[union_index[0]: union_index[-1]].index)
intersect_col = intersect_col.intersection(price.columns)
else:
price = pd.DataFrame(index=union_index, columns=intersect_col)
price.index = pd.to_datetime(price.index)
if rank_reindex is not None:
union_index = union_index.union(rank_reindex.index)
intersect_col = intersect_col.intersection(rank_reindex.columns)
entry = self_reindex.reindex(union_index, method='ffill')[
intersect_col].ffill().fillna(False)
exit = exit_reindex.reindex(union_index, method='ffill')[
intersect_col].ffill().fillna(False)
if price is not None:
price = price.reindex(union_index, method='ffill')[intersect_col]
if rank_reindex is not None:
rank_reindex = rank_reindex.reindex(
union_index, method='ffill')[intersect_col]
else:
rank_reindex = pd.DataFrame(
1, index=union_index, columns=intersect_col)
rank_reindex = rank_reindex.replace([np.inf, -np.inf], np.nan)
max_rank = rank_reindex.max().max()
min_rank = rank_reindex.min().min()
rank_reindex = (rank_reindex - min_rank) / (max_rank - min_rank)
rank_reindex.fillna(0, inplace=True)
def rotate_stocks(ret, entry, exit, nstocks_limit, stop_loss=-np.inf, take_profit=np.inf, price=None, ranking=None):
nstocks = 0
ret[0][np.argsort(entry[0], kind='stable')[-nstocks_limit:]] = 1
ret[0][exit[0] == 1] = 0
ret[0][entry[0] == 0] = 0
entry_price = np.empty(entry.shape[1])
entry_price[:] = np.nan
for i in range(1, entry.shape[0]):
# regitser entry price
if stop_loss != -np.inf or take_profit != np.inf:
is_entry = ((ret[i-2] == 0) if i >
1 else (ret[i-1] == 1))
is_waiting_for_entry = np.isnan(
entry_price) & (ret[i-1] == 1)
is_entry |= is_waiting_for_entry
entry_price[is_entry == 1] = price[i][is_entry == 1]
# check stop_loss and take_profit
returns = price[i] / entry_price
stop = (returns > 1 + abs(take_profit)
) | (returns < 1 - abs(stop_loss))
exit[i] |= stop
# run signal
rank = (entry[i] * ranking[i] + ret[i-1] * 3)
rank[exit[i] == 1] = -1
rank[(entry[i] == 0) & (ret[i-1] == 0)] = -1
ret[i][np.argsort(rank)[-nstocks_limit:]] = 1
ret[i][rank == -1] = 0
return ret
ret = pd.DataFrame(0, index=entry.index, columns=entry.columns)
ret = rotate_stocks(ret.values,
entry.astype(int).values,
exit.astype(int).values,
nstocks_limit,
stop_loss,
take_profit,
price=price.values,
ranking=rank_reindex.values)
return pd.DataFrame(ret, index=entry.index, columns=entry.columns).astype(bool)
|
index_str_to_date
財務月季報索引格式轉換
將以下資料的索引轉換成datetime格式:
財務季報 (ex:2022-Q1) 從文字格式轉為財報電子檔資料上傳日。
通常使用情境為對不同週期的dataframe做reindex,常用於以公告截止日作為訊號產生日。
RETURNS |
DESCRIPTION |
DataFrame
|
|
Examples:
data.get('financial_statement:現金及約當現金').index_str_to_date()
Source code in finlab/dataframe.py
| def index_str_to_date(self):
"""財務月季報索引格式轉換
將以下資料的索引轉換成datetime格式:
財務季報 (ex:2022-Q1) 從文字格式轉為財報電子檔資料上傳日。
通常使用情境為對不同週期的dataframe做reindex,常用於以公告截止日作為訊號產生日。
Returns:
(pd.DataFrame): data
Examples:
```py
data.get('financial_statement:現金及約當現金').index_str_to_date()
```
"""
if len(self.index) == 0 or not isinstance(self.index[0], str):
return self
if self.index[0].find('M') != -1:
return self._index_str_to_date_month()
elif self.index[0].find('Q') != -1:
if self.index[0].find('US-ALL') != -1:
return self._index_str_to_date_season(market='us_stock_all')
elif self.index[0].find('US') != -1:
return self._index_str_to_date_season(market='us_stock')
else:
return self._index_str_to_date_season()
return self
|
industry_rank
industry_rank(categories=None)
計算產業 ranking 排名,0 代表產業內最低,1 代表產業內最高
Args:
categories (list of str): 欲考慮的產業,ex: ['貿易百貨', '雲端運算'],預設為全產業,請參考 data.get('security_industry_themes')
中的產業項目。
Examples:
本意比產業排名分數
from finlab import data
pe = data.get('price_earning_ratio:本益比')
pe_rank = pe.industry_rank()
print(pe_rank)
Source code in finlab/dataframe.py
| def industry_rank(self, categories=None):
"""計算產業 ranking 排名,0 代表產業內最低,1 代表產業內最高
Args:
categories (list of str): 欲考慮的產業,ex: ['貿易百貨', '雲端運算'],預設為全產業,請參考 `data.get('security_industry_themes')` 中的產業項目。
Examples:
本意比產業排名分數
```py
from finlab import data
pe = data.get('price_earning_ratio:本益比')
pe_rank = pe.industry_rank()
print(pe_rank)
```
"""
from finlab import data
themes = (data.get('security_industry_themes')
.copy() # 複製
.assign(category=lambda self: self.category
.apply(lambda s: eval(s))) # 從文字格式轉成陣列格
.explode('category') # 展開資料
)
categories = (categories
or set(themes.category[themes.category.str.find(':') == -1]))
def calc_rank(ind):
stock_ids = themes.stock_id[themes.category == ind]
return (self[list(stock_ids)].pipe(lambda self: self.rank(axis=1, pct=True)))
return (pd.concat([calc_rank(ind) for ind in categories], axis=1)
.groupby(level=0, axis=1).mean())
|
is_entry
進場點
取進場訊號點,若符合條件的值則為True,反之為False。
Returns:
(pd.DataFrame): data
Examples:
策略為每日收盤價前10高,取進場點。
from finlab import data
data.get('price:收盤價').is_largest(10).is_entry()
Source code in finlab/dataframe.py
| def is_entry(self):
"""進場點
取進場訊號點,若符合條件的值則為True,反之為False。
Returns:
(pd.DataFrame): data
Examples:
策略為每日收盤價前10高,取進場點。
```py
from finlab import data
data.get('price:收盤價').is_largest(10).is_entry()
```
"""
return (self & ~self.shift(fill_value=False))
|
is_exit
出場點
取出場訊號點,若符合條件的值則為 True,反之為 False。
Returns:
(pd.DataFrame): data
Examples:
策略為每日收盤價前10高,取出場點。
from finlab import data
data.get('price:收盤價').is_largest(10).is_exit()
Source code in finlab/dataframe.py
| def is_exit(self):
"""出場點
取出場訊號點,若符合條件的值則為 True,反之為 False。
Returns:
(pd.DataFrame): data
Examples:
策略為每日收盤價前10高,取出場點。
```py
from finlab import data
data.get('price:收盤價').is_largest(10).is_exit()
```
"""
return (~self & self.shift(fill_value=False))
|
is_largest
取每列前 n 筆大的數值
若符合 True
,反之為 False
。用來篩選每天數值最大的股票。
Args:
n (positive-int): 設定每列前 n 筆大的數值。
Returns:
(pd.DataFrame): data
Examples:
每季 ROA 前 10 名的股票
from finlab import data
roa = data.get('fundamental_features:ROA稅後息前')
good_stocks = roa.is_largest(10)
Source code in finlab/dataframe.py
| def is_largest(self, n):
"""取每列前 n 筆大的數值
若符合 `True` ,反之為 `False` 。用來篩選每天數值最大的股票。
<img src="https://i.ibb.co/8rh3tbt/is-largest.png" alt="is-largest">
Args:
n (positive-int): 設定每列前 n 筆大的數值。
Returns:
(pd.DataFrame): data
Examples:
每季 ROA 前 10 名的股票
```py
from finlab import data
roa = data.get('fundamental_features:ROA稅後息前')
good_stocks = roa.is_largest(10)
```
"""
return self.astype(float).apply(lambda s: s.nlargest(n), axis=1).reindex_like(self).notna()
|
is_smallest
取每列前 n 筆小的數值
若符合 True
,反之為 False
。用來篩選每天數值最小的股票。
Args:
n (positive-int): 設定每列前 n 筆小的數值。
Returns:
(pd.DataFrame): data
Examples:
股價淨值比最小的 10 檔股票
from finlab import data
pb = data.get('price_earning_ratio:股價淨值比')
cheap_stocks = pb.is_smallest(10)
Source code in finlab/dataframe.py
| def is_smallest(self, n):
"""取每列前 n 筆小的數值
若符合 `True` ,反之為 `False` 。用來篩選每天數值最小的股票。
Args:
n (positive-int): 設定每列前 n 筆小的數值。
Returns:
(pd.DataFrame): data
Examples:
股價淨值比最小的 10 檔股票
```py
from finlab import data
pb = data.get('price_earning_ratio:股價淨值比')
cheap_stocks = pb.is_smallest(10)
```
"""
return self.astype(float).apply(lambda s: s.nsmallest(n), axis=1).reindex_like(self).notna()
|
quantile_row
股票當天數值分位數
取得每列c定分位數的值。
Args:
c (positive-int): 設定每列 n 定分位數的值。
Returns:
(pd.DataFrame): data
Examples:
取每日股價前90%分位數
from finlab import data
data.get('price:收盤價').quantile_row(0.9)
Source code in finlab/dataframe.py
| def quantile_row(self, c):
"""股票當天數值分位數
取得每列c定分位數的值。
Args:
c (positive-int): 設定每列 n 定分位數的值。
Returns:
(pd.DataFrame): data
Examples:
取每日股價前90%分位數
```py
from finlab import data
data.get('price:收盤價').quantile_row(0.9)
```
"""
s = self.index_str_to_date().quantile(c, axis=1)
return s
|
rise
數值上升中
取是否比前第n筆高,若符合條件的值則為True,反之為False。
Args:
n (positive-int): 設定比較前第n筆高。
Returns:
(pd.DataFrame): data
Examples:
收盤價是否高於10日前股價
from finlab import data
data.get('price:收盤價').rise(10)
Source code in finlab/dataframe.py
| def rise(self, n=1):
"""數值上升中
取是否比前第n筆高,若符合條件的值則為True,反之為False。
<img src="https://i.ibb.co/Y72bN5v/Screen-Shot-2021-10-26-at-6-43-41-AM.png" alt="Screen-Shot-2021-10-26-at-6-43-41-AM">
Args:
n (positive-int): 設定比較前第n筆高。
Returns:
(pd.DataFrame): data
Examples:
收盤價是否高於10日前股價
```py
from finlab import data
data.get('price:收盤價').rise(10)
```
"""
return self > self.shift(n)
|
sustain
sustain(nwindow, nsatisfy=None)
持續 N 天滿足條件
取移動 nwindow 筆加總大於等於nsatisfy,若符合條件的值則為True,反之為False。
PARAMETER |
DESCRIPTION |
nwindow |
TYPE:
positive - int
|
nsatisfy |
TYPE:
positive - int
DEFAULT:
None
|
Returns:
(pd.DataFrame): data
Examples:
收盤價是否連兩日上漲
from finlab import data
data.get('price:收盤價').rise().sustain(2)
Source code in finlab/dataframe.py
| def sustain(self, nwindow, nsatisfy=None):
"""持續 N 天滿足條件
取移動 nwindow 筆加總大於等於nsatisfy,若符合條件的值則為True,反之為False。
Args:
nwindow (positive-int): 設定移動窗格。
nsatisfy (positive-int): 設定移動窗格計算後最低滿足數值。
Returns:
(pd.DataFrame): data
Examples:
收盤價是否連兩日上漲
```py
from finlab import data
data.get('price:收盤價').rise().sustain(2)
```
"""
nsatisfy = nsatisfy or nwindow
return self.rolling(nwindow).sum() >= nsatisfy
|