[Solution] OpenAI高频题 GPU Credits — 3 Parts 详解

最近在准备 OpenAI 的面试，整理了一道高频题 GPU Credits。题目是根据面经回忆重构的（非原题，练手用）。核心是一个带过期时间的额度账本，三个 part 层层加码，分享一下思路和代码。
题意
三个 API：

• 1. add_credit(credit_id, amount, timestamp, expiration)

  复制代码
  ：发放一笔额度，在闭区间
  1. [timestamp, timestamp+expiration]

  复制代码
  内有效。
• 1. subtract(amount, timestamp)

  复制代码
  ：从当前有效的额度里扣，按最早过期优先，可以跨多笔扣；永远不报错——扣超了变成 debt，后续 timestamp 更大的 grant 先还 debt。
• 1. get_balance(timestamp)

  复制代码
  ：把所有
  1. timestamp <= 查询时间

  复制代码
  的调用按 timestamp 顺序（不是调用顺序）重放一遍，
  1. v = 剩余有效额度 − 未还 debt

  复制代码
  ；
  1. v < 0

  复制代码
  返回 None，否则返回 v（v=0 时返回整数 0）。

约束：timestamp 互不相同；add + subtract ≤ 2e4，query ≤ 2e4。
核心洞察
这题本质是一个 event-sourced ledger：balance 不是存出来的，是按 timestamp 重放算出来的。所有 tricky 的地方都来自三件事：

• grant 是闭区间有效（
  1. t == expiration

  复制代码
  那一刻还能用）；
• debt 不是"最后统一减一下"——grant 到达时先还 debt，这个顺序会影响过期 corner case；
• query 看的是 timestamp 顺序，不是到达顺序。

Part 1：到达有序
事件按 timestamp 到达。维护一个按 expiration 的 min-heap + 一个 debt 计数器，一遍过：add 先还 debt，有剩再入堆；subtract 先踢掉

1. exp < t

复制代码

的过期 grant，再按最早过期扣，扣不完变 debt；query 踢掉过期，

1. sum(heap) − debt

复制代码

。每个操作 O(log n)。
Part 2：到达无序
调用顺序和 timestamp 顺序不一致。标准做法：append-only event log，每次 query 把

1. ts <= T

复制代码

的事件过滤出来按 timestamp 排序重放。每次 query O(U log U)。简单但 query 多了贵——这就是 Part 3 的动机。
Part 3：query 单调 + 事件只增
加两个前提：query 的 timestamp 单调不减；新事件的 timestamp 永远大于已回答的最大 query 时间（late event 直接拒绝）。
做法：unsorted pending buffer + cursor。每次 query 只把

1. ts <= T

复制代码

的事件挑出来排序、重放一次、推进 cursor。每个事件只被排序和重放一次，总 O(U log U + Q)。
这也是常问的 follow-up（"GPU Credit II 怎么优化避免每次从 T0 replay"）的答案：checkpoint——账本状态推进到已回答的最大时间，之后只算增量。
Follow-up 讨论

• undo 一次 subtract：得记住这次扣动了哪些 grant（各扣了多少）以及 debt 变化了多少，再逆操作恢复。注意 grant 可能已过期或被后续扣过，faithful 的 undo 需要版本化。
• timestamp 撞车：题设保证唯一；真撞了，约定同一 timestamp 先 add 后 subtract。
• late add_credit：记一个 max_answered，
  1. ts <= max_answered

  复制代码
  的事件会改写已回答的答案，直接拒绝（或退化到 Part 2 重放）。

易错点

• 闭区间：expire 条件是
  1. exp < t

  复制代码
  ，不是
  1. <=

  复制代码
  。
  1. t == expiration

  复制代码
  时 grant 依然有效。
• debt 语义：grant 到达时先还 debt，不是查询时统一减。超扣之后来的 grant 先填坑，填完剩下的才算余额；剩下的过期了就是 0（不是 None）。
• 1. get_balance

  复制代码
  只有
  1. v < 0

  复制代码
  才返回 None；
  1. v == 0

  复制代码
  返回整数 0。
• Part 2/3 一定要按 timestamp 重放，不能按到达顺序。

代码

1. import heapq, itertools
3. class _Ledger:
4. """核心重放引擎: 事件必须按 timestamp 顺序喂入"""
5. def __init__(self):
6. self._heap = [] # (expiration, seq, remaining) min-heap
7. self._seq = itertools.count()
8. self._debt = 0
10. def _expire(self, t):
11. # 闭区间: t == expiration 时 grant 依然有效, 只有 exp < t 才过期
12. while self._heap and self._heap[0][0] < t:
13. heapq.heappop(self._heap)
15. def add_credit(self, credit_id, amount, timestamp, expiration):
16. if self._debt: # 后到的 grant 先还 debt
17. repay = min(self._debt, amount)
18. self._debt -= repay
19. amount -= repay
20. if amount:
21. heapq.heappush(self._heap, (timestamp + expiration, next(self._seq), amount))
23. def subtract(self, amount, timestamp):
24. self._expire(timestamp)
25. while amount and self._heap: # 最早过期优先, 可跨 grant
26. exp, seq, rem = heapq.heappop(self._heap)
27. take = min(rem, amount)
28. rem, amount = rem - take, amount - take
29. if rem:
30. heapq.heappush(self._heap, (exp, seq, rem))
31. if amount:
32. self._debt += amount # 永不报错, 超扣变 debt
34. def get_balance(self, timestamp):
35. self._expire(timestamp)
36. v = sum(rem for _, _, rem in self._heap) - self._debt
37. return None if v < 0 else v # v==0 返回 0, 不是 None
39. class LedgerP1(_Ledger):
40. """Part 1: 到达有序, 直接维护, 每次操作 O(log n)"""
42. class LedgerP2:
43. """Part 2: 到达无序. append-only log, 每次 query 过滤+排序+重放"""
44. def __init__(self):
45. self._log = []
46. def add_credit(self, credit_id, amount, timestamp, expiration):
47. self._log.append((timestamp, 0, "add", (credit_id, amount, expiration)))
48. def subtract(self, amount, timestamp):
49. self._log.append((timestamp, 1, "sub", amount))
50. def get_balance(self, timestamp):
51. ledger = _Ledger()
52. evs = sorted((e for e in self._log if e[0] <= timestamp),
53. key=lambda e: (e[0], e[1]))
54. for ts, _, kind, p in evs:
55. if kind == "add":
56. ledger.add_credit(p[0], p[1], ts, p[2])
57. else:
58. ledger.subtract(p, ts)
59. return ledger.get_balance(timestamp)
61. class LedgerP3:
62. """Part 3: query 单调不减 + 事件只增(晚于已回答的最大时间).
63. pending buffer + cursor: 每个事件只排序/重放一次, 总 O(U log U + Q)"""
64. def __init__(self):
65. self._pending = []
66. self._max_answered = float("-inf")
67. self._ledger = _Ledger()
68. def _push(self, ts, k, kind, p):
69. if ts <= self._max_answered:
70. raise ValueError("late event")
71. self._pending.append((ts, k, kind, p))
72. def add_credit(self, credit_id, amount, timestamp, expiration):
73. self._push(timestamp, 0, "add", (credit_id, amount, expiration))
74. def subtract(self, amount, timestamp):
75. self._push(timestamp, 1, "sub", amount)
76. def get_balance(self, timestamp):
77. assert timestamp >= self._max_answered
78. due = sorted([e for e in self._pending if e[0] <= timestamp],
79. key=lambda e: (e[0], e[1]))
80. self._pending = [e for e in self._pending if e[0] > timestamp]
81. for ts, _, kind, p in due:
82. if kind == "add":
83. self._ledger.add_credit(p[0], p[1], ts, p[2])
84. else:
85. self._ledger.subtract(p, ts)
86. self._max_answered = timestamp
87. return self._ledger.get_balance(timestamp)

复制代码

三个 part 跑了 300 组随机对拍（和暴力重放逐个 query 对比），都一致。有问题欢迎讨论。

同一个 timestamp 里 add 和 subtract 的处理顺序你是怎么定的？先加后扣还是先扣后加，感觉会影响 debt 和过期额度怎么算。

匿名用户 发表于 2026-10-06 10:44:00
同一个 timestamp 里 add 和 subtract 的处理顺序你是怎么定的？先加后扣还是先扣后加，感觉会影响 debt 和过期额度怎么算。

好问题，这个顺序确实会影响结果，我的实现里定的是同一 timestamp 先 add 后 subtract。

原因：grant 的有效区间是闭区间 [ts, ts+exp]，t 时刻加上去的额度在 t 时刻就是可用的，所以同一时刻的 subtract 应该能扣到它。

顺序不同过期额度的计算确实不一样。举个例子：
- 池子里有一个长期有效的 G1
- add(G2, 100, t, exp=0)，只在 t 这一刻有效
- subtract(60, t)
- 先加后扣：按最早过期先扣，60 从 G2 扣，G2 剩 40（t 之后过期），G1 一分没动
- 先扣后加：60 从 G1 扣，G1 永久少了 60；G2 的 100 加进来但 t 一过就过期
- t+1 时刻的余额两种顺序算出来不一样

debt 的处理：
- subtract 永远不报错，池子里不够扣的部分记成 debt
- 之后 timestamp 更大的 add 会先还 debt，剩下的才进池子
- 还掉的 debt 不会因为用来还款的那个 grant 过期而复活

另外说明一下：我复原的题面里 timestamp 是唯一的，所以同 timestamp 的顺序是实现上的确定性选择。上面的例子可以直接跑代码验证。