如何统计pandas dataframe最近7行满足条件的行数并生成0/1标记列

完整可运行代码 #

import pandas as pd
import numpy as np

def function1(df):
    # 逐行生成基础布尔判断序列
    close_lt_upper = df['close'] < df['upper']
    close_lt_mid = df['close'] < df['mid']
    close_gt_mid = df['close'] > df['mid']
    
    # 过去7行（不含当前行，和原有shift逻辑完全对齐）全部小于upper
    no_upper_breach = close_lt_upper.shift(1).rolling(7).sum() == 7
    # 过去7行至少2行小于mid
    two_below_middle = close_lt_mid.shift(1).rolling(7).sum() >= 2
    # 过去7行至少2行大于mid
    two_above_middle = close_gt_mid.shift(1).rolling(7).sum() >= 2
    
    # 合并三个条件，布尔值转1/0，前7行无足够数据的空值统一填充为0
    df['function1'] = (no_upper_breach & two_below_middle & two_above_middle).fillna(0).astype(int)
    return df

问题1：计数逻辑说明 #

• 首先逐行生成收盘价小于mid、大于mid的布尔序列，布尔值在数值运算中会自动转换为1（True）和0（False）
• 用pandas的rolling(7)创建长度为7的滚动窗口，配合sum()方法即可统计窗口内满足条件的行数
• 判定统计结果≥2，就可以得到“至少2行满足条件”的布尔序列
• 如果你的业务逻辑需要包含当前行统计最近7行，直接删除代码里所有的.shift(1)即可

问题2：条件合并逻辑说明 #

• 三个条件返回的都是布尔类型序列，直接用&做按位与运算，只有三个条件同时为True时，最终结果才为True
• 用astype(int)可以直接将布尔值转换为1（True）和0（False），前7行没有足够历史数据产生的NaN值，用fillna(0)统一填充为0即可

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