Calculate intermediate highs and lows given a minimum price movement threshold

I'm looking to get the high and low reversals/pivots in a price series given a minimum price movement threshold and wondering if there are any existing python libraries that can do this. Essentially, I'm looking for something that would return the indexes of the arrows that could be customized for more or less sensitivity. That is, if the threshold were higher, there would be fewer reversal arrows and more arrows for a lower threshold.

Anyone know of an existing library or otherwise easy way to do this?

I recently wrote a non-vectorised, looped Octave function to do just this, code below

## Copyright (C) 2019 dekalog
##
## This program is free software: you can redistribute it and/or modify it
## the Free Software Foundation, either version 3 of the License, or
## (at your option) any later version.
##
## This program is distributed in the hope that it will be useful, but
## WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
## GNU General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with this program.  If not, see

## -*- texinfo -*-
## @deftypefn {} {@var{[ tps , } @var{smooth ]} =} turning_point_filter(@var{ price }, @var{n_bar })
##
## Finds peaks and troughs in the PRICE sequence, determined by looking N_BARS forwards and
## backwards along PRICE sequence from each PRICE point in the sequence.
##
## A peak (trough) is determined by a PRICE point being higher (lower) than
## the N_BARS on either side of it. If N_BAR is not given, the default value is 2.
##
## Internally the function performs some checks to ensure that:
##
## 1) the peaks and troughs form an alternating sequence, and
##
## 2) adjacent peaks and troughs are separated by at least one bar.
##
## @seealso{}
## @end deftypefn

## Author: dekalog <dekalog@dekalog>
## Created: 2019-10-08

function [ tps , smooth ] = turning_point_filter( price , n_bar )

## ensure price is a column vector
if ( size( price , 1 ) == 1 && size( price , 2 ) > 1 )
price = price' ;
endif

## get n_bar
if ( nargin == 1 ) ## no user supplied n_bar
n_bar = 2 ;
endif

tps = zeros( size( price , 1 ) , 2 ) ;
B = [ ( 1 : n_bar ) fliplr( ( 1 : n_bar ) ) ] ; B = B ./ sum( B)  ;
smooth = filter( B , 1 , price ) ;
smooth = filter( [ 0.5 0.5 ] , 1 , smooth ) ;
smooth = shift( smooth , -n_bar ) ;
last_peak_ix = 1 ; last_trough_ix = 1 ;

for ii = n_bar + 1 : size( price , 1 ) - n_bar

if( smooth( ii ) > smooth( ii - 1 ) && smooth( ii ) > smooth( ii + 1 ) ) ## a possible peak?

[ ~ , max_ix ] = max( smooth( ii - n_bar : ii + n_bar ) ) ;
if( max_ix == n_bar + 1 )
[ ~ , max_ix ] = max( price( ii - n_bar : ii + n_bar ) ) ;
ix_correction = max_ix - ( n_bar + 1 ) ;
new_peak_ix = ii + ix_correction ;

if( last_peak_ix <= last_trough_ix && new_peak_ix > last_trough_ix ) ## alternating peak, trough and peak?

if( new_peak_ix - last_trough_ix > 1 ) ## and not too close to previous trough
tps( new_peak_ix , 1 ) = 1 ;
last_peak_ix = new_peak_ix ;
endif

elseif( last_peak_ix > last_trough_ix && new_peak_ix > last_trough_ix ) ## non alternating trough, peak and peak?

if( price( new_peak_ix ) > price( last_peak_ix ) ) ## a new higher peak?
tps( last_peak_ix , 1 ) = 0 ;
tps( new_peak_ix , 1 ) = 1 ;
last_peak_ix = new_peak_ix ;
endif

endif

endif

elseif( smooth( ii ) < smooth( ii - 1 ) && smooth( ii ) < smooth( ii + 1 ) ) ## a possible trough?

[ ~ , min_ix ] = min( smooth( ii - n_bar : ii + n_bar ) ) ;
if( min_ix == n_bar + 1 )
[ ~ , min_ix ] = min( price( ii - n_bar : ii + n_bar ) ) ;
ix_correction = min_ix - ( n_bar + 1 ) ;
new_trough_ix = ii + ix_correction ;

if( last_trough_ix <= last_peak_ix && new_trough_ix > last_peak_ix ) ## alternating trough, peak and trough?

if( new_trough_ix - last_peak_ix > 1 ) ## and not too close to previous peak
tps( new_trough_ix , 2 ) = 1 ;
last_trough_ix = new_trough_ix ;
endif

elseif( last_trough_ix > last_peak_ix && new_trough_ix > last_peak_ix ) ## non alternating peak, trough and trough?

if( price( new_trough_ix ) < price( last_trough_ix ) ) ## a new lower trough?
tps( last_trough_ix , 2 ) = 0 ;
tps( new_trough_ix , 2 ) = 1 ;
last_trough_ix = new_trough_ix ;
endif

endif

endif

else
## do nothing
endif

endfor ## end of ii loop

endfunction


and an example plot Because it is a non-vectorised function, the logic in the loop should easily be able to be converted into Python. Please note that the function looks forward along the price series and so would not be suitable for online use, however it would be suitable for offline use in, for example, creating training data labels for machine learning purposes.

• Thanks for your answer. I ended up getting it to work and had a chart that looked almost identical to yours in terms of the vertical lines. I’ll post my answer this evening. Commented Mar 5, 2020 at 17:07
• Also, the reason I asked the question is exactly what you mentioned in the last sentence of your answer. Seemingly, capturing more information in the label values is preferable to something like simply predicting the price at t + x. I just had the idea last night, so I haven’t thought in depth about how to quantify a full range of values. Something like a stochastic indicator in reverse for the label values might be a start, but I’d be curious to hear if you’ve had success with anything that you’d like to share (but understandable if you prefer not to). Commented Mar 5, 2020 at 17:25
• @SuperCodeBrah If you're thinking along the lines of a stochastic indicator you might be interested in a Turning Point Oscillator as described in the paper at arxiv.org/abs/1209.0127 Commented Mar 6, 2020 at 15:40

This appears to work for python. Anyone seeing this might want to customize how the beginning and end are handled. I'm planning to truncate the data from both sides, so I really just needed the middle swings (the main logic section below).

def get_price_swings(prices, threshold):

last_swing = None # 1 or -1
high_swings = [0]
low_swings = [0]
last_price = prices[0]

for i, price in enumerate(prices):

# section is optional to set initial swing high/low from initial price
if len(high_swings) == 1 and price >= prices[high_swings[-1]]:
high_swings[0] = i
last_swing = 1
elif len(low_swings) == 1 and price <= prices[low_swings[-1]]:
low_swings[0] = i
last_swing = -1

# main logic
elif price - threshold > prices[low_swings[-1]] and price > last_price:
if last_swing == 1:
if price > prices[high_swings[-1]]:
high_swings[-1] = i
else:
high_swings.append(i)
last_swing = 1

elif price + threshold < prices[high_swings[-1]] and price < last_price:
if last_swing == -1:
if price < prices[low_swings[-1]]:
low_swings[-1] = i
else:
low_swings.append(i)
last_swing = -1

last_price = price

# section is optional to set last period as a high or low
last_period = len(prices) - 1
if high_swings[-1] > low_swings[-1] and high_swings[-1] < last_period:
low_swings.append(last_period)
elif low_swings[-1] > high_swings[-1] and low_swings[-1] < last_period:
high_swings.append(last_period)

return high_swings, low_swings