scipy峰谷检测后处理函数无法保证交替模式问题求助

You're attempting to detect alternating peaks and valleys in a mechanical signal for polymer behavior modeling, but your current post-processing can't eliminate consecutive peaks/valleys even after multiple runs. The core requirement is strict alternation: every pair of consecutive peaks must have exactly one valley in between, and vice versa.

Your current approach processes peaks and valleys as separate ordered lists, then cleans up consecutive extremes within each list. This misses critical cross-list ordering checks—for example, a "consecutive peak" in the peaks list might have a valley positioned between them in the valleys list, but your logic doesn't account for relative positions across both lists. Additionally, naive multiple passes can't resolve overlapping or nested extreme points that break alternation.

Instead of handling peaks and valleys separately, merge them into a single sorted sequence of events, then enforce alternation by iterating through this combined list. This ensures you check the order of extremes in the context of the entire signal timeline.

Step-by-Step Implementation #

1. Combine and sort peaks/valleys: Create a list of tuples containing each extreme's index, type (peak/valley), and corresponding y-value. Sort this list by index to maintain timeline order.
2. Enforce alternation: Traverse the sorted list, building a valid sequence where each element alternates between peak and valley. When consecutive elements are the same type, keep the more extreme one (higher value for peaks, lower for valleys).
3. Adjust to start/end with peak: Trim the valid sequence to ensure it starts and ends with a peak as required.

Example Code #

import numpy as np
from scipy.signal import find_peaks

def process_peaks_and_valleys(y_data_array, peaks_indices, valleys_indices):
    # Create combined list of (index, type, value)
    peaks = [(idx, 'peak', y_data_array[idx]) for idx in peaks_indices]
    valleys = [(idx, 'valley', y_data_array[idx]) for idx in valleys_indices]
    combined = peaks + valleys
    
    # Sort combined list by index to maintain timeline order
    combined_sorted = sorted(combined, key=lambda x: x[0])
    
    valid_sequence = []
    for event in combined_sorted:
        if not valid_sequence:
            # Initialize with peak if available; skip initial valley
            if event[1] == 'peak':
                valid_sequence.append(event)
            continue
        
        last_event = valid_sequence[-1]
        if event[1] != last_event[1]:
            # Alternating type: add to valid sequence
            valid_sequence.append(event)
        else:
            # Same type: retain the more extreme value
            if event[1] == 'peak':
                # Keep the higher peak
                if event[2] > last_event[2]:
                    valid_sequence[-1] = event
            else:
                # Keep the lower valley
                if event[2] < last_event[2]:
                    valid_sequence[-1] = event
    
    # Ensure sequence starts and ends with peak
    if valid_sequence and valid_sequence[0][1] != 'peak':
        valid_sequence.pop(0)
    if valid_sequence and valid_sequence[-1][1] != 'peak':
        valid_sequence.pop()
    
    # Split back into peaks and valleys indices
    processed_peaks = [event[0] for event in valid_sequence if event[1] == 'peak']
    processed_valleys = [event[0] for event in valid_sequence if event[1] == 'valley']
    
    return processed_peaks, processed_valleys

# Usage
# y_data_array = your signal data array
# peaks_indices = find_peaks(y_data_array, distance=20, width=10, rel_height=1)[0]
# valleys_indices = find_peaks(-y_data_array, distance=20, width=10, rel_height=1)[0]
# cleaned_peaks, cleaned_valleys = process_peaks_and_valleys(y_data_array, peaks_indices, valleys_indices)

Key Improvements #

• Timeline-aware processing: Sorting all extremes together ensures you check the actual order of events in the signal, not just within separate peak/valley lists.
• Single-pass resolution: This logic resolves consecutive extremes in one pass by comparing adjacent events in the full timeline, eliminating the need for multiple runs.
• Extreme prioritization: When consecutive peaks/valleys are found, the most extreme one (highest peak, lowest valley) is retained, aligning with typical mechanical signal analysis needs.

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