Implement a simplified version of a banking system. All operations have a timestamp parameter (stringified ms, range 1 to 10⁹, strictly increasing).
Level 1 — Account Creation, Deposit, Transfer
create_account(timestamp, account_id) -> booldeposit(timestamp, account_id, amount) -> int | Nonetransfer(timestamp, source_account_id, target_account_id, amount) -> int | None
Level 2 — Top Spenders
top_spenders(timestamp, n) -> list[str]— topnaccounts by highest outgoing total (transfer source + pay). Format"account_id(total_outgoing)". Tiebreak alphabetically.
Level 3 — Scheduled Payments with 2% Cashback
pay(timestamp, account_id, amount) -> str | None— withdrawamount. Provides 2% cashback (rounded down) refunded 24 hours later (86_400_000ms). Returns"payment{ordinal}". Cashback processes before any other transaction at its refund timestamp.get_payment_status(timestamp, account_id, payment_id) -> str | None—"IN_PROGRESS"or"CASHBACK_RECEIVED".
Level 4 — Account Merging with Balance History
merge_accounts(timestamp, account_id_1, account_id_2) -> bool— mergeaccount_id_2intoaccount_id_1. Pending cashbacks for account 2 still refund to account 1. Queries with oldpayment_idmust still work via account 1.get_balance(timestamp, account_id, time_at) -> int | None— historical balance after operations attime_at.
解法
四级状态机。账户字典维护 balance + outgoing + history。Cashback 用按 (refund_ts, payment_id) 排序的小顶堆，每次操作开头先 flush，确保同一时间戳的退款先于用户操作发生。合并用 merged_to 重定向表(并查集风格)让旧账户 id 和 payment id 仍能解析。get_balance 在账户历史列表上二分。多数操作均摊 O(log n)。设计题，只给 Python 实现。
import heapq, bisect
from collections import defaultdict
from typing import Optional, List
class Bank:
DAY_MS = 86_400_000
def __init__(self):
self.accounts = {}
self.payments = {}
self.payment_counter = 0
self.cashback_heap = []
self.merged_to = {}
def _resolve(self, aid):
while aid in self.merged_to: aid = self.merged_to[aid]
return aid
def _flush_cashback(self, ts):
while self.cashback_heap and self.cashback_heap[0][0] <= ts:
refund_ts, pid = heapq.heappop(self.cashback_heap)
pay = self.payments[pid]
if pay["status"] == "CASHBACK_RECEIVED": continue
aid = self._resolve(pay["account"])
if aid not in self.accounts: continue
refund = pay["amount"] // 50
acc = self.accounts[aid]
acc["balance"] += refund
acc["history"].append((refund_ts, acc["balance"]))
pay["status"] = "CASHBACK_RECEIVED"
def create_account(self, ts, aid):
self._flush_cashback(ts)
if aid in self.accounts: return False
self.accounts[aid] = {"balance": 0, "outgoing": 0, "history": [(ts, 0)]}
return True
def deposit(self, ts, aid, amount):
self._flush_cashback(ts)
aid = self._resolve(aid)
if aid not in self.accounts: return None
acc = self.accounts[aid]
acc["balance"] += amount
acc["history"].append((ts, acc["balance"]))
return acc["balance"]
def transfer(self, ts, src, dst, amount):
self._flush_cashback(ts)
src = self._resolve(src); dst = self._resolve(dst)
if src == dst or src not in self.accounts or dst not in self.accounts: return None
sa = self.accounts[src]
if sa["balance"] < amount: return None
sa["balance"] -= amount; sa["outgoing"] += amount
self.accounts[dst]["balance"] += amount
sa["history"].append((ts, sa["balance"]))
self.accounts[dst]["history"].append((ts, self.accounts[dst]["balance"]))
return sa["balance"]
def top_spenders(self, ts, n):
self._flush_cashback(ts)
items = sorted(
((aid, acc["outgoing"]) for aid, acc in self.accounts.items()),
key=lambda x: (-x[1], x[0])
)
return [f"{aid}({amt})" for aid, amt in items[:n]]
def pay(self, ts, aid, amount):
self._flush_cashback(ts)
aid = self._resolve(aid)
if aid not in self.accounts: return None
acc = self.accounts[aid]
if acc["balance"] < amount: return None
acc["balance"] -= amount; acc["outgoing"] += amount
acc["history"].append((ts, acc["balance"]))
self.payment_counter += 1
pid = f"payment{self.payment_counter}"
cashback_ts = ts + self.DAY_MS
self.payments[pid] = {"account": aid, "amount": amount, "cashback_ts": cashback_ts, "status": "IN_PROGRESS"}
heapq.heappush(self.cashback_heap, (cashback_ts, pid))
return pid
def get_payment_status(self, ts, aid, pid):
self._flush_cashback(ts)
aid = self._resolve(aid)
if aid not in self.accounts or pid not in self.payments: return None
if self._resolve(self.payments[pid]["account"]) != aid: return None
return self.payments[pid]["status"]
def merge_accounts(self, ts, a1, a2):
self._flush_cashback(ts)
a1 = self._resolve(a1); a2 = self._resolve(a2)
if a1 == a2 or a1 not in self.accounts or a2 not in self.accounts: return False
acc1 = self.accounts[a1]; acc2 = self.accounts[a2]
acc1["balance"] += acc2["balance"]
acc1["outgoing"] += acc2["outgoing"]
merged_hist = sorted(acc1["history"] + acc2["history"], key=lambda x: x[0])
clean = []
for tts, bal in merged_hist:
if clean and clean[-1][0] == tts:
clean[-1] = (tts, bal)
else:
clean.append((tts, bal))
acc1["history"] = clean
acc1["history"].append((ts, acc1["balance"]))
self.merged_to[a2] = a1
del self.accounts[a2]
return True
def get_balance(self, ts, aid, time_at):
self._flush_cashback(ts)
aid = self._resolve(aid)
if aid not in self.accounts: return None
hist = self.accounts[aid]["history"]
keys = [h[0] for h in hist]
idx = bisect.bisect_right(keys, time_at) - 1
if idx < 0: return None
return hist[idx][1]You've been asked to program a bot for a popular bank. Every request has a timestamp in seconds (strictly increasing).
"deposit <timestamp> <holder_id> <amount>""withdraw <timestamp> <holder_id> <amount>"— withdraw + 2% cashback rounded down, refunded 24h (86400sec) later. If the same timestamp triggers both a withdraw and a cashback, cashback happens first.
Invalid requests: invalid account number, or insufficient funds. Return final balances after last request, or [-request_id] (1-based) on first invalid.
解法
对请求单次扫描，待退款用按 refund_ts 排序的小顶堆。每条请求先 flush 所有 refund_ts ≤ ts 的退款再处理。首次非法立即返回 [-i](1 起)。复杂度 O((R + C) log C)，C 为退款数。
import heapq
def solution(balances, requests):
n_acc = len(balances)
bal = list(balances)
cashback = []
def flush(ts):
while cashback and cashback[0][0] <= ts:
rt, idx, amt = heapq.heappop(cashback)
bal[idx] += amt
for i, req in enumerate(requests, 1):
parts = req.split()
op, ts, hid, amt = parts[0], int(parts[1]), int(parts[2]), int(parts[3])
flush(ts)
if hid < 1 or hid > n_acc:
return [-i]
idx = hid - 1
if op == "deposit":
bal[idx] += amt
elif op == "withdraw":
if bal[idx] < amt: return [-i]
bal[idx] -= amt
cashback_amt = (amt * 2) // 100
heapq.heappush(cashback, (ts + 86400, idx, cashback_amt))
else:
return [-i]
return balSame 4-level structure as Problem 1, signatures use Optional<Integer> / Optional<String>.
Level 1
createAccount(ts, accountId) -> booleanOptional<Integer> deposit(ts, accountId, amount)Optional<Integer> transferMoney(ts, sourceAccountId, targetAccountId, amount)
Level 2 — Ranking
topSpenders(ts, n)— topnaccounts by total outgoing. Format"accountId(total)".
Level 3 — Scheduled Payments
schedulePayment(ts, accountId, amount, delay)— schedule a payment that executes atts + delay. ReturnspaymentNid.cancelScheduledPayment(ts, paymentId)— only if stillSCHEDULED.
Level 4 — Merge Accounts
boolean mergeAccounts(ts, accountId1, accountId2)Optional<Integer> getBalance(ts, accountId, timeAt)
解法
与第 1 题同样的 4 级骨架，Level 3 把 cashback 换成显式的 schedulePayment / cancelScheduledPayment API；待执行付款用同样的小顶堆。状态机：SCHEDULED → EXECUTED / FAILED / CANCELLED。设计题，只给 Python 骨架。
import heapq
class Bank2:
def __init__(self):
self.accounts = {}
self.scheduled = {}
self.pay_counter = 0
self.queue = []
self.merged = {}
def _resolve(self, aid):
while aid in self.merged: aid = self.merged[aid]
return aid
def _flush_scheduled(self, ts):
while self.queue and self.queue[0][0] <= ts:
et, pid = heapq.heappop(self.queue)
p = self.scheduled[pid]
if p["status"] != "SCHEDULED": continue
aid = self._resolve(p["account"])
if aid not in self.accounts:
p["status"] = "CANCELLED"; continue
acc = self.accounts[aid]
if acc["balance"] < p["amount"]:
p["status"] = "FAILED"; continue
acc["balance"] -= p["amount"]
acc["outgoing"] += p["amount"]
acc["history"].append((et, acc["balance"]))
p["status"] = "EXECUTED"
def schedulePayment(self, ts, aid, amt, delay):
self._flush_scheduled(ts)
aid = self._resolve(aid)
if aid not in self.accounts: return None
self.pay_counter += 1
pid = f"payment{self.pay_counter}"
exec_ts = ts + delay
self.scheduled[pid] = {"account": aid, "amount": amt, "exec_ts": exec_ts, "status": "SCHEDULED"}
heapq.heappush(self.queue, (exec_ts, pid))
return pid
def cancelScheduledPayment(self, ts, pid):
self._flush_scheduled(ts)
if pid not in self.scheduled: return False
if self.scheduled[pid]["status"] != "SCHEDULED": return False
self.scheduled[pid]["status"] = "CANCELLED"
return True