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_mav
The 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)
