实时时间序列峰值检测优化及谷值提取方法咨询

问题概述 #

当前使用find_peaks_cwt处理随机时间序列的实时峰值检测时，存在精度不足、延迟高、重绘卡顿的问题，预处理Ricker小波未改善效果，同时需要实现实时谷值检测。

峰值检测优化方案 #

• 替换find_peaks_cwt为find_peaks：find_peaks_cwt基于连续小波变换，计算成本高且实时性差，改用scipy.signal.find_peaks，它直接基于极值点判断，速度更快、精度更可控：
  • 通过height参数过滤低幅值噪声峰
  • distance参数避免相邻过近的重复峰
  • prominence参数识别"真正有意义"的峰值（排除局部小凸起）
• 优化实时绘图逻辑：原代码每次调用plt.cla()重绘整个画布导致卡顿，改为保留坐标轴对象，仅更新数据：
  • 初始化时创建line和peak_scatter对象
  • 动画函数中仅更新这两个对象的set_data，避免全量重绘
• 限制历史数据长度（可选）：实时场景中无需保留全部历史数据，可设置固定窗口长度（如最近200个点），减少计算和绘图压力

实时谷值检测方法 #

谷值本质是信号的局部最小值，只需将信号取反后调用find_peaks，再将结果映射回原信号即可。

优化后的完整代码 #

import random
import matplotlib.pyplot as plt
from itertools import count
from matplotlib.animation import FuncAnimation
from scipy.signal import find_peaks
import numpy as np

def random_timeseries(initial_value: float, volatility: float, count: int) -> list:
    time_series = [initial_value, ]
    for _ in range(count):
        time_series.append(time_series[-1] + initial_value * random.gauss(0, 1) * volatility)
    return time_series

ts = random_timeseries(1.2, 0.15, 1000)

x_data = []
y_data = []
index = count()

# 初始化绘图对象，避免每次重绘
fig, ax = plt.subplots()
line, = ax.plot([], [], label='Time Series')
peak_scatter, = ax.plot([], [], "x", color='r', label='Peaks')
valley_scatter, = ax.plot([], [], "o", color='g', label='Valleys')
ax.set_title("Random Time Series with Peaks & Valleys")
ax.set_xlabel("Time")
ax.set_ylabel("Value")
ax.legend()

def animate(i):
    current_index = next(index)
    if current_index >= len(ts):
        return line, peak_scatter, valley_scatter
    
    x_data.append(current_index)
    y_data.append(ts[current_index])
    
    # 实时峰值检测：用统计量动态设置阈值过滤噪声
    peaks, _ = find_peaks(y_data, height=np.mean(y_data)+0.5*np.std(y_data), distance=5)
    # 实时谷值检测：信号取反后复用峰值检测逻辑
    valleys, _ = find_peaks([-y for y in y_data], height=-np.mean(y_data)+0.5*np.std(y_data), distance=5)
    
    # 更新数据，避免重绘整个坐标轴
    line.set_data(x_data, y_data)
    peak_scatter.set_data(peaks, [y_data[p] for p in peaks])
    valley_scatter.set_data(valleys, [y_data[v] for v in valleys])
    
    # 自动调整x轴显示窗口
    ax.set_xlim(max(0, current_index-200), current_index+5)
    ax.set_ylim(min(y_data)-0.2, max(y_data)+0.2)
    
    return line, peak_scatter, valley_scatter

ani = FuncAnimation(fig, animate, interval=100, blit=True)

plt.tight_layout()
plt.show()

关键优化点说明 #

• 性能提升：find_peaks比find_peaks_cwt快一个数量级，适配实时场景
• 绘图流畅：使用blit=True和对象数据更新，消除重绘卡顿
• 精度控制：通过统计量（均值+标准差）动态设置阈值，适配信号波动
• 谷值实现：取反信号复用峰值检测逻辑，无需额外算法

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