Python中保留时间序列最近10分钟数据的高效方案及标准差计算

一、优化现有deque的使用方式 #

• 针对时序数据递增的特性，每次新增数据时优先清理过期元素：由于新数据的时间戳一定晚于旧数据，只需从队首移除所有早于「当前时间-10分钟」的元素，无需遍历整个队列。这种方式下，deque的popleft()是O(1)操作，清理效率极高，适配突发数据后空闲的场景。
  示例代码：

from collections import deque
import time

window = deque()
WINDOW_DURATION = 600  # 10分钟，单位秒

def add_data(timestamp, value):
    # 清理过期数据
    cutoff = time.time() - WINDOW_DURATION
    while window and window[0][0] < cutoff:
        window.popleft()
    # 添加新数据
    window.append((timestamp, value))

def calculate_std():
    if not window:
        return 0.0
    values = [v for _, v in window]
    n = len(values)
    sum_x = sum(values)
    sum_sq = sum(v*v for v in values)
    variance = (sum_sq / n) - (sum_x / n)**2
    return variance**0.5 if variance > 0 else 0.0

二、替代数据结构：SortedList（推荐） #

如果存在数据乱序插入的情况，sortedcontainers库中的SortedList是更优选择——它支持O(log n)的插入、查找和批量删除操作，纯Python实现无需编译，社区维护活跃度远高于FastRBTree。
示例代码：

from sortedcontainers import SortedList
import time

window = SortedList(key=lambda x: x[0])
WINDOW_DURATION = 600

def add_data(timestamp, value):
    window.add((timestamp, value))
    # 批量删除过期数据
    cutoff = time.time() - WINDOW_DURATION
    idx = window.bisect_left((cutoff, 0))
    del window[:idx]

def calculate_std():
    if not window:
        return 0.0
    n = len(window)
    sum_x = sum(v for _, v in window)
    sum_sq = sum(v*v for _, v in window)
    variance = (sum_sq / n) - (sum_x / n)**2
    return variance**0.5 if variance > 0 else 0.0

三、针对标准差计算的极致优化：维护滚动统计量 #

你的核心需求是计算固定时段的标准差，完全无需存储所有数据点。只需维护三个统计值：数据点数量count、数据总和sum_x、数据平方和sum_sq，配合存储（时间戳、数值、数值平方）的队列，清理过期数据时同步更新统计量，计算标准差为O(1)操作，内存占用极低。
示例代码：

from collections import deque
import time

WINDOW_DURATION = 600
count = 0
sum_x = 0.0
sum_sq = 0.0
window = deque()

def add_data(timestamp, value):
    global count, sum_x, sum_sq
    # 清理过期数据并更新统计量
    cutoff = time.time() - WINDOW_DURATION
    while window and window[0][0] < cutoff:
        ts, v, v_sq = window.popleft()
        count -= 1
        sum_x -= v
        sum_sq -= v_sq
    # 添加新数据并更新统计量
    v_sq = value * value
    window.append((timestamp, value, v_sq))
    count += 1
    sum_x += value
    sum_sq += v_sq

def calculate_std():
    if count == 0:
        return 0.0
    mean = sum_x / count
    variance = (sum_sq / count) - mean**2
    return variance**0.5 if variance > 0 else 0.0

这个方案完美适配「数据点越少标准差越小」的需求，统计量会随窗口内数据的增减自动更新。

四、关于FastRBTree的现状 #

FastRBTree作为C扩展的红黑树实现目前仍可使用，但存在安装依赖编译环境、社区维护活跃度低的问题。如果没有极端性能要求，SortedList是更易用、更稳妥的选择；若需极致插入/删除性能且能接受编译依赖，FastRBTree依然可用。

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