Data Structures Jobs in Banking — Role & Responsibilities

# transaction_queue.py — a real pattern used in banking systems
# A FIFO queue with idempotency and retry support.
# This is the kind of structure you would be asked to design in a banking interview.

from collections import deque
from dataclasses import dataclass, field
from typing import Deque, Set
import time


@dataclass
class Transaction:
    txn_id: str
    account_id: str
    amount: float
    retries: int = field(default=0)


class TransactionQueue:
    """A FIFO queue for processing bank transactions with idempotency."""

    def __init__(self) -> None:
        self._queue: Deque[Transaction] = deque()
        self._seen: Set[str] = set()  # prevents duplicate processing
        self._dead_letter: Deque[Transaction] = deque()

    def enqueue(self, txn: Transaction) -> bool:
        if txn.txn_id in self._seen:
            return False  # already processed or queued
        self._seen.add(txn.txn_id)
        self._queue.append(txn)
        return True

    def process_next(self, handler) -> Transaction | None:
        if not self._queue:
            return None
        txn = self._queue.popleft()
        try:
            handler(txn)
            return txn
        except Exception:
            txn.retries += 1
            if txn.retries < 3:
                self._queue.append(txn)  # retry later
            else:
                self._dead_letter.append(txn)  # move to dead-letter queue
            return txn

    def pending(self) -> int:
        return len(self._queue)

    def dead_letters(self) -> int:
        return len(self._dead_letter)


# Why banks care about this structure:
# - FIFO ordering matters for transaction consistency.
# - Idempotency prevents duplicate payments — a common real-world failure.
# - Retries with a dead-letter queue are standard in payment systems.
# - The same pattern appears in interviews and in production code.
#
# What you learn by building it:
# - How to design a queue with constraints (ordering + idempotency).
# - How to reason about failure and retry.
# - How to write code that a reviewer can audit.

# fraud_graph.py — detecting suspicious transaction networks
# Banks model transactions as a graph: accounts are nodes, transactions are edges.
# This is a simplified version of a real fraud detection approach.

from collections import defaultdict, deque
from typing import Dict, List, Set


class TransactionGraph:
    def __init__(self) -> None:
        self._edges: Dict[str, List[tuple[str, float]]] = defaultdict(list)

    def add_transaction(self, sender: str, receiver: str, amount: float) -> None:
        self._edges[sender].append((receiver, amount))

    def find_cycles(self, start: str, max_depth: int = 3) -> List[List[str]]:
        """Find money-laundering style cycles up to max_depth."""
        results: List[List[str]] = []
        visited: Set[str] = set()

        def dfs(node: str, path: List[str]) -> None:
            if len(path) > max_depth:
                return
            if node in visited and len(path) > 1:
                if path[0] == node:
                    results.append(path[:])
                return
            visited.add(node)
            for neighbor, _ in self._edges.get(node, []):
                dfs(neighbor, path + [neighbor])
            visited.remove(node)

        dfs(start, [start])
        return results

    def high_value_cluster(self, threshold: float = 100000.0) -> List[str]:
        """Find accounts that receive many high-value transfers."""
        counts: Dict[str, int] = defaultdict(int)
        for sender, neighbors in self._edges.items():
            for receiver, amount in neighbors:
                if amount >= threshold:
                    counts[receiver] += 1
        return [acct for acct, c in counts.items() if c >= 3]


# Why this matters in banking:
# - Fraud is relational — it shows up in graphs, not single rows.
# - Cycles and high-value clusters are two classic detection patterns.
# - Interviews for fraud/ML engineering roles often ask graph problems.
#
# What you learn:
# - How to model transactions as a graph.
# - How to run DFS with depth limits safely.
# - How to translate a business rule into a graph query.

# customer_lookup.py — a hash-map-based customer lookup system
# Banks need O(1) lookup by account number, customer ID, and mobile number.
# This pattern appears in every core banking system.

from dataclasses import dataclass
from typing import Dict, Optional


@dataclass
class Customer:
    customer_id: str
    name: str
    account_number: str
    mobile: str
    balance: float


class CustomerIndex:
    def __init__(self) -> None:
        self._by_id: Dict[str, Customer] = {}
        self._by_account: Dict[str, Customer] = {}
        self._by_mobile: Dict[str, Customer] = {}

    def add(self, customer: Customer) -> None:
        if customer.account_number in self._by_account:
            raise ValueError("Duplicate account number")
        self._by_id[customer.customer_id] = customer
        self._by_account[customer.account_number] = customer
        self._by_mobile[customer.mobile] = customer

    def by_account(self, account_number: str) -> Optional[Customer]:
        return self._by_account.get(account_number)

    def by_mobile(self, mobile: str) -> Optional[Customer]:
        return self._by_mobile.get(mobile)

    def by_id(self, customer_id: str) -> Optional[Customer]:
        return self._by_id.get(customer_id)


# Why banks care:
# - Customer lookup must be fast and consistent.
# - Duplicate accounts must be rejected at write time.
# - Different access patterns (account vs mobile) need different indexes.
#
# What you learn:
# - How hash maps support O(1) lookup.
# - Why indexes on multiple keys are a standard design.
# - How to enforce correctness at the data structure level.