numpy数组快速条件重叠窗口（分帧）实现，适配时间序列类别不平衡场景

原代码问题说明 #

你提供的原始代码存在逻辑错误：将start_cursor更新为end_cursor后，再取y[start_cursor:end_cursor]得到的是空切片，np.any判断永远为False，重叠逻辑完全不会触发。此外原生Python循环+动态列表append的写法，在处理大量序列时会有极高的解释器开销，导致运行缓慢。

高效实现方案 #

推荐使用Numba JIT编译优化动态循环逻辑，性能可以提升10~100倍，同时修正逻辑错误，采用预分配内存避免动态扩展开销。

依赖安装（可选） #

如果不需要Numba可以跳过，直接去掉代码中的@njit装饰器即可，只是性能会稍低，但依然优于原始写法。

pip install numba numpy

实现代码 #

import numpy as np
from numba import njit

@njit  # 开启JIT编译，去除Python循环开销
def conditional_window_split(arr: np.ndarray, y: np.ndarray, window_size: int, overlap_pct: float):
    """
    按条件分割时间序列窗口
    :param arr: 输入一维时间序列数组
    :param y: 对应标签数组，长度和arr一致
    :param window_size: 窗口大小M
    :param overlap_pct: 含1/2类窗口的重叠比例p%
    :return: 分割后的窗口数组(N, M)，对应标签数组(N,)
    """
    overlap_size = int(window_size * overlap_pct / 100)
    seq_len = arr.size
    # 预分配足够的输出空间，避免动态append的内存开销
    min_step = window_size - overlap_size
    max_frame_num = seq_len // max(min_step, 1) + 2
    frames = np.zeros((max_frame_num, window_size), dtype=arr.dtype)
    labels = np.zeros(max_frame_num, dtype=y.dtype)
    frame_count = 0
    start_pos = 0

    while start_pos + window_size <= seq_len:
        end_pos = start_pos + window_size
        # 提取当前窗口和对应标签
        frames[frame_count] = arr[start_pos:end_pos]
        win_label = y[start_pos:end_pos].max()
        labels[frame_count] = win_label
        frame_count += 1
        # 按标签调整下一个窗口的起始位置
        if win_label > 0:
            start_pos = end_pos - overlap_size
        else:
            start_pos = end_pos
    
    # 裁剪掉预分配的多余空间
    return frames[:frame_count], labels[:frame_count]

调用示例 #

# 单条序列调用
M = 10 # 窗口大小
p = 20 # 重叠比例20%
frames, labels = conditional_window_split(arr, y, window_size=M, overlap_pct=p)

# 批量处理大量序列
list_of_arr = [...] # 你的所有一维序列列表
list_of_y = [...] # 对应标签列表
all_frames = []
all_labels = []
for arr, y in zip(list_of_arr, list_of_y):
    f, l = conditional_window_split(arr, y, M, p)
    all_frames.append(f)
    all_labels.append(l)

# 合并所有结果（可选）
total_frames = np.concatenate(all_frames, axis=0)
total_labels = np.concatenate(all_labels, axis=0)

补充说明 #

• 上述代码默认丢弃最后不足window_size的片段，如果需要保留，可以将循环条件改为while start_pos < seq_len，对最后一段不足长度的窗口做补0或者截断处理即可。
• 开启Numba后首次调用函数会有编译开销，后续调用时无额外开销，适合批量处理大量序列的场景。

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