Pandas: segmentação em zigue-zague de dados com base em mínimos-máximos locais

10

Eu tenho dados de séries temporais. Gerando dados

date_rng = pd.date_range('2019-01-01', freq='s', periods=400)
df = pd.DataFrame(np.random.lognormal(.005, .5,size=(len(date_rng), 3)),
                  columns=['data1', 'data2', 'data3'],
                  index= date_rng)
s = df['data1']

Quero criar uma linha em zigue-zague conectando-se entre o máximo local e o mínimo local, que satisfaça a condição de que, no eixo y, |highest - lowest value|de cada linha em zigue-zague deve exceder uma porcentagem (digamos 20%) da distância da anterior linha zig-zag, E um valor pré-estabelecido k (diga 1.2)

Eu posso encontrar os extremos locais usando este código:

# Find peaks(max).
peak_indexes = signal.argrelextrema(s.values, np.greater)
peak_indexes = peak_indexes[0]

# Find valleys(min).
valley_indexes = signal.argrelextrema(s.values, np.less)
valley_indexes = valley_indexes[0]
# Merge peaks and valleys data points using pandas.
df_peaks = pd.DataFrame({'date': s.index[peak_indexes], 'zigzag_y': s[peak_indexes]})
df_valleys = pd.DataFrame({'date': s.index[valley_indexes], 'zigzag_y': s[valley_indexes]})
df_peaks_valleys = pd.concat([df_peaks, df_valleys], axis=0, ignore_index=True, sort=True)

# Sort peak and valley datapoints by date.
df_peaks_valleys = df_peaks_valleys.sort_values(by=['date'])

mas não sei como aplicar a condição de limite a ela. Por favor, informe-me sobre como aplicar essa condição.

Como os dados podem conter milhões de registros de data e hora, é altamente recomendável um cálculo eficiente

Para uma descrição mais clara: insira a descrição da imagem aqui

Exemplo de saída, dos meus dados:

 # Instantiate axes.
(fig, ax) = plt.subplots()
# Plot zigzag trendline.
ax.plot(df_peaks_valleys['date'].values, df_peaks_valleys['zigzag_y'].values, 
                                                        color='red', label="Zigzag")

# Plot original line.
ax.plot(s.index, s, linestyle='dashed', color='black', label="Org. line", linewidth=1)

# Format time.
ax.xaxis_date()
ax.xaxis.set_major_formatter(mdates.DateFormatter("%Y-%m-%d"))

plt.gcf().autofmt_xdate()   # Beautify the x-labels
plt.autoscale(tight=True)

plt.legend(loc='best')
plt.grid(True, linestyle='dashed')

insira a descrição da imagem aqui

Minha saída desejada (algo semelhante a isso, o zigue-zague apenas conecta os segmentos significativos) insira a descrição da imagem aqui

Thanh Nguyen
fonte

Respostas:

3

Respondi ao meu melhor entendimento da questão. No entanto, não está claro como a variável K influencia o filtro.

Você deseja filtrar os extremos com base em uma condição de execução. Suponho que você queira marcar todos os extremos cuja distância relativa ao último extremo marcado seja maior que p%. Suponho ainda que você sempre considere o primeiro elemento das séries temporais um ponto válido / relevante.

Eu implementei isso com a seguinte função de filtro:

def filter(values, percentage):
    previous = values[0] 
    mask = [True]
    for value in values[1:]: 
        relative_difference = np.abs(value - previous)/previous
        if relative_difference > percentage:
            previous = value
            mask.append(True)
        else:
            mask.append(False)
    return mask

Para executar seu código, primeiro importo dependências:

from scipy import signal
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.dates as mdates

Para tornar o código reproduzível, eu corrijo a semente aleatória:

np.random.seed(0)

O resto daqui é copypasta. Observe que diminui a quantidade de amostra para tornar o resultado claro.

date_rng = pd.date_range('2019-01-01', freq='s', periods=30)
df = pd.DataFrame(np.random.lognormal(.005, .5,size=(len(date_rng), 3)),
                  columns=['data1', 'data2', 'data3'],
                  index= date_rng)
s = df['data1']
# Find peaks(max).
peak_indexes = signal.argrelextrema(s.values, np.greater)
peak_indexes = peak_indexes[0]
# Find valleys(min).
valley_indexes = signal.argrelextrema(s.values, np.less)
valley_indexes = valley_indexes[0]
# Merge peaks and valleys data points using pandas.
df_peaks = pd.DataFrame({'date': s.index[peak_indexes], 'zigzag_y': s[peak_indexes]})
df_valleys = pd.DataFrame({'date': s.index[valley_indexes], 'zigzag_y': s[valley_indexes]})
df_peaks_valleys = pd.concat([df_peaks, df_valleys], axis=0, ignore_index=True, sort=True)
# Sort peak and valley datapoints by date.
df_peaks_valleys = df_peaks_valleys.sort_values(by=['date'])

Então usamos a função de filtro:

p = 0.2 # 20% 
filter_mask = filter(df_peaks_valleys.zigzag_y, p)
filtered = df_peaks_valleys[filter_mask]

E plote como você fez tanto o enredo anterior quanto os extremos recém-filtrados:

 # Instantiate axes.
(fig, ax) = plt.subplots(figsize=(10,10))
# Plot zigzag trendline.
ax.plot(df_peaks_valleys['date'].values, df_peaks_valleys['zigzag_y'].values, 
                                                        color='red', label="Extrema")
# Plot zigzag trendline.
ax.plot(filtered['date'].values, filtered['zigzag_y'].values, 
                                                        color='blue', label="ZigZag")

# Plot original line.
ax.plot(s.index, s, linestyle='dashed', color='black', label="Org. line", linewidth=1)

# Format time.
ax.xaxis_date()
ax.xaxis.set_major_formatter(mdates.DateFormatter("%Y-%m-%d"))

plt.gcf().autofmt_xdate()   # Beautify the x-labels
plt.autoscale(tight=True)

plt.legend(loc='best')
plt.grid(True, linestyle='dashed')

insira a descrição da imagem aqui

EDIT :

Se você quiser considerar o primeiro e o último ponto como válidos, poderá adaptar a função de filtro da seguinte maneira:

def filter(values, percentage):
    # the first value is always valid
    previous = values[0] 
    mask = [True]
    # evaluate all points from the second to (n-1)th
    for value in values[1:-1]: 
        relative_difference = np.abs(value - previous)/previous
        if relative_difference > percentage:
            previous = value
            mask.append(True)
        else:
            mask.append(False)
    # the last value is always valid
    mask.append(True)
    return mask
Nikolas Rieble
fonte
oi, obrigado pela ótima resposta. Sim, sua suposição está correta "marque todos os extremos cuja distância relativa ao último extremo marcado seja maior que p%.", E o primeiro e o último ponto sempre devem ser considerados. Eu verifiquei sua resposta, às vezes faltou o último ponto, você poderia me ajudar nisso?
Thanh Nguyen
3

Você pode usar a funcionalidade de rolagem do Pandas para criar os extremos locais. Isso simplifica um pouco o código comparado à sua abordagem Scipy.

Funções para encontrar os extremos:

def islocalmax(x):
    """Both neighbors are lower,
    assumes a centered window of size 3"""
    return (x[0] < x[1]) & (x[2] < x[1])

def islocalmin(x):
    """Both neighbors are higher,
    assumes a centered window of size 3"""
    return (x[0] > x[1]) & (x[2] > x[1])

def isextrema(x):
    return islocalmax(x) or islocalmin(x)

A função para criar o ziguezague, pode ser aplicada no Dataframe de uma só vez (em cada coluna), mas isso apresentará os NaNs, pois os carimbos de data e hora retornados serão diferentes para cada coluna. Você pode descartá-las facilmente mais tarde, como mostrado no exemplo abaixo, ou simplesmente aplicar a função em uma única coluna no Dataframe.

Observe que descomentei o teste em um limite k, não tenho certeza se entendi completamente essa parte corretamente. Você pode incluí-lo se a diferença absoluta entre o extremo anterior e o atual precisar ser maior que k:& (ext_val.diff().abs() > k)

Também não tenho certeza se o zigue-zague final sempre deve passar de uma alta original para uma baixa ou vice-versa. Eu presumi que deveria, caso contrário, você pode remover a segunda pesquisa por extrema no final da função.

def create_zigzag(col, p=0.2, k=1.2):

    # Find the local min/max
    # converting to bool converts NaN to True, which makes it include the endpoints    
    ext_loc = col.rolling(3, center=True).apply(isextrema, raw=False).astype(np.bool_)

    # extract values at local min/max
    ext_val = col[ext_loc]

    # filter locations based on threshold
    thres_ext_loc = (ext_val.diff().abs() > (ext_val.shift(-1).abs() * p)) #& (ext_val.diff().abs() > k)

    # Keep the endpoints
    thres_ext_loc.iloc[0] = True
    thres_ext_loc.iloc[-1] = True

    thres_ext_loc = thres_ext_loc[thres_ext_loc]

    # extract values at filtered locations 
    thres_ext_val = col.loc[thres_ext_loc.index]

    # again search the extrema to force the zigzag to always go from high > low or vice versa,
    # never low > low, or high > high
    ext_loc = thres_ext_val.rolling(3, center=True).apply(isextrema, raw=False).astype(np.bool_)
    thres_ext_val  =thres_ext_val[ext_loc]

    return thres_ext_val

Gere alguns dados de amostra:

date_rng = pd.date_range('2019-01-01', freq='s', periods=35)

df = pd.DataFrame(np.random.randn(len(date_rng), 3),
                  columns=['data1', 'data2', 'data3'],
                  index= date_rng)

df = df.cumsum()

Aplique a função e extraia o resultado para a coluna 'data1':

dfzigzag = df.apply(create_zigzag)
data1_zigzag = dfzigzag['data1'].dropna()

Visualize o resultado:

fig, axs = plt.subplots(figsize=(10, 3))

axs.plot(df.data1, 'ko-', ms=4, label='original')
axs.plot(data1_zigzag, 'ro-', ms=4, label='zigzag')
axs.legend()

insira a descrição da imagem aqui

Rutger Kassies
fonte
obrigado pela sua resposta. Eu quero perguntar sobre esta linha (ext_val.diff().abs() > (ext_val.shift(-1).abs() * p)), como eu entendo, você está comparando a distância entre dois pontos e p%o último ponto, estou certo? Porque eu quero comparar cada segmento em zigue-zague com o segmento anterior e repita até que a condição seja satisfeita.
Thanh Nguyen