Stock on the move: momentum about blog

[waldren]

Kerberos thanks for the bug find. I have been trying to learn Backtrader and found Terry' work from a blog post on Backtrader from May 2019. I have been working on incorporating both codes into the BackTrader strategy.

`
'''
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 to function call
  • Got an error when no present
  • Assuming that is redundant and do not have to subset <the_array>
• In init
  • fixed typo (add

    ) on call to params <momentum_period>

  • added: self.stock_under_idx_mav_filter but did not work so reverted to checking in rebalance/reposition methods
• In added: >= self.p.stock_period
• In and changed <self.datas> to <self.stocks> to exclude the idx datafeed
• Fixed bugs in selling stocks as it checked self.data for positions not d

'''
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  # Was unable to make this a filter

    # 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(d).size:  # changed self.data to d
            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(d).size:   # changed self.data to d
            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)

`