Intermediate Topics
2025-10-09
OOP is a tool, not a religion. Use the minimum structure that makes change cheap.
Order vs Tradeself everywhere?selfthis; explicit is better than implicitPosition.add(pos, 10) is how pos.add(10) is resolved__new__(cls, ...) allocates a new instance (rarely overridden)__init__(self, ...) initializes the instance state__post_init__ exists for dataclasses to validate/compute derived fields__init__ simple; do not perform heavy I/O or network calls here__repr__, __str__ — debugging and display__len__, __iter__, __contains__ — collection-like behavior__eq__, __lt__, __hash__ — comparisons and dict/set keys__enter__, __exit__ — context manager protocolclass Quote:
def __init__(self, bids: list[float], asks: list[float]):
self.bids = bids
self.asks = asks
def __repr__(self): # helpful for debugging
return f"Quote(bid={max(self.bids, default=None)}, ask={min(self.asks, default=None)})"
def __len__(self): # treat as sized container of price levels
return len(self.bids) + len(self.asks)__repr__ early—it pays off in logs and notebooksdataclasses__init__, __repr__, comparisons; supports frozen=True for immutabilityslots=True in Python 3.10+ dataclasses for memory savings on many instancesif-elif chainsabc module to define interfacesabstractmethod is a method that must be implemented by subclassesfrom abc import ABC, abstractmethod
class FeeModel(ABC):
@abstractmethod
def fee(self, notional: float) -> float: ...
class FixedBpsFee(FeeModel):
def __init__(self, bps: float):
self.bps = bps
def fee(self, notional: float) -> float:
return notional * self.bps / 10_000
class MinFloorFee(FeeModel):
def __init__(self, bps: float, minimum: float):
self.bps = bps
self.minimum = minimum
def fee(self, notional: float) -> float:
return max(notional * self.bps / 10_000, self.minimum)
class Broker:
def __init__(self, fee_model: FeeModel):
self.fee_model = fee_model
def commission(self, notional: float) -> float:
return self.fee_model.fee(notional)Broker(FixedBpsFee(1.2)) vs Broker(MinFloorFee(0.8, 2.0))class JsonMixin:
def to_json(self) -> str:
import json
return json.dumps(self.__dict__)
class CsvMixin:
def to_csv(self) -> str:
return ",".join(f"{k}={v}" for k, v in self.__dict__.items())
class Portfolio(JsonMixin, CsvMixin):
def __init__(self, name: str, positions: dict[str, int]):
self.name = name
self.positions = positionsPortfolio.mro() → [Portfolio, JsonMixin, CsvMixin, object]class BrokerageAccount:
def __init__(self):
self._balance = 0.0 # protected by convention, _ attributes are considered "private"
@property
def balance(self) -> float:
return self._balance
def deposit(self, amount: float) -> None:
if amount <= 0:
raise ValueError("deposit must be positive")
self._balance += amount
def withdraw(self, amount: float) -> None:
if amount <= 0:
raise ValueError("withdraw must be positive")
if amount > self._balance:
raise ValueError("insufficient funds")
self._balance -= amountbalance >= 0_protected, __name for name-mangling)OrderRequest (dataclass)FeeModel strategiesBrokerageAccount for cash managementfrom dataclasses import dataclass
@dataclass
class Fill:
""" A filled order/trade """
symbol: str
qty: int
price: float
class ExecutionEngine:
"""
Simple execution engine that processes orders through a broker and updates an account.
"""
def __init__(self, broker: Broker, account: BrokerageAccount):
self.broker = broker
self.account = account
def execute(self, req: OrderRequest, price: float) -> Fill:
notional = abs(req.qty) * price
fee = self.broker.commission(notional)
cash_change = -notional - fee if req.side == "BUY" else notional - fee
if self.account.balance + cash_change < 0:
raise ValueError("order would make balance negative")
if cash_change > 0:
self.account.deposit(cash_change)
else:
self.account.withdraw(-cash_change)
return Fill(req.symbol, req.qty, price)What is a backtester?
What classes would you create for a backtester? Be creative
DataHandler class, examplefrom abc import ABC, abstractmethod
import pandas as pd
from typing import Dict, List # typing is a module that provides support for type hints
class DataHandler(ABC):
""" Abstract base class for data handling """
@abstractmethod
def get_latest_bar(self, symbol: str) -> pd.Series:
""" Returns the latest bar for a given symbol """
pass
@abstractmethod
def get_latest_bars(self, symbol: str, N: int = 1) -> List[pd.Series]:
""" Returns the latest N bars for a given symbol """
pass
@abstractmethod
def update_bars(self) -> None:
""" Updates the bars to the next time step """
passDataHandlerclass HistoricCSVDataHandler(DataHandler):
""" Handles data from CSV files """
def __init__(self, csv_dir: str, symbols: List[str]):
self.csv_dir = csv_dir
self.symbols = symbols
self.data: Dict[str, pd.DataFrame] = {} # type: ignore
self.latest_data: Dict[str, List[pd.Series]] = {symbol: [] for symbol in symbols}
self._load_data()
def _load_data(self) -> None:
""" Loads data from CSV files into the data dictionary """
for symbol in self.symbols:
file_path = f"{self.csv_dir}/{symbol}.csv"
self.data[symbol] = pd.read_csv(file_path, index_col="date", parse_dates=True)
self.data[symbol].sort_index(inplace=True)
self.latest_data[symbol] = []
self._update_latest_data(symbol)
def _update_latest_data(self, symbol: str) -> None:
""" Updates the latest data for a given symbol """
self.latest_data[symbol] = self.data[symbol].iloc[-1].to_dict()
def get_latest_bar(self, symbol: str) -> pd.Series:
""" Returns the latest bar for a given symbol """
return pd.Series(self.latest_data[symbol])
def get_latest_bars(self, symbol: str, N: int = 1) -> List[pd.Series]:
""" Returns the latest N bars for a given symbol """
return [pd.Series(self.latest_data[symbol]) for _ in range(N)]
def update_bars(self) -> None:
""" Updates the bars to the next time step """
for symbol in self.symbols:
if not self.data[symbol].empty:
self.latest_data[symbol] = self.data[symbol].iloc[-1].to_dict()
self.data[symbol] = self.data[symbol].iloc[:-1]Strategy class exampleLet’s first focus on the Signal.
from abc import ABC, abstractmethod
from typing import Dict, List
import pandas as pd
from dataclasses import dataclass
from enum import Enum # Enum is a module that provides support for enumerations
class SignalType(Enum):
""" Enumeration for signal types """
BUY = "BUY"
SELL = "SELL"
HOLD = "HOLD"
EXIT = "EXIT"
@dataclass
class Signal:
""" Represents a trading signal """
symbol: str
signal_type: SignalType
price: float
volume: float
def __post_init__(self):
if self.volume <= 0:
raise ValueError("Volume must be positive")
if self.signal_type not in SignalType:
raise ValueError(f"Invalid signal type: {self.signal_type}")Strategy class example (continued)class MovingAverageCrossoverStrategy(Strategy):
""" Simple moving average crossover strategy """
def __init__(self, short_window: int = 40, long_window: int = 100):
self.short_window = short_window
self.long_window = long_window
def generate_signals(self, data: Dict[str, pd.DataFrame]) -> List[Signal]:
signals = []
for symbol, df in data.items():
df['short_mavg'] = df['close'].rolling(window=self.short_window, min_periods=1).mean()
df['long_mavg'] = df['close'].rolling(window=self.long_window, min_periods=1).mean()
if df['short_mavg'].iloc[-1] > df['long_mavg'].iloc[-1]:
signals.append(Signal(symbol=symbol, signal_type=SignalType.BUY, price=df['close'].iloc[-1], volume=100))
elif df['short_mavg'].iloc[-1] < df['long_mavg'].iloc[-1]:
signals.append(Signal(symbol=symbol, signal_type=SignalType.SELL, price=df['close'].iloc[-1], volume=100))
else:
signals.append(Signal(symbol=symbol, signal_type=SignalType.HOLD, price=df['close'].iloc[-1], volume=0))
return signalsfrom dataclasses import dataclass, field
from typing import Dict
@dataclass
class Portfolio:
cash: float = 0.0
positions: Dict[str, int] = field(default_factory=dict)
def qty(self, symbol: str) -> int:
return self.positions.get(symbol, 0)
def apply_fill(self, fill: Fill, fee: float) -> None:
sign = 1 if fill.qty > 0 else -1
notional = abs(fill.qty) * fill.price
# BUY reduces cash, SELL increases cash
self.cash += (-notional - fee) if sign > 0 else (notional - fee)
self.positions[fill.symbol] = self.qty(fill.symbol) + fill.qty
if self.positions[fill.symbol] == 0:
self.positions.pop(fill.symbol)
def market_value(self, prices: Dict[str, float]) -> float:
return sum(self.qty(sym) * prices.get(sym, 0.0) for sym in self.positions)
def equity(self, prices: Dict[str, float]) -> float:
return self.cash + self.market_value(prices)from typing import Tuple
class BacktestBroker:
def __init__(self, fee_model: FeeModel, slippage_bps: float = 0.0):
self.fee_model = fee_model
self.slippage_bps = slippage_bps
def _apply_slippage(self, price: float, side: SignalType) -> float:
# BUY pays up, SELL gets less
adj = price * (self.slippage_bps / 10_000.0)
return price + adj if side == SignalType.BUY else price - adj
def order_for_signal(self, sig: Signal) -> OrderRequest:
if sig.signal_type == SignalType.HOLD or sig.volume == 0:
return OrderRequest(symbol=sig.symbol, qty=0, side="HOLD")
side = "BUY" if sig.signal_type == SignalType.BUY else "SELL"
qty = int(sig.volume) if side == "BUY" else -int(sig.volume)
return OrderRequest(symbol=sig.symbol, qty=qty, side=side)
def execute(self, req: OrderRequest, mkt_price: float) -> Tuple[Fill, float]:
if req.qty == 0:
return Fill(req.symbol, 0, mkt_price), 0.0
traded_price = self._apply_slippage(mkt_price, SignalType.BUY if req.qty > 0 else SignalType.SELL)
notional = abs(req.qty) * traded_price
fee = self.fee_model.fee(notional)
return Fill(req.symbol, req.qty, traded_price), feeimport math
from typing import List, Dict
class PerformanceMetrics:
def __init__(self, equity_curve: List[float]):
self.equity = equity_curve
self.returns = [0.0] + [
(self.equity[i] / self.equity[i-1] - 1.0) for i in range(1, len(self.equity))
]
def summary(self, periods_per_year: int = 252) -> Dict[str, float]:
if not self.equity or len(self.equity) < 2:
return {"total_return": 0.0, "vol": 0.0, "sharpe": 0.0, "max_dd": 0.0}
total_return = self.equity[-1] / self.equity[0] - 1.0
mean = sum(self.returns[1:]) / max(1, len(self.returns) - 1)
var = sum((r - mean) ** 2 for r in self.returns[1:]) / max(1, len(self.returns) - 2)
vol = math.sqrt(var) * math.sqrt(periods_per_year)
sharpe = (mean * periods_per_year) / vol if vol > 0 else 0.0
# max drawdown
peak = self.equity[0]
max_dd = 0.0
for v in self.equity:
peak = max(peak, v)
dd = (v / peak) - 1.0
max_dd = min(max_dd, dd)
return {
"total_return": total_return,
"vol": vol,
"sharpe": sharpe,
"max_dd": max_dd,
}from typing import Dict, List
class Backtester:
def __init__(self, data: Dict[str, pd.DataFrame], strategy: Strategy, broker: BacktestBroker, portfolio: Portfolio):
self.data = data
self.strategy = strategy
self.broker = broker
self.portfolio = portfolio
self.equity_curve: List[float] = []
def _prices_snapshot(self, t: int) -> Dict[str, float]:
return {sym: df['close'].iloc[t] for sym, df in self.data.items() if t < len(df)}
def run(self) -> PerformanceMetrics:
# assume all symbols share the same index length for brevity
T = min(len(df) for df in self.data.values())
for t in range(T):
# 1) Generate signals using history up to t
hist = {sym: df.iloc[: t + 1] for sym, df in self.data.items()}
signals = self.strategy.generate_signals(hist)
prices = self._prices_snapshot(t)
# 2) Convert to order requests and execute
for sig in signals:
req = self.broker.order_for_signal(sig)
if req.qty == 0:
continue
mkt_px = prices.get(req.symbol)
if mkt_px is None:
continue
fill, fee = self.broker.execute(req, mkt_px)
self.portfolio.apply_fill(fill, fee)
# 3) Record equity
self.equity_curve.append(self.portfolio.equity(prices))
return PerformanceMetrics(self.equity_curve)# Inputs
symbols = ["AAPL", "MSFT"]
data = {s: pd.DataFrame({"close": pd.Series([100, 101, 102, 101, 103])}) for s in symbols}
strategy = MovingAverageCrossoverStrategy(short_window=2, long_window=3)
broker = BacktestBroker(FixedBpsFee(1.0), slippage_bps=2.0)
portfolio = Portfolio(cash=100_000)
bt = Backtester(data, strategy, broker, portfolio)
metrics = bt.run().summary()
print({k: round(v, 4) for k, v in metrics.items()})Tip: Swap the fee model or tweak slippage to stress test costs; or add a risk cap mixin that vetoes oversize orders.