SQL & SQLAlchemy

Relational databases, SQL fundamentals, and mapping Python classes to database tables

Every production financial system persists data — trades, positions, accounts, audit trails, and risk metrics — in a relational database. Whether it is PostgreSQL at a hedge fund, SQLite in an algorithmic trading bot, or Oracle at a tier-one bank, the structured query language is the same: SQL.

This chapter teaches you to design normalised schemas for financial data, write efficient queries, and use transactions to guarantee that a fund transfer either completes fully or is entirely reversed — never partially applied. You will learn the SQL that actually runs on trading desks: GROUP BY for position aggregation, JOIN for linking instruments to trades, and INDEX for sub-millisecond lookups across millions of rows.

Then SQLAlchemy — the standard Python ORM — connects your Python classes directly to PostgreSQL tables. You will define models, run queries, and manage transactions with the same patterns used in fintech APIs, risk systems, and settlement engines worldwide. Every example in this chapter uses the banking, trading, and portfolio management context you have built throughout this course.

Relational Databases & SQL Fundamentals

A relational database organises data into tables (rows and columns), with explicit relationships between them. Every major financial firm uses one: PostgreSQL at hedge funds, Oracle at investment banks, SQL Server at prime brokers, SQLite in trading bots and mobile apps.

Why relational databases in finance?

ACID guarantees — trades either fully settle or fully reverse; no partial states
Referential integrity — a trade cannot reference a non-existent instrument; the database enforces this
SQL — a universal, declarative language for querying millions of rows in milliseconds
Audit trail — every INSERT, UPDATE, and DELETE is a structured, queryable event

Core SQL data types for financial data:

INTEGER — quantity, trade ID, account number
REAL / FLOAT — prices, rates (use with care — binary rounding; prefer NUMERIC/DECIMAL in PostgreSQL for exact money)
NUMERIC(15,4) — monetary amounts, NAV, notional — exact decimal, no rounding errors
TEXT / VARCHAR — symbol, ISIN, BIC, name
TIMESTAMP / TEXT — trade date, settlement date, timestamps

Constraints — database-enforced rules:

PRIMARY KEY — unique identifier per row; implicitly NOT NULL; automatically indexed
FOREIGN KEY — references a primary key in another table; prevents orphaned records (e.g. a trade with no instrument)
NOT NULL — column must always have a value; use for required fields like symbol and price
UNIQUE — no two rows can share the same value; e.g. ISIN, account number
CHECK — arbitrary condition: CHECK (qty > 0), CHECK (side IN ('BUY','SELL'))

Core DML (Data Manipulation Language):

INSERT INTO table (cols) VALUES (...) — add a row
SELECT cols FROM table WHERE ... ORDER BY ... LIMIT n — query rows
UPDATE table SET col = val WHERE ... — modify rows; always include a WHERE clause
DELETE FROM table WHERE ... — remove rows; always include a WHERE clause

Python

import sqlite3 # ── Connect (PostgreSQL in production: psycopg2 / SQLAlchemy) ─────────────────conn = sqlite3.connect(":memory:")cur  = conn.cursor() # ── Create trades table ───────────────────────────────────────────────────────cur.execute("""    CREATE TABLE trades (        id         INTEGER PRIMARY KEY AUTOINCREMENT,        symbol     TEXT    NOT NULL,        side       TEXT    NOT NULL CHECK (side IN ('BUY', 'SELL')),        qty        INTEGER NOT NULL CHECK (qty > 0),        price      REAL    NOT NULL CHECK (price > 0),        desk       TEXT    NOT NULL DEFAULT 'Equities',        trade_date TEXT    NOT NULL,        settled    INTEGER NOT NULL DEFAULT 0    )""") # ── INSERT — parameterised queries (? placeholders prevent SQL injection) ─────trades = [    ("AAPL", "BUY",  100, 172.50, "Equities",    "2024-01-10"),    ("MSFT", "BUY",   50, 380.00, "Equities",    "2024-01-11"),    ("TSLA", "BUY",   30, 225.00, "Derivatives", "2024-01-16"),    ("AAPL", "BUY",   50, 178.00, "Equities",    "2024-01-15"),    ("AAPL", "SELL",  80, 186.55, "Equities",    "2024-02-01"),    ("MSFT", "BUY",   25, 410.00, "Equities",    "2024-02-05"),    ("NVDA", "BUY",   20, 480.00, "Equities",    "2024-02-10"),    ("TSLA", "SELL",  15, 195.40, "Derivatives", "2024-02-12"),]cur.executemany(    "INSERT INTO trades (symbol, side, qty, price, desk, trade_date) VALUES (?,?,?,?,?,?)",    trades,)conn.commit() # ── SELECT — filter, sort, limit ──────────────────────────────────────────────print("=== All BUY trades sorted by notional ===")print(f"  {'Symbol':<6} {'Qty':>5} {'Price':>8} {'Notional':>12} {'Desk'}")print("  " + "-" * 48)cur.execute("""    SELECT symbol, qty, price, qty * price AS notional, desk    FROM   trades    WHERE  side = 'BUY'    ORDER  BY notional DESC""")for sym, qty, price, notional, desk in cur.fetchall():    print(f"  {sym:<6} {qty:>5} {price:>8.2f} {notional:>12,.2f}  {desk}") # ── UPDATE — mark Jan trades as settled ───────────────────────────────────────cur.execute("UPDATE trades SET settled = 1 WHERE trade_date < '2024-02-01'")print(f"\nSettled {cur.rowcount} rows") # ── SELECT with IS/boolean filter ─────────────────────────────────────────────cur.execute("SELECT COUNT(*) FROM trades WHERE settled = 0")print(f"Pending settlement: {cur.fetchone()[0]} trades") conn.close()

Filtering, Sorting & Aggregations

Filtering with WHERE is how you extract meaningful subsets from millions of rows. In fintech, WHERE clauses appear in every report — filtering trades by date range, accounts by balance threshold, positions by desk or currency.

Filtering operators:

AND, OR, NOT — combine conditions
BETWEEN low AND high — inclusive range; use for date ranges and price bands
IN ('BUY', 'SELL') — match against a fixed set of values
LIKE 'GB%' — pattern match; % is wildcard (slow on large tables without full-text index)
IS NULL / IS NOT NULL — test for absence of value
COALESCE(col, default) — return first non-NULL value; essential when LEFT JOINs produce NULLs

Aggregate functions — computing portfolio-level metrics:

COUNT(*) — total rows; COUNT(col) — non-NULL rows only
SUM(qty * price) — total notional, total P&L
AVG(price) — average execution price
MIN / MAX — best/worst execution, date range

GROUP BY and HAVING:

GROUP BY symbol — collapse rows with the same symbol into one output row
HAVING SUM(qty) > 100 — filter after aggregation (WHERE filters before)
Always include every non-aggregate SELECT column in GROUP BY

CASE WHEN — conditional logic in SQL: essential for computing net positions (BUY adds, SELL subtracts) and categorising risk buckets directly in the query.

Python

import sqlite3 conn = sqlite3.connect(":memory:")cur  = conn.cursor() cur.executescript("""    CREATE TABLE trades (        id INTEGER PRIMARY KEY AUTOINCREMENT,        symbol TEXT NOT NULL, side TEXT NOT NULL,        qty INTEGER NOT NULL, price REAL NOT NULL,        desk TEXT NOT NULL, trade_date TEXT NOT NULL    );    INSERT INTO trades (symbol,side,qty,price,desk,trade_date) VALUES        ('AAPL','BUY', 100,172.50,'Equities',   '2024-01-10'),        ('AAPL','BUY',  50,178.00,'Equities',   '2024-01-15'),        ('AAPL','SELL', 80,186.55,'Equities',   '2024-02-01'),        ('MSFT','BUY',  50,380.00,'Equities',   '2024-01-11'),        ('MSFT','BUY',  25,410.00,'Equities',   '2024-02-05'),        ('TSLA','BUY',  30,225.00,'Derivatives','2024-01-16'),        ('TSLA','SELL', 15,195.40,'Derivatives','2024-02-12'),        ('NVDA','BUY',  20,480.00,'Equities',   '2024-02-10'),        ('BP',  'BUY',  40,467.00,'Equities',   '2024-02-14');""") # ── Notional by desk (GROUP BY + SUM) ─────────────────────────────────────────print("=== Notional traded by desk ===")cur.execute("""    SELECT desk, COUNT(*) AS trades, ROUND(SUM(qty * price), 2) AS notional    FROM   trades    GROUP  BY desk    ORDER  BY notional DESC""")for desk, count, notional in cur.fetchall():    print(f"  {desk:<14} {count} trades   notional = {notional:>12,.2f}") # ── Net position per symbol (CASE WHEN: BUY adds, SELL subtracts) ─────────────print("\n=== Net positions (BUY qty − SELL qty) ===")cur.execute("""    SELECT        symbol,        SUM(CASE WHEN side='BUY'  THEN qty ELSE  0   END) AS bought,        SUM(CASE WHEN side='SELL' THEN qty ELSE  0   END) AS sold,        SUM(CASE WHEN side='BUY'  THEN qty ELSE -qty END) AS net_qty    FROM  trades    GROUP BY symbol    ORDER BY net_qty DESC""")print(f"  {'Symbol':<6} {'Bought':>8} {'Sold':>6} {'Net':>6}")print("  " + "-" * 30)for sym, bought, sold, net in cur.fetchall():    print(f"  {sym:<6} {bought:>8} {sold:>6} {net:>6}") # ── HAVING: only symbols with net position > 50 ───────────────────────────────print("\n=== Symbols with net position > 50 (HAVING) ===")cur.execute("""    SELECT symbol, SUM(CASE WHEN side='BUY' THEN qty ELSE -qty END) AS net    FROM   trades    GROUP  BY symbol    HAVING net > 50    ORDER  BY net DESC""")for sym, net in cur.fetchall():    print(f"  {sym:<6}  net = {net}") conn.close()

Joins — Linking Tables

Normalisation eliminates data duplication by splitting information across related tables. In a trading system, instrument metadata (name, sector, currency, ISIN) is stored once in an instruments table and referenced by thousands of trades via its symbol — rather than repeating the full name and ISIN on every row.

JOIN types:

INNER JOIN — returns only rows where the join condition matches in both tables. A trade joined to instruments only appears if the instrument exists. Use for: position reports, trade confirmations where you only care about matched data.
LEFT JOIN (LEFT OUTER JOIN) — returns all rows from the left table, plus matched rows from the right. If no match, right-side columns are NULL. Use for: listing all instruments including those with no current position, or all clients including those with no recent trades.
FULL OUTER JOIN — all rows from both tables; NULLs where no match. Less common; use for reconciliation reports.

JOIN syntax:

FROM trades t INNER JOIN instruments i ON t.symbol = i.symbol
Table aliases (t, i) keep queries concise — required when the same column name exists in both tables
COALESCE(p.qty, 0) — convert NULL from a LEFT JOIN into a meaningful default (e.g. zero position)

Multi-table joins are the backbone of risk reports: join trades to instruments to accounts to get a complete view of a client's exposure across asset classes and desks.

Python

import sqlite3 conn = sqlite3.connect(":memory:")cur  = conn.cursor() cur.executescript("""    CREATE TABLE instruments (        symbol   TEXT PRIMARY KEY,        name     TEXT NOT NULL,        sector   TEXT NOT NULL,        currency TEXT NOT NULL DEFAULT 'USD'    );    CREATE TABLE positions (        id         INTEGER PRIMARY KEY AUTOINCREMENT,        symbol     TEXT NOT NULL REFERENCES instruments(symbol),        qty        INTEGER NOT NULL,        avg_cost   REAL NOT NULL,        last_price REAL NOT NULL    );    INSERT INTO instruments VALUES        ('AAPL','Apple Inc',       'Technology', 'USD'),        ('MSFT','Microsoft Corp',  'Technology', 'USD'),        ('JPM', 'JPMorgan Chase',  'Financials', 'USD'),        ('SHEL','Shell plc',       'Energy',     'GBP'),        ('BP',  'BP plc',          'Energy',     'GBP');    INSERT INTO positions (symbol, qty, avg_cost, last_price) VALUES        ('AAPL', 150, 174.33, 186.55),        ('MSFT',  75, 386.67, 415.30),        ('JPM',   60, 190.00, 198.20);    -- SHEL and BP have no positions: will show NULL in LEFT JOIN""") # ── INNER JOIN: only instruments with an open position ────────────────────────print("=== Open Positions — INNER JOIN ===")cur.execute("""    SELECT i.symbol, i.name, i.sector, i.currency,           p.qty, p.avg_cost, p.last_price,           ROUND((p.last_price - p.avg_cost) * p.qty, 2) AS pnl    FROM   positions p    INNER  JOIN instruments i ON p.symbol = i.symbol    ORDER  BY pnl DESC""")print(f"  {'Sym':<5} {'Name':<18} {'Sector':<12} {'Qty':>4} {'Avg':>7} {'Last':>7} {'P&L':>10}")print("  " + "-"*62)for sym, name, sector, ccy, qty, avg, last, pnl in cur.fetchall():    sign = "+" if pnl >= 0 else ""    print(f"  {sym:<5} {name:<18} {sector:<12} {qty:>4} {avg:>7.2f} {last:>7.2f} {sign}{pnl:>9,.2f}") # ── LEFT JOIN: all instruments, NULLs for untraded ────────────────────────────print("\n=== Instrument Universe — LEFT JOIN (incl. untraded) ===")cur.execute("""    SELECT i.symbol, i.name, i.sector, i.currency,           COALESCE(p.qty, 0) AS qty, p.last_price    FROM   instruments i    LEFT   JOIN positions p ON i.symbol = p.symbol    ORDER  BY i.symbol""")for sym, name, sector, ccy, qty, last in cur.fetchall():    status = f"qty={qty:>3}  last={last:.2f}" if qty else "— no position"    print(f"  {sym:<5} {name:<18} {ccy}  {status}") conn.close()

Indexes, Transactions & ACID

Indexes are pre-sorted lookup structures the database maintains alongside your data. Without an index, every query scans every row — O(n). With a B-tree index on trade_date, the database jumps directly to matching rows — O(log n). On a table with 50 million trades, this is the difference between 30 seconds and 5 milliseconds.

When to create an index:

Columns frequently used in WHERE clauses: symbol, trade_date, account_id
Columns used in JOIN ON conditions: foreign keys
Columns used in ORDER BY when sorting large result sets
Avoid indexing columns that change constantly — index maintenance adds overhead to every INSERT/UPDATE

ACID — the four guarantees of a transaction:

Atomicity — all operations in a transaction succeed together or none do. A fund transfer (debit Account A, credit Account B) is one atomic unit — you will never see a debit without the corresponding credit.
Consistency — every transaction moves the database from one valid state to another. Constraints (CHECK, FOREIGN KEY, NOT NULL) are enforced; a trade with a negative quantity is rejected.
Isolation — concurrent transactions do not interfere. At READ COMMITTED (default in PostgreSQL), each query sees only committed data. At SERIALIZABLE, transactions execute as if sequential — required for settlement and NAV calculations.
Durability — once committed, data survives crashes. The database writes to a WAL (Write-Ahead Log) before confirming the commit.

Transaction commands: BEGIN / START TRANSACTION — opens a transaction; COMMIT — makes all changes permanent; ROLLBACK — undoes all changes back to the last commit. In Python's sqlite3, a connection is in autocommit mode by default unless you call BEGIN explicitly or use the connection as a context manager.

Python

import sqlite3 conn = sqlite3.connect(":memory:")cur  = conn.cursor() # ── Schema: accounts table + index on account name ────────────────────────────cur.executescript("""    CREATE TABLE accounts (        id      INTEGER PRIMARY KEY AUTOINCREMENT,        name    TEXT UNIQUE NOT NULL,        balance REAL NOT NULL DEFAULT 0.0 CHECK (balance >= 0)    );    CREATE TABLE transfers (        id         INTEGER PRIMARY KEY AUTOINCREMENT,        from_acct  TEXT NOT NULL,        to_acct    TEXT NOT NULL,        amount     REAL NOT NULL,        ts         TEXT NOT NULL DEFAULT (datetime('now'))    );    -- Index: fast lookup by account name in WHERE / JOIN clauses    CREATE INDEX idx_accounts_name ON accounts (name);     INSERT INTO accounts (name, balance) VALUES        ('Client-GoldmanSachs',  500000.00),        ('Client-BridgewaterLP', 250000.00),        ('Firm-ClearingAccount',       0.00);""") # ── ATOMIC FUND TRANSFER — commit on success, rollback on failure ──────────────def transfer(from_name: str, to_name: str, amount: float) -> bool:    """Debit from_name and credit to_name atomically.    Both succeed or neither does — ACID Atomicity in practice."""    try:        cur.execute("BEGIN")        # Step 1: read source balance        cur.execute("SELECT balance FROM accounts WHERE name = ?", (from_name,))        row = cur.fetchone()        if not row:            raise ValueError(f"Account '{from_name}' not found")        if row[0] < amount:            raise ValueError(f"Insufficient funds: {row[0]:,.2f} < {amount:,.2f}")        # Step 2: debit source        cur.execute("UPDATE accounts SET balance = balance - ? WHERE name = ?",                    (amount, from_name))        # Step 3: credit destination        cur.execute("UPDATE accounts SET balance = balance + ? WHERE name = ?",                    (amount, to_name))        # Step 4: log transfer        cur.execute("INSERT INTO transfers (from_acct, to_acct, amount) VALUES (?,?,?)",                    (from_name, to_name, amount))        conn.commit()        print(f"  OK  {from_name} -> {to_name}  amount={amount:>10,.2f}")        return True    except Exception as e:        conn.rollback()        print(f"  FAIL ({e}) — rolled back")        return False print("=== Fund transfers ===")transfer("Client-GoldmanSachs",  "Firm-ClearingAccount", 150_000)transfer("Client-BridgewaterLP", "Firm-ClearingAccount",  75_000)transfer("Client-GoldmanSachs",  "Firm-ClearingAccount", 400_000)  # insufficient print("\n=== Final balances ===")for row in cur.execute("SELECT name, balance FROM accounts ORDER BY name"):    print(f"  {row[0]:<28} {row[1]:>12,.2f}") print("\n=== Transfer log ===")for row in cur.execute("SELECT from_acct, to_acct, amount FROM transfers"):    print(f"  {row[0]} -> {row[1]}  {row[2]:,.2f}") conn.close()

SQLAlchemy — Engine, Core & ORM

SQLAlchemy is Python's standard database toolkit. It has two layers you use at different levels of abstraction:

SQLAlchemy Core — a thin layer over raw SQL. You write SQL explicitly but use Python objects for connection management, parameterised queries, and connection pooling. The main benefit: safe parameterised queries with any SQL dialect.
SQLAlchemy ORM — maps Python classes to database tables. A Trade class becomes a trades table; instances become rows. You write Python instead of SQL for most operations, but can drop to SQL when needed.

Engine and connection URL:

create_engine("postgresql+psycopg2://user:pass@host:5432/db") — creates the engine (does not connect yet)
The engine manages a connection pool — reuses existing connections rather than creating a new TCP connection for every query. In a trading system handling 10,000 queries/second, connection overhead would be catastrophic without pooling.
pool_size=10 — persistent connections; max_overflow=20 — burst capacity

Declarative ORM — mapping classes to tables:

class Trade(Base): __tablename__ = "trades" — the class is the table definition
Column(Integer, primary_key=True), Column(String(10), nullable=False), Column(Numeric(15,4)) — column definitions with types and constraints
ForeignKey("instruments.symbol") — enforces referential integrity at the Python layer and DB layer simultaneously
relationship("Instrument", back_populates="trades") — lets you navigate: trade.instrument.name instead of a JOIN query

Session — the unit-of-work pattern:

session.add(obj) — stages an INSERT; not yet sent to DB
session.flush() — sends SQL to DB but does not commit; changes are visible only within this transaction
session.commit() — makes all changes permanent; visible to all connections
session.rollback() — discards all unflushed/uncommitted changes
session.query(Trade).filter(Trade.symbol == "AAPL").all() — SELECT with ORM filter

SQL injection — always use parameterised queries: Never build queries with f-strings or string concatenation when the value comes from user input. SQLAlchemy's ORM and text("... :param") with bound parameters are both safe. Direct string interpolation (f"WHERE symbol = '{user_input}'") is dangerous — a value like '; DROP TABLE trades; -- would execute arbitrary SQL.

Python

# SQLAlchemy ORM — production Python for PostgreSQL# Requires: pip install sqlalchemy psycopg2-binary# This code is a reference example — it connects to a real PostgreSQL database.# The browser REPL uses sqlite3 directly (see REPL tab). from sqlalchemy import (    create_engine, Column, Integer, String, Float, DateTime,    ForeignKey, Numeric, Index, text,)from sqlalchemy.orm import DeclarativeBase, Session, relationshipfrom datetime import datetime # ── 1. Engine — one per application, manages the connection pool ───────────────engine = create_engine(    "postgresql+psycopg2://trader:password@localhost:5432/tradedb",    pool_size=10,       # keep 10 connections open    max_overflow=20,    # allow 20 burst connections under load    echo=False,         # set True to log all generated SQL (debug mode)) # ── 2. Declarative base — all ORM models inherit from Base ────────────────────class Base(DeclarativeBase):    pass class Instrument(Base):    __tablename__ = "instruments"    symbol   = Column(String(10),  primary_key=True)    name     = Column(String(100), nullable=False)    sector   = Column(String(50))    currency = Column(String(3),   nullable=False, default="USD")    trades   = relationship("Trade", back_populates="instrument") class Trade(Base):    __tablename__ = "trades"    __table_args__ = (        Index("idx_trades_symbol", "symbol"),      # fast lookup by symbol        Index("idx_trades_date",   "trade_date"),  # fast lookup by date range    )    id         = Column(Integer,        primary_key=True, autoincrement=True)    symbol     = Column(String(10),     ForeignKey("instruments.symbol"), nullable=False)    side       = Column(String(4),      nullable=False)    # BUY | SELL    qty        = Column(Integer,        nullable=False)    price      = Column(Numeric(15, 4), nullable=False)    # exact decimal    desk       = Column(String(50),     default="Equities")    trade_date = Column(DateTime,       default=datetime.utcnow)    instrument = relationship("Instrument", back_populates="trades") # ── 3. Create all tables (idempotent: skips if already exist) ─────────────────Base.metadata.create_all(engine) # ── 4. Session — unit-of-work; always use as context manager ──────────────────with Session(engine) as session:    # INSERT: add instruments and trades    aapl = Instrument(symbol="AAPL", name="Apple Inc", sector="Technology")    msft = Instrument(symbol="MSFT", name="Microsoft", sector="Technology")    t1   = Trade(symbol="AAPL", side="BUY",  qty=100, price=172.50)    t2   = Trade(symbol="MSFT", side="BUY",  qty=50,  price=380.00)    t3   = Trade(symbol="AAPL", side="SELL", qty=40,  price=186.55)    session.add_all([aapl, msft, t1, t2, t3])    session.commit()     # SELECT with ORM filter — safe; SQLAlchemy parameterises automatically    buy_trades = (        session.query(Trade)        .filter(Trade.side == "BUY")        .order_by(Trade.price.desc())        .all()    )    for t in buy_trades:        print(f"  {t.symbol}  {t.side}  qty={t.qty}  price={float(t.price):.2f}")     # Raw SQL with bound parameters — safe (never use f-strings with user input)    result = session.execute(        text("SELECT symbol, SUM(qty) AS net FROM trades WHERE side = :side GROUP BY symbol"),        {"side": "BUY"},    )    print("\nNet BUY qty:", {row.symbol: row.net for row in result})     # UPDATE via ORM    t1.price = 173.00    session.commit()     # DELETE    session.delete(t3)    session.commit()

Integration — Trade Blotter & Position Engine

This section assembles everything from the chapter into a production-style system: a Trade Blotter backed by SQLite (swap to PostgreSQL with a one-line change), a Position Engine that calculates net positions via SQL aggregation, and a P&L report that prices those positions with current market data.

Design patterns used:

Repository pattern — all database logic is encapsulated in TradeBlotter. Business logic never touches SQL directly; it calls methods. This makes the data layer swappable (SQLite → PostgreSQL → MongoDB) without changing business code.
sqlite3.Row as row_factory — rows behave like dicts: row["symbol"] instead of row[0]. Prevents bugs when columns are reordered.
NULLIF(denominator, 0) — prevents division by zero in average cost calculation (if a symbol has zero BUY qty, the denominator is NULL and the result is NULL, not a crash).
Indexes on symbol and trade_date — queries that aggregate by symbol or filter by date range use the index; no full scans.

Average cost calculation: SUM(qty * price) / SUM(qty) over all BUY trades — the weighted average cost method used in trade accounting and P&L attribution. SELL trades are not included in the average cost (they reduce the position but don't change the cost basis of remaining shares under the average cost method).

Extending this pattern: add settlement_date, currency, FX rate lookups, audit triggers, and you have the core of a real prime brokerage trade capture system. The same repository pattern scales from SQLite to Postgres to a distributed database — only the connection string changes.

Python

import sqlite3 # ── Schema ─────────────────────────────────────────────────────────────────────SCHEMA = """CREATE TABLE IF NOT EXISTS instruments (    symbol   TEXT PRIMARY KEY,    name     TEXT NOT NULL,    sector   TEXT NOT NULL DEFAULT 'Equity',    currency TEXT NOT NULL DEFAULT 'USD');CREATE TABLE IF NOT EXISTS trades (    id         INTEGER PRIMARY KEY AUTOINCREMENT,    symbol     TEXT    NOT NULL REFERENCES instruments(symbol),    side       TEXT    NOT NULL CHECK (side IN ('BUY','SELL')),    qty        INTEGER NOT NULL CHECK (qty > 0),    price      REAL    NOT NULL CHECK (price > 0),    desk       TEXT    NOT NULL DEFAULT 'Equities',    trade_date TEXT    NOT NULL);CREATE INDEX IF NOT EXISTS idx_trades_symbol ON trades (symbol);CREATE INDEX IF NOT EXISTS idx_trades_date   ON trades (trade_date);""" # ── Repository ─────────────────────────────────────────────────────────────────class TradeBlotter:    def __init__(self, db_path: str = ":memory:"):        self.conn = sqlite3.connect(db_path)        self.conn.row_factory = sqlite3.Row   # access columns by name        self.conn.executescript(SCHEMA)     def add_instrument(self, symbol: str, name: str, sector: str = "Equity", currency: str = "USD"):        self.conn.execute(            "INSERT OR IGNORE INTO instruments (symbol, name, sector, currency) VALUES (?,?,?,?)",            (symbol, name, sector, currency),        )        self.conn.commit()     def record_trade(self, symbol: str, side: str, qty: int, price: float,                     trade_date: str, desk: str = "Equities"):        self.conn.execute(            "INSERT INTO trades (symbol,side,qty,price,desk,trade_date) VALUES (?,?,?,?,?,?)",            (symbol, side, qty, price, desk, trade_date),        )        self.conn.commit()     def positions(self):        """Net qty and weighted average cost per symbol (BUY − SELL)."""        return self.conn.execute("""            SELECT                symbol,                SUM(CASE WHEN side='BUY'  THEN qty ELSE -qty END) AS net_qty,                ROUND(                    SUM(CASE WHEN side='BUY' THEN qty * price ELSE 0 END)                    / NULLIF(SUM(CASE WHEN side='BUY' THEN qty ELSE 0 END), 0),                4) AS avg_cost            FROM  trades            GROUP BY symbol            HAVING net_qty > 0            ORDER  BY net_qty DESC        """).fetchall()     def pnl_report(self, current_prices: dict):        rows = self.positions()        print(f"  {'Symbol':<6} {'Net Qty':>8} {'Avg Cost':>9} {'Last':>9} {'P&L':>12} {'Return':>8}")        print("  " + "-" * 58)        total_pnl = 0.0        for r in rows:            last = current_prices.get(r["symbol"], r["avg_cost"])            pnl  = round((last - r["avg_cost"]) * r["net_qty"], 2)            ret  = round((last / r["avg_cost"] - 1) * 100, 2)            total_pnl += pnl            sign = "+" if pnl >= 0 else ""            ret_sign = "+" if ret >= 0 else ""            print(f"  {r['symbol']:<6} {r['net_qty']:>8} {r['avg_cost']:>9.2f} "                  f"{last:>9.2f} {sign}{pnl:>11,.2f} {ret_sign}{ret:>7.2f}%")        sign = "+" if total_pnl >= 0 else ""        print("  " + "-" * 58)        print(f"  {'Total P&L':>36}  {sign}{total_pnl:>11,.2f}") # ── Demo — populate blotter and generate P&L report ───────────────────────────blotter = TradeBlotter() for sym, name, sector in [    ("AAPL", "Apple Inc",       "Technology"),    ("MSFT", "Microsoft Corp",  "Technology"),    ("NVDA", "NVIDIA Corp",     "Technology"),    ("TSLA", "Tesla Inc",       "Consumer"),    ("JPM",  "JPMorgan Chase",  "Financials"),]:    blotter.add_instrument(sym, name, sector) for sym, side, qty, price, date in [    ("AAPL","BUY",  100,172.50,"2024-01-10"), ("AAPL","BUY",   50,178.00,"2024-01-15"),    ("MSFT","BUY",   50,380.00,"2024-01-11"), ("MSFT","BUY",   25,410.00,"2024-02-05"),    ("NVDA","BUY",   20,480.00,"2024-02-10"), ("TSLA","BUY",   30,225.00,"2024-01-16"),    ("JPM", "BUY",   40,190.00,"2024-01-20"), ("AAPL","SELL",  60,186.55,"2024-02-01"),    ("TSLA","SELL",  10,195.40,"2024-02-12"), ("JPM", "SELL",  10,198.20,"2024-02-20"),]:    blotter.record_trade(sym, side, qty, price, date) print("=== Trade Blotter — P&L Report ===")blotter.pnl_report({    "AAPL": 189.30,    "MSFT": 415.30,    "NVDA": 520.00,    "TSLA": 195.40,    "JPM":  201.50,})

Common Table Expressions (CTEs)

A Common Table Expression (CTE) using the WITH clause lets you create temporary result sets that exist just for one query. In finance, queries often require multiple steps: first calculating a daily metric, then aggregating it weekly, and finally comparing it to a benchmark.

Benefits of CTEs:

Readability: CTEs read top-to-bottom, unlike nested subqueries which read inside-out.
Reusability: You can define a CTE once and reference it multiple times in the main query.

For example, if you need to calculate the total P&L for accounts that have more than 5 trades, you can first use a CTE to find active accounts, then join that CTE to the trades table.

Python

# ── CTE Example ─────────────────────────────────────────────────────────────import sqlite3 conn = sqlite3.connect(":memory:")cur = conn.cursor() cur.executescript("""    CREATE TABLE trades (id INTEGER, account TEXT, amount REAL);    INSERT INTO trades VALUES        (1, 'Acc-A', 150), (2, 'Acc-A', -50),        (3, 'Acc-B', 200), (4, 'Acc-C', 300);""") cur.execute("""    WITH account_totals AS (        SELECT account, SUM(amount) AS total        FROM trades        GROUP BY account    )    SELECT account, total    FROM account_totals    WHERE total >= 150    ORDER BY total DESC;""") print("Accounts with total >= 150:")for row in cur.fetchall():    print(f"  Account: {row[0]}, Total: {row[1]}")

Window Functions

Window Functions perform calculations across a set of rows that are related to the current row, without collapsing them into a single output row like GROUP BY does. They are essential for running totals, moving averages, and ranking.

Syntax components:

OVER() — defines the window
PARTITION BY — divides the window into groups (like GROUP BY, but keeps all rows)
ORDER BY — orders the rows within the partition (crucial for running totals)

Examples include ROW_NUMBER() to deduplicate records, SUM() OVER (ORDER BY date) for a cumulative P&L curve, and LAG() or LEAD() to calculate day-over-day price differences.

Python

# ── Window Function Example ──────────────────────────────────────────────────import sqlite3 conn = sqlite3.connect(":memory:")cur = conn.cursor() cur.executescript("""    CREATE TABLE prices (date TEXT, symbol TEXT, price REAL);    INSERT INTO prices VALUES        ('2024-01-01', 'AAPL', 150),        ('2024-01-02', 'AAPL', 152),        ('2024-01-03', 'AAPL', 149),        ('2024-01-01', 'MSFT', 300),        ('2024-01-02', 'MSFT', 305);""") # Calculate day-over-day difference using LAG()cur.execute("""    SELECT date, symbol, price,           price - LAG(price) OVER (PARTITION BY symbol ORDER BY date) AS daily_change    FROM prices    ORDER BY symbol, date;""") print(f"{'Date':<12} {'Sym':<5} {'Price':<6} {'Change'}")print("-" * 35)for row in cur.fetchall():    change = f"{row[3]:+.2f}" if row[3] is not None else "N/A"    print(f"{row[0]:<12} {row[1]:<5} {row[2]:<6.2f} {change}")

PythonRuns entirely in your browser — nothing is sent to a server.

[{"label":"1 · SQL Basics","code":"# ── SQL Basics — Trade Table with INSERT, SELECT, UPDATE ─────\nimport sqlite3\n\nconn = sqlite3.connect(\":memory:\")\ncur  = conn.cursor()\n\ncur.execute(\"\"\"\n    CREATE TABLE trades (\n        id         INTEGER PRIMARY KEY AUTOINCREMENT,\n        symbol     TEXT    NOT NULL,\n        side       TEXT    NOT NULL CHECK (side IN ('BUY','SELL')),\n        qty        INTEGER NOT NULL,\n        price      REAL    NOT NULL,\n        desk       TEXT    NOT NULL DEFAULT 'Equities',\n        trade_date TEXT    NOT NULL,\n        settled    INTEGER NOT NULL DEFAULT 0\n    )\n\"\"\")\n\ncur.executemany(\n    \"INSERT INTO trades (symbol,side,qty,price,desk,trade_date) VALUES (?,?,?,?,?,?)\",\n    [\n        (\"AAPL\",\"BUY\", 100,172.50,\"Equities\",   \"2024-01-10\"),\n        (\"MSFT\",\"BUY\",  50,380.00,\"Equities\",   \"2024-01-11\"),\n        (\"TSLA\",\"BUY\",  30,225.00,\"Derivatives\",\"2024-01-16\"),\n        (\"AAPL\",\"BUY\",  50,178.00,\"Equities\",   \"2024-01-15\"),\n        (\"AAPL\",\"SELL\", 80,186.55,\"Equities\",   \"2024-02-01\"),\n        (\"MSFT\",\"BUY\",  25,410.00,\"Equities\",   \"2024-02-05\"),\n        (\"NVDA\",\"BUY\",  20,480.00,\"Equities\",   \"2024-02-10\"),\n        (\"TSLA\",\"SELL\", 15,195.40,\"Derivatives\",\"2024-02-12\"),\n        (\"JPM\", \"BUY\",  40,190.00,\"Financials\", \"2024-02-14\"),\n    ]\n)\nconn.commit()\n\n# ── 1. SELECT all BUY trades, sorted by notional ──────────────\nprint(\"=== BUY trades by notional ===\")\nprint(f\"  {'#':<3} {'Symbol':<6} {'Qty':>5} {'Price':>8} {'Notional':>12}  Desk\")\nprint(\"  \" + \"-\"*52)\ncur.execute(\"\"\"\n    SELECT id, symbol, qty, price, qty*price AS notional, desk\n    FROM   trades\n    WHERE  side = 'BUY'\n    ORDER  BY notional DESC\n\"\"\")\nfor id_, sym, qty, price, notional, desk in cur.fetchall():\n    print(f\"  {id_:<3} {sym:<6} {qty:>5} {price:>8.2f} {notional:>12,.2f}  {desk}\")\n\n# ── 2. SELECT with BETWEEN for date range ─────────────────────\nprint(\"\\n=== Trades in January 2024 ===\")\ncur.execute(\"\"\"\n    SELECT symbol, side, qty, price, trade_date\n    FROM   trades\n    WHERE  trade_date BETWEEN '2024-01-01' AND '2024-01-31'\n    ORDER  BY trade_date, symbol\n\"\"\")\nfor sym, side, qty, price, dt in cur.fetchall():\n    print(f\"  {dt}  {sym:<5} {side}  qty={qty:>4}  price={price:.2f}\")\n\n# ── 3. UPDATE — mark Jan as settled ───────────────────────────\ncur.execute(\"UPDATE trades SET settled = 1 WHERE trade_date < '2024-02-01'\")\nconn.commit()\npending = cur.execute(\"SELECT COUNT(*) FROM trades WHERE settled=0\").fetchone()[0]\nsettled = cur.execute(\"SELECT COUNT(*) FROM trades WHERE settled=1\").fetchone()[0]\nprint(f\"\\nSettled: {settled}  |  Pending: {pending}\")\n\nconn.close()"},{"label":"2 · Aggregations","code":"# ── Aggregations — Position Engine with GROUP BY & CASE WHEN ──\nimport sqlite3\n\nconn = sqlite3.connect(\":memory:\")\ncur  = conn.cursor()\n\ncur.executescript(\"\"\"\n    CREATE TABLE trades (\n        id INTEGER PRIMARY KEY AUTOINCREMENT,\n        symbol TEXT NOT NULL, side TEXT NOT NULL,\n        qty INTEGER NOT NULL, price REAL NOT NULL,\n        desk TEXT NOT NULL, trade_date TEXT NOT NULL\n    );\n    INSERT INTO trades (symbol,side,qty,price,desk,trade_date) VALUES\n        ('AAPL','BUY', 100,172.50,'Equities',   '2024-01-10'),\n        ('AAPL','BUY',  50,178.00,'Equities',   '2024-01-15'),\n        ('AAPL','SELL', 80,186.55,'Equities',   '2024-02-01'),\n        ('MSFT','BUY',  50,380.00,'Equities',   '2024-01-11'),\n        ('MSFT','BUY',  25,410.00,'Equities',   '2024-02-05'),\n        ('TSLA','BUY',  30,225.00,'Derivatives','2024-01-16'),\n        ('TSLA','SELL', 15,195.40,'Derivatives','2024-02-12'),\n        ('NVDA','BUY',  20,480.00,'Equities',   '2024-02-10'),\n        ('JPM', 'BUY',  40,190.00,'Financials', '2024-02-14'),\n        ('BP',  'BUY',  60,467.00,'Equities',   '2024-02-18');\n\"\"\")\n\n# ── 1. Notional by desk (GROUP BY + SUM) ──────────────────────\nprint(\"=== Notional traded by desk ===\")\ncur.execute(\"\"\"\n    SELECT desk, COUNT(*) AS trades,\n           ROUND(SUM(qty * price), 2) AS notional\n    FROM   trades\n    GROUP  BY desk\n    ORDER  BY notional DESC\n\"\"\")\nfor desk, count, notional in cur.fetchall():\n    print(f\"  {desk:<14} {count} trades  notional = {notional:>12,.2f}\")\n\n# ── 2. Net position per symbol (CASE WHEN) ────────────────────\nprint(\"\\n=== Net positions (weighted avg cost) ===\")\ncur.execute(\"\"\"\n    SELECT\n        symbol,\n        SUM(CASE WHEN side='BUY'  THEN qty ELSE  0   END)  AS bought,\n        SUM(CASE WHEN side='SELL' THEN qty ELSE  0   END)  AS sold,\n        SUM(CASE WHEN side='BUY'  THEN qty ELSE -qty END)  AS net_qty,\n        ROUND(\n            SUM(CASE WHEN side='BUY' THEN qty*price ELSE 0 END)\n            / NULLIF(SUM(CASE WHEN side='BUY' THEN qty ELSE 0 END), 0),\n        2) AS avg_cost\n    FROM  trades\n    GROUP BY symbol\n    ORDER BY net_qty DESC\n\"\"\")\nprint(f\"  {'Symbol':<6} {'Bought':>7} {'Sold':>6} {'Net':>5}  {'Avg Cost':>9}\")\nprint(\"  \" + \"-\"*40)\nfor sym, bought, sold, net, avg in cur.fetchall():\n    print(f\"  {sym:<6} {bought:>7} {sold:>6} {net:>5}  {avg:>9.2f}\")\n\n# ── 3. HAVING: only large net positions ───────────────────────\nprint(\"\\n=== Net positions > 50 shares (HAVING) ===\")\ncur.execute(\"\"\"\n    SELECT symbol,\n           SUM(CASE WHEN side='BUY' THEN qty ELSE -qty END) AS net\n    FROM   trades\n    GROUP  BY symbol\n    HAVING net > 50\n    ORDER  BY net DESC\n\"\"\")\nfor sym, net in cur.fetchall():\n    print(f\"  {sym:<6}  net = {net}\")\n\nconn.close()"},{"label":"3 · Joins & Blotter","code":"# ── Multi-table Joins + Full Trade Blotter ────────────────────\nimport sqlite3\n\nconn = sqlite3.connect(\":memory:\")\nconn.row_factory = sqlite3.Row   # access columns by name\ncur  = conn.cursor()\n\ncur.executescript(\"\"\"\n    CREATE TABLE instruments (\n        symbol   TEXT PRIMARY KEY,\n        name     TEXT NOT NULL,\n        sector   TEXT NOT NULL,\n        currency TEXT NOT NULL DEFAULT 'USD'\n    );\n    CREATE TABLE trades (\n        id         INTEGER PRIMARY KEY AUTOINCREMENT,\n        symbol     TEXT    NOT NULL REFERENCES instruments(symbol),\n        side       TEXT    NOT NULL,\n        qty        INTEGER NOT NULL,\n        price      REAL    NOT NULL,\n        trade_date TEXT    NOT NULL\n    );\n    INSERT INTO instruments VALUES\n        ('AAPL','Apple Inc',       'Technology','USD'),\n        ('MSFT','Microsoft Corp',  'Technology','USD'),\n        ('NVDA','NVIDIA Corp',     'Technology','USD'),\n        ('TSLA','Tesla Inc',       'Consumer',  'USD'),\n        ('JPM', 'JPMorgan Chase',  'Financials','USD'),\n        ('SHEL','Shell plc',       'Energy',    'GBP');\n    INSERT INTO trades (symbol,side,qty,price,trade_date) VALUES\n        ('AAPL','BUY', 100,172.50,'2024-01-10'),\n        ('AAPL','BUY',  50,178.00,'2024-01-15'),\n        ('AAPL','SELL', 60,186.55,'2024-02-01'),\n        ('MSFT','BUY',  50,380.00,'2024-01-11'),\n        ('MSFT','BUY',  25,410.00,'2024-02-05'),\n        ('NVDA','BUY',  20,480.00,'2024-02-10'),\n        ('TSLA','BUY',  30,225.00,'2024-01-16'),\n        ('TSLA','SELL', 10,195.40,'2024-02-12'),\n        ('JPM', 'BUY',  40,190.00,'2024-02-14');\n    -- SHEL has no trades: will appear only in LEFT JOIN\n\"\"\")\n\n# ── 1. INNER JOIN: position report with instrument name ───────\nprint(\"=== Net positions — INNER JOIN (trades ⨝ instruments) ===\")\ncur.execute(\"\"\"\n    SELECT i.symbol, i.name, i.sector, i.currency,\n           SUM(CASE WHEN t.side='BUY' THEN t.qty ELSE -t.qty END) AS net_qty,\n           ROUND(\n               SUM(CASE WHEN t.side='BUY' THEN t.qty*t.price ELSE 0 END)\n               / NULLIF(SUM(CASE WHEN t.side='BUY' THEN t.qty ELSE 0 END),0),\n           2) AS avg_cost\n    FROM   trades t\n    INNER  JOIN instruments i ON t.symbol = i.symbol\n    GROUP  BY t.symbol\n    HAVING net_qty > 0\n    ORDER  BY net_qty DESC\n\"\"\")\nprint(f\"  {'Sym':<5} {'Name':<18} {'Sector':<12} {'Ccy':<4} {'Net Qty':>7} {'Avg Cost':>9}\")\nprint(\"  \" + \"-\"*60)\nfor r in cur.fetchall():\n    print(f\"  {r['symbol']:<5} {r['name']:<18} {r['sector']:<12} \"\n          f\"{r['currency']:<4} {r['net_qty']:>7} {r['avg_cost']:>9.2f}\")\n\n# ── 2. LEFT JOIN: full instrument universe incl. untraded ─────\nprint(\"\\n=== Instrument universe — LEFT JOIN (all incl. SHEL) ===\")\ncur.execute(\"\"\"\n    SELECT i.symbol, i.name, i.currency,\n           COALESCE(SUM(CASE WHEN t.side='BUY' THEN t.qty ELSE -t.qty END), 0) AS net\n    FROM   instruments i\n    LEFT   JOIN trades t ON i.symbol = t.symbol\n    GROUP  BY i.symbol\n    ORDER  BY i.symbol\n\"\"\")\nfor r in cur.fetchall():\n    status = f\"net={r['net']:>4}\" if r['net'] else \"no trades\"\n    print(f\"  {r['symbol']:<5} {r['name']:<20} {r['currency']}  {status}\")\n\n# ── 3. P&L with current prices ────────────────────────────────\nprices = {\"AAPL\":189.30,\"MSFT\":415.30,\"NVDA\":520.00,\"TSLA\":195.40,\"JPM\":201.50}\ncur.execute(\"\"\"\n    SELECT t.symbol,\n           SUM(CASE WHEN t.side='BUY' THEN t.qty ELSE -t.qty END) AS net_qty,\n           ROUND(SUM(CASE WHEN t.side='BUY' THEN t.qty*t.price ELSE 0 END)\n                 / NULLIF(SUM(CASE WHEN t.side='BUY' THEN t.qty ELSE 0 END),0),2) AS avg_cost\n    FROM trades t GROUP BY t.symbol HAVING net_qty > 0\n\"\"\")\npositions = cur.fetchall()\nprint(\"\\n=== P&L Report ===\")\nprint(f\"  {'Symbol':<6} {'Net Qty':>7} {'Avg Cost':>9} {'Last':>8} {'P&L':>11}\")\nprint(\"  \" + \"-\"*46)\ntotal = 0.0\nfor r in positions:\n    last = prices.get(r[\"symbol\"], r[\"avg_cost\"])\n    pnl  = round((last - r[\"avg_cost\"]) * r[\"net_qty\"], 2)\n    total += pnl\n    sign = \"+\" if pnl >= 0 else \"\"\n    print(f\"  {r['symbol']:<6} {r['net_qty']:>7} {r['avg_cost']:>9.2f} \"\n          f\"{last:>8.2f} {sign}{pnl:>10,.2f}\")\nsign = \"+\" if total >= 0 else \"\"\nprint(f\"  {'Total P&L':>34}  {sign}{total:>10,.2f}\")\n\nconn.close()"},{"label":"4 · Advanced SQL","code":"# ── Advanced SQL: CTEs and Window Functions ─────────────────\nimport sqlite3\n\nconn = sqlite3.connect(\":memory:\")\ncur = conn.cursor()\n\ncur.executescript(\"\"\"\n    CREATE TABLE trades (\n        id INTEGER PRIMARY KEY,\n        account TEXT,\n        trade_date TEXT,\n        pnl REAL\n    );\n    INSERT INTO trades (account, trade_date, pnl) VALUES\n        ('Alpha', '2024-03-01', 1500.00),\n        ('Alpha', '2024-03-02', -200.00),\n        ('Alpha', '2024-03-03', 800.00),\n        ('Beta',  '2024-03-01', -500.00),\n        ('Beta',  '2024-03-02', 1200.00);\n\"\"\")\n\n# 1. CTE: Find top performing accounts, then get their details\nprint(\"=== CTE: Trades for accounts with >1000 Total P&L ===\")\ncur.execute(\"\"\"\n    WITH top_accounts AS (\n        SELECT account, SUM(pnl) as total_pnl\n        FROM trades\n        GROUP BY account\n        HAVING total_pnl > 1000\n    )\n    SELECT t.account, t.trade_date, t.pnl\n    FROM trades t\n    INNER JOIN top_accounts ta ON t.account = ta.account\n    ORDER BY t.account, t.trade_date;\n\"\"\")\nfor row in cur.fetchall():\n    print(f\"  {row[0]:<8} {row[1]:<12} {row[2]:>8.2f}\")\n\n# 2. Window Function: Running Total (Cumulative P&L)\nprint(\"\\n=== Window Function: Cumulative P&L by Account ===\")\ncur.execute(\"\"\"\n    SELECT account, trade_date, pnl,\n           SUM(pnl) OVER (\n               PARTITION BY account \n               ORDER BY trade_date\n           ) as cumulative_pnl\n    FROM trades\n    ORDER BY account, trade_date;\n\"\"\")\nprint(f\"  {'Account':<8} {'Date':<12} {'Daily P&L':>10} {'Running Tot':>12}\")\nprint(\"  \" + \"-\" * 44)\nfor acc, dt, pnl, cumul in cur.fetchall():\n    print(f\"  {acc:<8} {dt:<12} {pnl:>10.2f} {cumul:>12.2f}\")\n"}]

[{"label":"1 · SQL Basics","code":"# ── SQL Basics — Trade Table with INSERT, SELECT, UPDATE ─────\nimport sqlite3\n\nconn = sqlite3.connect(\":memory:\")\ncur  = conn.cursor()\n\ncur.execute(\"\"\"\n    CREATE TABLE trades (\n        id         INTEGER PRIMARY KEY AUTOINCREMENT,\n        symbol     TEXT    NOT NULL,\n        side       TEXT    NOT NULL CHECK (side IN ('BUY','SELL')),\n        qty        INTEGER NOT NULL,\n        price      REAL    NOT NULL,\n        desk       TEXT    NOT NULL DEFAULT 'Equities',\n        trade_date TEXT    NOT NULL,\n        settled    INTEGER NOT NULL DEFAULT 0\n    )\n\"\"\")\n\ncur.executemany(\n    \"INSERT INTO trades (symbol,side,qty,price,desk,trade_date) VALUES (?,?,?,?,?,?)\",\n    [\n        (\"AAPL\",\"BUY\", 100,172.50,\"Equities\",   \"2024-01-10\"),\n        (\"MSFT\",\"BUY\",  50,380.00,\"Equities\",   \"2024-01-11\"),\n        (\"TSLA\",\"BUY\",  30,225.00,\"Derivatives\",\"2024-01-16\"),\n        (\"AAPL\",\"BUY\",  50,178.00,\"Equities\",   \"2024-01-15\"),\n        (\"AAPL\",\"SELL\", 80,186.55,\"Equities\",   \"2024-02-01\"),\n        (\"MSFT\",\"BUY\",  25,410.00,\"Equities\",   \"2024-02-05\"),\n        (\"NVDA\",\"BUY\",  20,480.00,\"Equities\",   \"2024-02-10\"),\n        (\"TSLA\",\"SELL\", 15,195.40,\"Derivatives\",\"2024-02-12\"),\n        (\"JPM\", \"BUY\",  40,190.00,\"Financials\", \"2024-02-14\"),\n    ]\n)\nconn.commit()\n\n# ── 1. SELECT all BUY trades, sorted by notional ──────────────\nprint(\"=== BUY trades by notional ===\")\nprint(f\"  {'#':<3} {'Symbol':<6} {'Qty':>5} {'Price':>8} {'Notional':>12}  Desk\")\nprint(\"  \" + \"-\"*52)\ncur.execute(\"\"\"\n    SELECT id, symbol, qty, price, qty*price AS notional, desk\n    FROM   trades\n    WHERE  side = 'BUY'\n    ORDER  BY notional DESC\n\"\"\")\nfor id_, sym, qty, price, notional, desk in cur.fetchall():\n    print(f\"  {id_:<3} {sym:<6} {qty:>5} {price:>8.2f} {notional:>12,.2f}  {desk}\")\n\n# ── 2. SELECT with BETWEEN for date range ─────────────────────\nprint(\"\\n=== Trades in January 2024 ===\")\ncur.execute(\"\"\"\n    SELECT symbol, side, qty, price, trade_date\n    FROM   trades\n    WHERE  trade_date BETWEEN '2024-01-01' AND '2024-01-31'\n    ORDER  BY trade_date, symbol\n\"\"\")\nfor sym, side, qty, price, dt in cur.fetchall():\n    print(f\"  {dt}  {sym:<5} {side}  qty={qty:>4}  price={price:.2f}\")\n\n# ── 3. UPDATE — mark Jan as settled ───────────────────────────\ncur.execute(\"UPDATE trades SET settled = 1 WHERE trade_date < '2024-02-01'\")\nconn.commit()\npending = cur.execute(\"SELECT COUNT(*) FROM trades WHERE settled=0\").fetchone()[0]\nsettled = cur.execute(\"SELECT COUNT(*) FROM trades WHERE settled=1\").fetchone()[0]\nprint(f\"\\nSettled: {settled}  |  Pending: {pending}\")\n\nconn.close()"},{"label":"2 · Aggregations","code":"# ── Aggregations — Position Engine with GROUP BY & CASE WHEN ──\nimport sqlite3\n\nconn = sqlite3.connect(\":memory:\")\ncur  = conn.cursor()\n\ncur.executescript(\"\"\"\n    CREATE TABLE trades (\n        id INTEGER PRIMARY KEY AUTOINCREMENT,\n        symbol TEXT NOT NULL, side TEXT NOT NULL,\n        qty INTEGER NOT NULL, price REAL NOT NULL,\n        desk TEXT NOT NULL, trade_date TEXT NOT NULL\n    );\n    INSERT INTO trades (symbol,side,qty,price,desk,trade_date) VALUES\n        ('AAPL','BUY', 100,172.50,'Equities',   '2024-01-10'),\n        ('AAPL','BUY',  50,178.00,'Equities',   '2024-01-15'),\n        ('AAPL','SELL', 80,186.55,'Equities',   '2024-02-01'),\n        ('MSFT','BUY',  50,380.00,'Equities',   '2024-01-11'),\n        ('MSFT','BUY',  25,410.00,'Equities',   '2024-02-05'),\n        ('TSLA','BUY',  30,225.00,'Derivatives','2024-01-16'),\n        ('TSLA','SELL', 15,195.40,'Derivatives','2024-02-12'),\n        ('NVDA','BUY',  20,480.00,'Equities',   '2024-02-10'),\n        ('JPM', 'BUY',  40,190.00,'Financials', '2024-02-14'),\n        ('BP',  'BUY',  60,467.00,'Equities',   '2024-02-18');\n\"\"\")\n\n# ── 1. Notional by desk (GROUP BY + SUM) ──────────────────────\nprint(\"=== Notional traded by desk ===\")\ncur.execute(\"\"\"\n    SELECT desk, COUNT(*) AS trades,\n           ROUND(SUM(qty * price), 2) AS notional\n    FROM   trades\n    GROUP  BY desk\n    ORDER  BY notional DESC\n\"\"\")\nfor desk, count, notional in cur.fetchall():\n    print(f\"  {desk:<14} {count} trades  notional = {notional:>12,.2f}\")\n\n# ── 2. Net position per symbol (CASE WHEN) ────────────────────\nprint(\"\\n=== Net positions (weighted avg cost) ===\")\ncur.execute(\"\"\"\n    SELECT\n        symbol,\n        SUM(CASE WHEN side='BUY'  THEN qty ELSE  0   END)  AS bought,\n        SUM(CASE WHEN side='SELL' THEN qty ELSE  0   END)  AS sold,\n        SUM(CASE WHEN side='BUY'  THEN qty ELSE -qty END)  AS net_qty,\n        ROUND(\n            SUM(CASE WHEN side='BUY' THEN qty*price ELSE 0 END)\n            / NULLIF(SUM(CASE WHEN side='BUY' THEN qty ELSE 0 END), 0),\n        2) AS avg_cost\n    FROM  trades\n    GROUP BY symbol\n    ORDER BY net_qty DESC\n\"\"\")\nprint(f\"  {'Symbol':<6} {'Bought':>7} {'Sold':>6} {'Net':>5}  {'Avg Cost':>9}\")\nprint(\"  \" + \"-\"*40)\nfor sym, bought, sold, net, avg in cur.fetchall():\n    print(f\"  {sym:<6} {bought:>7} {sold:>6} {net:>5}  {avg:>9.2f}\")\n\n# ── 3. HAVING: only large net positions ───────────────────────\nprint(\"\\n=== Net positions > 50 shares (HAVING) ===\")\ncur.execute(\"\"\"\n    SELECT symbol,\n           SUM(CASE WHEN side='BUY' THEN qty ELSE -qty END) AS net\n    FROM   trades\n    GROUP  BY symbol\n    HAVING net > 50\n    ORDER  BY net DESC\n\"\"\")\nfor sym, net in cur.fetchall():\n    print(f\"  {sym:<6}  net = {net}\")\n\nconn.close()"},{"label":"3 · Joins & Blotter","code":"# ── Multi-table Joins + Full Trade Blotter ────────────────────\nimport sqlite3\n\nconn = sqlite3.connect(\":memory:\")\nconn.row_factory = sqlite3.Row   # access columns by name\ncur  = conn.cursor()\n\ncur.executescript(\"\"\"\n    CREATE TABLE instruments (\n        symbol   TEXT PRIMARY KEY,\n        name     TEXT NOT NULL,\n        sector   TEXT NOT NULL,\n        currency TEXT NOT NULL DEFAULT 'USD'\n    );\n    CREATE TABLE trades (\n        id         INTEGER PRIMARY KEY AUTOINCREMENT,\n        symbol     TEXT    NOT NULL REFERENCES instruments(symbol),\n        side       TEXT    NOT NULL,\n        qty        INTEGER NOT NULL,\n        price      REAL    NOT NULL,\n        trade_date TEXT    NOT NULL\n    );\n    INSERT INTO instruments VALUES\n        ('AAPL','Apple Inc',       'Technology','USD'),\n        ('MSFT','Microsoft Corp',  'Technology','USD'),\n        ('NVDA','NVIDIA Corp',     'Technology','USD'),\n        ('TSLA','Tesla Inc',       'Consumer',  'USD'),\n        ('JPM', 'JPMorgan Chase',  'Financials','USD'),\n        ('SHEL','Shell plc',       'Energy',    'GBP');\n    INSERT INTO trades (symbol,side,qty,price,trade_date) VALUES\n        ('AAPL','BUY', 100,172.50,'2024-01-10'),\n        ('AAPL','BUY',  50,178.00,'2024-01-15'),\n        ('AAPL','SELL', 60,186.55,'2024-02-01'),\n        ('MSFT','BUY',  50,380.00,'2024-01-11'),\n        ('MSFT','BUY',  25,410.00,'2024-02-05'),\n        ('NVDA','BUY',  20,480.00,'2024-02-10'),\n        ('TSLA','BUY',  30,225.00,'2024-01-16'),\n        ('TSLA','SELL', 10,195.40,'2024-02-12'),\n        ('JPM', 'BUY',  40,190.00,'2024-02-14');\n    -- SHEL has no trades: will appear only in LEFT JOIN\n\"\"\")\n\n# ── 1. INNER JOIN: position report with instrument name ───────\nprint(\"=== Net positions — INNER JOIN (trades ⨝ instruments) ===\")\ncur.execute(\"\"\"\n    SELECT i.symbol, i.name, i.sector, i.currency,\n           SUM(CASE WHEN t.side='BUY' THEN t.qty ELSE -t.qty END) AS net_qty,\n           ROUND(\n               SUM(CASE WHEN t.side='BUY' THEN t.qty*t.price ELSE 0 END)\n               / NULLIF(SUM(CASE WHEN t.side='BUY' THEN t.qty ELSE 0 END),0),\n           2) AS avg_cost\n    FROM   trades t\n    INNER  JOIN instruments i ON t.symbol = i.symbol\n    GROUP  BY t.symbol\n    HAVING net_qty > 0\n    ORDER  BY net_qty DESC\n\"\"\")\nprint(f\"  {'Sym':<5} {'Name':<18} {'Sector':<12} {'Ccy':<4} {'Net Qty':>7} {'Avg Cost':>9}\")\nprint(\"  \" + \"-\"*60)\nfor r in cur.fetchall():\n    print(f\"  {r['symbol']:<5} {r['name']:<18} {r['sector']:<12} \"\n          f\"{r['currency']:<4} {r['net_qty']:>7} {r['avg_cost']:>9.2f}\")\n\n# ── 2. LEFT JOIN: full instrument universe incl. untraded ─────\nprint(\"\\n=== Instrument universe — LEFT JOIN (all incl. SHEL) ===\")\ncur.execute(\"\"\"\n    SELECT i.symbol, i.name, i.currency,\n           COALESCE(SUM(CASE WHEN t.side='BUY' THEN t.qty ELSE -t.qty END), 0) AS net\n    FROM   instruments i\n    LEFT   JOIN trades t ON i.symbol = t.symbol\n    GROUP  BY i.symbol\n    ORDER  BY i.symbol\n\"\"\")\nfor r in cur.fetchall():\n    status = f\"net={r['net']:>4}\" if r['net'] else \"no trades\"\n    print(f\"  {r['symbol']:<5} {r['name']:<20} {r['currency']}  {status}\")\n\n# ── 3. P&L with current prices ────────────────────────────────\nprices = {\"AAPL\":189.30,\"MSFT\":415.30,\"NVDA\":520.00,\"TSLA\":195.40,\"JPM\":201.50}\ncur.execute(\"\"\"\n    SELECT t.symbol,\n           SUM(CASE WHEN t.side='BUY' THEN t.qty ELSE -t.qty END) AS net_qty,\n           ROUND(SUM(CASE WHEN t.side='BUY' THEN t.qty*t.price ELSE 0 END)\n                 / NULLIF(SUM(CASE WHEN t.side='BUY' THEN t.qty ELSE 0 END),0),2) AS avg_cost\n    FROM trades t GROUP BY t.symbol HAVING net_qty > 0\n\"\"\")\npositions = cur.fetchall()\nprint(\"\\n=== P&L Report ===\")\nprint(f\"  {'Symbol':<6} {'Net Qty':>7} {'Avg Cost':>9} {'Last':>8} {'P&L':>11}\")\nprint(\"  \" + \"-\"*46)\ntotal = 0.0\nfor r in positions:\n    last = prices.get(r[\"symbol\"], r[\"avg_cost\"])\n    pnl  = round((last - r[\"avg_cost\"]) * r[\"net_qty\"], 2)\n    total += pnl\n    sign = \"+\" if pnl >= 0 else \"\"\n    print(f\"  {r['symbol']:<6} {r['net_qty']:>7} {r['avg_cost']:>9.2f} \"\n          f\"{last:>8.2f} {sign}{pnl:>10,.2f}\")\nsign = \"+\" if total >= 0 else \"\"\nprint(f\"  {'Total P&L':>34}  {sign}{total:>10,.2f}\")\n\nconn.close()"},{"label":"4 · Advanced SQL","code":"# ── Advanced SQL: CTEs and Window Functions ─────────────────\nimport sqlite3\n\nconn = sqlite3.connect(\":memory:\")\ncur = conn.cursor()\n\ncur.executescript(\"\"\"\n    CREATE TABLE trades (\n        id INTEGER PRIMARY KEY,\n        account TEXT,\n        trade_date TEXT,\n        pnl REAL\n    );\n    INSERT INTO trades (account, trade_date, pnl) VALUES\n        ('Alpha', '2024-03-01', 1500.00),\n        ('Alpha', '2024-03-02', -200.00),\n        ('Alpha', '2024-03-03', 800.00),\n        ('Beta',  '2024-03-01', -500.00),\n        ('Beta',  '2024-03-02', 1200.00);\n\"\"\")\n\n# 1. CTE: Find top performing accounts, then get their details\nprint(\"=== CTE: Trades for accounts with >1000 Total P&L ===\")\ncur.execute(\"\"\"\n    WITH top_accounts AS (\n        SELECT account, SUM(pnl) as total_pnl\n        FROM trades\n        GROUP BY account\n        HAVING total_pnl > 1000\n    )\n    SELECT t.account, t.trade_date, t.pnl\n    FROM trades t\n    INNER JOIN top_accounts ta ON t.account = ta.account\n    ORDER BY t.account, t.trade_date;\n\"\"\")\nfor row in cur.fetchall():\n    print(f\"  {row[0]:<8} {row[1]:<12} {row[2]:>8.2f}\")\n\n# 2. Window Function: Running Total (Cumulative P&L)\nprint(\"\\n=== Window Function: Cumulative P&L by Account ===\")\ncur.execute(\"\"\"\n    SELECT account, trade_date, pnl,\n           SUM(pnl) OVER (\n               PARTITION BY account \n               ORDER BY trade_date\n           ) as cumulative_pnl\n    FROM trades\n    ORDER BY account, trade_date;\n\"\"\")\nprint(f\"  {'Account':<8} {'Date':<12} {'Daily P&L':>10} {'Running Tot':>12}\")\nprint(\"  \" + \"-\" * 44)\nfor acc, dt, pnl, cumul in cur.fetchall():\n    print(f\"  {acc:<8} {dt:<12} {pnl:>10.2f} {cumul:>12.2f}\")\n"}]

Practice Questions

Question 1

What two properties does a PRIMARY KEY constraint guarantee on a database column?

The column is indexed and stored in sorted order
The column value is unique across all rows and never NULL
The column references a foreign table and is automatically validated
The column is the fastest to query and requires no index

Question 2

A compliance report needs every client account, plus their total trade count if they have traded in the last 30 days (NULL if they have not traded). Which JOIN type is correct?

INNER JOIN — returns only clients who have traded
LEFT JOIN — returns all clients; NULL trade count for inactive accounts
CROSS JOIN — combines every client with every trade row
RIGHT JOIN — returns all trades regardless of whether the client exists

Question 3

A fund transfer must debit £50,000 from Client A and credit £50,000 to Client B. The credit step fails due to a constraint violation. Which database feature ensures Client A's balance is restored to its original value?

A CHECK constraint on the balance column
A transaction with ROLLBACK on error — ACID Atomicity ensures all-or-nothing
A UNIQUE constraint on the account ID column
A stored procedure that validates both accounts before executing

Question 4

In SQLAlchemy ORM, after calling session.flush(), are the changes visible to another database connection before session.commit() is called?

Yes — flush() immediately makes changes visible to all connections
No — flush() sends SQL to the DB but changes remain within the current transaction; other connections see them only after commit()
Yes — flush() and commit() are identical in SQLAlchemy
No — flush() only validates Python objects without touching the database

Question 5

Which of the following SQL queries is vulnerable to SQL injection if 'symbol' comes from user input?

cur.execute("SELECT * FROM trades WHERE symbol = ?", (symbol,))
cur.execute(f"SELECT * FROM trades WHERE symbol = '{symbol}'")
session.query(Trade).filter(Trade.symbol == symbol).all()
cur.execute("SELECT * FROM trades WHERE symbol = :s", {"s": symbol})

Question 6

A trades table has 50 million rows. The query SELECT * FROM trades WHERE trade_date = '2024-03-15' takes 45 seconds. What is the most effective first step?

Add more RAM to the database server
Create a B-tree index on the trade_date column — reduces lookup from O(n) full scan to O(log n)
Rewrite the query to use LIMIT 1000
Partition the trades table by year into separate tables