Pandas/numpy：行值达rolling mean倍数且满足最小窗口时切换指标值

需要为多列DataFrame的每一列生成取值为1或-1的指标，规则如下：

• 当当前行值达到滚动均值的指定倍数，且自上次重置后滚动窗口的观测数量已达到最小窗口长度时，重置滚动均值的计算起点，并将指标值切换为相反值（1变-1，-1变1）
• 初始指标值为1

本次示例使用参数：倍数=2，最小窗口长度=4

初始DataFrame #

import pandas as pd

df = pd.DataFrame(
    {
        "A": [0.1, 0.1, 0.15, 0.1, 0.1, 0.7, 0.1, 0.1, 0.5, 1, 0.1, 0.1],
        "B": [0.1, 0.1, 0.4, 0.1, 0.8, 0.1, 0.1, 0.1, 0.1, 0.1, 0.9, 0.1],
    },
    index=[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11],
)
print(df)

输出：

A    B
0   0.10  0.1
1   0.10  0.1
2   0.15  0.4
3   0.10  0.1
4   0.10  0.8
5   0.70  0.1
6   0.10  0.1
7   0.10  0.1
8   0.50  0.1
9   1.00  0.1
10  0.10  0.9
11  0.10  0.1

期望输出结果 #

A  B
0   1  1
1   1  1
2   1  1
3   1  1
4   1 -1
5  -1 -1
6  -1 -1
7  -1 -1
8  -1 -1
9   1 -1
10  1  1
11  1  1

逻辑说明（以列A为例） #

• 首次重置发生在第5行：当前值0.7 ≥ 2 × 第0-5行的均值（≈0.208），且窗口长度为6（满足最小窗口4的要求），因此指标切换为-1，滚动均值的计算起点重置为第6行
• 第8行无变化：此时滚动窗口仅包含第6-8行共3个观测值，未达到最小窗口长度，不触发重置
• 第二次重置发生在第9行：当前值1 ≥ 2 × 第6-9行的均值（0.425），且窗口长度为4（满足要求），因此指标切换为1，滚动均值计算起点重置为第10行

解决方案代码 #

def generate_indicator(series, multiplier=2, min_window=4, initial_indicator=1):
    indicator = []
    current_start = 0
    current_ind = initial_indicator
    for i, val in enumerate(series):
        window_length = i - current_start + 1
        if window_length >= min_window:
            window_mean = series[current_start:i+1].mean()
            if val >= multiplier * window_mean:
                current_ind *= -1
                current_start = i + 1
        indicator.append(current_ind)
    return pd.Series(indicator, index=series.index)

# 对每一列应用函数生成指标
result_df = df.apply(generate_indicator, multiplier=2, min_window=4)
print(result_df)

代码说明 #

• generate_indicator函数专门处理单列数据：
  • 用current_start跟踪当前滚动窗口的起始索引，current_ind记录当前指标值
  • 逐行遍历数据，计算当前窗口长度，若满足最小窗口要求则计算窗口均值
  • 当当前值达到阈值时，切换指标值并重置窗口起始索引
• 通过df.apply将函数批量应用到DataFrame的每一列，快速生成指标结果

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