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

发表评论 取消回复