利用Rolling窗口实现DataFrame阈值突破首索引判定与标签赋值

首先定义三个DataFrame：

import numpy as np
import pandas as pd

np.random.seed(42)
first_frame = pd.DataFrame(np.random.randn(100))
top_frame = pd.DataFrame(np.random.randn(100))
bottom_frame = pd.DataFrame(np.random.randn(100))

同时定义用于标记的DataFrame：

label_frame = pd.DataFrame(np.zeros(100))

需要按以下规则填充label_frame：

1. 若first_frame[i:i+5]中存在大于top_frame[i]的值，且没有小于bottom_frame[i]的值，标记label_frame[i]为1；
2. 若first_frame[i:i+5]中存在小于bottom_frame[i]的值，且没有大于top_frame[i]的值，标记label_frame[i]为-1；
3. 若first_frame[i:i+5]同时存在大于top_frame[i]和小于bottom_frame[i]的值，则根据首次突破的阈值判定：先突破top_frame[i]标记为1，先突破bottom_frame[i]标记为-1；
4. 未突破任何阈值则标记为0；
5. 对整个DataFrame以滑动窗口方式执行上述逻辑，依次处理每个first_frame[i:i+5]与对应的top_frame[i]、bottom_frame[i]。

已尝试的方法 #

目前通过.rolling函数实现了基础的阈值突破判断：

# 判断窗口是否突破top阈值
first_frame.rolling(5, min_periods=5).max().shift(-4) > top_frame
# 判断窗口是否突破bottom阈值
first_frame.rolling(5, min_periods=5).min().shift(-4) < bottom_frame

但无法处理同时突破上下阈值时获取首次突破顺序的场景，尝试结合自定义函数与apply但不确定写法。

示例场景 #

对于i=0的窗口：

first_frame[0:5] > top_frame.iloc[0]
      0
0  True
1  True
2  True
3  True
4  True

first_frame[0:5] < bottom_frame.iloc[0]
      0
0  False
1   True
2  False
3  False
4   True

该窗口同时突破了上下阈值，但第0条数据先突破top_frame[0]，因此label_frame[0]应标记为1。

可以通过自定义滚动窗口函数结合遍历实现完整逻辑，核心是对每个窗口单独判断突破情况与首次突破顺序：

def determine_label(window, top_val, bottom_val):
    # 标记窗口内每个元素是否突破top/bottom阈值
    cross_top = window > top_val
    cross_bottom = window < bottom_val

    # 仅突破top阈值
    if cross_top.any() and not cross_bottom.any():
        return 1
    # 仅突破bottom阈值
    elif cross_bottom.any() and not cross_top.any():
        return -1
    # 同时突破两者，比较首次突破的位置
    elif cross_top.any() and cross_bottom.any():
        # 获取窗口内首次突破top的索引
        first_top_idx = cross_top.argmax()
        # 获取窗口内首次突破bottom的索引
        first_bottom_idx = cross_bottom.argmax()
        # 哪个先突破就返回对应标记
        return 1 if first_top_idx < first_bottom_idx else -1
    # 未突破任何阈值
    else:
        return 0

# 遍历所有可形成完整5元素窗口的位置
for i in range(len(first_frame) - 4):
    # 提取当前窗口数据
    window_data = first_frame.iloc[i:i+5, 0].values
    # 获取对应位置的top和bottom阈值
    current_top = top_frame.iloc[i, 0]
    current_bottom = bottom_frame.iloc[i, 0]
    # 计算标记值并赋值
    label_frame.iloc[i, 0] = determine_label(window_data, current_top, current_bottom)

# 最后4个无法形成完整窗口的位置保持初始0值（可根据需求调整）

代码说明 #

1. 自定义函数determine_label接收窗口数据、对应阈值，按规则返回标记值；
2. 遍历所有有效窗口位置（共96个，100-4），逐个提取窗口与阈值进行计算；
3. 使用argmax()获取首次突破的索引——布尔数组中第一个True的位置即为首次突破点；
4. 最后4个位置因无法形成5元素窗口，保持初始的0值，若需支持短窗口可修改逻辑并调整min_periods参数。

示例验证 #

针对上述示例场景，传入窗口数据、top_val=top_frame.iloc[0,0]、bottom_val=bottom_frame.iloc[0,0]：

• cross_top第一个元素为True，cross_bottom第一个元素为False；
• first_top_idx=0，first_bottom_idx=1；
• 因0 < 1，返回1，符合预期。

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