3. 舍内温度序列的异常跳变识别与清洗

问题拆解：三类异常在序列上的不同“指纹”

舍内温度以 1 分钟为采样周期连续采集时，原始序列中常见的异常并非同一种形态，清洗策略必须按形态区分，否则会把真实的快速升温（如开启暖风机）误判为尖峰。

• 短时尖峰：单点或连续 1~2 点偏离邻域，随后立即回到原水平，典型成因为电磁干扰、探头接触抖动。
• 通信丢包：序列出现缺失点，或网关重试后写入重复值/保持值，表现为“平台段”或时间戳间隔异常。
• 传感器漂移：长时间缓慢单向偏移，单点变化率很小，但滑动中位数与同舍参考点（或回风温度）的偏差持续扩大。

因此本例采用两级判据：先用滑动中位数抑制孤立尖峰，再用变化率阈值拦截跳变；漂移则通过“中位数残差持续超限”单独标记，不直接删除，而是转为维护工单。

可运行的清洗实现

下面用 Python 标准库完成整条链路，不依赖第三方包，便于直接放到网关侧或边缘盒子上运行。核心是三个函数：重采样对齐、滑动中位数清洗、漂移标记。

from statistics import median

SAMPLE_MIN = 1          # 采样周期，分钟
WIN = 5                 # 滑动中位数窗口（奇数）
RATE_LIMIT = 0.8        # 变化率阈值，℃/min
RESID_LIMIT = 0.5       # 中位数残差阈值，℃
DRIFT_WIN = 30          # 漂移判定连续超限点数
DRIFT_LIMIT = 1.2       # 漂移残差阈值，℃

def resample(raw, sample_min=SAMPLE_MIN):
    """raw: [(timestamp_min, value or None)]，按整数分钟对齐，缺失填 None"""
    table = {t: v for t, v in raw}
    if not table:
        return []
    t0, t1 = min(table), max(table)
    return [(t, table.get(t)) for t in range(t0, t1 + 1, sample_min)]

def clean(series):
    """返回清洗后序列与异常记录"""
    values = [v for _, v in series]
    out, events = [], []
    for i, (t, v) in enumerate(series):
        if v is None:
            events.append((t, "丢包", None))
            out.append((t, None))
            continue
        lo, hi = max(0, i - WIN // 2), min(len(values), i + WIN // 2 + 1)
        window = [x for x in values[lo:hi] if x is not None]
        m = median(window) if window else v
        resid = v - m
        prev = next((values[j] for j in range(i - 1, -1, -1)
                     if values[j] is not None), v)
        rate = abs(v - prev)
        if abs(resid) > RESID_LIMIT and rate > RATE_LIMIT:
            events.append((t, "尖峰", round(resid, 2)))
            out.append((t, round(m, 2)))
        else:
            out.append((t, v))
    return out, events

def mark_drift(cleaned, ref):
    """ref: 同舍参考序列（回风温度），返回漂移区间"""
    flags, run = [], 0
    for (t, v), (_, r) in zip(cleaned, ref):
        if v is None or r is None:
            run = 0
            continue
        if abs(v - r) > DRIFT_LIMIT:
            run += 1
            if run >= DRIFT_WIN:
                flags.append(t)
        else:
            run = 0
    return flags

用一段教学示例数据验证判据

构造 20 分钟序列：第 8 分钟注入 +3.0 ℃ 尖峰，第 12~14 分钟模拟丢包（None），第 16 分钟起叠加缓慢漂移。以下数值为按上述参数直接推出的预期结果，用于核对实现是否符合判据。

raw = [(i, 24.0 + 0.05 * i) for i in range(20)]
raw[7] = (7, 27.0)          # 尖峰
raw[11] = (11, None)        # 丢包
raw[12] = (12, None)
raw[13] = (13, None)
for i in range(15, 20):     # 漂移叠加
    raw[i] = (i, raw[i][1] + 0.15 * (i - 14))

series = resample(raw)
cleaned, events = clean(series)
print(events)
# 预期：[(7, '尖峰', 2.9x), (11, '丢包', None), (12, '丢包', None), (13, '丢包', None)]

清洗前后对环控决策的对比