當價格和您的指標向相反方向移動時，就會出現背離。例如，您使用 RSI 進行交易，它上次在 80 處達到峰值，現在在 70 處達到峰值。當 RSI 達到 80 時，您交易的標的證券價格為 14 美元，現在達到新的峰值 18 美元。這是一個背離。

由于峰值的趨勢，交易者將價格稱為“更高的高點”，將 RSI 稱為“更低的高點”。技術交易者通過視覺跟蹤但很難復制，因為并不總是清楚究竟是什么造就了“峰值”。我們提供了一種算法來檢測交易的波峰和波谷，我們將在下面深入探討構建 RSI 背離策略的細節時加以利用。

作為入場信號的背離

背離通常被稱為“看跌”或“看漲”。看跌背離就像我們在上面的例子中看到的那樣。我們有一個動量指標在價格之前減弱，這給了我們一個做空的點。看漲背離是我們的動量指標出現較高的低點，但價格較低的低點。

根據這種解釋，背離是領先指標——背離發生在價格行為確認之前。在實踐中，實現這一點更具挑戰性，因為您會發現自己正在尋找價格和指標的峰值，并且直到經過一段時間后才能確認該值是峰值，因此您可以查看該值是否下降。

無論如何，讓我們用一些代碼來說明這是如何工作的！

檢測背離

第一步將需要導入幾個包。

import numpy as np import pandas as pd import matplotlib.pyplot as plt import yfinance as yf from scipy.signal import argrelextrema from collections import deque

argrelextrema?用于檢測 SciPy 信號處理庫中的峰值，而?deque?就像一個固定長度的列表，如果超過它的長度，它將刪除最舊的條目并保留新條目。我們將使用第一個來發現數據中的極值，然后循環遍歷它們并保留高于先前條目的點。

為了找出極值，我們需要傳遞一個名為?order?的參數。這定義了我們實際需要在峰的兩側有多少個點來標記峰。因此，當?order=5?時，我們需要一些東西成為左右 5 個數據點內的最高點。我們提供的另一個參數是?K，它只是一個整數，用于確定我們想要識別多少個連續峰值以確定更高的高點趨勢。

下面給出了完整的、更高的檢測功能。

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 highshigh_idx = argrelextrema(data, np.greater, order=order)[0]highs = data[high_idx]# Ensure consecutive highs are higher than previous highsextrema = []ex_deque = deque(maxlen=K)for i, idx in enumerate(high_idx):if i == 0:ex_deque.append(idx)continueif highs[i] < highs[i-1]:ex_deque.clear()ex_deque.append(idx)if len(ex_deque) == K:extrema.append(ex_deque.copy())return extrema

這將返回包含峰值索引的雙端隊列列表。為了獲得用于識別背離的所有相關組合，我們需要四個這樣的函數，一個用于更高的高點（上圖）、更低的低點、更低的高點和更高的低點。它們中的每一個的邏輯都是相同的，我們只是在第 9 行將 np.greater 更改為 np.less 并在第 18 行更改不等號以獲得我們想要的行為。

我們需要一些數據，因此我們將從 Yahoo! 使用 yfinance 包的金融 API。我將使用埃克森美孚 (XOM)，因為它在過去幾十年中經歷了相當多的繁榮和蕭條。

start = '2011-01-01' end = '2011-07-31'ticker = 'XOM' yfObj = yf.Ticker(ticker) data = yfObj.history(start=start, end=end) # Drop unused columns data.drop(['Open', 'High', 'Low', 'Volume', 'Dividends', 'Stock Splits'], axis=1, inplace=True)

現在我們可以計算所有的極值并繪制結果。

from matplotlib.lines import Line2D # For legendprice = data['Close'].values dates = data.index# Get higher highs, lower lows, etc. order = 5 hh = getHigherHighs(price, order) lh = getLowerHighs(price, order) ll = getLowerLows(price, order) hl = getHigherLows(price, order)# Get confirmation indices hh_idx = np.array([i[1] + order for i in hh]) lh_idx = np.array([i[1] + order for i in lh]) ll_idx = np.array([i[1] + order for i in ll]) hl_idx = np.array([i[1] + order for i in hl])# Plot results colors = plt.rcParams['axes.prop_cycle'].by_key()['color']plt.figure(figsize=(12, 8)) plt.plot(data['Close']) plt.scatter(dates[hh_idx], price[hh_idx-order], marker='^', c=colors[1]) plt.scatter(dates[lh_idx], price[lh_idx-order], marker='v', c=colors[2]) plt.scatter(dates[ll_idx], price[ll_idx-order], marker='v', c=colors[3]) plt.scatter(dates[hl_idx], price[hl_idx-order], marker='^', c=colors[4]) _ = [plt.plot(dates[i], price[i], c=colors[1]) for i in hh] _ = [plt.plot(dates[i], price[i], c=colors[2]) for i in lh] _ = [plt.plot(dates[i], price[i], c=colors[3]) for i in ll] _ = [plt.plot(dates[i], price[i], c=colors[4]) for i in hl]plt.xlabel('Date') plt.ylabel('Price ($)') plt.title(f'Potential Divergence Points for {ticker} Closing Price') legend_elements = [Line2D([0], [0], color=colors[0], label='Close'),Line2D([0], [0], color=colors[1], label='Higher Highs'),Line2D([0], [0], color='w', marker='^',markersize=10,markerfacecolor=colors[1],label='Higher High Confirmation'),Line2D([0], [0], color=colors[2], label='Higher Lows'),Line2D([0], [0], color='w', marker='^',markersize=10,markerfacecolor=colors[2],label='Higher Lows Confirmation'),Line2D([0], [0], color=colors[3], label='Lower Lows'),Line2D([0], [0], color='w', marker='v',markersize=10,markerfacecolor=colors[3],label='Lower Lows Confirmation'),Line2D([0], [0], color=colors[4], label='Lower Highs'),Line2D([0], [0], color='w', marker='^',markersize=10,markerfacecolor=colors[4],label='Lower Highs Confirmation') ] plt.legend(handles=legend_elements, bbox_to_anchor=(1, 0.65)) plt.show()

在這個圖中，我們提取了所有潛在的分歧點，并將高點和低點映射到價格。另外，請注意我為每個峰值繪制了確認點。我們不知道峰值是否真的是峰值，直到我們給它幾天（在這種情況下為 5 天）看看價格接下來會發生什么。

價格圖表只是背離所需的一半，我們還需要應用一個指標。借鑒考夫曼出色的交易系統和方法，我們應該使用某種動量指標。我們將繼續應用 RSI，盡管 MACD、隨機指標等也適用。

RSI 的峰值和谷值

RSI 最常被解釋為當該值高于中心線 (RSI=50) 時表現出上升勢頭，而當它低于中心線時表現出下降勢頭。如果我們有一系列高于 50 的較小峰值，則可能表明動能減弱，而低于 50 的一系列不斷增加的谷可能表明我們可以交易的動能增加。

我們的下一步是計算 RSI，然后應用與上述相同的技術來提取相關的極值。

def calcRSI(data, P=14):data['diff_close'] = data['Close'] - data['Close'].shift(1)data['gain'] = np.where(data['diff_close']>0, data['diff_close'], 0)data['loss'] = np.where(data['diff_close']<0, np.abs(data['diff_close']), 0)data[['init_avg_gain', 'init_avg_loss']] = data[['gain', 'loss']].rolling(P).mean()avg_gain = np.zeros(len(data))avg_loss = np.zeros(len(data))for i, _row in enumerate(data.iterrows()):row = _row[1]if i < P - 1:last_row = row.copy()continueelif i == P-1:avg_gain[i] += row['init_avg_gain']avg_loss[i] += row['init_avg_loss']else:avg_gain[i] += ((P - 1) * avg_gain[i-1] + row['gain']) / Pavg_loss[i] += ((P - 1) * avg_loss[i-1] + row['loss']) / Plast_row = row.copy()data['avg_gain'] = avg_gaindata['avg_loss'] = avg_lossdata['RS'] = data['avg_gain'] / data['avg_loss']data['RSI'] = 100 - 100 / (1 + data['RS'])return data

有了該功能，我們可以將 RSI 及其相關列添加到我們的數據框中：

data = calcRSI(data.copy()) # Get values to mark RSI highs/lows and plot rsi_hh = getHigherHighs(rsi, order) rsi_lh = getLowerHighs(rsi, order) rsi_ll = getLowerLows(rsi, order) rsi_hl = getHigherLows(rsi, order)

我們將遵循與上述相同的格式來繪制我們的結果：

fig, ax = plt.subplots(2, figsize=(20, 12), sharex=True) ax[0].plot(data['Close']) ax[0].scatter(dates[hh_idx], price[hh_idx-order], marker='^', c=colors[1]) ax[0].scatter(dates[lh_idx], price[lh_idx-order],marker='v', c=colors[2]) ax[0].scatter(dates[hl_idx], price[hl_idx-order],marker='^', c=colors[3]) ax[0].scatter(dates[ll_idx], price[ll_idx-order],marker='v', c=colors[4]) _ = [ax[0].plot(dates[i], price[i], c=colors[1]) for i in hh] _ = [ax[0].plot(dates[i], price[i], c=colors[2]) for i in lh] _ = [ax[0].plot(dates[i], price[i], c=colors[3]) for i in hl] _ = [ax[0].plot(dates[i], price[i], c=colors[4]) for i in ll]ax[0].set_ylabel('Price ($)') ax[0].set_title(f'Price and Potential Divergence Points for {ticker}') ax[0].legend(handles=legend_elements)ax[1].plot(data['RSI']) ax[1].scatter(dates[rsi_hh_idx], rsi[rsi_hh_idx-order], marker='^', c=colors[1]) ax[1].scatter(dates[rsi_lh_idx], rsi[rsi_lh_idx-order],marker='v', c=colors[2]) ax[1].scatter(dates[rsi_hl_idx], rsi[rsi_hl_idx-order],marker='^', c=colors[3]) ax[1].scatter(dates[rsi_ll_idx], rsi[rsi_ll_idx-order],marker='v', c=colors[4]) _ = [ax[1].plot(dates[i], rsi[i], c=colors[1]) for i in rsi_hh] _ = [ax[1].plot(dates[i], rsi[i], c=colors[2]) for i in rsi_lh] _ = [ax[1].plot(dates[i], rsi[i], c=colors[3]) for i in rsi_hl] _ = [ax[1].plot(dates[i], rsi[i], c=colors[4]) for i in rsi_ll]ax[1].set_ylabel('RSI') ax[1].set_title(f'RSI and Potential Divergence Points for {ticker}') ax[1].set_xlabel('Date')plt.tight_layout() plt.show()

這只是一個短暫的 7 個月窗口，因此我們可以清楚地看到價格和 RSI 的走勢，因此只有一個背離可見。我們在 RSI 圖表（橙色，向上的三角形）上看到 6 月中旬在價格圖表（藍色，向下的三角形）中一系列較低的低點中間確認更高的低點。我們不權衡圖表，所以讓我們把一個算法放在一起來測試這個 RSI 背離模型。

建立 RSI 發散模型

到目前為止，我們有一些通用規則來識別我們有背離的情況，但我們仍然需要進入和退出規則。首先，我們可以求助于 Kaufmann 出色的交易系統和方法，在那里他列出了一個示例策略，其中包含以下規則：

如果指標高于目標水平（例如 RSI = 50），則在確定背離時輸入頭寸。如果指標背離消失，則退出。如果我們在價格創出更高的高點而 RSI 創出更低的高點時做空，那么我們的 RSI 會移動到更高的高點，那么我們就出局了。一旦指標達到目標水平就退出。允許背離轉換為趨勢位置。為此，我們使用單獨的趨勢指標（例如 EMA 交叉），如果趨勢與背離方向相同，我們將持有頭寸。如果背離消失但趨勢繼續，我們持有，并僅在趨勢消失時退出。我們將根據 Kaufmann 規則構建兩種模型，一種僅交易背離（規則 1-3），另一種具有背離加趨勢（所有 4 條規則）。當然，您可以根據自己的需要隨意修改這些，并自己嘗試各種方法。

接下來，我將構建一些輔助函數來標記我們的峰值。第一組將修改我們的?getHigherHighs?函數組的輸出。這些是為上述可視化而構建的，但我們只需要為我們的模型提取趨勢的確認點。另請注意，由于我們正在向索引添加順序，因此我們可能會獲得會引發索引錯誤的確認點，因此我們會刪除任何大于我們擁有的數據點數量的索引。

四個函數如下：

def getHHIndex(data: np.array, order=5, K=2):extrema = getHigherHighs(data, order, K)idx = np.array([i[-1] + order for i in extrema])return idx[np.where(idx<len(data))]def getLHIndex(data: np.array, order=5, K=2):extrema = getLowerHighs(data, order, K)idx = np.array([i[-1] + order for i in extrema])return idx[np.where(idx<len(data))]def getLLIndex(data: np.array, order=5, K=2):extrema = getLowerLows(data, order, K)idx = np.array([i[-1] + order for i in extrema])return idx[np.where(idx<len(data))]def getHLIndex(data: np.array, order=5, K=2):extrema = getHigherLows(data, order, K)idx = np.array([i[-1] + order for i in extrema])return idx[np.where(idx<len(data))]

為了減少重寫代碼，我將引入一個名為?getPeaks?的函數，它獲取我們的數據幀并將我們的高點和低點的輸出編碼為列向量。它將使用我們上面定義的四個函數，并從我們觸及更高高點到?Close_highs?列分配值 1。如果我們的高點在確認較低的高點后呈下降趨勢，那么我們在同一列中用 -1 標記。它會為低點做同樣的事情。記住哪些值為 1 哪些值為 -1 很重要，因此如果趨勢正在增加（更高的高點或更高的低點），我將其設為 1，如果趨勢正在下降（更低的高點或更低的低點），我將其設為 1 ）。

def getPeaks(data, key='Close', order=5, K=2):vals = data[key].valueshh_idx = getHHIndex(vals, order, K)lh_idx = getLHIndex(vals, order, K)ll_idx = getLLIndex(vals, order, K)hl_idx = getHLIndex(vals, order, K)data[f'{key}_highs'] = np.nandata[f'{key}_highs'][hh_idx] = 1data[f'{key}_highs'][lh_idx] = -1data[f'{key}_highs'] = data[f'{key}_highs'].ffill().fillna(0)data[f'{key}_lows'] = np.nandata[f'{key}_lows'][ll_idx] = 1data[f'{key}_lows'][hl_idx] = -1data[f'{key}_lows'] = data[f'{key}_highs'].ffill().fillna(0)return data

最后，我們可以制定我們的戰略。在這里，我們只是遵循上面列出的前 3 條規則。

def RSIDivergenceStrategy(data, P=14, order=5, K=2):'''Go long/short on price and RSI divergence.- Long if price to lower low and RSI to higher low with RSI < 50- Short if price to higher high and RSI to lower high with RSI > 50Sell if divergence disappears.Sell if the RSI crosses the centerline.'''data = getPeaks(data, key='Close', order=order, K=K)data = calcRSI(data, P=P)data = getPeaks(data, key='RSI', order=order, K=K)position = np.zeros(data.shape[0])# position[:] = np.nanfor i, (t, row) in enumerate(data.iterrows()):if np.isnan(row['RSI']):continue# If no position is onif position[i-1] == 0:# Buy if indicator to higher low and price to lower lowif row['Close_lows'] == -1 and row['RSI_lows'] == 1:if row['RSI'] < 50:position[i] = 1entry_rsi = row['RSI'].copy()# Short if price to higher high and indicator to lower highelif row['Close_highs'] == 1 and row['RSI_highs'] == -1:if row['RSI'] > 50:position[i] = -1entry_rsi = row['RSI'].copy()# If current position is longelif position[i-1] == 1:if row['RSI'] < 50 and row['RSI'] < entry_rsi:position[i] = 1# If current position is shortelif position[i-1] == -1:if row['RSI'] < 50 and row['RSI'] > entry_rsi:position[i] = -1data['position'] = positionreturn calcReturns(data)def calcReturns(df):# Helper function to avoid repeating too much codedf['returns'] = df['Close'] / df['Close'].shift(1)df['log_returns'] = np.log(df['returns'])df['strat_returns'] = df['position'].shift(1) * df['returns']df['strat_log_returns'] = df['position'].shift(1) * df['log_returns']df['cum_returns'] = np.exp(df['log_returns'].cumsum()) - 1df['strat_cum_returns'] = np.exp(df['strat_log_returns'].cumsum()) - 1df['peak'] = df['cum_returns'].cummax()df['strat_peak'] = df['strat_cum_returns'].cummax()return df

關于退出條件需要注意的一件事，我們要等待趨勢的變化。我沒有等待 5 天來確認 RSI 的峰值，而是添加了一個條件，即如果 RSI 跌破我們的多頭倉位的入場 RSI 或高于我們的空頭倉位的入場 RSI，我們應該退出。這是有效的，因為如果我們在 RSI 的較低高點做空，那么如果情況逆轉，我們將退出。如果 RSI 收于我們的入場 RSI 上方，那么要么成為更高的高點，從而打破我們的趨勢，要么更高的高點仍將到來。設置這個條件只會讓我們更快地退出交易。

好了，解釋夠了，讓我們用 2000-2020 年的數據測試一下。

start = '2000-01-01' end = '2020-12-31' data = yfObj.history(start=start, end=end) # Drop unused columns data.drop(['Open', 'High', 'Low', 'Volume', 'Dividends', 'Stock Splits'], axis=1, inplace=True)df_div = RSIDivergenceStrategy(data.copy())plt.figure(figsize=(12, 8)) plt.plot(df_div['cum_returns'] * 100, label='Buy-and-Hold') plt.plot(df_div['strat_cum_returns'] * 100, label='RSI Divergence') plt.xlabel('Date') plt.ylabel('Returns (%)') plt.title(f'Buy-and-Hold and RSI Divergence Returns for {ticker}') plt.legend() plt.show()df_stats = pd.DataFrame(getStratStats(df_div['log_returns']), index=['Buy and Hold']) df_stats = pd.concat([df_stats, pd.DataFrame(getStratStats(df_div['strat_log_returns']),index=['Divergence'])])df_stats

最后，背離策略的表現優于買入并持有的策略（忽略埃克森美孚支付的股息）。它的波動性較小，跌幅較小，但在 2004 年至 2020 年期間表現不佳。換句話說，在 2020 年突破之前，你會等待 16 年，而這種策略看起來像是對底層證券的虧損。這種策略可能在其他地方更有效或適合多元化的投資組合，但至少在這種情況下， 純 RSI 背離策略看起來不太好。

RSI 背離和趨勢

對于下一個模型，讓我們采用 Kaufman 的建議并應用趨勢轉換。為此，我們將選擇 EMA 交叉。因此，該模型將像我們上面看到的背離模型一樣進行交易，但會檢查我們的 EMA 交叉所指示的趨勢。如果我們做多且 EMA1 > EMA2，我們將保持該頭寸。

EMA 計算代碼和策略如下：

def _calcEMA(P, last_ema, N):return (P - last_ema) * (2 / (N + 1)) + last_emadef calcEMA(data, N):# Initialize seriesdata['SMA_' + str(N)] = data['Close'].rolling(N).mean()ema = np.zeros(len(data))for i, _row in enumerate(data.iterrows()):row = _row[1]if i < N:ema[i] += row['SMA_' + str(N)]else:ema[i] += _calcEMA(row['Close'], ema[i-1], N)data['EMA_' + str(N)] = ema.copy()return datadef RSIDivergenceWithTrendStrategy(data, P=14, order=5, K=2, EMA1=50, EMA2=200):'''Go long/short on price and RSI divergence.- Long if price to lower low and RSI to higher low with RSI < 50- Short if price to higher high and RSI to lower high with RSI > 50Sell if divergence disappears or if the RSI crosses the centerline, unlessthere is a trend in the same direction.'''data = getPeaks(data, key='Close', order=order, K=K)data = calcRSI(data, P=P)data = getPeaks(data, key='RSI', order=order, K=K)data = calcEMA(data, EMA1)data = calcEMA(data, EMA2)position = np.zeros(data.shape[0])# position[:] = np.nanfor i, (t, row) in enumerate(data.iterrows()):if np.isnan(row['RSI']):continue# If no position is onif position[i-1] == 0:# Buy if indicator to higher high and price to lower highif row['Close_lows'] == -1 and row['RSI_lows'] == 1:if row['RSI'] < 50:position[i] = 1entry_rsi = row['RSI'].copy()# Short if price to higher high and indicator to lower highelif row['Close_highs'] == 1 and row['RSI_highs'] == -1:if row['RSI'] > 50:position[i] = -1entry_rsi = row['RSI'].copy()# If current position is longelif position[i-1] == 1:if row['RSI'] < 50 and row['RSI'] < entry_rsi:position[i] = 1elif row[f'EMA_{EMA1}'] > row[f'EMA_{EMA2}']:position[i] = 1# If current position is shortelif position[i-1] == -1:if row['RSI'] < 50 and row['RSI'] > entry_rsi:position[i] = -1elif row[f'EMA_{EMA1}'] < row[f'EMA_{EMA2}']:position[i] = -1data['position'] = positionreturn calcReturns(data)

在我們的數據上運行這個模型，我們得到：

plt.figure(figsize=(12, 8)) plt.plot(df_trend['cum_returns'] * 100, label=f'Buy-and-Hold') plt.plot(df_trend['strat_cum_returns'] * 100, label='RSI Div + Trend') plt.xlabel('Date') plt.ylabel('Returns (%)') plt.title(f'Buy-and-Hold and Divergence with Trend Returns for {ticker}') plt.legend() plt.show()df_trend = RSIDivergenceWithTrendStrategy(data.copy()) df_stats = pd.concat([df_stats, pd.DataFrame(getStratStats(df_trend['strat_log_returns']), index=['Div + Trend'])]) df_stats

添加我們的趨勢指標大大增加了我們的回報。它以較低的波動性（盡管比 RSI 背離策略更多）和較高的風險調整回報。最大回撤小于底層證券經歷的最大回撤，而且持續時間更短。

你準備好交易了嗎？

我們研究了兩種 RSI 背離策略的編碼和交易，一種很好，另一種則不然。這是否意味著您應該出去用 EMA 交叉交易 RSI 背離？

在這里給你一些關于這些指標的想法和解釋。這些快速的回測很有用，因為您可以了解如何測試想法，并且可以在各種證券和市場上進行測試以開始縮小選擇范圍。也許更嚴格的測試表明 RSI 背離是您系統中真正有價值的部分，而趨勢模型是一個異常值。除非你測試它，否則你永遠不會知道！

掃描本文最下方二維碼獲取全部完整源碼和Jupyter Notebook 文件打包下載。

↓↓長按掃碼獲取完整源碼↓↓

總結

以上是生活随笔為你收集整理的用 Python 实现 RSI 指标线与股价的顶背离、底背离的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。