服务器机柜环境监测数据降噪：网口温湿度传感器异常值过滤实践

┌────────────────────────────────────────────────────────────────────┐
│  类型        │ 特征                    │ 成因                    │
├──────────────┼─────────────────────────┼────────────────────────┤
│ ① 物理越界  │ 温度<-40℃ 或 >85℃     │ 通信帧解析错误/寄存器   │
│              │ 湿度<0% 或 >100%       │ 偏移量配置错误          │
│ ② 突变跳点  │ 相邻采样变化>5%RH/3℃   │ 电磁干扰/气流冲击       │
│              │ 且 30s 内恢复           │                        │
│ ③ 冻结（粘滞）│ 连续 N 个值完全相同   │ 传感器死机/通信缓存复用 │
│ ④ 趋势矛盾  │ 空调未变而温度持续漂移 │ 探头老化/安装松动       │
│ ⑤ 统计离群  │ 偏离滑动窗口均值>3σ   │ 综合噪声               │
└────────────────────────────────────────────────────────────────────┘

from collections import deque
import statistics

class AnomalyDetector:
    def __init__(self, window_size=20, max_frozen=5, max_jump_h=5.0, max_jump_t=3.0):
        self.window = deque(maxlen=window_size)
        self.max_frozen = max_frozen
        self.max_jump_h = max_jump_h
        self.max_jump_t = max_jump_t
        self.consecutive_same = 0
        self.last_value = None

    def check(self, reading: dict) -> tuple:
        """返回 (is_valid, reason, corrected_value)"""
        t = reading.get('temperature', 0)
        h = reading.get('humidity', 0)
        ts = reading.get('timestamp', 0)

        # ① 物理越界
        if t < -40 or t > 85 or h < 0 or h > 100:
            return False, f'PHYSICAL_LIMIT: t={t} h={h}', None

        # ② 突变跳点
        if self.last_value:
            dt = abs(t - self.last_value['temperature'])
            dh = abs(h - self.last_value['humidity'])
            if dt > self.max_jump_t or dh > self.max_jump_h:
                return False, f'JUMP: dt={dt:.1f} dh={dh:.1f}', self.last_value

        # ③ 冻结检测
        if self.last_value and t == self.last_value['temperature'] and h == self.last_value['humidity']:
            self.consecutive_same += 1
            if self.consecutive_same >= self.max_frozen:
                return False, f'FROZEN: {self.consecutive_same} consecutive same', None
        else:
            self.consecutive_same = 0

        # ⑤ 统计离群
        if len(self.window) >= 10:
            recent_h = [v['humidity'] for v in self.window]
            mean_h = statistics.mean(recent_h)
            stdev_h = statistics.stdev(recent_h) if len(recent_h) > 1 else 1.0
            if abs(h - mean_h) > 3 * stdev_h:
                return False, f'OUTLIER: h={h} mean={mean_h:.1f} ±3σ={3*stdev_h:.1f}', None

        self.window.append({'temperature': t, 'humidity': h, 'timestamp': ts})
        self.last_value = {'temperature': t, 'humidity': h}
        return True, 'OK', {'temperature': t, 'humidity': h}

class MovingAverageFilter:
    def __init__(self, window=5):
        self.window = deque(maxlen=window)

    def filter(self, value: float) -> float:
        self.window.append(value)
        return sum(self.window) / len(self.window)

class EMAFilter:
    def __init__(self, alpha=0.3):
        self.alpha = alpha
        self.ema_t = None
        self.ema_h = None

    def filter(self, t: float, h: float) -> tuple:
        if self.ema_t is None:
            self.ema_t, self.ema_h = t, h
        else:
            self.ema_t = self.alpha * t + (1 - self.alpha) * self.ema_t
            self.ema_h = self.alpha * h + (1 - self.alpha) * self.ema_h
        return round(self.ema_t, 2), round(self.ema_h, 2)

class MedianFilter:
    def __init__(self, window=5):
        self.t_window = deque(maxlen=window)
        self.h_window = deque(maxlen=window)

    def filter(self, t: float, h: float) -> tuple:
        self.t_window.append(t)
        self.h_window.append(h)
        return (
            statistics.median(self.t_window),
            statistics.median(self.h_window)
        )

class KalmanFilter:
    def __init__(self, process_noise=0.01, measurement_noise=0.1):
        self.Q = process_noise      # 过程噪声
        self.R = measurement_noise  # 测量噪声
        self.x_t = None             # 温度估计值
        self.x_h = None             # 湿度估计值
        self.P_t = 1.0
        self.P_h = 1.0

    def filter(self, t: float, h: float) -> tuple:
        if self.x_t is None:
            self.x_t, self.x_h = t, h
            return round(t, 2), round(h, 2)

        # 预测
        # 温度
        P_ = self.P_t + self.Q
        K_t = P_ / (P_ + self.R)
        self.x_t = self.x_t + K_t * (t - self.x_t)
        self.P_t = (1 - K_t) * P_

        # 湿度
        P_ = self.P_h + self.Q
        K_h = P_ / (P_ + self.R)
        self.x_h = self.x_h + K_h * (h - self.x_h)
        self.P_h = (1 - K_h) * P_

        return round(self.x_t, 2), round(self.x_h, 2)

原始数据 → ①异常检测 → ②分类处理 → ③平滑滤波 → ④输出

① 异常检测：物理越界/突变/冻结/离群
② 分类处理：
   - 越界 → 直接丢弃，标记设备异常
   - 突变跳点 → 用上一有效值替代（不丢弃，防误判）
   - 冻结 → 连续 5 次相同则标记 STALE，停止参与控制
   - 离群 → 用 EMA 预测值替代
③ 平滑滤波：EMA + 中值联合（先中值去跳点，再 EMA 平滑）
④ 输出：标准化 JSON + 质量码

class DataDenoisingPipeline:
    def __init__(self):
        self.detector = AnomalyDetector(window_size=20)
        self.median = MedianFilter(window=5)
        self.ema = EMAFilter(alpha=0.3)
        self.kalman = KalmanFilter(process_noise=0.01, measurement_noise=0.1)
        self.last_valid = None

    def process(self, reading: dict) -> dict:
        t_raw = reading.get('temperature', 0)
        h_raw = reading.get('humidity', 0)

        # ① 异常检测
        valid, reason, replacement = self.detector.check(reading)
        if not valid:
            if 'FROZEN' in reason or 'PHYSICAL' in reason:
                return {
                    **reading, 'status': 'INVALID',
                    'reason': reason, 'quality': 0
                }
            elif 'JUMP' in reason or 'OUTLIER' in reason:
                # 用上一有效值或 EMA 预测替代
                if self.last_valid:
                    t_raw, h_raw = self.last_valid['temperature'], self.last_valid['humidity']

        # ② 中值滤波
        t_med, h_med = self.median.filter(t_raw, h_raw)

        # ③ EMA 平滑
        t_ema, h_ema = self.ema.filter(t_med, h_med)

        # ④ 卡尔曼（可选，高精度场景启用）
        t_kf, h_kf = self.kalman.filter(t_ema, h_ema)

        result = {
            **reading,
            'temperature_raw': t_raw,
            'temperature': t_kf,
            'humidity_raw': h_raw,
            'humidity': h_kf,
            'status': 'OK',
            'quality': 1,
            'filter_applied': reason if not valid else 'EMA+MEDIAN+KF'
        }
        self.last_valid = {'temperature': t_kf, 'humidity': h_kf}
        return result