基于显式阈值的Python峰谷增减趋势状态标记实现

需求说明 #

• 正趋势区间（Positive Regime）：从谷值到峰值的连续时段，整体增幅≥指定阈值，且中途最大降幅未超过该阈值
• 负趋势区间（Negative Regime）：从峰值到谷值的连续时段，整体降幅≥指定阈值，且中途最大增幅未超过该阈值
• 区间无重叠，首尾未达阈值的波动归为中性区间；末尾趋势需延伸至最终谷值/峰值（如阈值为2时，最后负区间需延伸至-0.80谷值，而非中间-0.73点）

实现代码 #

核心模块：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'):
    """
    识别时间序列中的峰值和谷值
    
    参数:
    - df (pd.DataFrame): 包含时间和数值列的数据集
    - time_column (str): 时间列名称
    - value_column (str): 数值列名称
    
    返回:
    - df_extrema (pd.DataFrame): 包含极值点及其类型的数据集
    """
    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)
    
    # 将首尾点加入极值列表（若未被识别为极值）
    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')
    
    # 生成极值数据集并按时间排序
    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'):
    """
    根据极值点之间的波动标注趋势区间
    
    参数:
    - df_extrema (pd.DataFrame): 包含极值点的数据集
    - threshold (float): 触发趋势区间的最小波动阈值
    - value_column (str): 数值列名称
    - time_column (str): 时间列名称
    
    返回:
    - regimes_df (pd.DataFrame): 标注好的趋势区间数据集
    """
    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'):
    """
    合并连续的同类型趋势区间
    """
    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'):
    """
    识别合并后趋势区间之间的空白区间（中性波动）
    """
    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'):
    """
    将空白区间转换为趋势区间，并再次合并连续同类型区间
    """
    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):
    """
    完整处理时间序列，识别、合并、标注趋势区间
    
    参数:
    - df (pd.DataFrame): 原始时间序列数据集
    - time_column (str): 时间列名称
    - value_column (str): 数值列名称
    - threshold (float): 触发趋势区间的最小波动阈值
    
    返回:
    - df_extrema (pd.DataFrame): 极值点数据集
    - regimes_df (pd.DataFrame): 初始标注的趋势区间
    - final_regimes_df (pd.DataFrame): 最终合并后的趋势区间
    """
    if time_column not in df.columns or value_column not in df.columns:
        raise ValueError(f"数据集必须包含'{time_column}'和'{value_column}'列。")
    
    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'):
    """
    可视化时间序列，标注极值点和趋势区间
    
    参数:
    - df (pd.DataFrame): 原始时间序列数据集
    - df_extrema (pd.DataFrame): 极值点数据集
    - final_regimes_df (pd.DataFrame): 最终趋势区间数据集
    - value_column (str): 数值列名称
    - time_column (str): 时间列名称
    
    返回:
    - None: 直接显示可视化图表
    """
    plt.figure(figsize=(14, 7))
    plt.plot(df[time_column], df[value_column], label=value_column, color='blue')
    
    # 绘制峰值和谷值
    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='峰值')
    plt.plot(troughs[time_column], troughs[value_column], 'gv', markersize=10, label='谷值')
    
    # 标注极值点数值
    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')
    
    # 趋势区间颜色映射
    color_map = {'Positive': 'green', 'Negative': 'orange'}
    
    # 绘制趋势区间阴影
    for _, regime in final_regimes_df.iterrows():
        plt.axvspan(regime['Start_Time'], regime['End_Time'],
                    color=color_map[regime['Regime_Type']], alpha=0.3)
    
    # 创建自定义图例
    positive_patch = mpatches.Patch(color='green', alpha=0.3, label='正趋势区间')
    negative_patch = mpatches.Patch(color='orange', alpha=0.3, label='负趋势区间')
    
    # 合并所有图例
    plt.legend(handles=[
        plt.Line2D([], [], color='blue', label=value_column),
        plt.Line2D([], [], marker='^', color='r', linestyle='None', markersize=10, label='峰值'),
        plt.Line2D([], [], marker='v', color='g', linestyle='None', markersize=10, label='谷值'),
        positive_patch,
        negative_patch
    ], loc='upper right')
    
    # 美化图表
    plt.title('带趋势区间标注的时间序列')
    plt.xlabel(time_column)
    plt.ylabel(value_column)
    plt.grid(True)
    plt.tight_layout()
    plt.show()

测试代码：test.ipynb #

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

# 生成测试数据
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})

# 处理时间序列，设置阈值为2
df_extrema, regimes_df, final_regimes_df = process_time_series(df, value_column='Value', threshold=2, time_column='Time')

# 可视化结果
plot_regimes(df, df_extrema, final_regimes_df, value_column='Value', time_column='Time')

关键逻辑说明 #

1. 极值识别：用scipy.signal.find_peaks识别峰值，取负值后识别谷值，同时强制将首尾点纳入极值列表，避免遗漏边界趋势。
2. 趋势标注：遍历相邻极值点，若波动幅度≥阈值，根据极值类型（谷→峰/峰→谷）标注正/负趋势区间。
3. 区间合并：合并连续的同类型趋势区间，确保区间连续无间隔；同时将区间间的空白波动转换为趋势区间并再次合并，覆盖全序列。
4. 可视化：绘制原始序列、极值点（带数值标注）和趋势区间阴影，直观展示识别结果。

内容的提问来源于stack exchange，提问作者cpage