如何从NumPy数组中提取指定数量的关键变化点以实现压缩？

方法1：基于窗口方差的自适应采样 #

核心思路：计算数组局部窗口的方差，方差越大代表该区域数据变化越剧烈，优先保留这些区域的点，最终凑够指定数量。

import numpy as np

def variance_based_sample(arr, n):
    arr_len = len(arr)
    if n >= arr_len:
        return np.arange(arr_len)
    
    # 设置滑动窗口大小，可根据数据规模调整
    window_size = max(2, arr_len // (n * 2))
    # 计算每个位置的局部窗口方差
    variances = np.array([
        np.var(arr[max(0, i - window_size//2):min(arr_len, i + window_size//2 + 1)]) 
        for i in range(arr_len)
    ])
    
    # 选出方差最大的n个索引并排序，保证顺序不变
    top_indices = np.argsort(variances)[-n:]
    top_indices.sort()
    return top_indices

# 示例调用
array = np.random.randn(100)  # 替换为你的目标数组
n = 40
selected_indices = variance_based_sample(array, n)
selected_points = array[selected_indices]

方法2：迭代调整epsilon的RDP算法 #

原RDP无法指定输出点数，可通过二分法迭代调整epsilon参数，直到输出点数满足要求。

from rdp import rdp  # 需先安装：pip install rdp

def rdp_fixed_points(arr, n):
    arr_len = len(arr)
    if n >= arr_len:
        return np.arange(arr_len)
    if n <= 2:
        return np.array([0, arr_len - 1])
    
    # 将一维数组转为二维点集（x为索引，y为数组值）
    points = np.column_stack((np.arange(arr_len), arr))
    
    # 二分法寻找合适的epsilon
    low = 0.0
    high = np.max(arr) - np.min(arr)
    best_indices = None
    
    for _ in range(20):  # 迭代20次足够收敛
        mid = (low + high) / 2
        simplified = rdp(points, epsilon=mid)
        simplified_indices = simplified[:, 0].astype(int)
        
        if len(simplified_indices) == n:
            return simplified_indices
        elif len(simplified_indices) > n:
            # 点数过多，增大epsilon简化更多点
            low = mid
        else:
            # 点数过少，减小epsilon保留更多点
            high = mid
        best_indices = simplified_indices
    
    # 若未刚好命中，调整最近结果凑够数量
    if len(best_indices) > n:
        return best_indices[:n]
    else:
        remaining = n - len(best_indices)
        mask = np.ones(arr_len, dtype=bool)
        mask[best_indices] = False
        remaining_indices = np.where(mask)[0]
        # 补充方差最大的剩余点
        variances = np.array([np.var(arr[max(0,i-1):min(arr_len,i+2)]) for i in remaining_indices])
        add_indices = remaining_indices[np.argsort(variances)[-remaining:]]
        final_indices = np.sort(np.concatenate([best_indices, add_indices]))
        return final_indices

# 示例调用
array = np.random.randn(100)
n = 40
selected_indices = rdp_fixed_points(array, n)
selected_points = array[selected_indices]

方法3：基于斜率变化的采样 #

计算相邻点的斜率变化率，变化率大的点对应数据突变位置，优先保留，再补充均匀采样点凑数。

def slope_based_sample(arr, n):
    arr_len = len(arr)
    if n >= arr_len:
        return np.arange(arr_len)
    
    # 计算相邻斜率及斜率变化值
    slopes = np.diff(arr) / np.diff(np.arange(arr_len))
    slope_changes = np.abs(np.diff(slopes))
    change_indices = np.arange(1, arr_len - 1)
    
    # 保留首尾点，再选斜率变化最大的k个点
    k = n - 2
    top_change_indices = change_indices[np.argsort(slope_changes)[-k:]]
    selected_indices = np.sort(np.concatenate([[0], top_change_indices, [arr_len - 1]]))
    
    # 若数量不足，补充随机选的剩余点
    if len(selected_indices) < n:
        extra = n - len(selected_indices)
        mask = np.ones(arr_len, dtype=bool)
        mask[selected_indices] = False
        extra_indices = np.random.choice(np.where(mask)[0], extra, replace=False)
        selected_indices = np.sort(np.concatenate([selected_indices, extra_indices]))
    
    return selected_indices

# 示例调用
array = np.random.randn(100)
n = 40
selected_indices = slope_based_sample(array, n)
selected_points = array[selected_indices]

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