Python DataFrame中二进制月度数据的异常检测与值预测方法问询

需求概述 #

需要为月度更新的二进制Exclusion字段（1=排除，0=纳入）构建异常检测机制，适配3000+实体的场景，核心诉求是基于历史数据（可限定最近N个月）预测下月最可能取值，标记不符合预期的变动，无需复杂机器学习模型。

预期检测规则参考：

• 接受与历史完全一致的取值（如实体A持续为1）
• 接受历史有变化但近期无波动的取值（如实体B后期稳定为1）
• 接受切换为历史占比更高的取值（如实体C从1切回0，0为历史主流值）
• 接受规律切换的取值（如实体D交替0/1）
• 标记首次出现的新值（如实体E首次出现0）
• 标记打破近期趋势的取值（如实体F打破连续3个月的1趋势切为0）

示例数据 #

import pandas as pd
d = {'Date': [20210101, 20210201,20210301,20210401,20210501,
              20210101, 20210201,20210301,20210401,20210501,
              20210101,20210201,20210301,20210401,20210501,
              20210101,20210201,20210301,20210401,20210501,
              20210101,20210201,20210301,20210401,20210501,
              20210101,20210201,20210301,20210401,20210501,], 
              
     'Entity': ['A','A','A','A','A',
                'B','B','B','B','B',
                'C','C','C','C','C',
                'D','D','D','D','D',
                'E','E','E','E','E',
                'F','F','F','F','F'],
    'Exclusion': [1,1,1,1,1,
                  0,0,0,1,1,
                  0,0,0,1,0,
                  1,0,1,0,1,
                  1,1,1,1,0,
                  0,1,1,1,0]}
df = pd.DataFrame(data=d)
# 转换日期格式并排序，确保时间顺序正确
df['Date'] = pd.to_datetime(df['Date'], format='%Y%m%d')
df = df.sort_values(['Entity', 'Date']).reset_index(drop=True)

简易解决方案 #

以下方案均基于Pandas分组操作，批量处理所有实体，代码易读易调整，适合新手。

方法1：近期频率多数投票（基础版） #

核心逻辑：取每个实体最近N个月的历史值，预测出现次数最多的取值；若最新值与预测值不符，标记为可疑。

def predict_most_frequent(group, window=6):
    # 获取最近window条历史数据
    recent_vals = group['Exclusion'].tail(window)
    # 取出现次数最多的值（平局时默认取1，可按需修改）
    pred_val = recent_vals.mode().iloc[0] if not recent_vals.mode().empty else 1
    latest_val = group['Exclusion'].iloc[-1]
    # 标记可疑：至少有2条历史数据，且最新值≠预测值
    group['Suspicious'] = 0
    if len(recent_vals) >= 2 and latest_val != pred_val:
        group['Suspicious'].iloc[-1] = 1
    return group

# 按实体分组处理
result_df = df.groupby('Entity').apply(predict_most_frequent, window=6).reset_index(drop=True)

# 查看可疑记录
print(result_df[result_df['Suspicious'] == 1])

方法2：趋势+频率结合（进阶版） #

针对打破近期连续趋势的情况（如实体F），优先判断近期连续取值，再结合多数投票：

def predict_trend_plus_freq(group, window=6, trend_window=3):
    recent_vals = group['Exclusion'].tail(window)
    latest_val = group['Exclusion'].iloc[-1]
    group['Suspicious'] = 0

    # 先检查近期连续趋势
    if len(recent_vals) >= trend_window:
        trend_vals = recent_vals.tail(trend_window)
        # 若最近trend_window个月取值完全一致
        if len(trend_vals.unique()) == 1:
            trend_pred = trend_vals.iloc[0]
            if latest_val != trend_pred:
                group['Suspicious'].iloc[-1] = 1
                return group
    
    # 无连续趋势时用多数投票判断
    pred_val = recent_vals.mode().iloc[0] if not recent_vals.mode().empty else 1
    if len(recent_vals) >= 2 and latest_val != pred_val:
        group['Suspicious'].iloc[-1] = 1
    return group

result_df = df.groupby('Entity').apply(predict_trend_plus_freq).reset_index(drop=True)

方法3：首次出现值检测（针对性版） #

针对首次出现新值的情况（如实体E），单独判断历史是否出现过该值，再结合频率：

def predict_with_first_occurrence(group, window=6):
    latest_val = group['Exclusion'].iloc[-1]
    group['Suspicious'] = 0

    # 检查历史中是否出现过最新值
    has_historical = (group['Exclusion'].iloc[:-1] == latest_val).any()
    if not has_historical:
        group['Suspicious'].iloc[-1] = 1
        return group
    
    # 结合近期多数投票判断
    recent_vals = group['Exclusion'].tail(window)
    pred_val = recent_vals.mode().iloc[0] if not recent_vals.mode().empty else 1
    if len(recent_vals) >= 2 and latest_val != pred_val:
        group['Suspicious'].iloc[-1] = 1
    return group

result_df = df.groupby('Entity').apply(predict_with_first_occurrence).reset_index(drop=True)

方案优势 #

• 批量处理：基于Pandas分组操作，轻松适配3000+实体的场景，运行效率高
• 易维护：代码逻辑清晰，可根据实际需求调整窗口大小、平局规则等
• 无依赖：无需额外机器学习库，仅用基础Pandas功能即可实现

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