Speeding Up Price and Indicator Divergence Detection
Summary
The document asks how to make a divergence detector faster on long time series. It defines divergence as two points where one signal rises while the other falls, then shows a Python routine that repeatedly searches a growing price history for local extrema, stores low indices, and compares nearby extrema against an indicator. The example identifies regular and hidden divergence patterns using price and indicator direction plus consistency checks on intervening lows.
The main performance issue is the repeated extrema search over every expanding prefix, which revisits much of the same data and can become costly. The post provides no optimized replacement, timing comparison, or validation of the detection rules; it is a request for help rather than a demonstrated solution. Its sample applies the method to average high-low price and MACD values from cryptocurrency candles, but the excerpt does not establish whether the implementation correctly handles all divergence cases or scales adequately.
Key ideas
- The example defines divergence as opposing movement between two signals over a pair of time points.
- It repeatedly recomputes local extrema over an expanding price history, creating substantial redundant work.
- It compares subsequent price lows and indicator values to label regular and hidden divergence patterns.
- The post supplies no optimized implementation or performance measurements.
Tags
Full text
# How to optimize the finding of divergences between 2 signals
# How to optimize the finding of divergences between 2 signals
I am trying to create an indicator that will find all the divergences between 2 signals.
(A divergence being defined as t1, t2 such that one signal increases between t1 and t2 while the other decreases between t1 and t2).
The output of the function so far looks like this
But the problem is that is painfully slow when I am trying to use it with long signals. Could any of you guys help me to make it faster if is possible?
UPDATE
Some sample data
My code:
```
import os
import numpy as np
import pandas as pd
from scipy.signal import argrelextrema
from sortedcontainers import SortedSet
from sqlalchemy import create_engine
from tqdm import tqdm
import pandas_ta
def find_divergence(price: pd.Series, indicator: pd.Series, width_divergence: int, order: int):
div = pd.DataFrame(index=price.index)
div[f'Berish_{width_divergence}_{order}'] = None
div[f'Berish_{width_divergence}_{order}'] = None
div[f'Bullish_idx_{width_divergence}_{order}'] = False
div[f'Berish_idx_{width_divergence}_{order}'] = False
prices_lows = SortedSet()
for i in range(len(price)):
price_data = price.values[0:i + 1]
data = argrelextrema(price_data, np.less_equal, order=order)[0]
if len(data) == 0:
continue
for j in range(len(data)):
elem = data[-1 - j]
if elem not in prices_lows:
prices_lows.add(elem)
else:
break
del data
del price_data
price_lows_idx = pd.Series(prices_lows)
for idx_1 in range(price_lows_idx.size):
min_price = price[price_lows_idx[idx_1]]
min_indicator = indicator[price_lows_idx[idx_1]]
for idx_2 in range(idx_1 + 1, idx_1 + width_divergence):
if idx_2 >= price_lows_idx.size:
break
if price[price_lows_idx[idx_2]] < min_price:
min_price = price[price_lows_idx[idx_2]]
if indicator[price_lows_idx[idx_2]] < min_indicator:
min_indicator = indicator[price_lows_idx[idx_2]]
consistency_price_rd = min_price == price[price_lows_idx[idx_2]]
consistency_indicator_rd = min_indicator == indicator[price_lows_idx[idx_1]]
consistency_price_hd = min_price == price[price_lows_idx[idx_1]]
consistency_indicator_hd = min_indicator == indicator[price_lows_idx[idx_2]]
diff_price = price[price_lows_idx[idx_1]] - price[price_lows_idx[idx_2]] # should be neg
diff_indicator = indicator[price_lows_idx[idx_1]] - indicator[price_lows_idx[idx_2]] # should be pos
is_regular_divergence = diff_price > 0 and diff_indicator < 0
is_hidden_divergence = diff_price < 0 and diff_indicator > 0
if is_regular_divergence and consistency_price_rd and consistency_indicator_rd:
div.loc[div.iloc[price_lows_idx[idx_2]].name, f'Bullish_{width_divergence}_{order}'] = f"{price_lows_idx[idx_1]} {price_lows_idx[idx_2]}"
div.loc[div.iloc[price_lows_idx[idx_2]].name, f'Bullish_idx_{width_divergence}_{order}'] = True
elif is_hidden_divergence and consistency_price_hd and consistency_indicator_hd:
div.loc[div.iloc[price_lows_idx[idx_2]].name, f'Berish_{width_divergence}_{order}'] = f"{price_lows_idx[idx_1]} {price_lows_idx[idx_2]}"
div.loc[div.iloc[price_lows_idx[idx_2]].name, f'Berish_idx_{width_divergence}_{order}'] = True
return div
if __name__ == "__main__":
source_files_path = "/home/vlad/Projects/__freqtrade-2022.3/user_data/data/binance"
files = os.listdir(source_files_path)
files_5m = [x for x in files if x.split("-")[1].split(".")[0] == "5m"]
db_name = "database_divergence.db"
if os.path.exists(db_name):
os.remove(db_name)
engine = create_engine(f'sqlite:///{db_name}', echo=False)
for pair in tqdm(files_5m):
full_name = os.path.join(source_files_path, pair)
data = pd.read_json(full_name)
data.rename(columns={0: 'date',
1: 'open',
2: 'high',
3: 'low',
4: 'close',
5: 'volume'},
inplace=True)
data['date'] = data['date'].values.astype(dtype='datetime64[ms]')
data = data.set_index('date')
data.ta.macd(append=True)
avg_price = (data['high'] + data['low']) / 2
width = 100
order = 10
divergences = find_divergences_V2_ALL(avg_price, data.MACD_12_26_9, width, order)
data = pd.concat([divergences, data], axis=1)
data.drop(data.columns.difference(divergences.columns), axis=1, inplace=True)
data.to_sql(f"{pair.split('-')[0]}_{width}_{order}", con=engine)
```Shown in full with attribution under the source's licence. Licence: CC BY-SA 4.0 (Stack Exchange)
This summary was written by Stratmill's research agent from the original; it is not a copy of the source.