Coding Problems — Finance Flavor

How to approach each problem out loud

1. Restate in your own words.
2. Clarify input size, sorted, duplicates, currency, decimal precision.
3. Example traced on paper.
4. Brute force stated even if discarded.
5. Pattern: hash table / two-pointer / prefix sum / sliding window / topo / sampling.
6. Complexity for both brute and chosen.
7. Code with narration.
8. Edge cases with finance awareness (currency, decimal, dupes).

1 — Reconcile Two Ledgers Mark practiced

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.

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

2 — Period-over-Period Variance with Rolling Windows Mark practiced

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.

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

3 — Anomaly Detection on a Transaction Stream Mark practiced

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.

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

4 — Idempotency-Key Design Mark practiced

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.

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

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
);
      

5 — Audit-Log Query Mark practiced

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."

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;

6 — Running Balance from a Transaction Stream Mark practiced

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).

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

7 — Stratified Sampling for SOX Controls Testing Mark practiced

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.

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

8 — Intercompany Elimination Order (Topo Sort) Mark practiced

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.

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))

9 — FX Triangulation Mark practiced

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).

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}")

10 — Merge Two Sorted Feeds Mark practiced

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.

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)

✓ Check yourself

Answer from memory before opening each one.

1. In the ledger-matching solution, why does each bank bucket hold a list and consume one entry per match, rather than a set of keys?

   Show answer

   Two identical payments on one day must not both match a single bank line. With a set, both ledger lines would match the same bank item. With a list and pop, the second ledger line finds the bucket empty and lands in the ledger-only output, which is correct. See reconcile two ledgers.

2. The stream detector flags amounts above five times the median of the last 100 and does not judge until it has 20 values. Why median, and why the warm-up?

   Show answer

   A median barely moves when one huge transaction arrives, while a mean would be dragged up and hide the next outlier. With little history the median is itself noisy, so early flags would be false alarms. See anomaly detection.

3. What goes into an idempotency key, what must stay out of it, and what should the server do when a key arrives twice?

   Show answer

   The key hashes the operation's intent in canonical form, with Decimals as strings. Timestamps, request identifiers and the model run id stay out, so a retried or re-run operation keeps its key. On a repeat with the same request hash the server returns the stored response, and on a different hash it raises. See idempotency-key design.

4. 55 entries exceed $100,000 and the sample total is 40. What does the stratified sampler return, and is that the right call?

   Show answer

   It returns all 55 and nothing from the lower bands, because the top band is sampled in full and the rest is capped at what is left. That follows the stated rule of testing every large entry, but the policy should say so, and the seed must be logged so the selection can be rerun. See stratified sampling.

5. Why does the elimination routine reverse the topological order, and what does it do if the hierarchy contains a cycle?

   Show answer

   Kahn's algorithm yields parents before children, and eliminations proceed from the leaves upward, so the order is reversed. A cycle leaves some entities unprocessed, and the routine raises an error. A cycle in an entity hierarchy is a data-quality fault, so halt. See elimination order.

6. You reuse the sorted-feed merge to combine a bank webhook stream and a scheduled ledger fetch. What does it silently assume, and how do you protect against a violation?

   Show answer

   It assumes both inputs are already sorted by timestamp. If one arrives out of order, the output is wrong but no error appears. Check that timestamps never decrease as you read each feed, and halt or re-sort when they do. See merging sorted feeds.

Takeaway

• State the restatement, clarifications, brute force, pattern, complexity and edge cases aloud on every problem, in that order.
• Matching problems use a hash on a composite key, and duplicates need lists or counts so each line consumes exactly one counterpart.
• A stream detector uses a bounded buffer per account, a robust statistic such as the median, and a warm-up period before it judges.
• An idempotency key hashes the operation's intent and nothing incidental, and the server stores the response and compares request hashes.
• Reproducible sampling needs a seed that is logged with the run, and a sampling policy that the control narrative cites.
• Dependency and conversion problems are graph problems: topological sort for ordering, with a cycle treated as a data fault, and breadth-first search for paths.
• Merging sorted feeds is only correct if both feeds are in fact sorted, so validate order and halt on a violation.

The code runs on data that has to arrive reliably, so the next chapter covers pipelines and the data model under the agents. → Data & Pipelines

Further reading

• random: Generate pseudo-random numbers — Python documentation. Seeding and sampling behaviour.
• collections: Container datatypes — Python documentation. defaultdict, deque and Counter used in the solutions.
• graphlib: Topological sorting — Python documentation. A standard-library topological sorter that raises on cycles.
• heapq: Heap queue algorithm — Python documentation. heapq.merge for merging many sorted feeds.