For code/output blocks: Use ``` (aka backtick or grave accent) in a single line before and after the block. See: http://commonmark.org/help/
Problem Implementing Momentum Strategy from Blog Post
-
I read the Backtrader blog post about how to "better" implement a monentum strategy from a post by Terry Koker - both from May 2019. My code is below with links to the two blog posts.
I am getting an error at:
self.rankings.sort(key=lambda d: self.inds[d]["mom"][0])I get a key error, but I can see the object in the debugger.
Any insight would be much appreciated.
''' Original code and algo from: https://teddykoker.com/2019/05/momentum-strategy-from-stocks-on-the-move-in-python/ Revised code from: https://www.backtrader.com/blog/2019-05-20-momentum-strategy/momentum-strategy/ Changes: - Monmentum func: added <period> to function call - Got an error when no <period> present - Assuming that <period> is redundant and do not have to subset <the_array> - In __init__ - fixed typo (add <p>) on call to params <momentum_period> - added: self.stock_under_idx_mav_filter - In <prenext> added: >= self.p.stock_period ''' import backtrader as bt import numpy as np from scipy.stats import linregress import collections def momentum_func(the_array, period): r = np.log(the_array) slope, _, rvalue, _, _ = linregress(np.arange(len(r)), r) annualized = (1 + slope) ** 252 return annualized * (rvalue ** 2) class MomentumIndicator(bt.ind.OperationN): lines = ('trend',) params = dict(period=50) func = momentum_func class MomentumStrategy(bt.Strategy): params = dict( momentum=MomentumIndicator, # 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, rebal_weekday=5 # rebalance 5 is Friday ) 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.p.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) self.stock_under_idx_mav_filter = self.datas[0] < self.idx_mav # Timer to support rebalancing weekcarry over in case of holiday 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 ) #List of stocks that have sufficient length (based on indicators) self.d_with_len = [] def notify_timer(self, timer, when, *args, **kwargs): self.rebalance_portfolio() def prenext(self): # Populate d_with_len self.d_with_len = [d for d in self.datas if len(d) >= self.p.stock_period] # 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): l = len(self) if l % 5 == 0: self.rebalance_portfolio() if l % 10 == 0: self.rebalance_positions() def rebalance_portfolio(self): # only look at data that we can have indicators for self.rankings = self.d_with_len self.rankings.sort(key=lambda d: self.inds[d]["mom"][0]) num_stocks = len(self.rankings) # 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]["mav"]: self.close(d) if self.stock_under_idx_mav_filter: return # buy stocks with remaining cash for i, d in enumerate(self.rankings[:int(num_stocks * 0.2)]): cash = self.broker.get_cash() value = self.broker.get_value() if cash <= 0: break if not self.getposition(self.data).size: size = value * 0.001 / self.inds[d]["vol"] self.buy(d, size=size) def rebalance_positions(self): num_stocks = len(self.rankings) if self.stock_under_idx_mav_filter: return # rebalance all stocks for i, d in enumerate(self.rankings[:int(num_stocks * 0.2)]): cash = self.broker.get_cash() value = self.broker.get_value() if cash <= 0: break size = value * 0.001 / self.inds[d]["vol"] self.order_target_size(d, size) -
The Error I get:
Exception has occurred: KeyError 'mom' File "/home/techydoc/Development/gitrepo/quant/trade-testing/strategies/MomentumStrategy.py", line 102, in <lambda> self.rankings.sort(key=lambda d: self.inds[d]["mom"][0]) File "/home/techydoc/Development/gitrepo/quant/trade-testing/strategies/MomentumStrategy.py", line 102, in rebalance_portfolio self.rankings.sort(key=lambda d: self.inds[d]["mom"][0]) File "/home/techydoc/Development/gitrepo/quant/trade-testing/strategies/MomentumStrategy.py", line 95, in next self.rebalance_portfolio() File "/home/techydoc/Development/gitrepo/quant/trade-testing/strategies/MomentumStrategy.py", line 83, in prenext self.next() File "/home/techydoc/Development/gitrepo/quant/trade-testing/runstrategy.py", line 32, in <module> results = cerebro.run() -
Your momemtum function is coming back with
nanvalues. You need to debug in your momentum function. -
@run-out Thank You.
-
@run-out I fixed the momentum function. I did not realize that it got passed the MomentumIndicator object and the array.
def momentum_func(ind, period): r = np.log(period) slope, _, rvalue, _, _ = linregress(np.arange(len(r)), r) annualized = (1 + slope) ** 252 return annualized * (rvalue ** 2)I still get the same errors. The momentum indicator array has NaN for the first 90 entries, but it is suppose to due to the period needed. I get the "prenext and nextstart, but I don't really understand the inner workings.
Any additional insight is much appreciated. (I will be researching prenext/nextstart)
-
I think I found the issue. The line causing the error is looking for momentum in the index datafeed, which will not have any indicators. Now I just need to figure out how to fix.
-
In prenext and nextstart the blog post used self.datas when it should be self.stocks to exclude the index. It now runs but back to my original problem of it will not computing any analyzers.
The updated code:
''' Original code and algo from: https://teddykoker.com/2019/05/momentum-strategy-from-stocks-on-the-move-in-python/ Revised code from: https://www.backtrader.com/blog/2019-05-20-momentum-strategy/momentum-strategy/ Changes: - Monmentum func added <period> to function call - Got an error when no <period> present - Assuming that <period> is redundant and do not have to subset <the_array> - In __init__ - fixed typo (add <p>) on call to params <momentum_period> - added: self.stock_under_idx_mav_filter - In <prenext> added: >= self.p.stock_period ''' import backtrader as bt import numpy as np from scipy.stats import linregress import collections def momentum_func(ind, period): r = np.log(period) slope, _, rvalue, _, _ = linregress(np.arange(len(r)), r) annualized = (1 + slope) ** 252 return annualized * (rvalue ** 2) class MomentumIndicator(bt.ind.OperationN): lines = ('trend',) params = dict(period=50) func = momentum_func class MomentumStrategy(bt.Strategy): params = dict( momentum=MomentumIndicator, # 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, rebal_weekday=5 # rebalance 5 is Friday ) 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.p.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) self.stock_under_idx_mav_filter = self.datas[0] < self.idx_mav # Timer to support rebalancing weekcarry over in case of holiday 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 ) #List of stocks that have sufficient length (based on indicators) self.d_with_len = [] def notify_timer(self, timer, when, *args, **kwargs): self.rebalance_portfolio() def prenext(self): # Populate d_with_len self.d_with_len = [d for d in self.stocks if len(d) >= self.p.stock_period] # 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.stocks # all data sets fulfill the guarantees now self.next() # delegate the work to next def next(self): l = len(self) if l % 5 == 0: self.rebalance_portfolio() if l % 10 == 0: self.rebalance_positions() def rebalance_portfolio(self): # only look at data that we can have indicators for self.rankings = self.d_with_len #if no stocks are ready return - Added but not sure if needed if(len(self.rankings) == 0): return self.rankings.sort(key=lambda d: self.inds[d]["mom"][0]) num_stocks = len(self.rankings) # 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]["mav"]: self.close(d) if self.stock_under_idx_mav_filter: return # buy stocks with remaining cash for i, d in enumerate(self.rankings[:int(num_stocks * 0.2)]): cash = self.broker.get_cash() value = self.broker.get_value() if cash <= 0: break if not self.getposition(self.data).size: size = value * 0.001 / self.inds[d]["vol"] self.buy(d, size=size) def rebalance_positions(self): num_stocks = len(self.rankings) if self.stock_under_idx_mav_filter: return # rebalance all stocks for i, d in enumerate(self.rankings[:int(num_stocks * 0.2)]): cash = self.broker.get_cash() value = self.broker.get_value() if cash <= 0: break size = value * 0.001 / self.inds[d]["vol"] self.order_target_size(d, size) -
I found the two remaining problems. (1) the filter to check if the index was below the SMA
self.stock_under_idx_mav_filter = self.datas[0] < self.idx_mavThe Backtrader blog post talked about pulling up the test into the init function. I tried that put I got it screwed up. I pushed the checks back down into the reposition and rebalance method and fixed (I think). Any pointers, how to make this filter in the init function would be welcomed. Or maybe this should be a cross over indicator.
The second problem was the logic to buy stocks using 0.2 as the max for each stock. In my test I only had 4 stocks, so the logic said do not buy. Changed to 0.3 and Bob's your Uncle!
Final code:
''' Original code and algo from: https://teddykoker.com/2019/05/momentum-strategy-from-stocks-on-the-move-in-python/ Revised code from: https://www.backtrader.com/blog/2019-05-20-momentum-strategy/momentum-strategy/ Changes: - Monmentum func added <period> to function call - Got an error when no <period> present - Assuming that <period> is redundant and do not have to subset <the_array> - In __init__ - fixed typo (add <p>) on call to params <momentum_period> - added: self.stock_under_idx_mav_filter - In <prenext> added: >= self.p.stock_period - In <prenext> and <nextstart> changed <self.datas> to <self.stocks> to exclude the idx datafeed ''' import backtrader as bt import numpy as np from scipy.stats import linregress import collections def momentum_func(ind, period): r = np.log(period) slope, _, rvalue, _, _ = linregress(np.arange(len(r)), r) annualized = (1 + slope) ** 252 return annualized * (rvalue ** 2) class MomentumIndicator(bt.ind.OperationN): lines = ('trend',) params = dict(period=50) func = momentum_func class MomentumStrategy(bt.Strategy): params = dict( momentum=MomentumIndicator, # 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, rebal_weekday=5 # rebalance 5 is Friday ) 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.p.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) #self.stock_under_idx_mav_filter = self.datas[0].open < self.idx_mav # Timer to support rebalancing weekcarry over in case of holiday 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 ) #List of stocks that have sufficient length (based on indicators) self.d_with_len = [] def notify_timer(self, timer, when, *args, **kwargs): self.rebalance_portfolio() def prenext(self): # Populate d_with_len self.d_with_len = [d for d in self.stocks if len(d) >= self.p.stock_period] # 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.stocks # all data sets fulfill the guarantees now self.next() # delegate the work to next def next(self): l = len(self) if l % 5 == 0: self.rebalance_portfolio() if l % 10 == 0: self.rebalance_positions() def rebalance_portfolio(self): # only look at data that we can have indicators for self.rankings = self.d_with_len #if no stocks are ready return - Added but not sure if needed if(len(self.rankings) == 0): return self.rankings.sort(key=lambda d: self.inds[d]["mom"][0]) num_stocks = len(self.rankings) # 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]["mav"]: self.close(d) if self.datas[0].open < self.idx_mav: #self.stock_under_idx_mav_filter: return # buy stocks with remaining cash for i, d in enumerate(self.rankings[:int(num_stocks * 0.3)]): cash = self.broker.get_cash() value = self.broker.get_value() if cash <= 0: break if not self.getposition(self.data).size: size = value * 0.001 / self.inds[d]["vol"] self.buy(d, size=size) def rebalance_positions(self): num_stocks = len(self.rankings) if self.datas[0].open < self.idx_mav: #self.stock_under_idx_mav_filter: return # rebalance all stocks for i, d in enumerate(self.rankings[:int(num_stocks * 0.2)]): cash = self.broker.get_cash() value = self.broker.get_value() if cash <= 0: break size = value * 0.001 / self.inds[d]["vol"] self.order_target_size(d, size)
