时间序列局部极大值与极小值检测方案咨询

待处理测试数据 #

+---------------------+------------------+--+ 
|      date_add       |           fnv_wa |  |
+---------------------+------------------+--+
| 2022-06-24 06:00:16 | 46.216866        |  |
| 2022-06-24 07:00:16 | 46.216866        |  |
| 2022-06-24 08:00:16 | 45.685139        |  |
| 2022-06-24 09:00:16 | 45.633936        |  |
| 2022-06-24 10:00:16 | 43.487337        |  |
| 2022-06-24 11:00:16 | 40.182756        |  |
| 2022-06-24 12:00:16 | 40.017330        |  |
| 2022-06-24 13:00:16 | 39.548623        |  |
| 2022-06-24 14:00:16 | 39.548623        |  |
| 2022-06-24 15:00:16 | 38.607271        |  |
| 2022-06-24 16:00:16 | 39.989759        |  |
| 2022-06-24 17:00:16 | 39.111426        |  |
| 2022-06-24 18:00:16 | 37.862854        |  |
| 2022-06-24 19:00:16 | 37.862854        |  |
| 2022-06-24 20:00:16 | 37.862854        |  |
| 2022-06-24 21:00:16 | 36.173146        |  |
| 2022-06-24 22:00:16 | 35.164835        |  |
+---------------------+------------------+--+

已尝试方案的缺陷 #

现有两种常用极值检测方案均无法完整识别所有极值，会漏检X≈10位置的局部极大值：

• 方案1：基于scipy.signal.argrelextrema实现
  问题：大窗口场景下识别失效，原代码如下：

  df['min'] = df.iloc[argrelextrema(df.data.values, np.less_equal,
                      order=n)[0]]['data']
  df['max'] = df.iloc[argrelextrema(df.data.values, np.greater_equal,
                      order=n)[0]]['data']
  

• 方案2：基于DataFrame shift相邻比较实现
  问题：连续相等值组成的平段场景下漏判极值，原代码如下：

  df['min'] = df.data[(df.data.shift(1) > df.data) & (df.data.shift(-1) > df.data)]
  df['max'] = df.data[(df.data.shift(1) < df.data) & (df.data.shift(-1) < df.data)]
  

漏检核心原因：两种方案均未兼容连续相等数值的平段场景，仅做单点前后值比较时，平段内的点不满足严格大于/小于相邻点的条件，就会被过滤。

可行实现方案 #

方案1：无额外依赖的平段兼容版shift实现 #

核心逻辑是先对连续相等的数值做分组，以平段为单位比较前后趋势，判断是否为极值，适合小规模数据集：

import pandas as pd
import numpy as np

# 为连续相等的数值段分配唯一分组ID
df['seg_id'] = (df['fnv_wa'].diff() != 0).cumsum()
# 聚合得到每个平段的代表值、相邻段数值
seg_info = df.groupby('seg_id')['fnv_wa'].first().reset_index()
seg_info['prev_val'] = seg_info['fnv_wa'].shift(1)
seg_info['next_val'] = seg_info['fnv_wa'].shift(-1)

# 识别极值段：比前后段都大是极大值，比前后段都小是极小值
max_seg = seg_info[(seg_info['fnv_wa'] > seg_info['prev_val']) & (seg_info['fnv_wa'] > seg_info['next_val'])]['seg_id']
min_seg = seg_info[(seg_info['fnv_wa'] < seg_info['prev_val']) & (seg_info['fnv_wa'] < seg_info['next_val'])]['seg_id']

# 映射回原数据表
df['local_max'] = np.where(df['seg_id'].isin(max_seg), df['fnv_wa'], np.nan)
df['local_min'] = np.where(df['seg_id'].isin(min_seg), df['fnv_wa'], np.nan)

针对提供的测试集，该方案可准确识别全部极值：

• 局部极大值：起始段46.216866、16点位置39.989759（即之前漏检的X≈10位置峰值）
• 局部极小值：15点位置38.607271、末尾段35.164835

方案2：scipy优化参数实现 #

换用scipy.signal.find_peaks函数，通过plateau_size参数开启平段极值识别，适合大规模时序数据，支持自定义窗口大小，无大窗口识别失效问题：

from scipy.signal import find_peaks
import numpy as np

vals = df['fnv_wa'].values
# 识别极大值，plateau_size设为1即可支持识别长度≥1的平段峰值
max_idx, _ = find_peaks(vals, plateau_size=1)
# 对数值取反后找峰值，即为原序列极小值
min_idx, _ = find_peaks(-vals, plateau_size=1)

# 写入结果
df['local_max_scipy'] = np.nan
df.loc[max_idx, 'local_max_scipy'] = vals[max_idx]
df['local_min_scipy'] = np.nan
df.loc[min_idx, 'local_min_scipy'] = vals[min_idx]

如果需要识别更大时间窗口的极值，只需要调整find_peaks的distance、wlen参数即可，无需修改核心逻辑。

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