How to calculate Connors RSI (CRSI)

Question

I’m trying to calculate Connor’s RSI (CRSI). RSI and ROC give correct results, but due to streaks I get incorrect results for CRSI. Does anyone know how to calculate it correctly?

def get_streaks_rsi(self, closing_prices, length):
        # logic tables
        series = pd.DataFrame(closing_prices)
        geq = series >= series.shift(1)  # True if rising
        eq = series == series.shift(1)  # True if equal
        logic_table = pd.concat([geq, eq], axis=1)

        streaks = [0]  # holds the streak duration, starts with 0

        for row in logic_table.iloc[1:].itertuples():  # iterate through logic table
            if row[2]:  # same value as before
                streaks.append(0)
                continue
            last_value = streaks[-1]
            if row[1]:  # higher value than before
                streaks.append(last_value + 1 if last_value >= 0 else 1)  # increase or reset to +1
            else:  # lower value than before
                streaks.append(last_value - 1 if last_value < 0 else -1)  # decrease or reset to -1

        streaks_numpy = np.array(streaks, dtype=np.float)
        streaks_rsi = talib.RSI(streaks_numpy, length)
        return streaks_rsi[-1]

    def get_connors_rsi(self, a, b, c):
        candles = self.client.futures_klines(symbol=self.symbol,
                                             interval=self.candles_time,
                                             limit=1500)
        closing_prices = np.array([float(candle[4]) for candle in candles])

        first_rsi = self.get_rsi(closing_prices, a)
        print('RSI:', first_rsi)
        second_rsi = self.get_streaks_rsi(closing_prices, b)
        print('STREAKS:', second_rsi)
        third_rsi = self.get_percent_rank(closing_prices, c)
        print('PERCENT_RANK:', third_rsi)
        connors_rsi = (first_rsi + second_rsi + third_rsi) / 3
        print('CONNORS RSI:', connors_rsi)
        return connors_rsi

Answer 1

I tried using your code. It seems to work when you save your 'streaks_numpy' as part of the dataframe that talib is going to be using. Im using a very similar library to talib and this is what worked for me. When i ran your code.

def get_streaks_rsi(closing_prices, length):
  # logic tables
  series = pd.DataFrame(closing_prices)
  geq = series >= series.shift(1)  # True if rising
  eq = series == series.shift(1)  # True if equal
  logic_table = pd.concat([geq, eq], axis=1)

  streaks = [0]  # holds the streak duration, starts with 0

  for row in logic_table.iloc[1:].itertuples():  # iterate through logic table
      if row[2]:  # same value as before
          streaks.append(0)
          continue
      last_value = streaks[-1]
      if row[1]:  # higher value than before
          streaks.append(last_value + 1 if last_value >= 0 else 1)  # increase or reset to +1
      else:  # lower value than before
          streaks.append(last_value - 1 if last_value < 0 else -1)  # decrease or reset to -1

  df['streaks_numpy'] = np.array(streaks, dtype=float)
  streaks_rsi = ta.momentum.rsi(df['streaks_numpy'], length)
  return streaks_rsi.iloc[-1]
get_streaks_rsi(df['Close'], 2)