2. 基于报警日志的报警风暴时间窗口定位

为什么不能只看“报警总数”

AMS 报警日志里，单条记录通常包含 时间戳、报警点标识（Tag）、报警类型（ALM/RTN）、优先级、来源设备/网段。报警风暴的本质不是“某天报警多”，而是在很短的时间跨度内，报警到达速率显著超过基线。因此定位风暴需要两步：先把离散时间戳聚合成“每分钟报警数”这样的速率序列，再用滑动窗口找出速率持续超阈值的区间，最后区分起始点（第一次越限的时刻）与持续时段（越限到回落到阈值以下的区间）。

时间戳聚类与滑动窗口的计算口径

设日志按时间升序排列，取统计粒度 Δt（工程上常用 1 min），第 k 个桶的报警数为 N(k)。滑动窗口宽度 W（例如 5 min），窗口内报警总数 S(k)=ΣN(i)，i 从 k−W+1 到 k。定义基线 B 为风暴前一段平稳期的每分钟报警数中位数，阈值 T = max(α·B, N_min)，其中 α 为放大倍数（如 5），N_min 为防止基线过低导致误触发的最小绝对量（如 10 条/min）。当 S(k) ≥ T·W 时判定该窗口为风暴窗口。为避免单点抖动造成多次起停，可要求连续 m 个窗口越限才确认起始点。

用一段可运行的脚本完成定位

下面用 Python 标准库实现上述口径，输入为从 AMS 导出的 CSV（字段：ts,tag,type,priority,source），输出风暴起始点、持续时段与峰值窗口。示例数据为教学构造，运行结果属于按给定输入直接推出的预期值，不是现场实测。

import csv
from datetime import datetime, timedelta
from collections import Counter

# ---- 参数（先定死，保证可复现）----
DT_MIN   = 1      # 分钟桶粒度
W        = 5      # 滑动窗口宽度（桶数）
ALPHA    = 5.0    # 相对基线放大倍数
N_MIN    = 10     # 每分钟最小绝对报警数
M_CONFIRM= 2      # 连续越限窗口数才确认起始

def load(path):
    rows = []
    with open(path, newline='', encoding='utf-8') as f:
        for r in csv.DictReader(f):
            r['ts'] = datetime.strptime(r['ts'], '%Y-%m-%d %H:%M:%S')
            rows.append(r)
    rows.sort(key=lambda x: x['ts'])
    return rows

def minute_buckets(rows):
    if not rows:
        return {}, None, None
    t0 = rows[0]['ts'].replace(second=0, microsecond=0)
    t1 = rows[-1]['ts'].replace(second=0, microsecond=0)
    n = int((t1 - t0).total_seconds() // 60) + 1
    cnt = Counter()
    for r in rows:
        k = int((r['ts'].replace(second=0, microsecond=0) - t0).total_seconds() // 60)
        cnt[k] += 1
    return {k: cnt.get(k, 0) for k in range(n)}, t0, n

def baseline(buckets, n):
    # 取前 20% 桶的中位数作为平稳期基线
    head = [buckets[k] for k in range(max(1, n // 5))]
    head.sort()
    return head[len(head) // 2]

def sliding(buckets, n):
    s, out = 0, []
    for k in range(n):
        s += buckets[k]
        if k >= W:
            s -= buckets[k - W]
        out.append(s if k >= W - 1 else None)
    return out

def locate(path):
    rows = load(path)
    buckets, t0, n = minute_buckets(rows)
    B = baseline(buckets, n)
    T = max(ALPHA * B, N_MIN)
    S = sliding(buckets, n)
    over = [k for k in range(n) if S[k] is not None and S[k] >= T * W]

    # 连续 m 个越限窗口才确认起始
    start = None
    for i in range(len(over) - M_CONFIRM + 1):
        if over[i + M_CONFIRM - 1] - over[i] == M_CONFIRM - 1:
            start = over[i] - W + 1
            break
    if start is None:
        return {'baseline': B, 'threshold': T, 'storm': None}

    end = start
    for k in range(start, n):
        if S[k] is not None and S[k] >= T * W:
            end = k
        elif k > end + W:
            break

    peak_k = max(range(start, end + 1), key=lambda k: S[k] or 0)
    return {
        'baseline': B,
        'threshold': T,
        'start': t0 + timedelta(minutes=start),
        'end':   t0 + timedelta(minutes=end),
        'peak_time': t0 + timedelta(minutes=peak_k),
        'peak_count': S[peak_k],
        'duration_min': end - start + 1,
    }

if __name__ == '__main__':
    print(locate('ams_alarm_log.csv'))

按图中示例数据（10:00–10:18，平稳期约 2–3 条/min，10:06–10:12 明显抬升）运行，预期输出形如：baseline≈2, threshold≈10, start=10:06, end=10:12, duration_min≈7。若实际日志的基线为 0（夜间无报警），则 T 由 N_MIN 兜底，避免 α·B=0 导致任何窗口都越限。

结果判读与常见坑

• 起始点偏移：滑动窗口是“回看”的，S(k) 首次越限的桶 k 对应的是窗口末端，真实起始点要回退 W−1 个桶，脚本里用 start = over[i] - W + 1 修正。
• 抖动导致多次起停：若日志中报警呈“爆发—短暂回落—再爆发”，可把 M_CONFIRM 调大，或对 S(k) 做 3 点中值平滑后再判越限。
• 重复报警放大计数：同一 Tag 的 ALM 未 RTN 前反复刷新会虚增 N(k)。定位阶段可先按 (tag, type) 去重，或只统计 ALM 边沿，避免把“刷新”误判为风暴。
• 时间基准统一：AMS 服务器时间、控制器时间、历史库时间可能存在秒级到分钟级偏差，定位前应确认日志时间戳来源，并在排障记录中注明，否则后续与趋势数据对齐会错位。