基于Python线性回归的一维数据流漂移检测置信度异常问题咨询

现有实现的问题 #

你当前的代码有几个明显的逻辑错误和设计偏差，是导致得分不符合预期的核心原因：

• 低级语法/命名错误
  测试代码中你定义的类是UnivariateDriftAnalysis，但实例化时用了不存在的LinearRegressionSPC；类里的得分计算方法是drift_detected，调用时用了不存在的alarm_score；最后打印的是未定义的result变量，这些错误会直接导致代码无法运行。
• 窗口索引逻辑错误
  你用全局索引x0[i:i+self.n_window]做回归的x变量，而且i仅在窗口满足长度时才递增，会导致x和窗口内的y完全错位，计算出来的p值本身就是错的。实际上回归只需要窗口内的相对索引0 ~ n_window-1即可，不需要关联全局位置。
• 得分映射逻辑缺失
  你直接返回1-p_value当置信度，这个值本身是「斜率不为0的概率」，是0~1的连续值，当然不会自动符合你要求的「无漂移时<=0.2、有漂移时>=0.9」的分段范围，你需要额外做阈值映射。
• 统计假设理解偏差
  线性回归的p值只能检测线性趋势类漂移，如果是阶跃式的突变漂移（比如你测试数据里直接从24_47跳到10004000的情况），用固定窗口的线性回归是很难检测到的，刚跳变的几个点计算出来的p值大概率达不到你的阈值要求。

符合要求的实现方案 #

第一步：修复基础逻辑，得到正确的p值 #

首先把代码里的基础错误修复，得到正确的「原假设（无漂移，斜率为0）成立的概率」，再添加得分映射逻辑适配你的要求：

import numpy as np
from scipy import stats

class UnivariateDriftAnalysis:
    def __init__(self, n_window, p_threshold=0.01):
        '''
        n_window: 滑动窗口长度
        p_threshold: 假设检验的显著性水平，p低于该值则判定存在漂移
        '''
        self.n_window = n_window
        self.p_threshold = p_threshold
        # 缓存窗口数据，用于流式更新
        self.window = []

    def calculate_score(self, data) -> list:
        '''批量计算所有位置的漂移置信度得分'''
        result = [np.NaN] * self.n_window
        for i in range(self.n_window, len(data)):
            window_data = data[i-self.n_window:i]
            # 窗口内用相对索引做回归
            x = np.arange(self.n_window)
            p_value = stats.linregress(x, window_data).pvalue
            # 映射到要求的得分范围
            score = self._map_score(p_value)
            result.append(score)
        return result

    def _map_score(self, p_value):
        '''把p值映射到要求的得分范围'''
        # 无漂移判定阈值：p>=0.8时认为无漂移，得分落在0~0.2区间
        # 有漂移判定阈值：p<=p_threshold时认为有漂移，得分落在0.9~1区间
        # 中间为过渡区间，得分线性落在0.2~0.9之间
        if p_value >= 0.8:
            return np.interp(p_value, [0.8, 1], [0.2, 0])
        elif p_value <= self.p_threshold:
            return np.interp(p_value, [0, self.p_threshold], [1, 0.9])
        else:
            return np.interp(p_value, [self.p_threshold, 0.8], [0.9, 0.2])

    def update(self, new_data_point) -> float:
        '''流式处理单个数据点，返回当前得分'''
        self.window.append(new_data_point)
        if len(self.window) < self.n_window:
            return np.NaN
        self.window = self.window[-self.n_window:]
        x = np.arange(self.n_window)
        p_value = stats.linregress(x, self.window).pvalue
        return self._map_score(p_value)

# 测试代码
np.random.seed(100)
n_window = 10
detector = UnivariateDriftAnalysis(n_window=n_window, p_threshold=0.01)
data = np.concatenate([np.random.randint(24, 47, 1500), np.random.randint(1000, 4000, 2000), np.random.randint(1, 5, 500)])
scores = detector.calculate_score(data)
print(scores[n_window:])

第二步：可选优化（针对阶跃漂移） #

如果你的场景里突变漂移占比高，建议额外加分布检验逻辑，比如把当前窗口和历史基准窗口的数值分布做KS检验，结合线性回归和分布检验两种p值计算最终置信度，能覆盖更多漂移场景。

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