如何在 Python 中实施 RSI 背离
How to implement RSI Divergence in Python
我想知道是否有涵盖 RSI-Divergence
的 Python 库(快速和慢速 RSI
之间的区别)或关于如何在 Python.
已提出问题:Programmatically detect RSI divergence. One of the answer suggests quantconnect forum 用于 Python 版本,但它不涵盖任何内容。
我找不到它的数学公式,但我找到了 RSI-Divergence in pine-script, as below, but I was not able to convert it into Python since its not possible to debug pine-script
using 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)
我在下一个 link 上找到了这个:
Back Testing RSI Divergence Strategy on FX
作者post使用指数移动平均线进行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 的第五列。然后添加接下来的两列:
- 第 6 列: 数据[:, 5] 将用于看涨背离,值为 0 或 1(开始买入)。
- 第 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
我稍微修改了上面的代码希望这对您有所帮助:
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
这是我在 this post 中找到的一个简明函数。每个 high/low 组合都有代码,它只需要改变不等式,并且可以根据您想要发现差异的连续峰值的数量轻松扩展。
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
这个 follow up post(同一站点)在此基础上构建了 RSI 背离策略,如果您想了解更多细节,可以对其进行回测。
我想知道是否有涵盖 RSI-Divergence
的 Python 库(快速和慢速 RSI
之间的区别)或关于如何在 Python.
已提出问题:Programmatically detect RSI divergence. One of the answer suggests quantconnect forum 用于 Python 版本,但它不涵盖任何内容。
我找不到它的数学公式,但我找到了 RSI-Divergence in pine-script, as below, but I was not able to convert it into Python since its not possible to debug pine-script
using 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)
我在下一个 link 上找到了这个: Back Testing RSI Divergence Strategy on FX
作者post使用指数移动平均线进行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 的第五列。然后添加接下来的两列:
- 第 6 列: 数据[:, 5] 将用于看涨背离,值为 0 或 1(开始买入)。
- 第 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
我稍微修改了上面的代码希望这对您有所帮助:
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
这是我在 this post 中找到的一个简明函数。每个 high/low 组合都有代码,它只需要改变不等式,并且可以根据您想要发现差异的连续峰值的数量轻松扩展。
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
这个 follow up post(同一站点)在此基础上构建了 RSI 背离策略,如果您想了解更多细节,可以对其进行回测。