For code/output blocks: Use ``` (aka backtick or grave accent) in a single line before and after the block. See: http://commonmark.org/help/

Momentum strategy not working properly



  • Hi,

    We are implementing the momentum strategy from "Stocks on the Move" for our Masters' thesis, trying to follow the Backtrader implementation from here: https://teddykoker.com/2019/05/momentum-strategy-from-stocks-on-the-move-in-python/
    Data is +4000 US stocks named with a SEDOL code. In addition, data consist of log-returns, so it has been necessary to create a "pseudo" close price. We don't have High-Low data, so it is not possible to use Average True Range for position sizes.
    1: It seems like we have issues with short selling, which we tried to stop with self.savedPositions as we are not interested in going short. It is a long-only strategy.
    2: For some reason it sometimes sells the same stock multiple times on the same day. It also prints out multiple buys on the same stock, on the same day and with same price and costs.
    3. Also, some sells give very high negative costs, mostly at the last sell when there have been multiple sells on the same stock on the same day.

    Hope someone can help us fix it. See code below and let me know if you have any questions.

    Multiple sells and negative costs:
    mom_error_3.PNG

    Momentum indicator:

    #Momentum indicator
    class Momentum(bt.ind.PeriodN):
        lines = ('trend', )
        params = dict(period = 90)
    
        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)
    

    Strategy:

    # Create a strategy 
    class Strategy(bt.Strategy):
        
        params = dict(
            momentum=Momentum,
            momentum_period=90,
    
            movav=bt.ind.SMA,
            idx_period=200,
            stock_period=100,
    
            volatr=bt.ind.ATR,
            vol_period=20, 
        )
        
        def __init__(self):
            
            self.d_with_len = []
            self.inds = defaultdict(dict)
            self.stocks = self.datas[1:]
            self.dicter = {}
            self.savedPositions = dict()
            
            # S&P500 MA
            self.idx_mav = self.p.movav(self.data0, period=self.p.idx_period)
            
            # Indicators
            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)
    
        def notify_order(self, order):
            if order.status in [order.Completed]:
                self.dicter[order.ref] = None
                
                if order.isbuy():
                    if order.data._name not in self.savedPositions.keys():
                        self.savedPositions[order.data._name] = bt.num2date(order.executed.dt)
                    print('BUY EXECUTED on %s, Price: %.2f, Cost: %.2f, Comm %.2f, Datetime: %s' %
                            (order.data._name,
                             order.executed.price,
                             order.executed.value,
                             order.executed.comm,
                             bt.num2date(order.executed.dt)))                    
                if order.issell():
                    if order.data._name in self.savedPositions.keys():
                        self.savedPositions.pop(order.data._name)
                    print('Sell EXECUTED on %s, Price: %.2f, Cost: %.2f, Comm %.2f, Datetime: %s' %
                            (order.data._name,
                             order.executed.price,
                             order.executed.value,
                             order.executed.comm,
                             bt.num2date(order.executed.dt)))
                
        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):
            
            # portfolio rebal every 5 days, positions rebal every 10 days
            l = len(self)
            if l % 5 == 0:
                self.rebalance_portfolio()
            if l % 10 == 0:
                self.rebalance_positions()
    
        def rebalance_portfolio(self):
            # momentum ranking
            self.rankings = list(filter(lambda d: len(d) > 100, self.stocks))
            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 > 0:
                    if (i > num_stocks * 0.05 or d < self.inds[d]["mav"]) and self.savedPositions.get(d._name, None) != None:
                        self.order = self.close(d, exectype=bt.Order.Close)
                        
            if self.data0 < self.idx_mav:
                return
    
            # buy stocks with remaining cash
            for i, d in enumerate(self.rankings[:int(num_stocks * 0.05)]):
                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"]
                    size = 1 / (len(self.stocks) * 0.05)
                    #self.buy(d, size=size, exectype=bt.Order.Close)    
                    self.order = self.order_target_percent(d, target=size, exectype=bt.Order.Close)
                    
                    
        def rebalance_positions(self):
            num_stocks = len(self.rankings)
            
            if self.data0 < self.idx_mav:
                return
    
            # rebalance all stocks
            for i, d in enumerate(self.rankings[:int(num_stocks * 0.05)]):
                cash = self.broker.get_cash()
                value = self.broker.get_value()
                if cash <= 0:
                    break
                #size = value * 0.001 #/ self.inds[d]["vol"]
                size = 1 / (len(self.stocks) * 0.05)
                self.order = self.order_target_percent(d, target=size, exectype=bt.Order.Close)
    

    Cerebro:

    if __name__ == '__main__':
        # Create cerebro entity
        cerebro = bt.Cerebro()
        
        # Add S&P500
        spy = pd.read_csv("spy.csv", index_col = "as_of", parse_dates = True)
        spy = spy.dropna()
        spy.index = pd.to_datetime(spy.index)
        spy = bt.feeds.PandasData(dataname = spy)
    
        cerebro.adddata(spy)  # add S&P 500 Index
        
        # Add data
        sedolList = []
        for sedol in sedols:
            
            pd_df = df_returns[df_returns['sedol'] == sedol].drop(columns=['sedol']).set_index('as_of')
            pd_df["close"] = np.exp(pd_df['val_num'].cumsum())
            pd_df = pd_df.drop(columns=['val_num'])
            pd_df.index = pd.to_datetime(pd_df.index)
            pd_df = pd_df.dropna()
            data = bt.feeds.PandasData(dataname = pd_df)
            
            if len(pd_df) > 100 and all(pd_df["close"] > 0.001): # data must be long enough to compute 100 day SMA
                sedolList.append(sedol)
                # Add data to Cerebro
                cerebro.adddata(data, name=sedol)
        
        # Set our desired starting cash
        cerebro.broker.set_cash(1000000)
        
        # Set the commission
        cerebro.broker.setcommission(commission=0.0001)
    
        # Add strategy
        cerebro.addstrategy(Strategy)
        cerebro.addanalyzer(bt.analyzers.PyFolio, _name='pyfolio', timeframe=bt.TimeFrame.Days)
        cerebro.addanalyzer(bt.analyzers.PositionsValue, headers=True, cash=True, _name='mypositions')
        
        # Print out the starting conditions
        print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue())
    
        # Run over everything
        results = cerebro.run()
    
        # Print out the final result
        print('Final Portfolio Value: %.2f' % cerebro.broker.getvalue())
    


  • @jean15ac said in Momentum strategy not working properly:

    if self.getposition(self.data).size > 0:

    maybe there is something wrong with

    if self.getposition(self.data).size > 0:
    

    may be it is your fuction:

    rebalance_positions
    rebalance_portfolio
    

    maybe it is the backtrader's bug.

    I often write similar code like you,but I don't find this quesiton.



  • @tianjixuetu said in Momentum strategy not working properly:

    @jean15ac said in Momentum strategy not working properly:

    if self.getposition(self.data).size > 0:

    maybe there is something wrong with

    if self.getposition(self.data).size > 0:
    

    may be it is your fuction:

    rebalance_positions
    rebalance_portfolio
    

    maybe it is the backtrader's bug.

    I often write similar code like you,but I don't find this quesiton.

    I'm sorry @tianjixuetu but that's really not a very helpful response, since you're basically saying the bug can originate from anywhere 😅

    I would like to know if there's a solution to this as well as I'm using a similar "long-only" setup for my strategy and using if self.getposition(self.data).size > 0: to prevent shorting as well.



  • @nistrup self.getposition(target_data).size,you can get the target_data's size.you use self.data,you only get the first data'size,do you understand?



  • The momentum indicator and Cerebro both look good. The problem is in Strategy. @tianjixuetu is correct when he said that you need to test the target_data and not the first data.

    More specifically, what's happening is that in rebalance_profolio this section of code:

            for i, d in enumerate(self.rankings):
                if self.getposition(self.data).size > 0:
                    if (i > num_stocks * 0.05 or d < self.inds[d]["mav"]) and self.savedPositions.get(d._name, None) != None:
                        self.order = self.close(d, exectype=bt.Order.Close)
    

    is iterating over a list of stocks (in ranked order) but it's determining the position size from just the first data set that it finds. You need something like the following (note lined 2):

            for i, d in enumerate(self.rankings):
                if self.getposition(d.data).size > 0:
                    if (i > num_stocks * 0.05 or d < self.inds[d]["mav"]) and self.savedPositions.get(d._name, None) != None:
                        self.order = self.close(d, exectype=bt.Order.Close)
    
    

    There appears to be a similar problem with the buy loop also.


Log in to reply
 

});