如何统计从0起始并回落至0的峰值？Python find_peaks用法

find_peaks的内置参数无法直接过滤出“必须从0起始、之后回落至0”的峰值，因为它仅关注峰值自身的局部特征（高度、突出度等），不校验峰值前后序列是否回到基线（0）。需要通过额外的过滤逻辑实现需求，具体步骤如下：

1. 先找出所有基线（0）的位置 #

由于浮点精度问题，避免直接判断array == 0，用小阈值识别接近0的点：

import numpy as np
zero_indices = np.where(np.abs(array) < 1e-6)[0]

2. 编写过滤函数验证峰值的前后基线 #

对每个候选峰值，检查其前后是否存在0点，确保峰值是“从0升起、回落至0”的独立峰：

def filter_valid_peaks(candidate_peaks):
    valid_peaks = []
    for peak in candidate_peaks:
        # 找峰值前最近的0点
        prev_zero = zero_indices[zero_indices < peak][-1] if len(zero_indices[zero_indices < peak]) > 0 else -1
        # 找峰值后最近的0点
        next_zero = zero_indices[zero_indices > peak][0] if len(zero_indices[zero_indices > peak]) > 0 else len(array)
        
        # 验证逻辑：
        # - 峰值前后都有0点，说明是完整的“0-峰值-0”区间
        # - 若为最后一个峰值，需确认序列末尾已回落至0
        if prev_zero != -1 and next_zero != len(array):
            valid_peaks.append(peak)
        elif next_zero == len(array) and np.abs(array[-1]) < 1e-6:
            valid_peaks.append(peak)
    return np.array(valid_peaks)

3. 结合原代码实现过滤 #

先获取候选峰值，再用过滤函数筛选有效峰值：

from scipy.signal import find_peaks
import matplotlib.pyplot as plt

# 获取原候选峰值
peaks1_candidate = find_peaks(array, height=(1,1.5), prominence=1)[0]
peaks1_5_candidate = find_peaks(array, height=(1.5,2), prominence=1.5)[0]
peaks2_candidate = find_peaks(array, height=2, prominence=2)[0]

# 过滤有效峰值
peaks1 = filter_valid_peaks(peaks1_candidate)
peaks1_5 = filter_valid_peaks(peaks1_5_candidate)
peaks2 = filter_valid_peaks(peaks2_candidate)

# 绘图代码不变
fig, ax = plt.subplots(figsize=(30, 10), dpi=80)
plt.plot(spi_neg['date'], spi["SPI-12"])
[plt.axvline(spi_neg.date.iloc[p], c='red', linewidth=0.3) for p in peaks1]
[plt.axvline(spi_neg.date.iloc[p], c='green', linewidth=0.3) for p in peaks1_5]
[plt.axvline(spi_neg.date.iloc[p], c='purple', linewidth=0.3) for p in peaks2]
plt.axhline(2, linestyle='dashed', linewidth=1)
plt.axhline(1.5, linestyle='dashed', linewidth=1)
plt.axhline(1, linestyle='dashed', linewidth=1)
plt.show()

可选优化：严格校验上升/下降趋势 #

如果需要确保峰值是从0上升到峰值，再下降回0，可以在过滤函数中添加趋势判断：

# 在filter_valid_peaks函数的if判断内添加：
# 检查从prev_zero到peak是上升趋势（非递减）
is_rising = np.all(array[prev_zero:peak+1] >= array[prev_zero:peak])
# 检查从peak到next_zero是下降趋势（非递增）
is_falling = np.all(array[peak:next_zero+1] <= array[peak:next_zero])
if is_rising and is_falling:
    valid_peaks.append(peak)

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