Python实战：高效处理实时数据流中的最小/最大值

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: # 逻辑错误：这里试图更新最小值，但与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
# 预期结果应为 min: -88, max: 98

import numpy as np

# 为了可复现性，设置随机种子
rng = np.random.default_rng(42)

# 模拟数据流的范围
stream_min_val = -100
stream_max_val = 100
# 生成一个模拟数据流
test_stream = rng.choice(np.arange(stream_min_val, stream_max_val + 1, dtype=int),
                         10,
                         replace=False)

# 正确初始化最小值和最大值
current_max = -float("inf")
current_min = float("inf")

# 模拟数据流处理
for i in test_stream:
    # 使用简单的if语句更新最大值
    if i > current_max:
        current_max = i
    # 使用简单的if语句更新最小值
    if i < current_min:
        current_min = i

print("模拟数据流:", test_stream)
print("实时追踪的最小值:", current_min)
print("实时追踪的最大值:", current_max)

# 示例输出:
# 模拟数据流: [ 97  49 -83  26 -15 -16  38 -82 -60  69]
# 实时追踪的最小值: -83
# 实时追踪的最大值: 97

import numpy as np
import timeit

rng = np.random.default_rng(42) # 保持随机种子一致

stream_min_val = -1000
stream_max_val = 1000
# 生成一个更大的模拟数据流进行性能测试
test_data = rng.choice(np.arange(stream_min_val, stream_max_val + 1, dtype=int),
                       500, # 500个元素
                       replace=False)

def update_with_ternary():
    """使用三元运算符更新最小值和最大值"""
    current_max = -float("inf")
    current_min = float("inf")
    for i in test_data:
        current_max = i if i > current_max else current_max
        current_min = i if i < current_min else current_min
    return current_min, current_max

def update_with_plain_if():
    """使用简单的if语句更新最小值和最大值"""
    current_max = -float("inf")
    current_min = float("inf")
    for i in test_data:
        if i > current_max:
            current_max = i
        if i < current_min:
            current_min = i
    return current_min, current_max

def update_with_minmax_funcs():
    """使用内置的min()和max()函数更新最小值和最大值"""
    current_max = -float("inf")
    current_min = float("inf")
    for i in test_data:
        current_max = max(i, current_max)
        # 注意：这里原问题答案中有一个潜在的错误，min(i, testmax) 应该是 min(i, current_min)
        # 修正为：
        current_min = min(i, current_min)
    return current_min, current_max

print("--- 性能测试结果 (500个元素) ---")
print(f"三元运算符方式: {timeit.timeit(update_with_ternary, number=10000):.3f} 秒")
print(f"简单if语句方式: {timeit.timeit(update_with_plain_if, number=10000):.3f} 秒")
print(f"min/max函数方式: {timeit.timeit(update_with_minmax_funcs, number=10000):.3f} 秒")

# 示例输出 (具体数值会因机器而异):
# --- 性能测试结果 (500个元素) ---
# 三元运算符方式: 0.554 秒
# 简单if语句方式: 0.506 秒
# min/max函数方式: 1.700 秒