4. 持续报警与陈旧报警识别规则实现

识别对象与判定口径

本节要交付的是一份“持续报警 + 陈旧报警”清单，按设备与工段分组。两类对象共用同一份标准化报警记录，但判定逻辑不同，必须分开计算再合并输出。

• 持续报警：同一报警位号（tag）在观察窗口内连续处于激活状态，且持续时间超过设定阈值 persist_min。判定依据是“激活区间长度”，不是报警条数。
• 陈旧报警：报警已产生但长时间未被操作员确认（ack），或确认后长时间未恢复。判定依据是“未确认时长”或“确认后未恢复时长”，阈值 stale_ack_min。

关键前提：原始报警日志通常是“事件流”（激活/确认/恢复三条独立记录），必须先按 tag 重建成状态区间，才能计算时长。直接对事件行做时间差是最常见的错误来源。

状态区间重建

对每个 tag 按时间排序，用状态机把事件流折叠成区间。下面给出可直接运行的 Python 实现，输入为 DataFrame，字段为 tag、event（ACT/ACK/CLR）、ts、device、section、priority。

import pandas as pd

def build_intervals(df, window_end):
    """把事件流重建为报警区间。返回每个区间的激活/确认/恢复时刻。"""
    df = df.sort_values(["tag", "ts"])
    rows = []
    for tag, g in df.groupby("tag", sort=False):
        active_since = None
        ack_at = None
        meta = g.iloc[0]
        for _, e in g.iterrows():
            if e.event == "ACT" and active_since is None:
                active_since, ack_at = e.ts, None
            elif e.event == "ACK" and active_since is not None and ack_at is None:
                ack_at = e.ts
            elif e.event == "CLR" and active_since is not None:
                rows.append(dict(tag=tag, device=meta.device, section=meta.section,
                                 priority=meta.priority, start=active_since,
                                 ack=ack_at, end=e.ts))
                active_since, ack_at = None, None
        if active_since is not None:  # 窗口结束时仍未恢复
            rows.append(dict(tag=tag, device=meta.device, section=meta.section,
                             priority=meta.priority, start=active_since,
                             ack=ack_at, end=window_end))
    return pd.DataFrame(rows)

注意 window_end 的传入：未恢复的报警必须用观察窗口终点截断，否则时长会被算成无穷大或负数。这是海洋平台长周期数据里最常见的边界问题。

规则判定与分组输出

def classify(intervals, persist_min=30, stale_ack_min=60):
    iv = intervals.copy()
    iv["dur_min"] = (iv.end - iv.start).dt.total_seconds() / 60
    iv["unack_min"] = iv.apply(
        lambda r: ((r.ack if pd.notna(r.ack) else r.end) - r.start).total_seconds() / 60, axis=1)

    iv["is_persist"] = iv.dur_min >= persist_min
    iv["is_stale"] = (iv.unack_min >= stale_ack_min) | \
                     ((iv.ack.notna()) & ((iv.end - iv.ack).dt.total_seconds() / 60 >= stale_ack_min))
    return iv

def grouped_report(iv):
    hit = iv[iv.is_persist | iv.is_stale].copy()
    hit["类型"] = hit.apply(
        lambda r: "持续+陈旧" if r.is_persist and r.is_stale
                  else ("持续" if r.is_persist else "陈旧"), axis=1)
    out = (hit.groupby(["section", "device", "tag", "类型"])
              .agg(区间数=("tag", "size"),
                   最长持续min=("dur_min", "max"),
                   最长未确认min=("unack_min", "max"),
                   最高优先级=("priority", "min"))
              .reset_index()
              .sort_values(["section", "device", "最长持续min"], ascending=[True, True, False]))
    return out

示例数据（教学假设）：某平台生产分离器 P-101 的高压报警在 08:00 激活、08:52 才确认、09:40 恢复，则持续 100 min、未确认 52 min；在 persist_min=30、stale_ack_min=60 下判为“持续”而非“陈旧”。若把 stale_ack_min 调到 45，则同时命中两类。阈值必须与平台实际班次和响应能力匹配，不能照搬默认值。

结果校验要点

清单生成后至少做三项核对：一是区间总数与原始 ACT 事件数是否一致（未恢复的应计入）；二是抽查 2~3 个 tag 的手工时间差与脚本结果是否吻合；三是确认“持续+陈旧”双命中项是否被正确标注，避免被单一类型覆盖。若某工段清单为空，先检查该工段是否在 section 字段中存在拼写差异，而不是直接判定无异常。