Create boolean flag in pandas from signal's crossings

Viewed 215

I would like to create a flag with a function and applying it to one column in a pandas dataframe. The intention of the function is to set the value 1 when the signal crosses upwards over -1 and resets the value to 0 when the signal crosses 1 downwards. Here is my code example: I just cant get the function to work

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
 
x = np.arange(0, 10, 0.01)
x2 = np.arange(0, 20, 0.02)
sin1 = np.sin(x)
sin2 = np.sin(x2)
x2 /= 2
sin3 = sin1 + sin2
 
 
df = pd.DataFrame(sin3)
#name signal column
df.columns = ['signal']
df.signal.plot()
 
 
def my_flag(x):
   
    #cross over -1
    ok1 = (x.iloc[-1] > -1)*1
    ok2 = (x.iloc[-2] < -1)*1
   
    activate = (ok1*ok2) > 0.5
    if activate:
        flag_activate = 1
    # OFF
    #cross under 1
    ok3 = (x.iloc[-1] <1)*1
    ok4 = (x.iloc[-2] > 1)*1
   
    inactivate = (ok3*ok4) > 0.5
    if inactivate:       
        flag_activate = 0
    # # add to df
   
    return flag_activate
 
df['the_flag'] = df['signal'].apply(my_flag)
 
#I have set the flag to 0 for plotting purposes for demo,
# should be replaced when my_flag function works
df['the_flag'] = 0
fig, (ax1,ax2) = plt.subplots(2)
ax1.plot(df['signal'])
ax1.set_title('signal')
y1 = -1
y2 = 1
ax1.axhline(y1,color='r') 

I have made a "cartoon picture" of what I would like the flag to llook like for a sine signal: enter image description here

2 Answers

We can first detect the -1 and +1 crossings whilst considering they should cross-up and cross-down, respectively. This can be done via shifting the signal to left and right by 1 and comparing against -/+ 1 with the crossing behaviour in mind:

neg_1_crossings = np.where((sin3[:-1] < -1) & (sin3[1:] > -1))[0]

pos_1_crossings = np.where((sin3[:-1] > +1) & (sin3[1:] < +1))[0]

For -1 cross-up's: First mask imposes previous values be less than -1, second one imposes next values be greater then -1. Similar for the +1, except operators flipped.

Now we have:

>>> neg_1_crossings
array([592], dtype=int64)

>>> pos_1_crossings
array([157, 785], dtype=int64)

I'd run for loops here to get the flag:

flag = np.zeros_like(sin3)

for neg_cross in neg_1_crossings:
    # a `neg_cross` raises the flag
    flag[neg_cross:] = 1
    
    for pos_cross in pos_1_crossings:
        if pos_cross > neg_cross:
            # once we hit a `pos_cross` later on, restrict the flag's ON
            # periods to be between the `neg_cross` and this `pos_cross`
            flag[pos_cross:] = 0
            
            # we are done with this `neg_cross`
            break

which gives

signal+flag

Overall:

def get_flag(col):
    """
    `col` is a pd.Series
    """
    # signal in numpy domain; also its shifted versions
    signal = col.to_numpy()
    sig_shifted_left = signal[1:]
    sig_shifted_right = signal[:-1]

    # detect crossings
    neg_1_crossings = np.where((sig_shifted_right < -1) & (sig_shifted_left > -1))[0]
    pos_1_crossings = np.where((sig_shifted_right > +1) & (sig_shifted_left < +1))[0]

    # form the `flag` signal
    flag = np.zeros_like(signal)

    for neg_cross in neg_1_crossings:
        # a `neg_cross` raises the flag
        flag[neg_cross:] = 1
        
        for pos_cross in pos_1_crossings:
            if pos_cross > neg_cross:
                # once we hit a `pos_cross` later on, restrict the flag's ON
                # periods to be between the `neg_cross` and this `pos_cross`
                flag[pos_cross:] = 0
                
                # we are done with this `neg_cross`
                break
     return flag

You can use shift and query to find where the signal crosses your interval boundaries

df["shifted"] = df.signal.shift(-1)

start = df.query("shifted <= -1 and signal >= -1")
stop = df.query("shifted <= 1 and signal >= 1")

then you can use these crossings to set your flag column, probably there's some more compact way to do this in pandas

df["flag"] = False
# pair each left boundary with the closest right one, if any
for l in start.index.values:
    try:
        r = stop.index.values[stop.index.values > l][0]
        df.loc[l:r, "flag"] = True
    except:
        continue

Let's see if this works:

df.signal.plot()
start.signal.plot(marker="o", lw=0)
stop.signal.plot(marker="o", lw=0)
df.flag.astype(int).plot()

plot

Related