Navigation

    Backtrader Community

    • Register
    • Login
    • Search
    • Categories
    • Recent
    • Tags
    • Popular
    • Users
    • Groups
    • Search
    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

    General Code/Help
    2
    8
    321
    Loading More Posts
    • Oldest to Newest
    • Newest to Oldest
    • Most Votes
    Reply
    • Reply as topic
    Log in to reply
    This topic has been deleted. Only users with topic management privileges can see it.
    • T
      techydoc last edited by

      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)
      
      
      1 Reply Last reply Reply Quote 0
      • T
        techydoc last edited by

        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()
        
        1 Reply Last reply Reply Quote 0
        • run-out
          run-out last edited by

          Your momemtum function is coming back with nan values. You need to debug in your momentum function.

          RunBacktest.com

          T 1 Reply Last reply Reply Quote 1
          • T
            techydoc @run-out last edited by

            @run-out Thank You.

            T 1 Reply Last reply Reply Quote 0
            • T
              techydoc @techydoc last edited by

              @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)

              T 1 Reply Last reply Reply Quote 0
              • T
                techydoc @techydoc last edited by

                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.

                T 1 Reply Last reply Reply Quote 0
                • T
                  techydoc @techydoc last edited by

                  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)
                  
                  T 1 Reply Last reply Reply Quote 0
                  • T
                    techydoc @techydoc last edited by

                    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)
                    
                    
                    1 Reply Last reply Reply Quote 1
                    • 1 / 1
                    • First post
                      Last post
                    Copyright © 2016, 2017, 2018, 2019, 2020, 2021 NodeBB Forums | Contributors