如何在 Python 中实现 RSI Divergence

Question

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

已经提出的问题：以编程方式检测 RSI 分歧。答案之一建议使用 Python 版本的quantconnect 论坛，但它没有涵盖任何内容。

我无法找到它的数学公式，但我能够在 pine-script 中找到RSI-Divergence，如下所示，但我无法将其转换为 Python，因为它无法pine-script使用tradingview进行调试。

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)

Answer 1

我在下一个链接中找到了这一点： 回溯测试 FX 上的 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 列：数据[:, 5] 将用于看涨背​​离，其值为 0 或 1（启动买入）。
2. 第 7 列：数据[:, 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