OOP Advanced, JSON & API Integration

ABCs, mixins, dunder methods, JSON pipelines and REST APIs

By this point you can write Python programs that handle errors, model entities as classes, and read files. Now we move to the patterns that appear in every professional production codebase.

Abstract Base Classes let large teams agree on interfaces at the language level — every account type, every trading strategy must implement the same methods, enforced at import time, not discovered at runtime. Mixins let you compose cross-cutting capabilities (audit logging, serialisation, rate limiting) without deep inheritance chains. Dunder methods make your objects behave like native Python types: sortable, iterable, addable.

Then the outside world. JSON is the universal language of APIs — every data provider speaks it. You will parse it, validate it, and round-trip it with custom types. Finally you will build a complete ETL pipeline — Extract, Transform, Load — the backbone of every trading system and data warehouse in production.

Abstract Base Classes — Financial Contracts

An Abstract Base Class (ABC) defines a contract: every subclass must implement all @abstractmethod members or Python raises TypeError at instantiation — before the first line of business logic ever runs.

  • Import: from abc import ABC, abstractmethod
  • Subclasses that skip any abstract method fail immediately at instantiation — not silently at runtime
  • Combine @abstractmethod with @property to enforce abstract computed attributes
  • Concrete methods in the ABC are inherited by all subclasses for free

Why it matters: A trading system with 20 account types using a plain base class will silently ship code that crashes only when a user triggers a rarely-called method. An ABC crashes at import time — testable, predictable, and safe.

Python
# Production Python uses: from abc import ABC, abstractmethod# Browser REPL uses NotImplementedError to enforce the same contract. class FinancialProduct:    def __init__(self, product_id, owner, balance=0.0):        self.product_id = product_id        self.owner      = owner        self._balance   = float(balance)     def apply_monthly_fee(self) -> float:        """Return delta: positive = gain, negative = cost."""        raise NotImplementedError(type(self).__name__ + " must implement apply_monthly_fee()")     def risk_rating(self) -> str:        """Return 'LOW', 'MEDIUM', or 'HIGH'."""        raise NotImplementedError(type(self).__name__ + " must implement risk_rating()")     @property    def balance(self): return self._balance     def deposit(self, amount):        self._balance += float(amount); return self     def __repr__(self):        return (f"{type(self).__name__}({self.product_id}, "                f"£{self._balance:,.2f}, {self.risk_rating()})") class SavingsAccount(FinancialProduct):    def __init__(self, pid, owner, bal=0.0, rate=0.045):        super().__init__(pid, owner, bal); self.rate = rate    def apply_monthly_fee(self):        interest = round(self._balance * self.rate / 12, 2)        self._balance += interest; return interest    def risk_rating(self): return "LOW" class LoanAccount(FinancialProduct):    def __init__(self, pid, owner, bal=0.0, rate=0.08):        super().__init__(pid, owner, bal); self.rate = rate    def apply_monthly_fee(self):        charge = round(self._balance * self.rate / 12, 2)        self._balance += charge; return -charge    def risk_rating(self): return "HIGH" if self._balance > 40_000 else "MEDIUM" class InvestmentFund(FinancialProduct):    def __init__(self, pid, owner, bal=0.0, vol=0.15):        super().__init__(pid, owner, bal); self.vol = vol    def apply_monthly_fee(self):        fee = round(self._balance * 0.001, 2)        self._balance -= fee; return -fee    def risk_rating(self): return "HIGH" if self.vol > 0.2 else "MEDIUM" products = [    SavingsAccount("SAV001", "Alice",  25_000, rate=0.048),    LoanAccount   ("LON001", "Bob",    38_000, rate=0.079),    InvestmentFund("INV001", "Carol",  80_000, vol=0.28),    InvestmentFund("INV002", "Dave",   45_000, vol=0.09),] print(f"{'ID':<8} {'Type':<18} {'Risk':<8} {'Balance':>12}  {'Delta':>10}")print("-" * 62)for p in products:    d = p.apply_monthly_fee()    print(f"{p.product_id:<8} {type(p).__name__:<18} {p.risk_rating():<8} "          f"£{p.balance:>10,.2f}  {'+'if d>=0 else ''}£{d:>8,.2f}") # Calling the base class raises NotImplementedError (same as ABC TypeError):# FinancialProduct("X", "Y").apply_monthly_fee()

Mixins & Multiple Inheritance

A mixin is a class that adds one focused capability — audit logging, JSON serialisation, rate limiting — without being a standalone base class. The rule: a mixin defines behaviour, never independent state.

  • Composition over inheritance: narrow mixins that compose ("has this capability") are more maintainable than deep hierarchies ("is a kind of")
  • MRO (Method Resolution Order): Python uses C3 linearisation — inspect with MyClass.__mro__. Convention: most-specific class leftmost
  • super(): always delegates to the next class in the MRO chain, not the immediate parent — essential for cooperative multiple inheritance
  • Naming convention: suffix mixins with Mixin — AuditMixin, SerializableMixin

Production pattern: Financial systems use mixins to add cross-cutting concerns (audit trails, retry logic, serialisation) to domain classes without coupling every subclass to unrelated infrastructure.

Python
import json class AuditMixin:    def __init__(self, *args, **kwargs):        super().__init__(*args, **kwargs)        self._audit_log = []    def log_event(self, action, detail):        self._audit_log.append(f"{action}: {detail}"); return self    def print_audit(self):        for entry in self._audit_log: print(f"  AUDIT  {entry}") class SerializableMixin:    def to_dict(self):        return {k: v for k, v in self.__dict__.items()                if not k.startswith("_")}    def to_json(self):        return json.dumps(self.to_dict(), default=str, indent=2) class AccountBase:    def __init__(self, acc_id, owner, balance=0.0):        self.acc_id   = acc_id        self.owner    = owner        self._balance = float(balance)    def deposit(self, amount):        self._balance += amount; return self    def withdraw(self, amount):        self._balance -= amount; return self    @property    def balance(self): return self._balance # Mixin order: most specific → least specific (left to right)class AuditedAccount(AuditMixin, SerializableMixin, AccountBase):    def deposit(self, amount):        super().deposit(amount)        self.log_event("DEPOSIT", f"+£{amount:,.2f}"); return self    def withdraw(self, amount):        super().withdraw(amount)        self.log_event("WITHDRAW", f"-£{amount:,.2f}"); return self print("MRO:", [c.__name__ for c in AuditedAccount.__mro__])acc = AuditedAccount("ACC001", "Alice Chen", 10_000)acc.deposit(5_000).withdraw(1_200).deposit(2_500)print(f"Balance : £{acc.balance:,.2f}")print(f"Pub data: {acc.to_dict()}")print("Audit trail:")acc.print_audit()

Dunder Methods — Industry-Standard Patterns

Dunder (double-underscore) methods let your custom objects behave like Python built-ins. len(portfolio) calls __len__; "AAPL" in portfolio calls __contains__; sorted(positions) calls __lt__.

  • Representation: __repr__ — unambiguous, for logs and the REPL  ·  __str__ — human-readable, for print()
  • Equality & ordering: __eq__ · __hash__ · __lt__ + @functools.total_ordering (auto-generates ≤, >, ≥)
  • Container protocol: __len__ · __contains__ (O(1) with dict backing) · __iter__ · __getitem__
  • Arithmetic: __add__ (a + b) · __iadd__ (a += b) · __mul__
  • Context manager: __enter__ / __exit__ — powers the with statement

Key rule: Always define __repr__. It appears in logs, tracebacks, and the REPL — include the object's ID and key state so any engineer can understand it without a debugger.

Python
import functools @functools.total_orderingclass Position:    def __init__(self, sym, qty, avg, price):        self.sym=sym; self.qty=qty; self.avg=avg; self.price=price    @property    def value(self): return self.qty * self.price    @property    def pnl(self): return round((self.price - self.avg) * self.qty, 2)    def __repr__(self):        return (f"  {self.sym:5} {self.qty:3}x  val:£{self.value:>9,.2f}"                f"  P&L:{'+'if self.pnl>=0 else ''}£{self.pnl:>7,.2f}")    def __eq__(self, other): return self.sym == other.sym    def __lt__(self, other): return self.value < other.value    def __add__(self, other):        """Merge two lots: dollar-cost-average the cost basis."""        total_qty = self.qty + other.qty        new_avg   = (self.avg*self.qty + other.avg*other.qty) / total_qty        return Position(self.sym, total_qty, round(new_avg, 4), self.price) class Portfolio:    def __init__(self, name): self.name=name; self._p={}    def add(self, sym, qty, avg, price):        self._p[sym]=Position(sym,qty,avg,price); return self    def __len__(self):         return len(self._p)    def __contains__(self, s): return s in self._p   # O(1) — dict lookup    def __iter__(self):        return iter(sorted(self._p.values(), reverse=True))    def __bool__(self):        return len(self._p) > 0    @property    def total_pnl(self): return sum(p.pnl for p in self._p.values())    def __repr__(self): return f"Portfolio({self.name!r}, {len(self)} positions)" pf = Portfolio("ALPHA")pf.add("AAPL",100,182.50,186.55).add("MSFT",50,415.25,430.00).add("TSLA",20,220.00,195.00) print(repr(pf))print(f"bool(pf)={bool(pf)}  len={len(pf)}  'AAPL' in pf={'AAPL' in pf}")print("Positions sorted by market value:")for pos in pf: print(pos)print(f"Total P&L: {'+'if pf.total_pnl>=0 else ''}£{pf.total_pnl:,.2f}") # Merge two lots using __add__lot1 = Position("NVDA", 20, 480.00, 520.00)lot2 = Position("NVDA", 10, 510.00, 520.00)merged = lot1 + lot2print(f"\nMerged NVDA lot: qty={merged.qty}  avg={merged.avg:.2f}  val=£{merged.value:,.2f}")

JSON — The Language of Financial APIs

Every financial data provider — Bloomberg, Alpha Vantage, OpenExchangeRates — speaks JSON. Python's json module handles the full round-trip, but JSON only knows strings, numbers, booleans, null, lists and objects — not Decimal, datetime, or custom classes.

  • json.loads(s) — parse a JSON string → Python dict  ·  json.dumps(d) — dict → JSON string
  • json.load(f) / json.dump(d, f) — read/write from a file object
  • Custom encoder: subclass json.JSONEncoder, override default(obj) to handle non-standard types
  • Custom decoder: pass object_hook=fn to json.loads() — called for every parsed dict, innermost-first
  • Quick fallback: json.dumps(d, default=str) converts unknown types to strings — good for logging, not for round-tripping

Always wrap API JSON parsing in try/except json.JSONDecodeError — external services send malformed payloads, truncated responses, or HTML error pages when they are down.

Python
import jsonfrom datetime import datetimefrom decimal import Decimal # ── Custom encoder: Decimal + datetime ───────────────────class FinancialEncoder(json.JSONEncoder):    def default(self, obj):        if isinstance(obj, Decimal):  return {"__decimal__": str(obj)}        if isinstance(obj, datetime): return {"__datetime__": obj.isoformat()}        return super().default(obj) # ── Custom decoder: reverse the transformation ────────────def financial_decoder(d):    if "__decimal__"  in d: return Decimal(d["__decimal__"])    if "__datetime__" in d: return datetime.fromisoformat(d["__datetime__"])    return d trade = {    "id":       "T001",    "symbol":   "AAPL",    "qty":      100,    "price":    Decimal("183.50"),    "executed": datetime(2024, 1, 16, 9, 31, 0),    "notional": Decimal("18350.00"),} encoded = json.dumps(trade, cls=FinancialEncoder, indent=2)print("Encoded JSON:\n" + encoded) decoded = json.loads(encoded, object_hook=financial_decoder)print(f"\nDecoded types:")print(f"  price    : {type(decoded['price']).__name__} = {decoded['price']}")print(f"  executed : {type(decoded['executed']).__name__} = {decoded['executed']}") # ── Error handling ─────────────────────────────────────────for bad in ['{"price": 183', "not json at all"]:    try:        json.loads(bad)    except json.JSONDecodeError as e:        print(f"\nJSONDecodeError on {bad!r}: {e}")

Consuming REST APIs — Market Data & FX Rates

Financial data arrives over HTTP. Python gives you two tools: the requests library (industry standard — OAuth2, retries, streaming, connection pooling) and urllib.request (stdlib — always available, works in the browser REPL).

  • 401 Unauthorized — wrong or expired API key; do not retry, rotate the key
  • 403 Forbidden — valid key but insufficient permissions (wrong plan, endpoint not subscribed)
  • 429 Too Many Requests — rate limited; back off exponentially: 1 s → 2 s → 4 s → 8 s; respect the Retry-After header
  • 500 / 503 — server error; retry with back-off, alert if it persists
  • Always validate response data against your expected schema — APIs change formats without notice

Note: the browser REPL cannot make external HTTP requests (CORS policy). The code below uses simulated API responses to demonstrate the full parsing and error-handling pattern.

Python
import json # ── Production pattern (requests library) ─────────────────# import requests# def fetch_json(url, params, api_key):#     headers = {"Authorization": f"Bearer {api_key}"}#     r = requests.get(url, params=params, headers=headers, timeout=10)#     r.raise_for_status()   # raises HTTPError on 4xx/5xx#     return r.json() # ── Browser REPL: simulated API responses ─────────────────SIMULATED = {    "crypto": json.dumps({        "bitcoin":  {"usd": 42350.18, "usd_24h_change":  2.14},        "ethereum": {"usd":  2284.55, "usd_24h_change": -0.87},        "solana":   {"usd":    88.32, "usd_24h_change":  4.21},    }),    "fx": json.dumps({        "base": "USD",        "rates": {"GBP": 0.7862, "EUR": 0.9184, "JPY": 148.25, "INR": 83.15},    }),} def parse_response(raw):    """Parse JSON; raise ValueError on bad payload."""    try:        return json.loads(raw)    except json.JSONDecodeError as e:        raise ValueError(f"Malformed API response: {e}") from e # Crypto pricescrypto = parse_response(SIMULATED["crypto"])print("Crypto prices (simulated CoinGecko):")for coin, d in crypto.items():    chg = d["usd_24h_change"]    arrow = "▲" if chg >= 0 else "▼"    print(f"  {coin.capitalize():<10}  £{d['usd']:>10,.2f}  24h: {chg:+.2f}%  {arrow}") # FX ratesfx = parse_response(SIMULATED["fx"])print(f"\nFX rates (base: {fx['base']}):")for ccy, rate in fx["rates"].items():    print(f"  1 USD = {rate:.4f} {ccy}")

ETL Pipeline — API → Transform → Report

ETL — Extract, Transform, Load — is the backbone of every trading system, data warehouse, and regulatory report. Each stage has exactly one job:

  1. Extract — fetch raw data; handle parse errors here; return (data, errors) — never raise on a partial failure
  2. Transform — validate, normalise, and enrich each record; bad records go to a reject log; good records continue
  3. Load — write clean data to storage (CSV, database); compute aggregates and produce the report
  • Never silently discard data: every rejected record must carry an ID and a reason — that is your audit trail
  • Idempotency: run the pipeline twice → same result; essential for recovery after partial failures
  • Single responsibility: each stage is independently testable; a bad API response is an Extract problem, not a Transform problem
Python
import jsonfrom decimal import Decimalfrom collections import defaultdict # Simulated raw trade stream (T003 has invalid price, T005 missing symbol)RAW_TRADES = [    '{"id":"T001","symbol":"AAPL","qty":100,"price":"183.50"}',    '{"id":"T002","symbol":"MSFT","qty":50, "price":"415.20"}',    '{"id":"T003","symbol":"TSLA","qty":20, "price":"BAD"}',    '{"id":"T004","symbol":"NVDA","qty":30, "price":"520.00"}',    '{"id":"T005",              "qty":10, "price":"186.55"}',    '{"id":"T006","symbol":"GOOGL","qty":15,"price":"140.80"}',] # ── Stage 1: Extract ──────────────────────────────────────def extract(records):    data, errors = [], []    for r in records:        try:    data.append(json.loads(r))        except json.JSONDecodeError as e: errors.append(str(e))    return data, errors # ── Stage 2: Transform ────────────────────────────────────def transform(records):    clean, rejected = [], []    for r in records:        rid = r.get("id", "?")        try:            if not r.get("symbol"): raise ValueError("missing symbol")            r["price"]    = Decimal(str(r["price"]))            r["notional"] = float(r["price"]) * r["qty"]            clean.append(r)        except Exception as e:            rejected.append({"id": rid, "reason": str(e)})    return clean, rejected # ── Stage 3: Load (report) ────────────────────────────────def load(records):    totals = defaultdict(lambda: {"notional": 0.0, "qty": 0})    for r in records:        totals[r["symbol"]]["notional"] += r["notional"]        totals[r["symbol"]]["qty"]      += r["qty"]     print(f"\n{'Symbol':<8} {'Qty':>6} {'Notional':>13}")    print("─" * 30)    grand = 0.0    for sym, t in sorted(totals.items()):        print(f"{sym:<8} {t['qty']:>6} {t['notional']:>13,.2f}")        grand += t["notional"]    print(f"{'TOTAL':>16} {grand:>13,.2f}") raw,   ex_err = extract(RAW_TRADES)clean, tr_err = transform(raw) print(f"Extracted : {len(raw)} ok   {len(ex_err)} parse errors")print(f"Transformed: {len(clean)} clean  {len(tr_err)} rejected")for r in tr_err: print(f"  ✗ {r['id']}: {r['reason']}")load(clean)

Integration — Institutional Portfolio System

This section assembles every concept from the chapter into a production-grade system — the architecture used by hedge funds, asset managers, and proprietary trading desks:

  • ABC — Instrument enforces asset_class() and margin_requirement() on every tradeable
  • Mixins — AuditMixin logs every trade automatically; no subclass needs to remember to call it
  • Dunder methods — Portfolio is a proper container: len(), in, iteration, and repr()
  • JSON config — portfolio parameters (manager, initial cash, position limits) loaded from a JSON string at runtime, not hardcoded
  • ETL pattern — trade execution follows Extract (validate inputs) → Transform (compute fills) → Load (record audit) stages
Python
import jsonfrom collections import namedtuple # ── Instrument base class (contract via NotImplementedError) ──# Production: from abc import ABC, abstractmethod + class Instrument(ABC)class Instrument:    def __init__(self, ticker, name): self.ticker=ticker; self.name=name    def asset_class(self):        raise NotImplementedError(type(self).__name__ + " must implement asset_class()")    def margin_requirement(self):        raise NotImplementedError(type(self).__name__ + " must implement margin_requirement()")    def __repr__(self): return f"{self.asset_class()}:{self.ticker}" class Equity(Instrument):    def asset_class(self): return "EQ"    def margin_requirement(self): return 0.20 class CryptoCurrency(Instrument):    def asset_class(self): return "CRYPTO"    def margin_requirement(self): return 0.50 class FixedIncome(Instrument):    def asset_class(self): return "FI"    def margin_requirement(self): return 0.05 Fill = namedtuple("Fill", ["ticker", "qty", "price", "notional"]) CONFIG = json.dumps({"manager": "Quant Alpha Fund I", "cash": 500_000, "max_pos": 4}) class InstitutionalPortfolio:    def __init__(self, config_json):        cfg           = json.loads(config_json)        self.manager  = cfg["manager"]        self._cash    = float(cfg["cash"])        self._max_pos = cfg["max_pos"]        self._pos     = {}        self._fills   = []    def __len__(self):         return len(self._pos)    def __contains__(self, t): return t in self._pos    def __iter__(self):        return iter(self._pos.items())    def __repr__(self):        return f"Portfolio({self.manager}, {len(self)} pos, cash=£{self._cash:,.0f})"    def execute(self, inst, qty, price):        cost = qty * price        if cost > self._cash:            print(f"  x {inst.ticker}: insufficient cash (need £{cost:,.0f})"); return        if inst.ticker not in self._pos and len(self._pos) >= self._max_pos:            print(f"  x {inst.ticker}: position limit ({self._max_pos}) reached"); return        self._cash -= cost        self._pos[inst.ticker] = self._pos.get(inst.ticker, 0) + qty        fill = Fill(inst.ticker, qty, price, cost)        self._fills.append(fill)        print(f"  + {fill.ticker:<6}  {qty:>3} x £{price:>8,.2f}  notional £{cost:>10,.2f}")    def report(self):        print("")        print(f"{'Ticker':<8} {'Qty':>6}")        print("-" * 16)        for ticker, qty in self:            print(f"{ticker:<8} {qty:>6}")        print(f"Cash remaining: £{self._cash:,.2f}")        print(f"Total fills   : {len(self._fills)}") instruments = {    "AAPL":  Equity("AAPL", "Apple Inc"),    "MSFT":  Equity("MSFT", "Microsoft"),    "BTC":   CryptoCurrency("BTC", "Bitcoin"),    "GILTS": FixedIncome("GILTS", "UK 10Y Gilt"),} pf = InstitutionalPortfolio(CONFIG)print(repr(pf))print("")pf.execute(instruments["AAPL"],  100, 183.50)pf.execute(instruments["MSFT"],   50, 415.20)pf.execute(instruments["BTC"],     1, 42350.00)pf.execute(instruments["GILTS"], 200, 98.50)pf.execute(instruments["MSFT"],   25, 418.00)pf.report()
PythonRuns entirely in your browser — nothing is sent to a server.

Practice Questions

Question 1

When exactly does Python raise TypeError for a subclass that misses an @abstractmethod?

  • When the missing method is first called at runtime
  • When the subclass is instantiated (created)
  • When the module is imported
  • Only when running in strict mode

Question 2

In class A(B, C, D), when code inside B calls super().foo(), which class does Python look in next?

  • The direct parent of B in the class hierarchy
  • The next class in A's MRO after B — typically C
  • Always object (the root class)
  • D, because it is last and therefore most general

Question 3

Why is defining __contains__ directly more efficient than relying on __iter__ for 'AAPL' in portfolio?

  • __contains__ compiles to faster bytecode
  • Without __contains__, Python falls back to __iter__ and checks each element — O(n) vs O(1) for a dict-backed __contains__
  • __iter__ is deprecated for membership testing
  • Python only calls __iter__ for lists, not custom objects

Question 4

You define only __repr__ on a class. What does Python use when print(obj) is called?

  • Python raises AttributeError — __str__ must be explicitly defined
  • Python falls back to __repr__ when __str__ is not defined
  • Python prints the object's memory address
  • Python calls __format__ instead

Question 5

What is the correct way to round-trip a Python Decimal through JSON without losing precision?

  • Use json.dumps(d, default=float) — float has enough precision for financial data
  • Encode Decimal as a string (e.g. {'__decimal__': '183.50'}) and decode it back with object_hook
  • JSON natively supports arbitrary-precision decimals
  • Multiply by 100, store as int, divide on decode

Question 6

What does the object_hook parameter in json.loads() do?

  • It validates the JSON against a schema before parsing
  • It is called for every parsed JSON object (dict), allowing custom type conversion during parsing
  • It catches JSONDecodeError automatically and returns None on failure
  • It speeds up parsing by skipping nested objects

Question 7

Why do Extract and Transform stages in an ETL pipeline return (data, errors) tuples instead of raising exceptions?

  • Returning tuples is required by the csv and json modules
  • Raising on the first bad record would halt the pipeline, losing all valid records that follow it
  • Exceptions are too slow to raise in a loop
  • ETL pipelines run in processes that cannot propagate exceptions

Question 8

An API returns HTTP 429 with a Retry-After: 30 header. What is the correct response?

  • Retry immediately — 429 is informational, not an error
  • Wait at least 30 seconds then retry; use exponential back-off if Retry-After is absent
  • Rotate the API key — 429 means the key is rate-limited permanently
  • Switch to a different API endpoint — 429 means the current endpoint is unavailable