高效算法需求：检测指定时段内数值升降是否达GAP_VALUE

核心问题分析 #

你当前的逐行取窗口子集计算极值的方法，时间复杂度为O(n*k)（n是总行数，k是窗口内平均行数），百万级数据下会因为重复遍历窗口导致CPU和内存占用过高，尤其是窗口较大时问题更突出。

优化方案：单调队列实现滑动窗口极值 #

前置预处理 #

• 将CSV中的datetime转换为数值型时间戳（如Unix秒/毫秒），便于快速计算时间范围边界
• 按时间戳对数据进行升序排序（若原数据已严格按时间顺序采集可跳过，但建议做格式校验）

单调队列原理 #

维护两个独立的单调队列：

• 最大值队列：队列元素按value单调递减，队首始终是当前窗口的最大值
• 最小值队列：队列元素按value单调递增，队首始终是当前窗口的最小值
  每个元素仅入队、出队各一次，整体时间复杂度降至O(n)，远优于原方法的O(n*k)。

具体实现步骤 #

1. 遍历每一条数据，先移除队列中超出TIME_GAP时间范围的元素（因数据已排序，只需从队首弹出）
2. 维护最大值队列：若当前元素的value大于队尾元素，弹出队尾，直到队列为空或队尾元素更大，再将当前元素加入队列
3. 维护最小值队列：若当前元素的value小于队尾元素，弹出队尾，直到队列为空或队尾元素更小，再将当前元素加入队列
4. 取两个队列的队首值计算VALUE_SHIFT，与GAP_VALUE比较，达标则记录结果

代码示例（Python） #

import csv
from collections import deque
import time

def detect_value_gap(csv_file_path, time_gap_sec, gap_threshold):
    # 读取并预处理数据（若数据量极大，可改为逐行流式处理）
    processed_data = []
    with open(csv_file_path, 'r', encoding='utf-8') as f:
        csv_reader = csv.reader(f)
        next(csv_reader)  # 跳过表头行，无表头则删除此句
        for row in csv_reader:
            try:
                value = float(row[0])
                # 注意：需根据CSV中实际的datetime格式调整strptime参数
                dt_timestamp = time.mktime(time.strptime(row[1], "%Y-%m-%d %H:%M:%S"))
                processed_data.append((dt_timestamp, value))
            except ValueError:
                continue  # 跳过格式错误的行
    
    # 按时间戳升序排序
    processed_data.sort(key=lambda x: x[0])
    
    max_deque = deque()  # 存储(时间戳, 数值)，单调递减
    min_deque = deque()  # 存储(时间戳, 数值)，单调递增
    trigger_records = []
    
    for curr_dt, curr_val in processed_data:
        time_lower_bound = curr_dt - time_gap_sec
        
        # 清理最大值队列中超出窗口的元素
        while max_deque and max_deque[0][0] < time_lower_bound:
            max_deque.popleft()
        # 维护最大值队列单调性
        while max_deque and max_deque[-1][1] <= curr_val:
            max_deque.pop()
        max_deque.append((curr_dt, curr_val))
        
        # 清理最小值队列中超出窗口的元素
        while min_deque and min_deque[0][0] < time_lower_bound:
            min_deque.popleft()
        # 维护最小值队列单调性
        while min_deque and min_deque[-1][1] >= curr_val:
            min_deque.pop()
        min_deque.append((curr_dt, curr_val))
        
        # 计算波动幅度并判断
        current_max = max_deque[0][1]
        current_min = min_deque[0][1]
        value_shift = abs(current_max - current_min)
        if value_shift >= gap_threshold:
            # 转换时间戳为可读格式，按需调整
            trigger_records.append((time.ctime(curr_dt), value_shift))
    
    return trigger_records

额外优化建议 #

• 流式处理：若数据量过大无法全量加载，可改为逐行读取处理，无需存储全量数据，仅保留队列和窗口边界内的历史数据
• 数据类型优化：使用更紧凑的数值类型（如numpy.float32替代Python原生float）减少内存占用
• Pandas替代方案：若熟悉Pandas，可将datetime设为索引后使用rolling(time_gap).agg(['max', 'min'])，底层已做优化，但超大数据量下手动实现单调队列内存效率更高
• 并行处理：若数据可按时间分片（确保分片重叠TIME_GAP时长），可多进程并行处理各分片，最后合并结果

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