如何在每分钟更新的CSV数据局部簇中识别峰值？

核心逻辑 #

你要识别的是连续上升趋势结束、首次出现下降时的前一个值，也就是每个上升簇的最高点。规则可简化为：

• 当数据从「持续上升」转为「下降」时，记录下降前的那个值作为峰值
• 保留所有检测到的峰值，不覆盖历史记录

实现步骤 #

1. 数据读取：针对每分钟更新的CSV，每次读取新增的时序数据点（确保数据按时间顺序排列）
2. 状态跟踪：维护三个核心状态：
   • 上一个数据点的值
   • 当前上升阶段的临时峰值
   • 是否处于上升趋势中
3. 峰值检测：遍历数据时：
   • 若当前值 > 上一个值：更新临时峰值，标记为上升状态
   • 若当前值 < 上一个值且处于上升状态：将临时峰值加入归档列表，重置上升状态
   • 若当前值等于上一个值：不改变状态（可根据需求调整规则）
4. 峰值归档：将检测到的峰值持久化存储（如写入文件、数据库）

代码示例（Python） #

基础峰值检测函数 #

def detect_local_peaks(data_sequence):
    peaks = []
    if len(data_sequence) < 3:
        return peaks  # 至少需要「升-升-降」或「升-平-降」三个点才能形成有效峰值
    
    prev_val = data_sequence[0]
    current_peak = prev_val
    is_rising = False
    
    for val in data_sequence[1:]:
        if val > prev_val:
            current_peak = val
            is_rising = True
        elif val < prev_val and is_rising:
            peaks.append(current_peak)
            is_rising = False
        # 相等值不触发状态变化，可根据业务需求修改
        prev_val = val
    
    return peaks

# 测试示例数据
sample_data = [100, 125, 150, 120, 135, 160, 180, 145, 165, 185, 200, 190, 180]
print("检测到的峰值:", detect_local_peaks(sample_data))  # 输出: [150, 180, 200]

适配CSV实时更新的监控逻辑 #

import time
import pandas as pd

def monitor_real_time_peaks(csv_path, interval=60):
    # 初始化状态变量，用于跨周期跟踪
    peaks = []
    prev_val = None
    current_peak = None
    is_rising = False
    last_row_idx = -1

    while True:
        # 读取CSV数据
        df = pd.read_csv(csv_path)
        data_col = df['value']  # 替换为你的数据列名
        new_rows = data_col.iloc[last_row_idx + 1:]
        
        if not new_rows.empty:
            for val in new_rows:
                if prev_val is None:
                    prev_val = val
                    current_peak = val
                    continue
                
                if val > prev_val:
                    current_peak = val
                    is_rising = True
                elif val < prev_val and is_rising:
                    peaks.append(current_peak)
                    is_rising = False
                
                prev_val = val
            
            last_row_idx = len(df) - 1
            print("已归档峰值:", peaks)
        
        time.sleep(interval)

# 调用示例（替换为你的CSV文件路径）
# monitor_real_time_peaks("your_timed_data.csv")

注意事项 #

• 相等值处理：如果业务允许「上升后持平再下降」也算峰值，可修改逻辑，将val < prev_val改为val <= prev_val，但需注意避免误判
• 边界处理：若CSV末尾是未完成的上升序列（如最后两个点是185,200），不会记录峰值，直到出现下降点
• 状态持久化：若程序需要重启，可将peaks、prev_val、is_rising等状态写入本地文件或数据库，避免丢失
• CSV顺序校验：必须确保CSV新数据是追加在末尾，且严格按时间顺序排列，否则检测逻辑会失效

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