Calcular máximos y mínimos intermedios dado un umbral mínimo de movimiento de precios

Question

Estoy buscando obtener los reversos/pivotes altos y bajos en una serie de precios dado un umbral mínimo de movimiento de precios y me pregunto si hay alguno existente python bibliotecas que pueden hacer esto. Esencialmente, estoy buscando algo que devuelva los índices de las flechas que puedan ser personalizados para mayor o menor sensibilidad. Es decir, si el umbral fuera más alto, habría menos flechas de inversión y más flechas para un umbral más bajo.

¿Alguien sabe de una biblioteca existente o de una manera fácil de hacer esto?

Answer 1

Recientemente escribí una función de Octave no vectorizada y en bucle para hacer precisamente esto, el código es el siguiente

## Copyright (C) 2019 dekalog
## 
## This program is free software: you can redistribute it and/or modify it
## under the terms of the GNU General Public License as published by
## 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
## <https://www.gnu.org/licenses/>.

## -*- 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

y un ejemplo de gráfico Al ser una función no vectorizada, la lógica del bucle debería poder convertirse fácilmente en Python. Tenga en cuenta que la función mira hacia adelante a lo largo de la serie de precios y por lo tanto no sería adecuado para el uso en línea, sin embargo, sería adecuado para el uso fuera de línea en, por ejemplo, la creación de etiquetas de datos de formación para fines de aprendizaje automático.

Answer 2

Esto parece funcionar para python. Cualquiera que vea esto podría querer personalizar cómo se manejan el principio y el final. Estoy planeando truncar los datos de ambos lados, por lo que realmente sólo necesitaba las oscilaciones del medio (el main logic sección más abajo).

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:
                # price already on upswing
                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:
                # price already on downswing
                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