实时计算无全量数组的数据流最值：代码问题及优化方案咨询

问题背景 #

拥有百万级连续数值数据流，无法存储全量数据，需实时计算截至当前的最小值与最大值，且数据范围未知。自行编写的Python代码计算结果有误，寻求更优解决方案，包括使用numpy、scipy库的实现方式。

错误代码及输出 #

import numpy as np
rng = np.random.default_rng()

test = rng.choice(np.arange(-100,100, dtype=int), 10, replace=False)
testmax = 0
testmin = 0
for i in test: # 模拟数据流
    if i < testmax:
        testmin = i
    if i > testmax:
        testmax = i
    if i < testmin:
        testmin = i


print(test, 'min: ',testmin, 'max: ', testmax)

错误输出示例：

[ 39 -32  61 -18 -53 -57 -69  98 -88 -47] min:  -47 max:  98 # 正确结果应为-88和98
[-65 -53   1   2  26 -62  82  70  39 -44] min:  -44 max:  82 # 正确结果应为-65和82

错误原因分析 #

1. 初始化错误：将testmax和testmin初始化为0，若数据流中所有值都小于0或大于0，初始值会干扰正确结果。
2. 逻辑判断错误：第一个条件if i < testmax就更新testmin完全不符合最小值的判断逻辑，最小值应与当前testmin直接比较。

解决方案 #

1. 基础Python实现（无第三方库） #

只需维护两个变量存储当前的最小值和最大值，每次处理数据流中的一个值时，仅需两次比较操作，内存占用可忽略，适合单条数据的实时流：

import numpy as np
rng = np.random.default_rng()

test = rng.choice(np.arange(-100,100, dtype=int), 10, replace=False)
# 初始化：取数据流第一个值作为初始min和max
it = iter(test)
testmin = testmax = next(it)
for i in it:
    if i < testmin:
        testmin = i
    if i > testmax:
        testmax = i

print(test, 'min: ', testmin, 'max: ', testmax)

2. numpy批量处理优化 #

若数据流是分批传入的numpy数组（比如每次处理几千条数据），可利用numpy的向量化运算提升效率，避免纯Python循环的性能损耗：

import numpy as np
rng = np.random.default_rng()

# 模拟分批数据流（比如分3批）
batches = [rng.choice(np.arange(-100,100, dtype=int), 10, replace=False) for _ in range(3)]

# 初始化
first_batch = batches[0]
current_min = first_batch.min()
current_max = first_batch.max()

# 处理后续批次
for batch in batches[1:]:
    batch_min = batch.min()
    batch_max = batch.max()
    current_min = min(current_min, batch_min)
    current_max = max(current_max, batch_max)

# 输出所有批次合并后的结果
all_data = np.concatenate(batches)
print(all_data, 'min: ', current_min, 'max: ', current_max)

3. scipy相关说明 #

scipy库没有专门针对实时流min/max计算的特殊函数，核心逻辑仍与上述方案一致——维护当前的最小/最大值，若需结合统计分析，可基于numpy的结果进一步处理。

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