如何针对前置条目高效标准化至[-1,1]区间的大规模数据？

问题背景 #

处理超10000条大规模数据，需基于滑动窗口（窗口大小为10，包含当前条目及前9条数据）将每条数据标准化至[-1,1]区间。原实现通过遍历每个窗口并重新计算窗口内的最小值、最大值，数据量较大时耗时显著，需从数学与通用算法层面优化流程，可结合numpy、pandas实现。

原简化代码：

def standardize_last(data):
    min_val = data[0]
    max_val = data[0]
    for i in range(0, len(data)):
        entry = data[i]
        if entry < min_val:
            min_val = entry
        elif entry > max_val:
            max_val = entry
    if min_val == max_val:
        return 0
    else:
        # 1. Shift range minimum to 0
        # 2. Divide value by range to get a standard value from 0 to 1
        # 3. Move to range from -1 to 1
        return (((data[-1] - min_val) / (max_val - min_val)) * 2) - 1

results = []
dataset = [2,2,2,2,2,2,2,2,2,-2,-1,0,1,2,3,2]

# Standardize
for i in range(9, len(dataset)):
    slice_data = dataset[i - 9:i + 1]
    val = standardize_last(slice_data)
    results.append(val)

print(results)

原输出：

[-1.0, -0.5, 0.0, 0.5, 1.0, 1.0, 0.6000000000000001]

优化方案 #

1. 数学公式化简 #

原标准化公式可等价化简，减少计算步骤且结果完全一致：
原公式：

(((x - window_min) / (window_max - window_min)) * 2) - 1

化简后：

(2 * x - window_min - window_max) / (window_max - window_min)

当window_min == window_max时返回0，该化简可减少一次乘法与一次减法操作，降低单步计算开销。

2. 滑动窗口min/max的算法优化（核心） #

原方法每次滑动窗口都遍历整个窗口计算min/max，时间复杂度为O(N*K)（N为数据量，K为窗口大小）。采用单调队列维护滑动窗口的min和max，可将时间复杂度降至O(N)，每个元素仅入队、出队各一次。

单调队列实现思路：

• 维护两个双端队列：
  • min_deque：存储窗口内元素的索引，保证队首对应窗口的最小值；
  • max_deque：存储窗口内元素的索引，保证队首对应窗口的最大值。
• 窗口滑动时：
  1. 移除队列中超出当前窗口范围的索引（索引小于当前窗口左边界）；
  2. 对于新加入窗口的元素，从min_deque尾部移除所有值大于当前元素的索引，再加入当前元素索引；从max_deque尾部移除所有值小于当前元素的索引，再加入当前元素索引；
  3. 队首元素即为当前窗口的min/max对应的索引，直接取值即可。

优化后代码实现：

from collections import deque

def sliding_window_standardize(dataset, window_size=10):
    n = len(dataset)
    if n < window_size:
        return []
    
    min_deque = deque()
    max_deque = deque()
    results = []
    
    # 初始化第一个窗口
    for i in range(window_size):
        # 维护min队列
        while min_deque and dataset[i] <= dataset[min_deque[-1]]:
            min_deque.pop()
        min_deque.append(i)
        
        # 维护max队列
        while max_deque and dataset[i] >= dataset[max_deque[-1]]:
            max_deque.pop()
        max_deque.append(i)
    
    # 处理第一个窗口的标准化值
    window_min = dataset[min_deque[0]]
    window_max = dataset[max_deque[0]]
    x = dataset[window_size-1]
    if window_min == window_max:
        results.append(0.0)
    else:
        results.append((2 * x - window_min - window_max) / (window_max - window_min))
    
    # 滑动窗口处理剩余元素
    for i in range(window_size, n):
        current_idx = i
        left_bound = current_idx - window_size + 1
        
        # 移除超出窗口的元素索引
        while min_deque and min_deque[0] < left_bound:
            min_deque.popleft()
        while max_deque and max_deque[0] < left_bound:
            max_deque.popleft()
        
        # 维护min队列
        while min_deque and dataset[current_idx] <= dataset[min_deque[-1]]:
            min_deque.pop()
        min_deque.append(current_idx)
        
        # 维护max队列
        while max_deque and dataset[current_idx] >= dataset[max_deque[-1]]:
            max_deque.pop()
        max_deque.append(current_idx)
        
        # 计算标准化值
        window_min = dataset[min_deque[0]]
        window_max = dataset[max_deque[0]]
        x = dataset[current_idx]
        if window_min == window_max:
            results.append(0.0)
        else:
            results.append((2 * x - window_min - window_max) / (window_max - window_min))
    
    return results

dataset = [2,2,2,2,2,2,2,2,2,-2,-1,0,1,2,3,2]
results = sliding_window_standardize(dataset)
print(results)

输出结果与原代码一致：

[-1.0, -0.5, 0.0, 0.5, 1.0, 1.0, 0.6000000000000001]

3. 结合pandas的实现（简化代码） #

利用pandas的rolling窗口功能，直接计算每个窗口的min、max，再应用标准化公式，底层已做滑动窗口统计优化：

import pandas as pd

dataset = [2,2,2,2,2,2,2,2,2,-2,-1,0,1,2,3,2]
df = pd.DataFrame({'value': dataset})

# 计算滑动窗口的min、max，窗口大小10，包含当前行
window = df['value'].rolling(window=10, min_periods=10)
df['window_min'] = window.min()
df['window_max'] = window.max()
df['standardized'] = df.apply(
    lambda row: 0.0 if row['window_min'] == row['window_max'] 
    else (2 * row['value'] - row['window_min'] - row['window_max']) / (row['window_max'] - row['window_min']),
    axis=1
)

# 取从第9个索引开始的结果
results = df['standardized'].iloc[9:].tolist()
print(results)

输出同样与原代码一致。

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