LinkedIn AlertMonitor：15 分钟窗口、整点统计与下一更高分钟

题目重点：把三种时间语义拆开，并用单调栈求每分钟右侧首个严格更大告警量；公开示例存在矛盾。

题意与假设

这道练习题保留了面经中值得训练的核心约束，并将需要澄清的条件单独列出。现场拿到变体时，先和面试官确认这些条件。

训练代码使用非负整数秒；15 分钟窗口为 [now-900, now)，按传入查询时刻计算；当前小时是 UTC 风格 floor(now/3600) 所在的 [start,start+3600)。题目一处写 0..3600、下一处写 3601..7200，与 3720→3600..7199 冲突，故不声称这是原题唯一边界。
spike 取最后 15 分钟内有告警的分钟，按绝对分钟 ID 升序排列，返回右侧第一个**严格更大**的下标；源码注释说“greater”，但样例 [1,2,5,2,2,1] 给第 3 位输出 4，只有“大于等于”才成立。本文按文字规则，结果为 [1,2,-1,-1,-1,-1]。
训练版允许乱序写入，故用简单事件列表保持正确性；高吞吐优化需要 watermark/迟到策略。

解题路线

15 分钟计数扫描事件并用半开区间判定；若时间戳单调且查询时间不回退，可按秒或分钟环形桶维护 O(1) 摊还更新/查询，但乱序数据需要修正策略。
严重程度分布按查询时刻所在的整点块统计三种 severity，缺失级别返回 0；生产可维护 hour×severity 聚合与保留期，防止全量扫描。
spike 先聚合最后窗口内的分钟计数，再从左到右维护计数非递增的下标栈；新计数严格大于栈顶时，将该位置答案设为当前下标。相等值保留在栈中，避免把相等误判为 greater。
写和读不能仅凭最新记录时刻随意移动时间窗口；显式 `now` 更易测试。若 API 无参数，则本文训练版使用最新事件秒+1，需在真题中核实查询时刻。

手动走一遍

在 t=900 查询，t=0 告警不计、t=899 计、t=900 不计。t=3720 的小时为 [3600,7200)。六个分钟计数 [1,2,5,2,2,1] 严格下一更大下标为 [1,2,-1,-1,-1,-1]，题目给出的第 3 位 4 与文字不符。

Python 参考实现

下面的代码按上文假设实现，适合本地运行并追加自己的测试用例。

代码区可横向滚动；使用键盘时，先聚焦代码区再按方向键查看。

from collections import defaultdict


class AlertMonitor:
    """Training contract: nonnegative integer seconds; arbitrary insertion order."""

    SEVERITIES = ("LOW", "MEDIUM", "HIGH")

    def __init__(self):
        self.events = []
        self.latest = None

    def record_alert(self, timestamp: int, severity: str) -> None:
        if timestamp < 0 or severity not in self.SEVERITIES:
            raise ValueError("invalid alert")
        self.events.append((timestamp, severity))
        self.latest = timestamp if self.latest is None else max(self.latest, timestamp)

    def report_alerts_last_15_min(self, now: int) -> int:
        if now < 0:
            raise ValueError("invalid query time")
        # Half-open [now - 900, now): events at exactly now are excluded.
        return sum(now - 900 <= ts < now for ts, _ in self.events)

    def report_severity_distribution(self, now: int) -> dict[str, int]:
        if now < 0:
            raise ValueError("invalid query time")
        start = (now // 3600) * 3600
        counts = dict.fromkeys(self.SEVERITIES, 0)
        for ts, severity in self.events:
            if start <= ts < start + 3600:
                counts[severity] += 1
        return counts

    def detect_alert_volume_spike(self, now: int | None = None) -> tuple[list[int], list[int]]:
        """Return (minute IDs, next-greater positions) in the last 15-minute window."""
        if now is None:
            now = 0 if self.latest is None else self.latest + 1
        if now < 0:
            raise ValueError("invalid query time")
        by_minute = defaultdict(int)
        for ts, _ in self.events:
            if now - 900 <= ts < now:
                by_minute[ts // 60] += 1
        minutes = sorted(by_minute)  # Source says each reported minute has an alert.
        counts = [by_minute[m] for m in minutes]
        next_greater = [-1] * len(counts)
        stack = []
        for i, count in enumerate(counts):
            while stack and count > counts[stack[-1]]:
                next_greater[stack.pop()] = i
            stack.append(i)
        return minutes, next_greater

复杂度

训练代码存 E 条事件 O(E)，每次窗口/小时查询 O(E)；spike 扫 E 条、排序至多 M 个分钟 O(E+M log M)，单调栈 O(M)。若连续时间和有限窗口成立，可用 900 秒环形桶把短窗计数降到 O(1) 查询，但需处理乱序、时间跳跃及并发。

易错点与追问

容易失分的边界

未说清窗口是 [now-900,now) 还是含 now。
把小时滚动窗口误当整点小时。
相等分钟量按严格 greater 弹栈。
照抄自相矛盾的 spike 示例而不指出。
只存最近 15 分钟却还承诺查询任意历史小时。

继续深挖

若允许迟到 2 小时的 alert，小时桶怎样修正？
spike 若定义为大于等于，栈条件怎样改？
多线程写入和查询如何给一致快照？
时间戳跨小时、跨天或查询过去时刻怎么办？