Error Handling, File I/O & Classes
Building robust, production-grade financial systems
Professional financial systems never crash silently. A payment processor must handle network failures, invalid inputs, and fraud alerts without losing data or corrupting balances. An algorithmic trading system must log every order to disk, recover gracefully from API errors, and model portfolios as objects with clear rules.
This chapter builds those foundations: exception handling that makes code resilient, file I/O for audit trails and reports, and object-oriented programming to model accounts, trades, and portfolios as real-world entities. Every example is drawn from live banking and trading system challenges.
Error Handling — try / except / else / finally
Financial systems must handle failures gracefully — a payment processor that crashes on bad input loses money and trust. Python's exception handling lets you respond to errors without crashing.
The four clauses:
try— the code that might failexcept ExceptionType as e— runs if that exception occurs; catch specific types first, general lastelse— runs only if no exception occurred (the happy path)finally— always runs, exception or not (audit log, close connections)
Best practice: catch the most specific exception first. Catching bare Exception hides bugs and makes debugging a nightmare.
# ── Simple: safe account lookup ───────────────────────────accounts = {"ACC001": {"balance": 5_000}, "ACC002": {"balance": 1_200}} def get_balance(acc_id): try: return accounts[acc_id]["balance"] except KeyError: return None # no crash, caller decides what to do print(get_balance("ACC001")) # 5000print(get_balance("ACC999")) # None — safe # ── Medium: multiple exception types + else + finally ─────def process_payment(balance, amount): try: if not isinstance(amount, (int, float)): raise TypeError(f"Amount must be numeric, got {type(amount).__name__}") if amount <= 0: raise ValueError("Amount must be positive") if amount > balance: raise ValueError(f"Insufficient: need £{amount:,.2f}, have £{balance:,.2f}") new_balance = balance - amount except TypeError as e: print(f" Type error : {e}") return None except ValueError as e: print(f" Value error : {e}") return None else: print(f" ✓ Approved — new balance: £{new_balance:,.2f}") return new_balance finally: print(f" [audit] payment attempt recorded") # always runs print("Test 1 — valid:")process_payment(5_000, 1_500)print("Test 2 — bad type:")process_payment(5_000, "abc")print("Test 3 — insufficient funds:")process_payment(5_000, 7_000) # ── Complex: re-raise + chained exceptions ─────────────────def transfer(sender, receiver, amount, accounts): try: if accounts[sender]["balance"] < amount: raise ValueError(f"Insufficient funds in {sender}") accounts[sender]["balance"] -= amount accounts[receiver]["balance"] += amount except KeyError as e: raise RuntimeError(f"Account not found: {e}") from e # chain except ValueError as e: print(f"Transfer rejected: {e}") else: print(f"✓ Transferred £{amount:,.2f}: {sender} → {receiver}") transfer("ACC001", "ACC002", 500, accounts)transfer("ACC001", "ACC999", 100, accounts) # KeyError → RuntimeErrorCustom Exceptions — Banking Error Hierarchy
Generic exceptions (ValueError, KeyError) carry no domain context. In a banking system you need to catch InsufficientFundsError vs FraudAlertError vs DailyLimitExceededError independently — each triggers a different response, UI message, and retry strategy.
Pattern:
- Create a base
BankingError(Exception)— catch-all for any banking error - Subclass for each specific failure mode
- Store domain data on the exception (
self.amount,self.risk_score) - Catch the most specific type first; fall through to
BankingErrorfor unexpected cases
This is the pattern used in every serious financial API — different HTTP status codes, different user messages, and different retry logic per error type.
# ── Define the hierarchy ──────────────────────────────────class BankingError(Exception): """Base — catch all banking errors with one handler.""" pass class InsufficientFundsError(BankingError): def __init__(self, account_id, amount, balance): self.account_id = account_id self.amount = amount self.balance = balance super().__init__( f"{account_id}: requested £{amount:,.2f}, " f"only £{balance:,.2f} available" ) class DailyLimitExceededError(BankingError): def __init__(self, account_id, attempted, remaining): super().__init__( f"{account_id}: daily limit hit — " f"£{remaining:,.2f} remaining, requested £{attempted:,.2f}" ) class FraudAlertError(BankingError): def __init__(self, account_id, risk_score): self.risk_score = risk_score super().__init__( f"{account_id}: transaction blocked (risk score {risk_score}/100)" ) # ── Business logic — raises specific exceptions ────────────def execute_transfer(acc_id, amount, balance, daily_remaining, risk_score): if risk_score > 75: raise FraudAlertError(acc_id, risk_score) if amount > daily_remaining: raise DailyLimitExceededError(acc_id, amount, daily_remaining) if amount > balance: raise InsufficientFundsError(acc_id, amount, balance) return balance - amount # ── Test every path ────────────────────────────────────────test_cases = [ ("ACC001", 500, 5_000, 10_000, 20), # ✓ approved ("ACC002", 500, 5_000, 10_000, 90), # 🚨 fraud ("ACC003", 8_000, 5_000, 5_000, 30), # ⚠ daily limit ("ACC004", 6_000, 5_000, 10_000, 10), # ⚠ insufficient] for acc_id, amount, balance, limit, risk in test_cases: try: new_bal = execute_transfer(acc_id, amount, balance, limit, risk) print(f" ✓ {acc_id}: approved — balance £{new_bal:,.2f}") except FraudAlertError as e: print(f" 🚨 FRAUD {e}") except DailyLimitExceededError as e: print(f" ⚠ LIMIT {e}") except InsufficientFundsError as e: print(f" ⚠ FUNDS {e}") except BankingError as e: print(f" ✗ ERROR {e}") # catch-all fallbackFile I/O — Audit Logs & CSV Reports
Financial systems produce two kinds of files constantly: audit logs (append-only records of every transaction) and CSV reports (exports for reconciliation, regulators, and analytics).
Always use the context manager:
with open("audit.log", "a") as f:
f.write(entry)
# file is automatically closed here — even if an exception occurs
File modes: "r" read, "w" write (overwrites), "a" append (never deletes), "r+" read+write
csv module: always use csv.writer / csv.DictReader — never manually split CSV strings (commas can appear inside quoted values).
In the browser REPL, files are written to an in-memory filesystem and are lost on page reload. In production they persist to disk.
import csv # ── Simple: append-only transaction audit log ──────────────transactions = [ ("2024-01-15", "ACC001", "CREDIT", 5_000.00, "Salary"), ("2024-01-15", "ACC001", "DEBIT", 1_500.00, "Rent"), ("2024-01-16", "ACC002", "CREDIT", 2_000.00, "Transfer in"), ("2024-01-16", "ACC002", "DEBIT", 250.00, "Card payment"),] # Append mode — never overwrites; safe for concurrent writerswith open("audit.log", "a") as f: for date, acc, type_, amount, desc in transactions: f.write(f"{date}|{acc}|{type_}|{amount:.2f}|{desc}\n") # Read backprint("Audit log:")with open("audit.log", "r") as f: for line in f: date, acc, tp, amount, desc = line.strip().split("|") sign = "+" if tp == "CREDIT" else "-" print(f" {date} {acc} {tp:6} {sign}£{float(amount):>8,.2f} {desc}") # ── Medium: CSV trade report ───────────────────────────────headers = ["date", "symbol", "side", "qty", "price", "value"]trade_data = [ ["2024-01-15", "AAPL", "BUY", 100, 182.50], ["2024-01-15", "TSLA", "SELL", 50, 210.00], ["2024-01-16", "MSFT", "BUY", 75, 415.25], ["2024-01-16", "NVDA", "BUY", 30, 485.00],] with open("trades.csv", "w", newline="") as f: writer = csv.writer(f) writer.writerow(headers) writer.writerows([r + [r[3] * r[4]] for r in trade_data]) print("\nTrade report:")total = 0.0with open("trades.csv") as f: for row in csv.DictReader(f): v = float(row["value"]) total += v print(f" {row['date']} {row['side']:4} {row['qty']:3}x {row['symbol']:5}: £{v:>10,.2f}")print(f" {'─'*40}\n Total: £{total:>10,.2f}") # ── Complex: exception-safe file loader ───────────────────def load_audit_log(filename): records, errors = [], [] try: with open(filename) as f: for i, line in enumerate(f, 1): try: date, acc, tp, amount, desc = line.strip().split("|") records.append({"date": date, "account": acc, "type": tp, "amount": float(amount)}) except ValueError: errors.append(f"Line {i}: malformed") except FileNotFoundError: print(f"Not found: {filename}") return [], [] return records, errors recs, errs = load_audit_log("audit.log")net = sum(r["amount"] if r["type"]=="CREDIT" else -r["amount"] for r in recs)print(f"\nLoaded {len(recs)} entries | net flow: £{net:,.2f}")Classes & Objects — The BankAccount
A class is a blueprint; an object is a specific instance. OOP is the natural fit for financial systems — an account has state (balance, history) and behaviour (deposit, withdraw, statement). All business rules live in one place: the class itself.
Key concepts:
__init__(self, ...)— constructor, runs when you create the objectself— the specific instance; Python passes it automatically- Instance attributes — data on this object:
self._balance - Methods — functions bound to the object:
acc.deposit(500) - Method chaining — return
selfto allowacc.deposit(500).withdraw(100) __repr__— defines whatprint(acc)shows
from datetime import datetime class BankAccount: """ Full-featured bank account. All business rules live here — encapsulated, testable, reusable. """ def __init__(self, account_id, owner, initial_balance=0.0): self.account_id = account_id self.owner = owner self._balance = float(initial_balance) self._transactions = [] self._created_at = datetime.now().strftime("%Y-%m-%d") def deposit(self, amount, description="Deposit"): if amount <= 0: raise ValueError("Deposit must be positive") self._balance += amount self._transactions.append(("CREDIT", amount, self._balance, description)) return self # enables method chaining def withdraw(self, amount, description="Withdrawal"): if amount <= 0: raise ValueError("Amount must be positive") if amount > self._balance: raise ValueError( f"Insufficient funds: £{self._balance:,.2f} available, " f"£{amount:,.2f} requested" ) self._balance -= amount self._transactions.append(("DEBIT", amount, self._balance, description)) return self @property def balance(self): """Read-only — balance only changes through deposit/withdraw.""" return self._balance @property def transaction_count(self): return len(self._transactions) def statement(self): print(f"\n{'─'*54}") print(f" {self.account_id} | {self.owner} | opened {self._created_at}") print(f"{'─'*54}") for type_, amount, running, desc in self._transactions: sign = "+" if type_ == "CREDIT" else "-" print(f" {type_:6} {sign}£{amount:>8,.2f} bal: £{running:>10,.2f} {desc}") print(f"{'─'*54}") print(f" Current balance: £{self._balance:,.2f}") def __repr__(self): return f"BankAccount({self.account_id!r}, {self.owner!r}, £{self._balance:,.2f})" # ── Using the class ────────────────────────────────────────acc = BankAccount("ACC001", "Alice Chen", 10_000) (acc .deposit(5_000, "Bonus payment") .deposit(2_500, "Freelance income") .withdraw(1_800, "Rent") .withdraw(450, "Council tax")) acc.statement()print(f"\n{acc.transaction_count} transactions logged") try: acc.withdraw(999_999)except ValueError as e: print(f"\nRejected: {e}") print(repr(acc))Properties, Encapsulation & Class Methods
Encapsulation keeps internal state private and exposes only a controlled interface. You can't set account._balance = 1_000_000 — you go through deposit() so all rules apply consistently.
Python access conventions:
_attr— single underscore: "internal, use the methods"__attr— double underscore: name-mangled, harder to access externally
Decorators:
@property— read access like an attribute, logic runs underneath@attr.setter— validates before allowing a write@classmethod— alternative constructors (from CSV, from API response)@staticmethod— utility functions that don't needselforcls
class SavingsAccount: """Savings account with tiered interest, encapsulated balance, factory methods.""" bank_name = "Steleios Bank" # class attribute — shared by all instances _base_rate = 0.045 TIER_MULTIPLIERS = {"standard": 1.0, "silver": 1.25, "gold": 1.5, "premium": 2.0} def __init__(self, account_id, owner, balance=0.0, tier="standard"): self._account_id = account_id self._owner = owner self._balance = float(balance) self._tier = tier.lower() # ── Properties ──────────────────────────────────────── @property def balance(self): return self._balance @property def interest_rate(self): """Computed from tier — no setter (business rule, not manual input).""" return self._base_rate * self.TIER_MULTIPLIERS.get(self._tier, 1.0) @property def annual_interest(self): return round(self._balance * self.interest_rate, 2) @property def tier(self): return self._tier @tier.setter def tier(self, new_tier): if new_tier not in self.TIER_MULTIPLIERS: raise ValueError( f"Invalid tier {new_tier!r}. Options: {list(self.TIER_MULTIPLIERS)}" ) self._tier = new_tier # ── Alternative constructor ──────────────────────────── @classmethod def from_csv_row(cls, row): """Create from CSV: 'ACC003,Carol,8000,standard'.""" acc_id, owner, balance, tier = row.split(",") return cls(acc_id.strip(), owner.strip(), float(balance), tier.strip()) # ── Utility (no self/cls needed) ────────────────────── @staticmethod def is_valid_sort_code(code): """Validate UK sort code XX-XX-XX.""" parts = code.split("-") return len(parts) == 3 and all(len(p) == 2 and p.isdigit() for p in parts) def __repr__(self): return (f"SavingsAccount({self._account_id!r}, {self._tier!r}, " f"£{self._balance:,.2f}, rate={self.interest_rate*100:.2f}%)") # ── Demonstrate ───────────────────────────────────────────accounts = [ SavingsAccount("SAV001", "Alice", 50_000, "gold"), SavingsAccount("SAV002", "Bob", 12_000, "silver"), SavingsAccount.from_csv_row("SAV003, Carol, 8000, standard"),] print(f"{'ID':<8} {'Tier':<10} {'Balance':>12} {'Rate':>7} {'Annual Interest':>16}")print("─" * 60)for acc in sorted(accounts, key=lambda a: a.annual_interest, reverse=True): print(f"{acc._account_id:<8} {acc.tier:<10} " f"£{acc.balance:>10,.2f} {acc.interest_rate*100:>5.2f}%" f" £{acc.annual_interest:>12,.2f}") # Tier upgrade with validationaccounts[1].tier = "gold"print(f"\nBob upgraded — new rate: {accounts[1].interest_rate*100:.2f}%")try: accounts[0].tier = "diamond"except ValueError as e: print(f"Tier error: {e}") print(f"Sort code 20-51-14: {SavingsAccount.is_valid_sort_code('20-51-14')}")print(f"Sort code 205114 : {SavingsAccount.is_valid_sort_code('205114')}")Inheritance — Financial Product Hierarchy
Inheritance lets you build specialised classes from a common base. In banking, all account types share core operations (deposit, withdraw, balance) but differ in fees, overdraft rules, and features.
Key mechanics:
class Child(Parent):— Child inherits all methods and attributes from Parentsuper().__init__(...)— call the parent's constructor to set up shared state- Method overriding — redefine a parent method in the child for specialised behaviour
- Polymorphism — a list of mixed types all respond to
apply_monthly_fee(); each does the right thing without an if/elif chain
# ── Base class — shared engine ────────────────────────────class Account: def __init__(self, account_id, owner, balance=0.0): self.account_id = account_id self.owner = owner self._balance = float(balance) def deposit(self, amount): if amount <= 0: raise ValueError("Deposit must be positive") self._balance += amount return self def withdraw(self, amount): if amount > self._balance: raise ValueError("Insufficient funds") self._balance -= amount return self @property def balance(self): return self._balance def monthly_fee(self): return 0.0 # override in subclasses def apply_monthly_fee(self): fee = self.monthly_fee() if fee > 0: self._balance -= fee return fee def __repr__(self): return f"{type(self).__name__}({self.account_id}, £{self._balance:,.2f})" # ── SavingsAccount ────────────────────────────────────────class SavingsAccount(Account): def __init__(self, account_id, owner, balance=0.0, rate=0.045): super().__init__(account_id, owner, balance) self.rate = rate def apply_interest(self): interest = round(self._balance * (self.rate / 12), 2) self._balance += interest return interest # ── CurrentAccount — fee waived above £1,000 ─────────────class CurrentAccount(Account): def __init__(self, account_id, owner, balance=0.0, overdraft=500.0): super().__init__(account_id, owner, balance) self.overdraft = overdraft def withdraw(self, amount): # OVERRIDE — allows overdraft if amount > self._balance + self.overdraft: raise ValueError(f"Overdraft limit £{self.overdraft:,.0f} exceeded") self._balance -= amount return self def monthly_fee(self): return 0.0 if self._balance >= 1_000 else 8.50 # ── PremiumAccount — no fees ──────────────────────────────class PremiumAccount(Account): pass # inherits everything; monthly_fee() returns 0 from base # ── Polymorphism in action ─────────────────────────────────accounts = [ SavingsAccount("SAV001", "Alice", 25_000, rate=0.048), SavingsAccount("SAV002", "Bob", 8_500, rate=0.045), CurrentAccount("CUR001", "Carol", 3_800), CurrentAccount("CUR002", "Dave", 750), # below threshold PremiumAccount("PRM001", "Eve", 120_000),] print(f"{'Account':<9} {'Type':<17} {'Balance':>12} {'Monthly'}")print("─" * 56)for acc in accounts: fee = acc.apply_monthly_fee() if isinstance(acc, SavingsAccount): interest = acc.apply_interest() label = f"+£{interest:.2f} interest" elif fee > 0: label = f"-£{fee:.2f} fee" else: label = "no charge" print(f"{acc.account_id:<9} {type(acc).__name__:<17} " f"£{acc.balance:>10,.2f} {label}") # Overdraft testcur = accounts[2]cur.withdraw(5_000) # into overdraftprint(f"\n{cur.account_id} overdraft: £{cur.balance:,.2f}")try: cur.withdraw(1_000) # exceeds limitexcept ValueError as e: print(f"Rejected: {e}")Integration — Mini Trading Portfolio System
This section assembles everything from the chapter into a cohesive system: custom exceptions for trading errors, CSV file logging for every order, and a Portfolio class that tracks cash, positions, and P&L — the same structure used in real algorithmic trading platforms.
Design principles demonstrated:
- Custom exceptions for each failure mode (InvalidSymbol, InsufficientCash, PositionNotFound)
- Encapsulation — cash and positions are private; only buy/sell can change them
- namedtuple for immutable order records — executed trades cannot be modified
- File logging — every order is appended to a CSV audit file on execution
- @property for computed unrealised P&L without storing stale data
from collections import namedtuple, defaultdictfrom datetime import datetimeimport csv # ── Exception hierarchy ────────────────────────────────────class TradingError(Exception): passclass InvalidSymbolError(TradingError): passclass InsufficientCashError(TradingError): passclass PositionNotFoundError(TradingError): pass # ── Immutable order record ─────────────────────────────────Order = namedtuple("Order", ["id", "symbol", "side", "qty", "price", "timestamp"]) # ── Portfolio class ────────────────────────────────────────class Portfolio: """Tracks cash, positions, and trade history. Logs all orders to CSV.""" VALID_SYMBOLS = {"AAPL", "TSLA", "MSFT", "NVDA", "META", "GOOGL"} def __init__(self, portfolio_id, cash=100_000.0): self.portfolio_id = portfolio_id self._cash = float(cash) self._positions = defaultdict(int) # symbol → qty self._avg_cost = {} # symbol → avg entry price self._history = [] self._counter = 0 with open("orders.csv", "w", newline="") as f: csv.writer(f).writerow(["id","symbol","side","qty","price","time"]) def buy(self, symbol, qty, price): sym = symbol.upper() if sym not in self.VALID_SYMBOLS: raise InvalidSymbolError(f"{sym} not in approved list") cost = qty * price if cost > self._cash: raise InsufficientCashError(f"Need £{cost:,.2f}, have £{self._cash:,.2f}") self._cash -= cost prev = self._positions[sym] self._avg_cost[sym] = (prev * self._avg_cost.get(sym, 0) + cost) / (prev + qty) self._positions[sym] += qty return self._record("BUY", sym, qty, price) def sell(self, symbol, qty, price): sym = symbol.upper() held = self._positions.get(sym, 0) if held < qty: raise PositionNotFoundError(f"Hold {held}x {sym}, cannot sell {qty}") self._cash += qty * price self._positions[sym] -= qty if self._positions[sym] == 0: del self._positions[sym]; del self._avg_cost[sym] return self._record("SELL", sym, qty, price) def _record(self, side, sym, qty, price): self._counter += 1 order = Order(f"ORD{self._counter:04d}", sym, side, qty, price, datetime.now().strftime("%H:%M:%S")) self._history.append(order) with open("orders.csv", "a", newline="") as f: csv.writer(f).writerow(list(order)) return order def unrealised_pnl(self, market_prices: dict) -> dict: return { sym: round((market_prices.get(sym, 0) - self._avg_cost[sym]) * qty, 2) for sym, qty in self._positions.items() } def report(self, market_prices=None): pnl = self.unrealised_pnl(market_prices or {}) print(f"\n{'═'*56}") print(f" Portfolio : {self.portfolio_id}") print(f" Cash : £{self._cash:>12,.2f}") if self._positions: print(f" Positions :") for sym, qty in sorted(self._positions.items()): p = pnl.get(sym, 0) print(f" {sym:6} {qty:4}x avg £{self._avg_cost[sym]:>8.2f}" f" P&L: {'+'if p>=0 else ''}£{p:>8,.2f}") print(f"{'═'*56}") # ── Run a trading session ──────────────────────────────────fund = Portfolio("ALPHA-FUND-1", cash=100_000)orders = [ ("buy", "AAPL", 100, 182.50), ("buy", "MSFT", 50, 415.25), ("buy", "NVDA", 30, 485.00), ("sell", "AAPL", 40, 195.00), # partial sell at profit] print("Order execution:")for action, sym, qty, price in orders: try: order = (fund.buy if action=="buy" else fund.sell)(sym, qty, price) print(f" ✓ {order.id}: {order.side:4} {order.qty:3}x {order.symbol:5} @ £{order.price:.2f}") except TradingError as e: print(f" ✗ {type(e).__name__}: {e}") # Invalid tradesfor bad in [("buy","XYZ",10,100), ("sell","GOOGL",5,150)]: try: (fund.buy if bad[0]=="buy" else fund.sell)(bad[1], bad[2], bad[3]) except (InvalidSymbolError, PositionNotFoundError) as e: print(f" ✗ {type(e).__name__}: {e}") fund.report(market_prices={"AAPL": 198.00, "MSFT": 430.00, "NVDA": 520.00}) print("\nAudit (orders.csv):")with open("orders.csv") as f: for row in csv.DictReader(f): print(f" {row['id']} {row['side']:4} {row['qty']:3}x " f"{row['symbol']:5} @ £{float(row['price']):.2f}")Practice Questions
Question 1
Which clause in a try/except block always executes, whether an exception occurred or not?
Question 2
What is the main advantage of raising InsufficientFundsError over a generic ValueError in a payment system?
Question 3
What does the single underscore in self._balance signal to other developers?
Question 4
What does @property allow you to do in Python?
Question 5
When writing a transaction audit log, why use open('audit.log', 'a') instead of open('audit.log', 'w')?
Question 6
What does super().__init__(...) do inside a subclass constructor?
Question 7
Which statement best describes polymorphism in the account hierarchy?
Question 8
What does 'with open("orders.csv") as f:' guarantee that f = open(...) alone does not?