Scipy峰值检测方法结果不一致：峰位偏移与缺失的修正需求

安全阿岳

2026-6-24

音频信号极值点检测问题修复方案

问题场景与现象

处理重采样后的音频信号时，使用_local_maxima_1d检测峰值和谷值出现以下问题：

仅少数极值点对（如(G1, R1)、(G2, R2)、(G3, R3)）位置正确；
多数检测出的峰值位置比实际超前1个索引；
重采样至原长度1/8时，信号峰部变钝、扁平化，导致部分峰值未被检测到；若改为重采样至1/4，缺失问题缓解，但不符合需求；
检测出的谷值位置比实际滞后1个索引；
调整select_by_peak_distance方法未解决问题，需针对极值点检测逻辑优化。

原问题代码

import numpy as np
import math
import matplotlib.pyplot as plt
from scipy.io import wavfile
from scipy.signal import resample
from scipy.signal._peak_finding_utils import _local_maxima_1d

def select_by_peak_distance(peaks, priority, distance):
    """
    评估哪些峰值满足距离条件。

    参数
    ----------
    peaks : ndarray
        `vector`中的峰值索引。
    priority : ndarray
        与`peaks`匹配的数组，用于确定每个峰值的优先级。优先级值更高的峰值会被保留，优先级低的则被剔除。
    distance : np.float64
        峰值之间必须保持的最小间距。

    返回
    -------
    keep : ndarray[bool]
        布尔掩码，满足距离条件的`peaks`对应的位置为True。
    """
    peaks_size = peaks.shape[0]
    # 向上取整，因为实际峰值间距只能是自然数
    distance = math.ceil(distance)
    keep = np.ones(peaks_size, dtype=np.uint8)  # 准备标记数组

    # 创建从`i`（按`priority`排序的`peaks`索引）到`j`（按位置排序的`peaks`索引）的映射。
    # 这允许按`priority`顺序遍历`peaks`和`keep`，同时仍能通过(`j` + 1)或(`j` - 1)访问相邻峰值。
    priority_to_position = np.argsort(priority)

    for i in range(peaks_size - 1, -1, -1):
        # 将`i`转换为`j`，指向当前要评估相邻峰值的峰值
        j = priority_to_position[i]
        if keep[j] == 0:
            # 跳过已标记为不保留的峰值
            continue

        k = j - 1
        # 标记需要移除的“前方”峰值，直到超过最小间距
        while 0 <= k and peaks[j] - peaks[k] < distance:
            keep[k] = 0
            k -= 1

        k = j + 1
        # 标记需要移除的“后方”峰值，直到超过最小间距
        while k < peaks_size and peaks[k] - peaks[j] < distance:
            keep[k] = 0
            k += 1

    return keep.astype(np.bool_)

sr, x = wavfile.read("jfk.wav")
x = x.astype(np.float64, order='C') / 32768.0 # 归一化到(-1.0, 1.0)
x = resample(x, len(x)//8)

peaks, _, _ = _local_maxima_1d(x)
# 筛选出所有大于0的峰值
peaks = np.array([p for p in peaks if x[p] > 0])
keep = select_by_peak_distance(peaks, x[peaks], 20)
peaks = peaks[keep]

valley, _, _ = _local_maxima_1d(-x)
# 筛选出所有小于0的谷值
valley = np.array([v for v in valley if x[v] < 0])
keep = select_by_peak_distance(valley, -x[valley], 20)
valley = valley[keep]

fig = plt.figure(figsize=(12, 6), dpi=80)
gs = fig.add_gridspec(1, hspace=0)
axs = gs.subplots()

axs.axhline(y = 0.0, color = 'lightgray', ls='-', lw=1.0) 

axs.plot(x, lw=1)
axs.plot(peaks, x[peaks], "o", color='green')
axs.plot(valley, x[valley], "o", color='red')

fig.tight_layout()
plt.show()
plt.close()

修复方案

1. 修正极值点偏移问题

重采样后的信号常出现平台型极值，_local_maxima_1d仅检测平台左边缘，导致峰值超前、谷值滞后。添加极值点调整函数，搜索平台区间并定位真实极值位置：

def adjust_peaks(x, peaks):
    adjusted = []
    for p in peaks:
        # 向右搜索平台右边界
        current_p = p
        while current_p + 1 < len(x) and x[current_p + 1] >= x[current_p]:
            current_p += 1
        # 向左搜索平台左边界
        left_p = p
        while left_p - 1 >= 0 and x[left_p - 1] >= x[left_p]:
            left_p -= 1
        # 取平台中最后一个最大值点（修正超前问题）
        plateau = np.arange(left_p, current_p + 1)
        max_val = x[plateau].max()
        final_p = plateau[x[plateau] == max_val][-1]
        adjusted.append(final_p)
    return np.array(adjusted)

def adjust_valleys(x, valleys):
    adjusted = []
    for v in valleys:
        # 向右搜索平台右边界
        current_v = v
        while current_v + 1 < len(x) and x[current_v + 1] <= x[current_v]:
            current_v += 1
        # 向左搜索平台左边界
        left_v = v
        while left_v - 1 >= 0 and x[left_v - 1] <= x[left_v]:
            left_v -= 1
        # 取平台中第一个最小值点（修正滞后问题）
        plateau = np.arange(left_v, current_v + 1)
        min_val = x[plateau].min()
        final_v = plateau[x[plateau] == min_val][0]
        adjusted.append(final_v)
    return np.array(adjusted)

2. 提升扁平化峰值检测率

替换_local_maxima_1d为scipy.signal.find_peaks，通过width参数支持平台型峰值检测，解决重采样导致的峰值缺失问题。

完整修复后代码

import numpy as np
import math
import matplotlib.pyplot as plt
from scipy.io import wavfile
from scipy.signal import resample, find_peaks
from scipy.signal._peak_finding_utils import _local_maxima_1d

def select_by_peak_distance(peaks, priority, distance):
    """
    评估哪些峰值满足距离条件。

    参数
    ----------
    peaks : ndarray
        `vector`中的峰值索引。
    priority : ndarray
        与`peaks`匹配的数组，用于确定每个峰值的优先级。优先级值更高的峰值会被保留，优先级低的则被剔除。
    distance : np.float64
        峰值之间必须保持的最小间距。

    返回
    -------
    keep : ndarray[bool]
        布尔掩码，满足距离条件的`peaks`对应的位置为True。
    """
    peaks_size = peaks.shape[0]
    # 向上取整，因为实际峰值间距只能是自然数
    distance = math.ceil(distance)
    keep = np.ones(peaks_size, dtype=np.uint8)  # 准备标记数组

    # 创建从`i`（按`priority`排序的`peaks`索引）到`j`（按位置排序的`peaks`索引）的映射。
    # 这允许按`priority`顺序遍历`peaks`和`keep`，同时仍能通过(`j` + 1)或(`j` - 1)访问相邻峰值。
    priority_to_position = np.argsort(priority)

    for i in range(peaks_size - 1, -1, -1):
        # 将`i`转换为`j`，指向当前要评估相邻峰值的峰值
        j = priority_to_position[i]
        if keep[j] == 0:
            # 跳过已标记为不保留的峰值
            continue

        k = j - 1
        # 标记需要移除的“前方”峰值，直到超过最小间距
        while 0 <= k and peaks[j] - peaks[k] < distance:
            keep[k] = 0
            k -= 1

        k = j + 1
        # 标记需要移除的“后方”峰值，直到超过最小间距
        while k < peaks_size and peaks[k] - peaks[j] < distance:
            keep[k] = 0
            k += 1

    return keep.astype(np.bool_)

def adjust_peaks(x, peaks):
    adjusted = []
    for p in peaks:
        # 向右搜索平台右边界
        current_p = p
        while current_p + 1 < len(x) and x[current_p + 1] >= x[current_p]:
            current_p += 1
        # 向左搜索平台左边界
        left_p = p
        while left_p - 1 >= 0 and x[left_p - 1] >= x[left_p]:
            left_p -= 1
        # 取平台中最后一个最大值点（修正超前问题）
        plateau = np.arange(left_p, current_p + 1)
        max_val = x[plateau].max()
        final_p = plateau[x[plateau] == max_val][-1]
        adjusted.append(final_p)
    return np.array(adjusted)

def adjust_valleys(x, valleys):
    adjusted = []
    for v in valleys:
        # 向右搜索平台右边界
        current_v = v
        while current_v + 1 < len(x) and x[current_v + 1] <= x[current_v]:
            current_v += 1
        # 向左搜索平台左边界
        left_v = v
        while left_v - 1 >= 0 and x[left_v - 1] <= x[left_v]:
            left_v -= 1
        # 取平台中第一个最小值点（修正滞后问题）
        plateau = np.arange(left_v, current_v + 1)
        min_val = x[plateau].min()
        final_v = plateau[x[plateau] == min_val][0]
        adjusted.append(final_v)
    return np.array(adjusted)

sr, x = wavfile.read("jfk.wav")
x = x.astype(np.float64, order='C') / 32768.0 # 归一化到(-1.0, 1.0)
x = resample(x, len(x)//8)

# 检测峰值，使用find_peaks支持平台型峰值
peaks, _ = find_peaks(x, height=0, width=1)
# 调整峰值位置
peaks = adjust_peaks(x, peaks)
keep = select_by_peak_distance(peaks, x[peaks], 20)
peaks = peaks[keep]

# 检测谷值，对应-x的峰值
valley, _ = find_peaks(-x, height=0, width=1)
valley = np.array([v for v in valley if x[v] < 0])
# 调整谷值位置
valley = adjust_valleys(x, valley)
keep = select_by_peak_distance(valley, -x[valley], 20)
valley = valley[keep]

fig = plt.figure(figsize=(12, 6), dpi=80)
gs = fig.add_gridspec(1, hspace=0)
axs = gs.subplots()

axs.axhline(y = 0.0, color = 'lightgray', ls='-', lw=1.0) 

axs.plot(x, lw=1)
axs.plot(peaks, x[peaks], "o", color='green')
axs.plot(valley, x[valley], "o", color='red')

fig.tight_layout()
plt.show()
plt.close()

方案说明

adjust_peaks和adjust_valleys函数通过搜索极值平台区间，定位真实的极值位置，解决峰值超前、谷值滞后的偏移问题；
使用find_peaks替代_local_maxima_1d，通过width参数识别平台型峰值，提升重采样后扁平化峰值的检测率；
保留原有的select_by_peak_distance函数，确保极值点间距符合要求。

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

Scipy峰值检测方法结果不一致：峰位偏移与缺失的修正需求

问题场景与现象 #

原问题代码 #

修复方案 #

1. 修正极值点偏移问题 #

2. 提升扁平化峰值检测率 #

完整修复后代码 #

方案说明 #