题目重点：把高频三段题合成一篇：累计订单费用 O(1) 查询、按结束时间堆结算、逐司机合并区间后扫描并发峰值。
题意与假设
这道练习题保留了面经中值得训练的核心约束，并将需要澄清的条件单独列出。现场拿到变体时，先和面试官确认这些条件。
- 时间为 Unix 秒，配送 (start,end) 满足 0<时长≤3h；展示例按半开 [start,end)；司机已注册且输入有效。
- 训练实现用 Decimal(str(hourly_rate)) 降低金额误差，公开 API 返回 float； 的该轮面试官明确说 cents 仍要保留小数，不能未经约定就把每单付款四舍五入到整数 cents。真实发薪必须在约定节点做舍入并使用货币单位/账务规则。
- `pay_up_to(cutoff)` 只结算当时已经记录且 end≤cutoff 的订单；之后迟到补录订单要再次调用才能结算。题目没有声明 cutoff 单调或延迟录入策略。
解题路线
- Part I：driver_id→费率；记录每笔配送时按 fee×秒数/3600 算其独立费用，同时更新 total 和 unpaid 两个累计值。配送重叠不影响费用，`getTotalCost()` 和 `getUnpaidAmount()` O(1)。
- Part II：未付配送放 (endTime,唯一序号) 最小堆；`pay_up_to` 反复弹出 end≤cutoff 的项，并从 unpaid 扣费。每单只入堆/出堆一次，所以重复或倒退 cutoff 不会二次付款。
- Part III：按司机收集过去 [now−86400,now) 的配送交集；先对同司机排序合并重叠或相接区间，再把每段作为 +1/-1 时间事件，按时间合并净变化，扫描最大值。这样同司机同时送两单只计一次。
- 若要返回最大区间而非数值，扫描相邻不同时间戳间的常值区段；同刻结束/开始按半开端点语义汇总后更新，避免瞬时虚高。数万司机×每周数百单时，Part III 每次全量扫描可考虑按时间分桶与增量索引，但先核对查询频率。
手动走一遍
司机 A 每小时 $10，同时做 0–3600 和 1800–5400 秒两单，各得 $10，总费用 $20。cutoff=3600 只结第一单、未付 $10；再次同 cutoff 不变。并发人数在 1800–3600 仍只算 A 一人；若 B 在该段上线，才是两人。
Python 参考实现
下面的代码按上文假设实现，适合本地运行并追加自己的测试用例。
代码区可横向滚动；使用键盘时，先聚焦代码区再按方向键查看。
from collections import defaultdict
from decimal import Decimal
import heapq
class DeliveryBilling:
"""In-memory, single-threaded training implementation of all three parts."""
def __init__(self) -> None:
self.rates: dict[int, Decimal] = {}
self.deliveries: list[tuple[int, int, int, Decimal]] = []
self.unpaid_heap: list[tuple[int, int]] = []
self.total = Decimal(0)
self.unpaid = Decimal(0)
def add_driver(self, driver_id: int, usd_hourly_rate: float) -> None:
if driver_id in self.rates or usd_hourly_rate < 0:
raise ValueError("invalid driver")
self.rates[driver_id] = Decimal(str(usd_hourly_rate))
def record_delivery(self, driver_id: int, start: int, end: int) -> None:
if driver_id not in self.rates or not (0 < end - start <= 10_800):
raise ValueError("invalid delivery")
payout = self.rates[driver_id] * Decimal(end - start) / Decimal(3600)
index = len(self.deliveries)
self.deliveries.append((driver_id, start, end, payout))
heapq.heappush(self.unpaid_heap, (end, index))
self.total += payout
self.unpaid += payout
def get_total_cost(self) -> float:
return float(self.total)
def pay_up_to(self, cutoff: int) -> None:
while self.unpaid_heap and self.unpaid_heap[0][0] <= cutoff:
_, index = heapq.heappop(self.unpaid_heap)
self.unpaid -= self.deliveries[index][3]
def get_unpaid_amount(self) -> float:
return float(self.unpaid)
def max_simultaneous_drivers_in_past_24_hours(self, now: int) -> int:
window_start = now - 86_400
by_driver: dict[int, list[tuple[int, int]]] = defaultdict(list)
for driver, start, end, _ in self.deliveries:
left, right = max(start, window_start), min(end, now)
if left < right:
by_driver[driver].append((left, right))
changes: dict[int, int] = defaultdict(int)
for intervals in by_driver.values():
intervals.sort()
left, right = intervals[0]
for next_left, next_right in intervals[1:]:
if next_left <= right:
right = max(right, next_right)
else:
changes[left] += 1
changes[right] -= 1
left, right = next_left, next_right
changes[left] += 1
changes[right] -= 1
active = maximum = 0
for time in sorted(changes):
active += changes[time]
maximum = max(maximum, active)
return maximum
复杂度
录单 O(log D) 因堆，累计查询 O(1)；一次付款 O(k log D) 弹出 k 单，全部付款累计 O(D log D)。过去 24h 统计若窗口内 D_w 单、G 位司机，排序合并和全局事件排序 O(D_w log D_w)，空间 O(D_w+G)。原始训练实现扫描全部已存订单 O(D)，长期运行需时间索引减小候选集。
易错点与追问
容易失分的边界
- 把重叠配送时长合并后计费，少付订单。
- 并发分析直接数 delivery，司机一人多单被重复算。
- 只按 start 扫描不按同刻 end 处理半开边界，出现虚假峰值。
- 截止结算时每次全表扫描或重复扣费。
- 用浮点金额累积结果当真实 payroll ledger 结算值。
继续深挖
- 新记录的订单 end 早于上一次已结 cutoff，如何补偿？
- 若司机费率有生效日期，recordDelivery 应按开始、结束还是各时间片费率计价？公开题库预览提费率变化但详情未给。
- 若 pay_up_to 与 recordDelivery 并发，锁粒度和线性化点如何选？
- 如何返回达到峰值的全部时间区间？
- 百万历史单、每秒多次 24h 查询时怎样维护滚动并发索引？
