如何使用Python实现RSI背离

Question

我想知道是否有任何Python库涵盖RSI-Divergence（快速和慢速RSI之间的差异）或关于如何在Python中实现其算法的指导。

已经提出了问题：程序化检测 RSI 分歧。其中一个答案建议使用QuantConnect论坛的Python版本，但它并没有涵盖任何内容。

我无法找到其数学公式，但我能够在pine脚本中找到RSI-Divergence，如下所示，但由于无法使用TradingView调试pine脚本，因此我无法将其转换为Python。

study(title="RSI Divergence", shorttitle="RSI Divergence")
src_fast = close, len_fast = input(5, minval=1, title="Length Fast RSI")
src_slow = close, len_slow = input(14,minval=1, title="Length Slow RSI")
up_fast = rma(max(change(src_fast), 0), len_fast)
down_fast = rma(-min(change(src_fast), 0), len_fast)
rsi_fast = down_fast == 0 ? 100 : up_fast == 0 ? 0 : 100 - (100 / (1 + up_fast / down_fast))
up_slow = rma(max(change(src_slow), 0), len_slow)
down_slow = rma(-min(change(src_slow), 0), len_slow)
rsi_slow = down_slow == 0 ? 100 : up_slow == 0 ? 0 : 100 - (100 / (1 + up_slow / down_slow))
divergence = rsi_fast - rsi_slow
plotdiv = plot(divergence, color = divergence > 0 ? lime:red, linewidth = 2)
band = hline(0)

- alper

Answer 1

我发现这篇文章在下面的链接中：“在 Python 上回测 RSI 分歧策略”。

文章作者使用指数移动平均法计算 RSI 值，使用以下代码：

'''
Assuming you have a pandas OHLC Dataframe downloaded from Metatrader 5 historical data. 
'''
# Get the difference in price from previous step
Data = pd.DataFrame(Data)
delta = Data.iloc[:, 3].diff()
delta = delta[1:]

# Make the positive gains (up) and negative gains (down) Series
up, down = delta.copy(), delta.copy()
up[up < 0] = 0
down[down > 0] = 0
roll_up = pd.stats.moments.ewma(up, lookback)
roll_down = pd.stats.moments.ewma(down.abs(), lookback)

# Calculate the SMA
roll_up = roll_up[lookback:]
roll_down = roll_down[lookback:]
Data = Data.iloc[lookback + 1:,].values

# Calculate the RSI based on SMA
RS = roll_up / roll_down
RSI = (100.0 - (100.0 / (1.0 + RS)))
RSI = np.array(RSI)
RSI = np.reshape(RSI, (-1, 1))

Data = np.concatenate((Data, RSI), axis = 1)

目前，我们拥有一个包含OHLC数据和第五列中RSI的数组。接下来添加了两个列：

1. 第6列：Data[:, 5] 用于识别看涨背离，并将具有值0或1（触发买入）。
2. 第7列：Data[:, 6] 用于识别看跌背离，并将具有值0或-1（触发卖空）。

使用以下变量：

lower_barrier = 30
upper_barrier = 70
width = 10

以下是代码：

# Bullish Divergence
for i in range(len(Data)):
   try:
       if Data[i, 4] < lower_barrier:
           for a in range(i + 1, i + width):
               if Data[a, 4] > lower_barrier:
                    for r in range(a + 1, a + width):
                       if Data[r, 4] < lower_barrier and \
                        Data[r, 4] > Data[i, 4] and Data[r, 3] < Data[i, 3]:
                            for s in range(r + 1, r + width): 
                                if Data[s, 4] > lower_barrier:
                                    Data[s + 1, 5] = 1
                                    break
                                else:
                                    continue
                        else:
                            continue
                    else:
                        continue
                else:
                    continue
  except IndexError:
        pass

# Bearish Divergence
for i in range(len(Data)):
   try:
       if Data[i, 4] > upper_barrier:
           for a in range(i + 1, i + width): 
               if Data[a, 4] < upper_barrier:
                   for r in range(a + 1, a + width):
                       if Data[r, 4] > upper_barrier and \
                       Data[r, 4] < Data[i, 4] and Data[r, 3] > Data[i, 3]:
                           for s in range(r + 1, r + width):
                               if Data[s, 4] < upper_barrier:
                                   Data[s + 1, 6] = -1
                                   break
                               else:
                                   continue
                       else:
                           continue
                   else:
                       continue
               else:
                   continue
   except IndexError:
       pass

- robbinc91

---

在哪里可以下载 pandas OHLC 数据框架数据集？还是有任何示例数据集可供使用（例如：BTC:USDT）？ - alper

也许您可以使用PyPiMetaTrader5。在这个链接中，您也可以找到如何使用它的方法。 - robbinc91

Answer 2

我稍微修改了上面的代码，希望这有所帮助:

lower_barrier = 30
upper_barrier = 70
width = 5
#Bullish Divergence
for i in range(len(Data)):

   try:
     if Data.iloc[i, 4] < lower_barrier:
         for a in range(i + 1, i + width):
             if Data.iloc[a, 4] > lower_barrier:
                  for r in range(a + 1, a + width):
                     if Data.iloc[r, 4] < lower_barrier and Data.iloc[r, 4] > Data.iloc[i, 4] and Data.iloc[r, 3] < Data.iloc[i, 3]:
                         for s in range(r + 1, r + width): 
                            if Data.iloc[s, 4] > lower_barrier:
                                print('Bullish above',Data.iloc[s+1,1])
                                Data.iloc[s + 1, 5] = 1
                                break
                            else:
                                continue
                    else:
                        continue
            else:
                continue
    else:
        continue
except IndexError:
    pass
#Bearish Divergence
for i in range(len(Data)):
try:
    if Data.iloc[i, 4] > upper_barrier:
        for a in range(i + 1, i + width): 
            if Data.iloc[a, 4] < upper_barrier:
                for r in range(a + 1, a + width):
                    if Data.iloc[r, 4] > upper_barrier and Data.iloc[r, 4] < Data.iloc[i, 4] and Data.iloc[r, 3] > Data.iloc[i, 3]:
                        for s in range(r + 1, r + width):
                            if Data.iloc[s, 4] < upper_barrier:
                                print('Bearish below',Data.iloc[s+1,2])
                                Data.iloc[s + 1, 6] = -1
                                break
                            else:
                                continue
                    else:
                        continue
                else:
                    continue
            else:
                continue
except IndexError:
    pass

- uzam mohammed

---

虽然这段代码块可能回答了问题，但最好能够提供一点解释为什么它能够做到这一点。 - nik7

你做了哪些更改？被接受答案的代码中有 bug 吗？ - alper

刚刚在他调用简单的Data[]处添加了Data.iloc[]。 - uzam mohammed

Answer 3

我认为@kieran-mackle已经掌握了查找价格和任何指标之间差异的逻辑。 请参阅以下文件以了解更多。

https://github.com/kieran-mackle/AutoTrader/blob/main/autotrader/indicators.py

- Venkatesh Kuppusamy

---

虽然此链接可能回答了问题，但最好在此处包含答案的基本部分并提供链接以供参考。如果链接页面发生更改，则仅有链接的答案可能会失效。- 来自审核 - bfontaine

Answer 4

我在这篇文章中找到了一个简洁的函数。对于每个高点/低点组合都有代码，只需要改变不等式就可以轻松扩展，以便根据您想要发现多少个连续峰值来进行分歧。

def getHigherHighs(data: np.array, order=5, K=2):
  '''
  Finds consecutive higher highs in price pattern.
  Must not be exceeded within the number of periods indicated by the width 
  parameter for the value to be confirmed.
  K determines how many consecutive highs need to be higher.
  '''
  # Get highs
  high_idx = argrelextrema(data, np.greater, order=order)[0]
  highs = data[high_idx]
  # Ensure consecutive highs are higher than previous highs
  extrema = []
  ex_deque = deque(maxlen=K)
  for i, idx in enumerate(high_idx):
    if i == 0:
      ex_deque.append(idx)
      continue
    if highs[i] < highs[i-1]:
      ex_deque.clear()
 
    ex_deque.append(idx)
    if len(ex_deque) == K:
      extrema.append(ex_deque.copy())
   
  return extrema

这篇后续文章（同一网站）在此基础上构建了一个RSI背离策略，并进行了回测，如果您想深入了解，请参考。

- hubbs5