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

Bullish and Bearish Divergence MACD/RSI



  • Hello,

    I'm trying to figure out how can I access these signals?

    To get it i thought for example to find first low, than the previous low indexes.
    Then get prices compare them. Then get indicator's values and compare them too. I got stuck at getting second low/high. Please help.

    self.high_first = btind.FindFirstIndexHighest(self.data.high, period=self.p.hl_period)
    self.high_second = btind.FindFirstIndexHighest(self.data.high(self.high_first-1), period=self.p.hl_period)
    self.low_first = btind.FindFirstIndexLowest(self.data.low, period=self.p.hl_period)
    self.low_second = btind.FindFirstIndexLowest(self.data.low(self.low_first-1), period=self.p.hl_period)


  • I might don't know something and I'm reinventing the wheel, It would be great if somebody point me the right direction.


  • administrators

    @coolernax said in Bullish and Bearish Divergence MACD/RSI:

    self.high_first = btind.FindFirstIndexHighest(self.data.high, period=self.p.hl_period)
    self.high_second = btind.FindFirstIndexHighest(self.data.high(self.high_first-1), period=self.p.hl_period)
    

    The problem is that self.data.high(self.high_first - 1) is not a valid construct. Because self.high_first will be determined at runtime, whereas a delayed line (using parenthesis) needs to know the value from how many periods ago it will returning.

    A potential coding

    class SecondHigh(bt.ind.PeriodN):
        lines = ('first_high', 'second_high',)
    
        def __init__(self):
            self.hf = self.lines.first_high  = bt.ind.FindFirstIndexHighest(self.data.high, period=self.p.period)
    
        def next(self):
            # 0 is valid return value for now, make sure to get at least 1 item
            iterable = self.data.get(size=self.hf[0] + 1)
            m = max(iterable)
            self.lines.second_high[0] = next(i for i, v in enumerate(reversed(iterable)) if v == m)
    

    Finding the index of the maximum is not made with the better iterable.index(max(iterable)) to comply with the idea in backtrader that the current index is 0.

    The FindFirstXXX family of functions return the indices in absolute value and this is also done here .

    Note: this is a handcrafted snippet which may have typos and not work at all.



  • @backtrader said in Bullish and Bearish Divergence MACD/RSI:

    @coolernax said in Bullish and Bearish Divergence MACD/RSI:

    self.high_first = btind.FindFirstIndexHighest(self.data.high, period=self.p.hl_period)
    self.high_second = btind.FindFirstIndexHighest(self.data.high(self.high_first-1), period=self.p.hl_period)
    

    The problem is that self.data.high(self.high_first - 1) is not a valid construct. Because self.high_first will be determined at runtime, whereas a delayed line (using parenthesis) needs to know the value from how many periods ago it will returning.

    A potential coding

    class SecondHigh(bt.ind.PeriodN):
        lines = ('first_high', 'second_high',)
    
        def __init__(self):
            self.hf = self.lines.first_high  = bt.ind.FindFirstIndexHighest(self.data.high, period=self.p.period)
    
        def next(self):
            # 0 is valid return value for now, make sure to get at least 1 item
            iterable = self.data.get(size=self.hf[0] + 1)
            m = max(iterable)
            self.lines.second_high[0] = next(i for i, v in enumerate(reversed(iterable)) if v == m)
    

    Finding the index of the maximum is not made with the better iterable.index(max(iterable)) to comply with the idea in backtrader that the current index is 0.

    The FindFirstXXX family of functions return the indices in absolute value and this is also done here .

    Note: this is a handcrafted snippet which may have typos and not work at all.

    Thanks I will try your example



  • @backtrader I created an indicator that works, but sometimes I face error like this:
    ValueError: max() arg is an empty sequence

    Please take a look
    Here is the code:

    class RSIDivergence(bt.ind.PeriodN):
        lines = ('signal',)
        params = dict(
            rsi_period=14,
            hl_period=100,
            hl_min=20
        )
    
        # plotinfo = dict(
        #     subplot=True,
        #     plotlinelabels=True, plotlinevalues=True, plotvaluetags=True,
        # )
    
        # plotlines = dict(
        #     high_first_price=dict(_name='high_first_price', color='red', ls='-'),
        #     high_second_price=dict(_name='high_second_price', color='blue', ls='-'),
        #     low_first_price=dict(_name='low_first_price', color='red', ls='-'),
        #     low_second_price=dict(_name='low_second_price', color='blue', ls='-'),
        # )
    
        def __init__(self):
            self.hfi = bt.ind.FindFirstIndexHighest(self.data.high, period=self.p.hl_period)
            self.lfi = bt.ind.FindFirstIndexLowest(self.data.low, period=self.p.hl_period)
            self.rsi = bt.ind.RSI_Safe(period=self.p.rsi_period)
    
        def signal_get(self):
            signal = 0
            if self.hfp >= self.hsp:
                if self.rsi[-int(self.hfi[0])] < self.rsi[-int(self.hsi)]:
                    print('bearish', self.rsi[-int(self.hfi[0])], self.rsi[-int(self.hsi)])
                    signal -= 1
    
            if self.lfp <= self.lsp:
                if self.rsi[-int(self.lfi[0])] > self.rsi[-int(self.lsi)]:
                    print('bullish', self.rsi[-int(self.hfi[0])], self.rsi[-int(self.hsi)])
                    signal += 1
    
            return signal
    
        def next(self):
            iterable = self.data.high.get(size=int(self.hfi[0]) + self.p.hl_min)
            m = max(iterable)
            self.hsi = next(i for i, v in enumerate(reversed(iterable)) if v == m) + self.p.hl_min
    
            iterable = self.data.low.get(size=int(self.lfi[0]) + self.p.hl_min)
            m = min(iterable)
            self.lsi = next(i for i, v in enumerate(reversed(iterable)) if v == m) + self.p.hl_min
    
            self.hfp = self.data.high[-int(self.hfi[0])]
            self.hsp = self.data.high[-int(self.hsi)]
            self.lfp = self.data.low[-int(self.lfi[0])]
            self.lsp = self.data.low[-int(self.lsi)]
    
            # self.lines.high_first_price[0] = self.hfp
            # self.lines.high_second_price[0] = self.hsp
            # self.lines.low_first_price[0] = self.lfp
            # self.lines.low_second_price[0] = self.lsp
            self.lines.signal[0] = self.signal_get()


  • @coolernax said in Bullish and Bearish Divergence MACD/RSI:

    iterable

    I have created the fix, hope you can include similar indicator for RSI/MACD in the main branch:

    class RSIDivergence(bt.ind.PeriodN):
        lines = ('signal',)
        params = dict(
            rsi_period=30,
            hl_period=100,
            hl_min=25
        )
    
        def __init__(self):
            self.hfi = bt.ind.FindFirstIndexHighest(self.data.high, period=self.p.hl_period)
            self.lfi = bt.ind.FindFirstIndexLowest(self.data.low, period=self.p.hl_period)
            self.rsi = bt.ind.RSI_Safe(period=self.p.rsi_period)
    
        def signal_get(self):
            signal = 0
            if self.hfp >= self.hsp:
                if self.rsi[-int(self.hfi[0])] < self.rsi[-int(self.hsi)]:
                    signal -= 1
    
            if self.lfp <= self.lsp:
                if self.rsi[-int(self.lfi[0])] > self.rsi[-int(self.lsi)]:
                    signal += 1
    
            return signal
    
        def next(self):
            h_iterable = self.data.get(size=self.p.hl_period, ago=-int(self.hfi[0]) - self.p.hl_min)
            l_iterable = self.data.get(size=self.p.hl_period, ago=-int(self.lfi[0]) - self.p.hl_min)
    
            if len(h_iterable) > 0 and len(l_iterable) > 0:
                m = max(h_iterable)
                self.hsi = next(i for i, v in enumerate(reversed(h_iterable)) if v == m) + int(self.hfi[0]) + self.p.hl_min
    
                m = min(l_iterable)
                self.lsi = next(i for i, v in enumerate(reversed(l_iterable)) if v == m) + int(self.lfi[0]) + self.p.hl_min
    
                self.hfp = self.data.high[-int(self.hfi[0])]
                self.hsp = self.data.high[-int(self.hsi)]
                self.lfp = self.data.low[-int(self.lfi[0])]
                self.lsp = self.data.low[-int(self.lsi)]
    
                self.lines.signal[0] = self.signal_get()
            else:
                self.lines.signal[0] = 0