Coding Problems — Finance Flavor
Ten problems picked for an AI Agents Architect (Finance) loop. Restate, clarify, name the pattern, code, edge-case. Reveal the answer only after you've tried.
How to approach each problem out loud
- Restate in your own words.
- Clarify input size, sorted, duplicates, currency, decimal precision.
- Example traced on paper.
- Brute force stated even if discarded.
- Pattern: hash table / two-pointer / prefix sum / sliding window / topo / sampling.
- Complexity for both brute and chosen.
- Code with narration.
- Edge cases with finance awareness (currency, decimal, dupes).
1 — Reconcile Two Ledgers
You are given two lists of transactions: gl (general ledger) and bank (bank statement). Each item has amount (Decimal), date (ISO), description (str), and id. Return three lists: matched pairs, GL-only, bank-only. Two transactions match if amount and date are equal.
Show solution
Pattern: hash table on composite key. Complexity: O(n + m) time, O(min(n,m)) space.
from collections import defaultdict
from decimal import Decimal
def reconcile(gl, bank):
# Bucket bank by (amount, date); allow duplicates by storing lists.
buckets = defaultdict(list)
for t in bank:
buckets[(t["amount"], t["date"])].append(t)
matched, gl_only = [], []
for g in gl:
key = (g["amount"], g["date"])
if buckets[key]:
b = buckets[key].pop() # one-to-one consume
matched.append((g, b))
else:
gl_only.append(g)
bank_only = [b for items in buckets.values() for b in items]
return matched, gl_only, bank_only
Edge cases: duplicates (use list, pop one-for-one); cross-currency (extend key with currency); date timezone (normalize to ISO date string before matching); Decimal precision (assert string-Decimal construction).
Senior moves: mention that real recon adds tolerance matching and fuzzy description matching on the residual — but state that's a follow-up after exact-match.
2 — Period-over-Period Variance with Rolling Windows
Given daily account balances [(account, date, balance)] spanning N days, for each account compute the 30-day rolling mean and flag any day where |balance - rolling_mean| > 3 * rolling_std.
Show solution
Pattern: sliding window with running mean/variance. Complexity: O(n) per account using Welford's online variance.
from collections import deque
from decimal import Decimal
import statistics
def flag_anomalies(rows, window=30, k=3):
by_acct = {}
for acct, date, bal in sorted(rows, key=lambda r: (r[0], r[1])):
by_acct.setdefault(acct, []).append((date, bal))
flags = []
for acct, series in by_acct.items():
win = deque()
for date, bal in series:
if len(win) >= window:
mu = statistics.fmean(float(x) for x in win)
sd = statistics.pstdev(float(x) for x in win) or 1e-9
if abs(float(bal) - mu) > k * sd:
flags.append((acct, date, bal))
win.append(bal)
if len(win) > window:
win.popleft()
return flags
Edge cases: too few days (skip until window full); zero variance (avoid div-by-zero); seasonality (3-sigma is naive; mention seasonal decomposition as follow-up).
Senior move: note that for production you'd use pandas rolling() but write the streaming version when asked for "no pandas" or for explainability.
3 — Anomaly Detection on a Transaction Stream
Transactions arrive one at a time as (timestamp, account, amount). After each event, return whether this transaction is anomalous: amount > 5x the median of this account's last 100 transactions.
Show solution
Pattern: bounded-buffer per account + median. For exact O(log n) median use two heaps; for interview, a sorted deque of recent values is fine and clearly correct.
from collections import defaultdict, deque
from sortedcontainers import SortedList
from decimal import Decimal
class StreamAnomalyDetector:
def __init__(self, window=100, multiplier=Decimal("5")):
self.window = window
self.k = multiplier
self.order = defaultdict(deque) # insertion order
self.sorted = defaultdict(SortedList) # sorted view
def ingest(self, account, amount) -> bool:
sl = self.sorted[account]
oq = self.order[account]
is_anomaly = False
if len(sl) >= 20: # only judge after some history
n = len(sl)
median = sl[n // 2]
if amount > self.k * median:
is_anomaly = True
sl.add(amount); oq.append(amount)
if len(oq) > self.window:
sl.remove(oq.popleft())
return is_anomaly
Edge cases: brand-new account with no history (don't flag); negative amounts (reversals); same-second duplicates (idempotency by event id, not amount).
Senior move: state that median is more robust than mean to whale transactions; suggest robust z-score (MAD) as a follow-up.
4 — Idempotency-Key Design
Design a function idempotency_key(operation) such that the same logical operation produces the same key across retries, and different logical operations produce different keys. Then design the server-side dedup table.
Show solution
Key design: deterministic hash of the operation's intent, not its incidental fields (timestamps, request ids).
import hashlib, json
from typing import Any
def idempotency_key(op_type: str, fields: dict[str, Any]) -> str:
# Canonical JSON: sorted keys, no whitespace, stringified Decimals.
canon = json.dumps({"op": op_type, "f": _normalize(fields)},
sort_keys=True, separators=(",", ":"))
return hashlib.sha256(canon.encode()).hexdigest()
def _normalize(d):
if isinstance(d, dict):
return {k: _normalize(v) for k, v in sorted(d.items())}
if isinstance(d, list):
return [_normalize(x) for x in d]
return str(d) if hasattr(d, "as_tuple") else d # Decimal → str
Server-side table:
CREATE TABLE idempotency (
key CHAR(64) PRIMARY KEY,
op_type TEXT NOT NULL,
request_hash CHAR(64) NOT NULL,
response_body JSONB,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
expires_at TIMESTAMPTZ NOT NULL
);
Flow: on each write, insert the key. If conflict, return the stored response. If request_hash differs from the new request, raise (different content, same key = client bug).
Senior moves: discuss TTL trade-offs (long retention for SOX but you don't want infinite); mention that the key should not depend on the model run id (so semantic retries dedupe); mention that the table must be in the same transactional boundary as the write to be effective.
5 — Audit-Log Query
Given an audit log table with millions of rows, write a SQL query for: "Every proposed journal entry by agent class 'bank_recon_v3' in April 2026, with approver, approval time, model version, and current state. Sort by amount descending."
Show solution
Pattern: filtered scan with indexed columns; CTE or window for the latest state per run.
WITH propose_events AS (
SELECT run_id,
(outputs->>'amount')::numeric AS amount,
model_id,
model_version_pin,
inputs_redacted->>'entity_id' AS entity,
approver_user_id,
approval_ts,
decision
FROM audit_log
WHERE agent_class = 'bank_recon_v3'
AND step_type = 'tool_call'
AND tool_name = 'propose_reconciling_item'
AND period = '2026-04'
),
latest AS (
SELECT run_id,
decision,
row_number() OVER (PARTITION BY run_id ORDER BY ts DESC) AS rn
FROM audit_log
WHERE agent_class = 'bank_recon_v3'
AND period = '2026-04'
)
SELECT p.run_id, p.entity, p.amount, p.model_id, p.model_version_pin,
p.approver_user_id, p.approval_ts, l.decision AS final_state
FROM propose_events p
JOIN latest l ON l.run_id = p.run_id AND l.rn = 1
ORDER BY p.amount DESC;
Senior moves: name the indexes you'd want ((agent_class, period), (run_id, ts)); mention partitioning by month for retention; remind that this is the kind of query SOX auditors will literally ask for.
6 — Running Balance from a Transaction Stream
Given a chronologically-sorted list of transactions (date, account, signed_amount) and a starting balance per account, produce the daily ending balance for each (account, date).
Show solution
Pattern: prefix sum per group. Complexity: O(n).
from collections import defaultdict
from decimal import Decimal
def daily_ending_balances(txns, opening):
balances = dict(opening) # account -> Decimal
last_date = {}
out = []
for date, account, amt in txns:
balances[account] = balances.get(account, Decimal(0)) + amt
last_date[account] = date
out.append((date, account, balances[account]))
return out
Edge cases: account that never appears in txns (carry opening); negative balance overdraft (allow, flag); same-date multiple txns (problem statement determines whether you emit intra-day or just end-of-day — clarify).
7 — Stratified Sampling for SOX Controls Testing
From a population of N journal entries, draw a sample for SOX testing: min(40, population) total, stratified so amounts > $100K are 100% sampled, $10K-$100K is 25%, smaller is the residual using uniform sampling. Reproducible with a seed.
Show solution
Pattern: stratify, then sample within strata with a seeded RNG.
import random
from decimal import Decimal
def stratified_sample(entries, seed: int, total=40,
high=Decimal("100000"), mid=Decimal("10000"),
mid_pct=0.25):
rng = random.Random(seed)
high_band = [e for e in entries if abs(e["amount"]) >= high]
mid_band = [e for e in entries if mid <= abs(e["amount"]) < high]
low_band = [e for e in entries if abs(e["amount"]) < mid]
sample = list(high_band) # 100% of high
mid_n = min(int(round(len(mid_band) * mid_pct)), max(0, total - len(sample)))
sample += rng.sample(mid_band, mid_n)
low_n = max(0, total - len(sample))
sample += rng.sample(low_band, min(low_n, len(low_band)))
return sample
Senior moves: the random seed is logged in the audit row so the sample can be reproduced — that's a real SOX requirement; mention you'd document the methodology in a stratification policy referenced by the control narrative.
8 — Intercompany Elimination Order (Topo Sort)
Consolidation requires eliminating intercompany pairs in dependency order. Given a graph of entities with intercompany positions and a parent-child structure, return an elimination order.
Show solution
Pattern: Kahn's algorithm for topological sort. Complexity: O(V+E).
from collections import defaultdict, deque
def elimination_order(entities, parent_of):
# parent_of: child -> parent
indeg = {e: 0 for e in entities}
children = defaultdict(list)
for child, parent in parent_of.items():
children[parent].append(child)
indeg[child] += 1
q = deque([e for e in entities if indeg[e] == 0])
order = []
while q:
node = q.popleft()
order.append(node)
for c in children[node]:
indeg[c] -= 1
if indeg[c] == 0:
q.append(c)
if len(order) != len(entities):
raise ValueError("Cycle in entity hierarchy")
# Eliminate from leaves up; reverse the topological order.
return list(reversed(order))
Senior moves: state that cycle = data quality issue, halt; mention real-world wrinkle that intercompany positions sometimes net before elimination (consolidation policy decision).
9 — FX Triangulation
Given direct FX rates as (from, to, rate) tuples, compute the rate from any currency A to any currency B, going through intermediate currencies if needed. Detect arbitrage cycles (product > 1.0 plus tolerance).
Show solution
Pattern: graph (BFS for shortest path / DFS for cycle detection with log-rates).
from collections import defaultdict, deque
from decimal import Decimal
def build_graph(rates):
g = defaultdict(dict)
for f, t, r in rates:
g[f][t] = Decimal(str(r))
g[t][f] = Decimal(1) / Decimal(str(r)) # inverse
return g
def convert(g, src, dst):
if src == dst: return Decimal(1)
seen = {src: Decimal(1)}
q = deque([src])
while q:
node = q.popleft()
for nxt, r in g[node].items():
if nxt in seen: continue
seen[nxt] = seen[node] * r
if nxt == dst:
return seen[nxt]
q.append(nxt)
raise ValueError(f"No path {src} → {dst}")
Senior moves: name that real treasury cares about bid/ask spread, not a single rate, so the model is a simplification; mention that arbitrage detection in production is a separate alerting concern.
10 — Merge Two Sorted Feeds
Given two iterables of transactions sorted by timestamp (one from the GL, one from the bank), produce a merged chronological stream. Memory must be O(1) in the input sizes.
Show solution
Pattern: classic two-pointer / heap merge. Complexity: O(n+m) time, O(1) extra.
def merge_sorted(a, b, key=lambda x: x["ts"]):
ia, ib = iter(a), iter(b)
sa = next(ia, None); sb = next(ib, None)
while sa is not None and sb is not None:
if key(sa) <= key(sb):
yield ("gl", sa); sa = next(ia, None)
else:
yield ("bank", sb); sb = next(ib, None)
while sa is not None:
yield ("gl", sa); sa = next(ia, None)
while sb is not None:
yield ("bank", sb); sb = next(ib, None)
Senior moves: extend trivially to N feeds with heapq.merge; mention this is the shape of every "reconcile two event streams" problem — the same algorithm reconciles bank webhook vs. NetSuite scheduled fetch.