在另一篇偉大的文章中,泰迪·科克(Teddy Koker)再次展示了演算法交易策略的發展之路:
- 研究優先應用
pandas - 回溯測試,然後使用
backtrader
榮譽!!!
該帖子可以在以下位置找到:
泰迪·科克(Teddy Koker)給我留言,問我是否可以評論 backtrader的用法。我的觀點可以 在下面看到。這隻是我個人的拙見,因為作為 backtrader 的作者,我對如何最好地使用這個平臺有偏見。
而我個人對如何制定某些結構的品味,不必與其他人喜歡使用平臺的方式相匹配。
注意
實際上,讓平臺 open 插入幾乎任何東西,並用不同的方式做同樣的事情,是一個有意識的決定,讓人們以他們認為合適的方式使用它(在平臺目標,語言可能性和我做出的失敗的設計決策的限制範圍內)。
在這裡,我們將只關注本來可以以不同的方式完成的事情。“不同”是否更好總是一個觀點問題。 backtrader 的作者並不總是必須正確使用「backtrader」進行開發實際上「更好」(因為實際開發必須適合開發人員而不是」backtrader」的作者)
參數:dict vs tuple of tuples
文件和/或博客中提供backtrader 的許多範例都使用 tuple of tuples 該模式作為參數。例如,從代碼中:
class Momentum(bt.Indicator):
lines = ('trend',)
params = (('period', 90),)
與這種範式一起,人們總是有機會使用adict。
class Momentum(bt.Indicator):
lines = ('trend',)
params = dict(period=90) # or params = {'period': 90}
隨著時間的流逝,這已經變得更容易使用,並成為作者的首選模式。
注意
作者更喜歡dict(period=90),更容易輸入,不需要引號。但是大括弧表示法 {'period': 90},是許多其他人的首選。
和 tuple 方法之間的dict根本區別:
-
tuple of tuples使用參數保留聲明的順序,這在枚舉它們時可能很重要。提示
對於Python
3.7中的預設有序字典,聲明順序應該沒有問題(3.6如果使用CPython,即使它是一個實現細節)
在下面作者修改的範例中dict ,將使用符號。
指標Momentum
在本文中,這是指標的定義方式
class Momentum(bt.Indicator):
lines = ('trend',)
params = (('period', 90),)
def __init__(self):
self.addminperiod(self.params.period)
def next(self):
returns = np.log(self.data.get(size=self.p.period))
x = np.arange(len(returns))
slope, _, rvalue, _, _ = linregress(x, returns)
annualized = (1 + slope) ** 252
self.lines.trend[0] = annualized * (rvalue ** 2)
使用力,即:使用已經存在的東西, PeriodN 如指標,它:
- 已經定義了一個
period參數,並知道如何將其傳遞給系統
因此,這可以更好
class Momentum(bt.ind.PeriodN):
lines = ('trend',)
params = dict(period=50)
def next(self):
...
我們已經跳過了為使用 addminperiod的唯一目的而定義__init__的需求,這應該只在特殊情況下使用。
為了繼續, backtrader 定義了一個OperationN 指標,該指標必須定義一個屬性,該屬性 func 將 period 作為參數傳遞柱線,並將返回值放入定義的 line中。
考慮到這一點,人們可以想像以下內容是潛在的代碼
def momentum_func(the_array):
r = np.log(the_array)
slope, _, rvalue, _, _ = linregress(np.arange(len(r)), r)
annualized = (1 + slope) ** 252
return annualized * (rvalue ** 2)
class Momentum(bt.ind.OperationN):
lines = ('trend',)
params = dict(period=50)
func = momentum_func
這意味著我們已經將指標的複雜性排除在指標之外。我們甚至可以從外部庫導入momentum_func ,如果底層函數發生變化,指標也無需更改即可反映新行為。作為獎金,我們有 純粹的 聲明性指標。否 __init__、否 addminperiod 和否 next
戰略
讓我們看一下這部分__init__ 。
class Strategy(bt.Strategy):
def __init__(self):
self.i = 0
self.inds = {}
self.spy = self.datas[0]
self.stocks = self.datas[1:]
self.spy_sma200 = bt.indicators.SimpleMovingAverage(self.spy.close,
period=200)
for d in self.stocks:
self.inds[d] = {}
self.inds[d]["momentum"] = Momentum(d.close,
period=90)
self.inds[d]["sma100"] = bt.indicators.SimpleMovingAverage(d.close,
period=100)
self.inds[d]["atr20"] = bt.indicators.ATR(d,
period=20)
關於樣式的一些內容:
-
盡可能使用參數,而不是固定值
-
使用較短的名稱和較短的名稱(例如,對於導入),在大多數情況下,它將提高可讀性
-
充分利用 Python
-
不要用於
closedata feed。通常傳遞data feed,它將使用 close。這可能看起來不相關,但是當嘗試在任何地方保持代碼通用時(如在指標中),它確實有所説明。
人們應該/應該考慮的第一件事是:如果可能的話,將所有內容作為參數。因此
class Strategy(bt.Strategy):
params = dict(
momentum=Momentum, # parametrize the momentum and its period
momentum_period=90,
movav=bt.ind.SMA, # parametrize the moving average and its periods
idx_period=200,
stock_period=100,
volatr=bt.ind.ATR, # parametrize the volatility and its period
vol_period=20,
)
def __init__(self):
# self.i = 0 # See below as to why the counter is commented out
self.inds = collections.defaultdict(dict) # avoid per data dct in for
# Use "self.data0" (or self.data) in the script to make the naming not
# fixed on this being a "spy" strategy. Keep things generic
# self.spy = self.datas[0]
self.stocks = self.datas[1:]
# Again ... remove the name "spy"
self.idx_mav = self.p.movav(self.data0, period=self.p.idx_period)
for d in self.stocks:
self.inds[d]['mom'] = self.p.momentum(d, period=self.momentum_period)
self.inds[d]['mav'] = self.p.movav(d, period=self.p.stock_period)
self.inds[d]['vol'] = self.p.volatr(d, period=self.p.vol_period)
params通過使用和更改一些命名約定,我們使__init__(以及隨之而來的策略)完全可定製和通用(沒有spy引用任何hwere)
next 及其 len
backtrader 嘗試在可能的情況下使用Python範例。它肯定會失敗,但它會嘗試。
讓我們看看在next
def next(self):
if self.i % 5 == 0:
self.rebalance_portfolio()
if self.i % 10 == 0:
self.rebalance_positions()
self.i += 1
這就是Pythonlen 範式説明的地方。讓我們使用它
def next(self):
l = len(self)
if l % 5 == 0:
self.rebalance_portfolio()
if l % 10 == 0:
self.rebalance_positions()
如您所見,沒有必要保留self.i 計數器。策略和大多數對象的長度由系統一直提供,計算和更新。
next 和 prenext
代碼包含此轉發
def prenext(self):
# call next() even when data is not available for all tickers
self.next()
而且進入時沒有 保障 next
def next(self):
if self.i % 5 == 0:
self.rebalance_portfolio()
...
好吧,我們知道正在使用無生存偏差的數據集,但通常不保護prenext => next 轉發不是一個好主意。
-
當所有緩衝區(指標、data feeds)至少可以提供數據點時,backtrader調用
next。100-bar移動平均線顯然只有在具有來自data feed的100個數據點時才會提供。這意味著在進入
next時,必須100 data points檢查data feed,移動平均線1 data point -
backtrader提供
prenext作為鉤子,讓開發人員在滿足上述保證之前訪問內容。例如,當多個data feeds正在運行並且它們的開始日期不同時,這很有用。開發人員可能希望在滿足next所有data feeds(和相關指標)的所有保證並首次被要求之前進行一些檢查或採取行動。
在一般情況下,prenext => next 轉發應該有一個這樣的保護裝置:
def prenext(self):
# call next() even when data is not available for all tickers
self.next()
def next(self):
d_with_len = [d for d in self.datas if len(d)]
...
這意味著只有 來自的self.datas子集d_with_len可以保證使用。
注意
類似的保護必須用於指標。
因為在策略的整個生命週期內進行這種計算似乎毫無意義,因此可以進行這樣的優化。
def __init__(self):
...
self.d_with_len = []
def prenext(self):
# Populate d_with_len
self.d_with_len = [d for d in self.datas if len(d)]
# call next() even when data is not available for all tickers
self.next()
def nextstart(self):
# This is called exactly ONCE, when next is 1st called and defaults to
# call `next`
self.d_with_len = self.datas # all data sets fulfill the guarantees now
self.next() # delegate the work to next
def next(self):
# we can now always work with self.d_with_len with no calculation
...
當滿足保證時,將停止調用保護計算prenext , nextstart 然後調用它,通過覆蓋它,我們可以重置 list 保存數據集的哪個,作為完整的數據集,即: self.datas
有了這個,所有的警衛都被移除了next。
next 帶計時器
雖然作者在這裡的意圖是每5/10天重新平衡(投資組合/頭寸),但這可能意味著每周/每兩周重新平衡。
如果出現以下情況,該len(self) % period 方法將失敗:
-
數據集不是在星期一開始的
-
在交易假期期間,這將使再平衡變得不一致
為了克服這一點,可以使用內置功能 backtrader
使用它們將確保在預期發生時進行再平衡。讓我們想像一下,其目的是在星期五重新平衡。
讓我們在我們的策略中params__init__加入一些魔力
class Strategy(bt.Strategy):
params = dict(
...
rebal_weekday=5, # rebalance 5 is Friday
)
def __init__(self):
...
self.add_timer(
when=bt.Timer.SESSION_START,
weekdays=[self.p.rebal_weekday],
weekcarry=True, # if a day isn't there, execute on the next
)
...
現在我們已經準備好知道什麼時候是星期五。即使星期五碰巧是交易假期,添加weekcarry=True 也可以確保我們在星期一收到通知(如果星期一也是假期,則為星期二或...
計時器的通知已接收notify_timer
def notify_timer(self, timer, when, *args, **kwargs):
self.rebalance_portfolio()
因為原始代碼中的每個柱線10也會發生一個rebalance_positions,所以可以:
-
添加第 2個計時器 ,也適用於星期五
-
使用計數器僅對每個第 2 個 調用執行操作,這甚至可以在計時器本身中使用
allow=callable參數
注意
計時器甚至可以更好地用於實現以下模式:
-
rebalance_portfolio每月第2和第 4個星期 五 -
rebalance_positions僅限每月第 4個星期五
一些額外功能
其他一些事情可能純粹是個人品味的問題。
個人品味 1
始終使用預先構建的比較,而不是在過程中next比較內容。例如,從代碼(多次使用)
if self.spy < self.spy_sma200:
return
我們可以做以下事情。第一個期間__init__
def __init__(self):
...
self.spy_filter = self.spe < self.spy_sma200
以及後來的
if self.spy_filter:
return
考慮到這一點,如果我們想改變spy_filter 條件,我們只需要在代碼中的 __init__ 多個位置執行此操作一次,而不必這樣做。
這同樣適用於這裡的其他比較d < self.inds[d]["sma100"] :
# sell stocks based on criteria
for i, d in enumerate(self.rankings):
if self.getposition(self.data).size:
if i > num_stocks * 0.2 or d < self.inds[d]["sma100"]:
self.close(d)
它也可以在期間__init__ 預先構建,因此更改為類似的東西
# sell stocks based on criteria
for i, d in enumerate(self.rankings):
if self.getposition(self.data).size:
if i > num_stocks * 0.2 or self.inds[d]['sma_signal']:
self.close(d)
個人品味 2
使所有內容都成為參數。例如,在上面的 lines 中,我們看到一個0.2 在代碼的幾個部分中使用的: 使其成為一個參數。與其他值(如 0.001 和 100 )相同(實際上已經建議將其作為創建移動平均線的參數)
將所有內容作為參數允許打包代碼,並通過更改策略的實例化而不是策略本身來嘗試不同的事情。