单调栈(Monotonic Stack)深度实战:从「下一个更大元素」的第一性原理、严格与非严格单调,到直方图最大矩形、接雨水、股票价格跨度与流式单调对齐(MoChA)的工程全解

单调栈是一种在遍历序列时让栈内元素始终保持单调(递增或递减)顺序的栈变体。它表面上只是一个「压栈前先弹出破坏顺序的元素」的小技巧,本质上却是对「每个元素左右第一个不满足某顺序约束的邻居」这一反复出现的查询做 O(n) amortized 常数化复用的经典手法。本文从第一性原理出发,推导它的不变量与正确性,给出四个基础查询(下一个/上一个更大/更小元素)的统一实现,拆解直方图最大矩形、接雨水、股票价格跨度等经典工程现场,并落到金融(订单簿价格档位、滑动窗口极值)、AI 推理(流式 ASR 的单调对齐 / Monotonic Chunkwise Attention)与系统(CLOCK_MONOTONIC、编译循环依赖分析)的真实场景,最后给出 12 项生产陷阱清单与一套可复现的 Python 工具箱。它是与本系列已发的 Trie、并查集、线段树与树状数组、基数树、堆与优先队列并列的「数据结构工程」拼图——补齐了「极值邻域 + 单调顺序」这一基础维度。

一、第一性原理:为什么需要单调栈

考虑一个最朴素的问题:对数组 a 中的每个元素,求它右侧第一个比它大的元素(Next Greater Element, NGE)。暴力做法是每个位置向后再扫一遍,整体 O(n²)。关键观察是:

当我们从左向右扫描时,若一个元素 a[j] 遇到了比它大的 a[i](i>j),那么对于所有满足 k 且 a[k] <= a[j] 的位置 k,a[i] 同样是其右侧第一个更大元素——a[j] 已经「挡不住」了,可以丢弃。

于是我们维护一个单调递减栈(从栈底到栈顶元素依次变小)。每遇到新元素 a[i]:

  1. 只要栈顶元素 < a[i],就弹出栈顶——被弹出的那个元素找到了它的 NGE = a[i];
  2. 把 a[i] 压入栈,保持栈的单调性。

每个元素最多被压入一次、弹出一次,因此整体线性的 O(n)。这就是单调栈的全部「魔法」:它把「对每个元素的重复扫描」压缩成对栈的进出操作。

核心不变量(invariant):遍历到位置 i 时,栈中保存的始终是「尚未找到各自 NGE 的那些元素」,且它们的下标递增、值递减。单调栈的威力,正来自这个不变量的可复用性。

二、单调递增栈 vs 单调递减栈,以及「存索引还是存值」

目标查询 栈的单调性 弹出条件 典型应用
下一个更大元素 (NGE) 单调递减 栈顶 < 当前 时弹出 温度、股票价格跨度
下一个更小元素 (NSE) 单调递增 栈顶 > 当前 时弹出 直方图最大矩形、接雨水
上一个更大/更小 反向遍历即可 同上,方向对称 区间边界、括号匹配变体

永远存索引,而不是值。 原因有三:(1) 需要下标来计算「距离/宽度」(如直方图矩形的宽度、股票跨度的天数);(2) 数组可能含重复值,单靠值无法定位;(3) 比较时通过 a[st[-1]] 取回值即可,不损失信息。这是 12 项生产陷阱里最高频的一条。

严格 vs 非严格单调:用 < 还是 <= 弹出,决定了「相等元素」如何处理。求 NGE 通常用 <(相等不算「更大」,让相等元素继续在栈中);求股票跨度时用 <= 把相等价格也弹掉(等价价格已失效)。选错符号是经典 off-by-one 来源。

三、四个基础查询的统一实现

下面用一套统一骨架实现 NGE / NSE / PGE(上一个更大)/ PSE(上一个更小)。它们只是「遍历方向」与「比较符号」的四种组合:


def _mono(arr, mode):
    """mode: 'NGE' | 'NSE' | 'PGE' | 'PSE' -> 返回每个位置对应的目标值(-1表示不存在)"""
    n = len(arr)
    res = [-1] * n
    st = []                     # 存下标,保持单调
    order = range(n) if mode[0] == 'N' else range(n - 1, -1, -1)
    greater = 'G' in mode       # True 找更大,False 找更小
    for i in order:
        while st:
            top = st[-1]
            if greater:
                ok = arr[top] < arr[i] if mode in ('NGE', 'PGE') else arr[top] <= arr[i]
            else:
                ok = arr[top] > arr[i] if mode in ('NSE', 'PSE') else arr[top] >= arr[i]
            if not ok:
                break
            res[top] = arr[i]
            st.pop()
        st.append(i)
    if mode[0] == 'P':         # 反向遍历时把结果翻回正向下标
        res = res[::-1]
    return res

def next_greater(arr):      return _mono(arr, 'NGE')
def next_smaller(arr):      return _mono(arr, 'NSE')
def prev_greater(arr):      return _mono(arr, 'PGE')
def prev_smaller(arr):      return _mono(arr, 'PSE')

注意反向遍历(PGE/PSE)后需要把结果翻转回正向顺序。更工程化的写法是为四个查询分别写清晰的小函数(见文末工具箱),避免 _mono 里层层 if 带来的可读性损耗——可读性本身就是生产陷阱之一。

四、经典工程现场

4.1 直方图最大矩形(Largest Rectangle in Histogram)

给定非负整数高度数组,求能勾勒出的最大矩形面积。O(n) 单调栈解法的关键技巧是哨兵:在数组末尾补一个高度 0,强制在最后把所有柱子弹出并计算。


def largest_rectangle(heights):
    st = []
    max_area = 0
    n = len(heights)
    for i in range(n + 1):
        h = heights[i] if i < n else 0          # 哨兵:末尾 0 触发清空
        while st and heights[st[-1]] > h:
            hh = heights[st.pop()]
            # 左边界 = 新栈顶的下一位;右边界 = i(被 h 挡住)
            w = i if not st else i - st[-1] - 1
            max_area = max(max_area, hh * w)
        st.append(i)
    return max_area

正确性:对每根柱子 heights[k],当它从栈中弹出时,i 是右侧第一根比它矮的柱子(右边界),st[-1] 是左侧第一根比它矮的柱子(左边界),于是以 heights[k] 为高的最大矩形宽度恰为 i - st[-1] - 1。

4.2 接雨水(Trapping Rain Water)

对每根柱子,它能接的雨水量由「左侧最高」与「右侧最高」的较小值减去自身高度决定。单调栈视角:维护一个单调递减栈(存下标),当遇到更高的柱子时,栈顶凹陷处(top)被左右两座更高的柱子夹住,可计算这一格的积水量。


def trap(height):
    st = []
    water = 0
    for i, h in enumerate(height):
        while st and height[i] > height[st[-1]]:
            top = st.pop()
            if not st:            # 左侧无更高柱子,构不成凹陷
                break
            dist = i - st[-1] - 1
            bounded = min(height[i], height[st[-1]]) - height[top]
            water += dist * bounded
        st.append(i)
    return water

4.3 股票价格跨度(Stock Span)—— 金融直击

「跨度」定义为当前价格之前连续多少天(含当天)价格不超过当前价格。这是 NGE 的变体:用单调递减栈存价格下标,弹出所有 <= 当前 的历史,跨度 = i - 新栈顶 - 1(栈空则 i+1)。这是量化里「N 日新高计数」「突破窗口」的基础算子。


def stock_span(prices):
    st = []
    spans = []
    for i, p in enumerate(prices):
        while st and prices[st[-1]] <= p:
            st.pop()
        span = (i - st[-1]) if st else (i + 1)
        st.append(i)
        spans.append(span)
    return spans

4.4 每日温度与「下一个更大」家族

LeetCode 739「每日温度」本质就是 NGE 的「距离版」:把返回值从「值」换成「下标差」。同一套 next_greater 逻辑,只需在弹出时记录 i - top 而非 arr[i]。这个家族还包括「小行星碰撞」(带符号的栈模拟)、「字典序最小去重字符串」(带频率计数的约束栈)等——它们共享「利用栈维护一种顺序不变量」的思想。

五、工程落地:金融、AI 推理与系统

5.1 金融:订单簿、滑动窗口极值与波动率突破

  • 订单簿价格档位单调性:限价订单簿的买/卖档天然按价格单调排列。求「某档上方第一档更优价格」「累计深度突破某阈值的价格边界」,与 NGE/NSE 同构;高频撮合中可用单调栈在 O(n) 内重建价格档的相邻关系。
  • 滑动窗口极值 → 引出单调队列:计算 VWAP 的时间窗、滚动波动率、流动性冲击的「过去 N 根 K 线最高/最低」,是单调队列(Monotonic Queue)的领地(见 5.4 的兄弟结构),它与单调栈共享「弹出破坏顺序者」的同一内核,但作用于滑动窗口而非静态序列。
  • 突破检测:把价格序列取 NGE 距离,可快速识别「已连续 N 日未创新高」的疲软标的,是动量/反转因子的轻量实现。

5.2 AI 推理:流式单调对齐(Monotonic Attention / MoChA)

这是单调栈思想在深度学习里最优雅的投影。单调注意力(Monotonic Attention)与单调分块注意力(Monotonic Chunkwise Attention, MoChA)用于流式语音识别(Streaming ASR)与实时字幕:解码器必须从左到右、不回跳地对齐源序列,满足「已消费的位置构成单调前缀」这一硬约束。这恰与单调栈的「不变量」一一对应——已处理边界单调推进、不可回退。MoChA 进一步把「单调边界 + 局部分块软注意力」结合,在延迟与准确率间取得平衡。把对齐问题建模为「在每个解码步决定继续推进还是消费当前源帧」,正是对单调顺序约束的工程化。同理,CTC / Transducer 的对齐搜索、流式 LLM 的「边生成边对齐外部知识」也隐含单调约束。

此外,目标检测中的 非极大值抑制(NMS) 虽是贪心而非严格单调,但其「按置信度排序后逐层剪枝低分重叠框」的消去顺序,与单调栈「按序弹出次优者」的思想同源;在 KV Cache 的重要性淘汰 heuristics 中,也常见基于单调分数阈值(如只保留分数高于某单调边界的 token)的剪枝策略。

5.3 系统:CLOCK_MONOTONIC 与编译循环依赖分析

  • CLOCK_MONOTONIC 的概念桥接:Linux 的 CLOCK_MONOTONIC 提供「只会前进、不受 NTP 回拨影响」的时间源。单调栈维护的「不变量只进不退」与单调时钟的语义同构——在超时、限流、调度器里用单调时钟正是为了避免「时间回退导致栈/队列顺序错乱」这类幽灵 bug。
  • 编译器循环依赖分析:依赖距离向量(dependence distance vector)按循环嵌套层级构成字典序单调序列;判断是否可向量化时,需要检查距离向量是否构成合法单调前缀,这与单调栈维护「前缀顺序不变量」如出一辙。
  • LSM-Tree 的 run 单调性:Leveled compaction 中同一 level 的 run 按 key 范围互不重叠且单调排列,合并时的「找下一个重叠 run」问题可借 NSE/下一个边界查询的同一思路实现。

5.4 兄弟结构:单调队列(Monotonic Queue)

单调栈解决「找每个元素的下一个/上一个极值邻居」;单调队列解决「滑动窗口内的最大/最小」。两者共享「弹出破坏顺序者」的内核,但单调队列用双端队列(deque)在窗口滑动时从队首踢出越界元素:


from collections import deque

def sliding_window_max(arr, k):
    dq = deque()           # 存下标,对应值单调递减
    out = []
    for i, x in enumerate(arr):
        while dq and arr[dq[-1]] <= x:
            dq.pop()
        dq.append(i)
        if dq[0] <= i - k:           # 队首越窗
            dq.popleft()
        if i >= k - 1:
            out.append(arr[dq[0]])
    return out

金融的滚动极值、限流器的「窗口内最大 QPS」、实时异常检测的「窗口峰值」都建立在它之上。认清两者的边界(静态序列 vs 滑动窗口),是避免在错误场景选错结构的常见失误。

六、复杂度与时空权衡

  • 时间:每个元素压入、弹出各一次,O(n)。若误用「每次重新扫描」则会退化为 O(n²)。
  • 空间:栈最多持有 n 个元素,O(n);单次查询额外开销极小。
  • 与线段树的关系:NGE/NSE 本质是「前缀/后缀极值查询」的退化情形;当查询带动态修改或需要区间任意位置极值时,应升级到线段树/树状数组(本系列已专文拆解)。单调栈胜在「无修改、纯静态序列、常数极小」。
  • 严格 vs 非严格:处理相等元素的符号选择会改变语义,必须在设计文档里写明。

七、12 项生产陷阱清单

  1. 存值不存索引:丢失位置信息,宽度/距离算不出——永远存下标。
  2. 严格性符号选错:< vs <= 决定相等如何处置,先写清语义再编码。
  3. 哨兵遗忘:直方图最大矩形漏末尾 0 哨兵,最后一根柱子面积算丢。
  4. 空栈访问:接雨水弹出后不检查 if not st 直接取 st[-1],越界崩溃。
  5. 返回顺序错位:PGE/PSE 反向遍历后忘记翻转结果,下标全部颠倒。
  6. 整数溢出:直方图面积 hh * w 在大数组上溢出 int,用 64 位累加。
  7. 负数/零高度:高度含 0 或负数时单调性假设崩塌,先过滤或特殊处理。
  8. 单调栈与单调队列混淆:静态序列用栈、滑动窗口用 deque,选型错误。
  9. 重复元素静默错误:相等值既不算更大也不算更小,导致「距离」偏差一格。
  10. 语言运行时差异:Python list 当栈 O(1) append/pop 没问题;Java 用 Deque/Stack 注意不要误用老 Stack 的同步开销;Go 用 slice+append。
  11. 超大流的内存:对无限流不能无限压栈,应改为「滑动窗口 + 定期 flush」或转单调队列。
  12. 可观测性缺失:生产环境只返回面积不记录「弹出次数/最大栈深」,线上抖动无从排查——加计数器。

八、可复现 Python 工具箱


from collections import deque

class MonotonicStack:
    """通用单调栈:direction='dec' 维护递减(找下一个更大),'inc' 维护递增(找下一个更小)"""
    def __init__(self, direction='dec', strict=True):
        self.st = []
        self.dir = direction          # 'dec' or 'inc'
        self.strict = strict          # True: 用 < / >;False: 用 <= / >=

    def push(self, value):
        if self.dir == 'dec':
            while self.st and (self.st[-1] < value if self.strict else self.st[-1] <= value):
                self.st.pop()
        else:
            while self.st and (self.st[-1] > value if self.strict else self.st[-1] >= value):
                self.st.pop()
        self.st.append(value)
        return list(self.st)

def next_greater(arr):
    res, st = [-1] * len(arr), []
    for i, x in enumerate(arr):
        while st and arr[st[-1]] < x:
            res[st.pop()] = x
        st.append(i)
    return res

def next_smaller(arr):
    res, st = [-1] * len(arr), []
    for i, x in enumerate(arr):
        while st and arr[st[-1]] > x:
            res[st.pop()] = x
        st.append(i)
    return res

def prev_greater(arr):
    res, st = [-1] * len(arr), []
    for i in range(len(arr) - 1, -1, -1):
        while st and arr[st[-1]] < arr[i]:
            res[st.pop()] = arr[i]
        st.append(i)
    return res[::-1]

def prev_smaller(arr):
    res, st = [-1] * len(arr), []
    for i in range(len(arr) - 1, -1, -1):
        while st and arr[st[-1]] > arr[i]:
            res[st.pop()] = arr[i]
        st.append(i)
    return res[::-1]

def largest_rectangle(heights):
    st, max_area, n = [], 0, len(heights)
    for i in range(n + 1):
        h = heights[i] if i < n else 0
        while st and heights[st[-1]] > h:
            hh = heights[st.pop()]
            w = i if not st else i - st[-1] - 1
            max_area = max(max_area, hh * w)
        st.append(i)
    return max_area

def trap(height):
    st, water = [], 0
    for i, h in enumerate(height):
        while st and height[i] > height[st[-1]]:
            top = st.pop()
            if not st:
                break
            water += (i - st[-1] - 1) * (min(height[i], height[st[-1]]) - height[top])
        st.append(i)
    return water

def stock_span(prices):
    st, spans = [], []
    for i, p in enumerate(prices):
        while st and prices[st[-1]] <= p:
            st.pop()
        spans.append((i - st[-1]) if st else (i + 1))
        st.append(i)
    return spans

def sliding_window_max(arr, k):
    dq, out = deque(), []
    for i, x in enumerate(arr):
        while dq and arr[dq[-1]] <= x:
            dq.pop()
        dq.append(i)
        if dq[0] <= i - k:
            dq.popleft()
        if i >= k - 1:
            out.append(arr[dq[0]])
    return out

if __name__ == "__main__":
    a = [2, 1, 2, 4, 3]
    print("NGE:", next_greater(a))        # [4, 2, 4, -1, -1]
    print("NSE:", next_smaller(a))        # [1, -1, 3, 3, -1]
    print("span:", stock_span([100, 80, 60, 70, 60, 75, 85]))  # [1,1,1,2,1,4,6]
    print("hist:", largest_rectangle([2, 1, 5, 6, 2, 3]))       # 10
    print("trap:", trap([0, 1, 0, 2, 1, 0, 1, 3, 2, 1, 2, 1])) # 6

小结

单调栈用「遍历时维持一个单调不变量」把 O(n²) 的邻域查询压到 O(n),其灵魂是每个元素只进栈一次、出栈一次。它与线段树/树状数组、堆、Trie、并查集、基数树同属「数据结构工程」系列,分别覆盖区间聚合、极值优先、前缀匹配、等价划分、路径压缩等正交维度。在金融里它是订单簿与突破检测的轻量算子,在 AI 推理里它是流式单调对齐(MoChA)的硬约束来源,在系统里它与 CLOCK_MONOTONIC 的「只进不退」语义同构。掌握「存索引、写清严格性、补对哨兵、分清栈与队列」这四句口诀,你就能在绝大多数单调顺序问题里一眼看出它的影子。

点赞(0) 打赏

评论列表 共有 0 条评论

暂无评论
立即
投稿
网站二维码

微信公众账号

微信扫一扫加关注

发表
评论
返回
顶部