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:
        testmin = i
    if i > testmax:
        testmax = i
    if i < testmin: # 这里的比较可能在testmin被错误赋值后才发生
        testmin = i

print (test, 'min: ', testmin, 'max: ', testmax)
# 示例输出可能为：[-65 -53   1   2  26 -62  82  70  39 -44] min:  -44 max:  82
# 实际应为：min: -65, max: 82

import numpy as np

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

# 模拟一个数据流，范围在-100到100之间
stream_min_val = -100
stream_max_val = 100
test_data_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")

# 模拟数据流处理
print(f"模拟数据流: {test_data_stream}")
for value in test_data_stream:
    # 更新最大值
    if value > current_max:
        current_max = value
    # 更新最小值
    if value < current_min:
        current_min = value

print(f"最终最小值: {current_min}, 最终最大值: {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_stream = rng.choice(np.arange(stream_min_val, stream_max_val + 1, dtype=int),
                              500,
                              replace=False)

def plain_if_else():
    """使用普通 if/else 语句更新最值"""
    local_max = -float("inf")
    local_min = float("inf")
    for i in test_data_stream:
        if i > local_max:
            local_max = i
        if i < local_min:
            local_min = i
    return local_min, local_max

def ternary_operator():
    """使用三元运算符更新最值"""
    local_max = -float("inf")
    local_min = float("inf")
    for i in test_data_stream:
        local_max = i if i > local_max else local_max
        local_min = i if i < local_min else local_min # 注意这里是 local_min 而非 local_max
    return local_min, local_max

def min_max_builtins():
    """使用内置 min()/max() 函数更新最值"""
    local_max = -float("inf")
    local_min = float("inf")
    for i in test_data_stream:
        local_max = max(i, local_max)
        local_min = min(i, local_min)
    return local_min, local_max

# 执行基准测试
print("--- 性能基准测试 (500个数据点) ---")
print(f"plain_if_else(): {timeit.timeit(plain_if_else, number=10000):.6f} seconds")
print(f"ternary_operator(): {timeit.timeit(ternary_operator, number=10000):.6f} seconds")
print(f"min_max_builtins(): {timeit.timeit(min_max_builtins, number=10000):.6f} seconds")

# 典型输出示例 (具体数值可能因环境而异):
# --- 性能基准测试 (500个数据点) ---
# plain_if_else(): 0.506000 seconds
# ternary_operator(): 0.554000 seconds
# min_max_builtins(): 1.700000 seconds