Threshold-Based Turning Point Regimes for Time Series

# Python Regime Labeling Using Explicit Threshold for Increase/Decrease from peak to trough


# Python Regime Labeling Using Explicit Threshold for Increase/Decrease from peak to trough












I am trying to find the longest stretches of time in a time series where the value from start to end increases by at least a certain threshold without any declines in the interim by at least that threshold. Think of the time series like a stock chart. Suppose the threshold is \$50. Find and label the longest stretches of time where the stock goes up by at least \$50 without going down by \$50 in the interim. Those long stretches are then labeled the "positive regime". Repeat the process for declines of at least \$50 without going up by \$50 in the interim and label those "negative regime".

Since positive regimes start at troughs and end at peaks, and negative regimes start at peaks and end at troughs, regimes should never overlap.

Below is an illustrated example of Positive and Negative Regimes with a threshold of 2. Note the first peak and trough are in an neutral regime since there is no decline greater than the threshold 2 prior to the start of the first positive regime. The final negative regime, however, should be extended to the -0.80 through though since -0.80 is less than -0.73 (where it currently stops) and there is no increase of at least 2 between the current end and the correct end.

How would you implement this in `python`? I have tried using `scipy.signal.find_peaks` but can't get it to work and I feel like there should be a simpler way I'm missing.

```
#test.ipynb

import numpy as np
import pandas as pd
from regimes import plot_regimes, process_time_series

# Sample Data Generation
np.random.seed(0)
time = np.arange(0, 100, 1)
value = np.sin(np.linspace(0, 10 * np.pi, 100)) + np.random.normal(0, 1, 100)
df = pd.DataFrame({'Time': time, 'Value': value})

# Process the time series data
df_extrema, regimes_df, final_regimes_df = process_time_series(df, value_column='Value', threshold=2, time_column='Time')

# Plot the final regimes along with peaks and troughs
plot_regimes(df, df_extrema, final_regimes_df, value_column='Value', time_column='Time')
```

```
# regimes.py
import pandas as pd
import numpy as np
from scipy.signal import find_peaks
import matplotlib.pyplot as plt
import matplotlib.patches as mpatches

def find_extrema(df, time_column='Time', value_column='Value'):
    """
    Identifies peaks and troughs in the given time series data.
    
    Parameters:
    - df (pd.DataFrame): DataFrame containing time and value columns.
    - time_column (str): Name of the column containing the time data.
    - value_column (str): Name of the column containing the time series values.
    
    Returns:
    - df_extrema (pd.DataFrame): DataFrame with extrema points and their types.
    """
    peaks, _ = find_peaks(df[value_column])
    troughs, _ = find_peaks(-df[value_column])
    
    extrema_indices = list(peaks) + list(troughs)
    extrema_types = ['Peak'] * len(peaks) + ['Trough'] * len(troughs)
    
    # Include first and last points as extrema
    if 0 not in extrema_indices:
        first_val = df.iloc[0][value_column]
        second_val = df.iloc[1][value_column]
        extrema_indices.append(0)
        extrema_types.append('Peak' if first_val > second_val else 'Trough')
    
    last_idx = len(df) - 1
    if last_idx not in extrema_indices:
        last_val = df.iloc[last_idx][value_column]
        second_last_val = df.iloc[last_idx - 1][value_column]
        extrema_indices.append(last_idx)
        extrema_types.append('Peak' if last_val > second_last_val else 'Trough')
    
    # Create the extrema DataFrame
    extrema = sorted(extrema_indices)
    sorted_extrema_types = [extrema_types[extrema_indices.index(idx)] for idx in extrema]
    df_extrema = df.iloc[extrema].copy().reset_index(drop=True)
    df_extrema['Type'] = sorted_extrema_types
    
    return df_extrema

def label_regimes(df_extrema, threshold=0.25, value_column='Value', time_column='Time'):
    """
    Labels regimes based on transitions between extrema points.
    
    Parameters:
    - df_extrema (pd.DataFrame): DataFrame containing extrema points and types.
    - threshold (float): Minimum height change to consider as a regime.
    - value_column (str): Column name for time series values.
    - time_column (str): Column name for time values.
    
    Returns:
    - regimes_df (pd.DataFrame): DataFrame containing labeled regimes.
    """
    regimes = []
    for i in range(len(df_extrema) - 1):
        current = df_extrema.iloc[i]
        next_ = df_extrema.iloc[i + 1]
        height_change = next_[value_column] - current[value_column]
        
        if abs(height_change) >= threshold:
            regime_type = 'Positive' if current['Type'] == 'Trough' and next_['Type'] == 'Peak' else 'Negative'
            regimes.append({
                'Start_Time': current[time_column],
                'End_Time': next_[time_column],
                'Start_Type': current['Type'],
                'End_Type': next_['Type'],
                'Height_Change': height_change if regime_type == 'Positive' else -height_change,
                'Regime_Type': regime_type
            })
    
    return pd.DataFrame(regimes)

def merge_consecutive_regimes(regimes_df, df, time_column='Time', value_column='Value'):
    """
    Merges consecutive regimes of the same type.
    """
    if regimes_df.empty:
        return regimes_df.copy()
    
    merged_regimes = []
    current_regime = regimes_df.iloc[0].copy()
    
    for i in range(1, len(regimes_df)):
        next_regime = regimes_df.iloc[i]
        
        if next_regime['Regime_Type'] == current_regime['Regime_Type']:
            current_regime['End_Time'] = next_regime['End_Time']
            current_regime['End_Type'] = next_regime['End_Type']
            start_value = df.loc[df[time_column] == current_regime['Start_Time'], value_column].values[0]
            end_value = df.loc[df[time_column] == current_regime['End_Time'], value_column].values[0]
            
            current_regime['Height_Change'] = (
                end_value - start_value if current_regime['Regime_Type'] == 'Positive' else start_value - end_value
            )
        else:
            merged_regimes.append(current_regime)
            current_regime = next_regime.copy()
    
    merged_regimes.append(current_regime)
    return pd.DataFrame(merged_regimes).reset_index(drop=True)

def calculate_whitespace(merged_regimes_df, df, time_column='Time', value_column='Value'):
    """
    Identifies gaps between merged regimes as whitespace.
    """
    whitespace = []
    for i in range(len(merged_regimes_df) - 1):
        current_regime = merged_regimes_df.iloc[i]
        next_regime = merged_regimes_df.iloc[i + 1]
        
        current_end_time = current_regime['End_Time']
        next_start_time = next_regime['Start_Time']
        
        if next_start_time > current_end_time:
            current_end_value = df.loc[df[time_column] == current_end_time, value_column].values[0]
            next_start_value = df.loc[df[time_column] == next_start_time, value_column].values[0]
            whitespace_type = 'Positive' if current_end_value < next_start_value else 'Negative'
            
            whitespace.append({
                'Start_Time': current_end_time,
                'End_Time': next_start_time,
                'Start_Height': current_end_value,
                'End_Height': next_start_value,
                'Whitespace_Type': whitespace_type
            })
    
    return pd.DataFrame(whitespace)

def add_whitespace_as_regimes_and_merge(merged_regimes_df, whitespace_df, df, time_column='Time', value_column='Value'):
    """
    Adds whitespace as new regimes and merges consecutive regimes of the same type.
    """
    if whitespace_df.empty:
        return merged_regimes_df.copy()
    
    whitespace_regimes = []
    for _, ws in whitespace_df.iterrows():
        start_time = ws['Start_Time']
        end_time = ws['End_Time']
        whitespace_type = ws['Whitespace_Type']
        start_type = 'Trough' if whitespace_type == 'Positive' else 'Peak'
        end_type = 'Peak' if whitespace_type == 'Positive' else 'Trough'
        height_change = ws['End_Height'] - ws['Start_Height'] if whitespace_type == 'Positive' else ws['Start_Height'] - ws['End_Height']
        
        whitespace_regimes.append({
            'Start_Time': start_time,
            'End_Time': end_time,
            'Start_Type': start_type,
            'End_Type': end_type,
            'Height_Change': height_change,
            'Regime_Type': whitespace_type
        })
    
    whitespace_regimes_df = pd.DataFrame(whitespace_regimes)
    combined_regimes = pd.concat([merged_regimes_df, whitespace_regimes_df], ignore_index=True)
    combined_regimes = combined_regimes.sort_values(by='Start_Time').reset_index(drop=True)
    
    return merge_consecutive_regimes(combined_regimes, df, time_column, value_column)

def process_time_series(df, time_column='Time', value_column='Value', threshold=0.25):
    """
    Processes the time series data to identify, merge, and label regimes.
    
    Parameters:
    - df (pd.DataFrame): DataFrame containing time and value data.
    - time_column (str): Column name for time data.
    - value_column (str): Column name for time series values.
    - threshold (float): Height change threshold to identify significant regimes.
    
    Returns:
    - df_extrema (pd.DataFrame): DataFrame of identified peaks and troughs.
    - regimes_df (pd.DataFrame): DataFrame of initially labeled regimes.
    - final_regimes_df (pd.DataFrame): DataFrame of final labeled regimes.
    """
    if time_column not in df.columns or value_column not in df.columns:
        raise ValueError(f"DataFrame must contain '{time_column}' and '{value_column}' columns.")
    
    df = df.sort_values(by=time_column).reset_index(drop=True)
    df_extrema = find_extrema(df, time_column, value_column)
    regimes_df = label_regimes(df_extrema, threshold, value_column, time_column)
    merged_regimes_df = merge_consecutive_regimes(regimes_df, df, time_column, value_column)
    whitespace_df = calculate_whitespace(merged_regimes_df, df, time_column, value_column)
    final_regimes_df = add_whitespace_as_regimes_and_merge(merged_regimes_df, whitespace_df, df, time_column, value_column)
    
    return df_extrema, regimes_df, final_regimes_df

def plot_regimes(df, df_extrema, final_regimes_df, value_column='Value', time_column='Time'):
    """
    Plots the time series with shaded regimes, highlighting peaks and troughs and labeling their heights.
    
    Parameters:
    - df (pd.DataFrame): Original time series data.
    - df_extrema (pd.DataFrame): DataFrame with identified peaks and troughs.
    - final_regimes_df (pd.DataFrame): DataFrame containing the final regimes.
    - value_column (str): Column name for time series values.
    - time_column (str): Column name for time.
    
    Returns:
    - None: Displays the plot.
    """
    plt.figure(figsize=(14, 7))
    plt.plot(df[time_column], df[value_column], label=value_column, color='blue')
    
    # Plot peaks and troughs
    peaks = df_extrema[df_extrema['Type'] == 'Peak']
    troughs = df_extrema[df_extrema['Type'] == 'Trough']
    
    plt.plot(peaks[time_column], peaks[value_column], 'r^', markersize=10, label='Peaks')
    plt.plot(troughs[time_column], troughs[value_column], 'gv', markersize=10, label='Troughs')
    
    # Annotate peak and trough heights
    for _, row in peaks.iterrows():
        plt.text(row[time_column], row[value_column], f"{row[value_column]:.2f}", 
                 fontsize=9, color='red', ha='center', va='bottom')
    
    for _, row in troughs.iterrows():
        plt.text(row[time_column], row[value_column], f"{row[value_column]:.2f}", 
                 fontsize=9, color='green', ha='center', va='top')
    
    # Define colors for regimes
    color_map = {'Positive': 'green', 'Negative': 'orange'}
    
    # Shade final regimes
    for _, regime in final_regimes_df.iterrows():
        plt.axvspan(regime['Start_Time'], regime['End_Time'],
                    color=color_map[regime['Regime_Type']], alpha=0.3)
    
    # Create custom legend for shaded regimes
    positive_patch = mpatches.Patch(color='green', alpha=0.3, label='Positive Regime')
    negative_patch = mpatches.Patch(color='orange', alpha=0.3, label='Negative Regime')
    
    # Combine all legends
    plt.legend(handles=[
        plt.Line2D([], [], color='blue', label=value_column),
        plt.Line2D([], [], marker='^', color='r', linestyle='None', markersize=10, label='Peaks'),
        plt.Line2D([], [], marker='v', color='g', linestyle='None', markersize=10, label='Troughs'),
        positive_patch,
        negative_patch
    ], loc='upper right')
    
    # Enhance plot
    plt.title('Time Series with Shaded Positive and Negative Regimes')
    plt.xlabel(time_column)
    plt.ylabel(value_column)
    plt.grid(True)
    plt.tight_layout()
    plt.show()
```