When I first built a momentum strategy in Python, I wrote it with pandas in an afternoon. Returns looked spectacular: 34% annualized, Sharpe of 2.1. I was ready to deploy.
Then I tried to model limit orders. And suddenly my vectorized backtest could not represent the idea of "wait for price to return to entry." It could not simulate fills partially, or charge realistic commissions on fills versus cancellations. My Sharpe of 2.1 was fiction — a consequence of peeking into the future and assuming every order filled at the closing price.
The vectorized backtest had optimized the wrong thing: simulation speed, not simulation fidelity.
This is the fork in the road every quant strategy developer eventually faces. Vectorized backtesting is fast, expressive, and well-suited for idea validation. But the moment your strategy involves contingent orders — anything more complex than "buy at market close, hold for N days" — you need an event-driven architecture. This article builds that architecture from first principles, implements a production-grade order state machine, and demonstrates a matching engine that models realistic fills.
The Fundamental Trade-off: Vectorized vs. Event-Driven
Before writing a line of code, it is worth understanding precisely what each paradigm optimizes for — and where each one fails.
Vectorized Backtesting: Speed at the Cost of Fidelity
Vectorized backtesting processes all historical data as numpy arrays or pandas DataFrames. Signals are computed on vector-aligned columns, and performance is excellent because the entire history is available to every calculation simultaneously.
This creates a structural problem: the strategy sees the future. If you write:
returns = prices.pct_change()
signal = returns.rolling(20).mean() > 0
position = signal.shift(1) # "shift" looks like avoiding look-ahead bias
The .shift(1) operation is an approximation of proper timing. It shifts the signal array by one row — but that shift happens in a batch, across the entire history, not in time sequence. During a live backtest, you would not know tomorrow's price to generate today's signal. In a vectorized backtest, you have already "seen" all prices.
Vectorized backtesting also struggles with:
- State-dependent orders: A stop-loss that activates only after a position is open requires state tracking that arrays do not naturally express.
- Partial fills: Simulating a market order that fills 60% of the desired quantity at one price and the remainder at another price requires sequential processing.
- Market impact: When your own orders move the market, you need a feedback loop that vectorized frameworks cannot provide without ugly hacks.
Event-Driven Backtesting: Fidelity at the Cost of Complexity
Event-driven backtesting processes history chronologically, one timestamp at a time. At each step, the engine:
- Receives new market data (tick or bar).
- Processes any pending events (order status updates, fills).
- Evaluates strategy logic to generate new signals.
- Sends orders to the execution layer.
This is structurally identical to a live trading system — the difference is only the data source (historical vs. real-time). The engine maintains explicit state: open positions, pending orders, account balance, unrealized PnL. Every action has a consequence that the engine tracks.
The cost is performance. Sequential processing cannot leverage numpy's vectorized operations. A 10-year backtest with minute-level data can take minutes instead of milliseconds.
The rule of thumb: Use vectorized backtesting for hypothesis generation and parameter sweeps. Use event-driven backtesting for strategy validation, signal timing verification, and any strategy where order lifecycle matters.
Architecture: The Six-Core Event System
An event-driven backtesting engine has six loosely coupled components, each with a single responsibility.
┌─────────────────────────────────────────────────────────────────┐
│ EVENT LOOP │
│ while not exhausted(data_feed): │
│ event = event_queue.get_next() │
│ dispatcher.route(event) │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ DataHandler → SignalEngine → Portfolio → Execution → Market │
└─────────────────────────────────────────────────────────────────┘
| Component | Responsibility |
|---|---|
EventQueue |
Maintains chronological ordering of all events (data, signal, order, fill, report) |
DataHandler |
Normalizes market data (bars, ticks) into a consistent event stream |
SignalEngine |
Consumes data events, evaluates strategy logic, emits SignalEvent |
Portfolio |
Tracks positions, computes equity curve, emits OrderEvent |
ExecutionHandler |
Simulates order routing and fill — the matching engine lives here |
PerformanceAnalyzer |
Computes metrics (Sharpe, drawdown, win rate) from fill events |
Implementing the Event Queue
The event queue is the backbone of the entire system. It must support:
- FIFO ordering: Events at the same timestamp are processed in insertion order.
- Event typing: The dispatcher routes by event type.
- Blocking get: The event loop blocks when the queue is empty.
import queue
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum, auto
from typing import Optional, Any
import heapq
class EventType(Enum):
MARKET = auto() # New price data available
SIGNAL = auto() # Strategy signal generated
ORDER = auto() # Order submitted
FILL = auto() # Order executed (fully or partially)
CANCEL = auto() # Order cancelled
REPORT = auto() # End-of-period performance snapshot
@dataclass(order=True)
class Event:
"""Base event class. Ordering is by (timestamp, sequence_number)."""
timestamp: datetime
sequence: int = field(compare=False)
event_type: EventType = field(compare=False)
payload: Any = field(compare=False, default=None)
event_id: str = field(compare=False, default_factory=lambda: uuid.uuid4().hex[:12])
def __post_init__(self):
# Heapq in Python is a min-heap; we want chronological ordering.
# Store negative timestamp to make earlier events pop first.
self._sort_key = self.timestamp
class EventQueue:
"""
Thread-safe, timestamp-ordered event queue.
Uses a heap-based priority queue internally for O(log n) insertion
and O(1) minimum retrieval. Events are immutable after insertion.
"""
_counter: int = 0
def __init__(self):
self._heap: list[Event] = []
self._sequence: int = 0
self._closed: bool = False
def put(self, timestamp: datetime, event_type: EventType, payload: Any = None) -> str:
"""Insert an event into the queue. Returns the event_id."""
if self._closed:
raise RuntimeError("Cannot add events to a closed queue")
EventQueue._sequence += 1
event = Event(
timestamp=timestamp,
sequence=EventQueue._sequence,
event_type=event_type,
payload=payload
)
heapq.heappush(self._heap, event)
return event.event_id
def get(self, block: bool = True, timeout: Optional[float] = None) -> Optional[Event]:
"""
Retrieve the next event by timestamp.
Args:
block: If True, block until an event is available.
timeout: Maximum seconds to wait (None = wait forever).
Returns:
The earliest event, or None if the queue is empty and not blocking.
Raises:
queue.Empty: If timeout expires without an event.
"""
if not self._heap:
if not block:
return None
raise queue.Empty("Event queue is empty")
return heapq.heappop(self._heap)
def peek(self) -> Optional[Event]:
"""View the next event without removing it."""
if not self._heap:
return None
return self._heap[0]
def close(self):
"""Permanently close the queue to new events."""
self._closed = True
def __len__(self) -> int:
return len(self._heap)
def __bool__(self) -> bool:
return bool(self._heap)
⚠️ Engineering note: This queue implementation uses a heap for timestamp ordering. If your backtest requires sub-millisecond event resolution or cross-thread communication, replace this with a lock-free ring buffer or use
asyncio.PriorityQueue. For standard historical backtesting, the heap is sufficient.
The Order State Machine
Orders are not binary (sent vs. filled). They have a rich lifecycle with transitions that must be tracked explicitly.
┌──────────┐
│ PENDING │ ← Order created, not yet submitted to broker
└────┬─────┘
│ submit()
▼
┌──────────┐
│ SUBMITTED│ ← Sent to exchange / execution layer
└────┬─────┘
│
├── cancel() → CANCELLED
│
├── fill(partial=True) → PARTIAL_FILL ──→ (more fills) ──→ FILLED
│
└── fill(full=True) → FILLED
from enum import Enum, auto
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
import uuid
class OrderStatus(Enum):
PENDING = auto()
SUBMITTED = auto()
PARTIAL_FILL = auto()
FILLED = auto()
CANCELLED = auto()
REJECTED = auto()
EXPIRED = auto()
class OrderSide(Enum):
BUY = auto()
SELL = auto()
class OrderType(Enum):
MARKET = auto()
LIMIT = auto()
STOP = auto()
STOP_LIMIT = auto()
@dataclass
class Order:
"""
Order entity with explicit state machine.
State transitions are validated at runtime. Attempting an
invalid transition raises StateTransitionError.
"""
symbol: str
side: OrderSide
order_type: OrderType
quantity: float
price: Optional[float] = None # For limit/stop orders
stop_price: Optional[float] = None # For stop-limit orders
timestamp: datetime = field(default_factory=datetime.now)
order_id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
# Mutable state — tracked separately for clarity
_status: OrderStatus = field(default=OrderStatus.PENDING, repr=False)
_filled_quantity: float = field(default=0.0, repr=False)
_avg_fill_price: float = field(default=0.0, repr=False)
_last_update: datetime = field(default_factory=datetime.now, repr=False)
_fill_history: list[dict] = field(default_factory=list, repr=False)
# Valid transitions map: current_status -> set of valid next statuses
_TRANSITIONS: dict[OrderStatus, set[OrderStatus]] = {
OrderStatus.PENDING: {OrderStatus.SUBMITTED, OrderStatus.CANCELLED},
OrderStatus.SUBMITTED: {OrderStatus.PARTIAL_FILL, OrderStatus.FILLED,
OrderStatus.CANCELLED, OrderStatus.REJECTED,
OrderStatus.EXPIRED},
OrderStatus.PARTIAL_FILL:{OrderStatus.PARTIAL_FILL, OrderStatus.FILLED,
OrderStatus.CANCELLED},
OrderStatus.FILLED: set(), # Terminal state
OrderStatus.CANCELLED: set(), # Terminal state
OrderStatus.REJECTED: set(), # Terminal state
OrderStatus.EXPIRED: set(), # Terminal state
}
@property
def status(self) -> OrderStatus:
return self._status
@property
def filled_quantity(self) -> float:
return self._filled_quantity
@property
def remaining_quantity(self) -> float:
return self.quantity - self._filled_quantity
@property
def avg_fill_price(self) -> float:
return self._avg_fill_price
def _transition_to(self, new_status: OrderStatus) -> None:
"""Validate and execute a state transition."""
valid = self._TRANSITIONS.get(self._status, set())
if new_status not in valid:
raise InvalidStateTransition(
f"Cannot transition from {self._status.name} to {new_status.name}"
)
self._status = new_status
self._last_update = datetime.now()
def submit(self) -> None:
"""Submit the order to the execution layer."""
self._transition_to(OrderStatus.SUBMITTED)
def fill(self, quantity: float, price: float, timestamp: datetime) -> None:
"""
Record a partial or full fill.
Args:
quantity: Number of shares/units filled in this event.
price: Execution price.
timestamp: Time of the fill.
"""
if quantity <= 0:
raise ValueError(f"Fill quantity must be positive, got {quantity}")
if quantity > self.remaining_quantity:
raise ValueError(
f"Fill quantity {quantity} exceeds remaining {self.remaining_quantity}"
)
# Update weighted average fill price
total_cost = (self._filled_quantity * self._avg_fill_price) + (quantity * price)
self._filled_quantity += quantity
self._avg_fill_price = total_cost / self._filled_quantity if self._filled_quantity > 0 else 0
# Record fill event
self._fill_history.append({
"timestamp": timestamp,
"quantity": quantity,
"price": price,
"remaining": self.remaining_quantity
})
# Determine new status
if self._filled_quantity >= self.quantity:
self._transition_to(OrderStatus.FILLED)
elif self._filled_quantity > 0:
self._transition_to(OrderStatus.PARTIAL_FILL)
self._last_update = timestamp
def cancel(self) -> None:
"""Cancel the order if it has not been fully filled."""
self._transition_to(OrderStatus.CANCELLED)
def reject(self, reason: str) -> None:
"""Reject the order (e.g., insufficient margin)."""
self._transition_to(OrderStatus.REJECTED)
def is_terminal(self) -> bool:
return self.status in {
OrderStatus.FILLED,
OrderStatus.CANCELLED,
OrderStatus.REJECTED,
OrderStatus.EXPIRED
}
def is_active(self) -> bool:
return self.status in {
OrderStatus.PENDING,
OrderStatus.SUBMITTED,
OrderStatus.PARTIAL_FILL
}
class InvalidStateTransition(Exception):
"""Raised when an order receives a command that is not valid for its current state."""
pass
The state machine above is not merely academic. Consider this scenario: your strategy submits a limit order to buy 1,000 shares of AAPL at $150.00. The market gaps down and fills 600 shares at $149.80. In a poorly designed backtest, you might accidentally "fill" the remaining 400 shares at $149.80 again — violating the state machine's invariant that you cannot fill more than remaining_quantity. The state machine catches this at runtime and raises a ValueError.
The Matching Engine: Simulating Realistic Fills
The matching engine determines whether and at what price orders fill. A naive matching engine fills all orders at the bar's close price. A realistic matching engine respects price-time priority, models slippage, and handles partial fills.
Price-Time Priority
In a price-time priority queue, orders are ranked first by price (best bid/ask), then by submission time. A buy order at $150.00 fills before a buy order at $149.99. Among orders at the same price, the earlier submission wins.
Market Order Matching
Market orders always fill against the best available counterparty price. For a buy market order, this is the best ask. For a sell market order, this is the best bid.
@dataclass
class Level:
"""A single price level in the order book."""
price: float
quantity: float
order_id: str
class OrderBookSide:
"""One side of a limit order book (bid or ask)."""
def __init__(self, is_bid: bool):
self.is_bid = is_bid
# Price -> list of (timestamp, order_id, quantity) tuples
self._levels: dict[float, list[tuple]] = {}
def add(self, price: float, quantity: float, order_id: str, timestamp: datetime) -> None:
if price not in self._levels:
self._levels[price] = []
self._levels[price].append((timestamp, order_id, quantity))
def best_price(self) -> Optional[float]:
if not self._levels:
return None
if self.is_bid:
return max(self._levels.keys())
else:
return min(self._levels.keys())
def best_level(self) -> Optional[Level]:
price = self.best_price()
if price is None:
return None
# Aggregate all orders at the best price
total_qty = sum(qty for _, _, qty in self._levels[price])
# Get the earliest order_id at this level
earliest_order = min(self._levels[price], key=lambda x: x[0])
return Level(price=price, quantity=total_qty, order_id=earliest_order[1])
def consume(self, price: float, quantity: float) -> list[tuple[str, float]]:
"""
Attempt to fill `quantity` at `price` (or better).
Returns a list of (order_id, filled_quantity) tuples.
"""
filled = []
remaining = quantity
if self.is_bid:
prices = sorted(self._levels.keys(), reverse=True) # Best bid first
else:
prices = sorted(self._levels.keys()) # Best ask first
for p in prices:
if self.is_bid and p > price: # Price better than limit — fill
pass
elif not self.is_bid and p < price: # Price better than limit — fill
pass
else:
continue # Skip prices outside the limit
while remaining > 0 and self._levels[p]:
ts, order_id, qty = self._levels[p][0]
fill_qty = min(qty, remaining)
filled.append((order_id, fill_qty))
remaining -= fill_qty
if fill_qty == qty:
self._levels[p].pop(0)
else:
self._levels[p][0] = (ts, order_id, qty - fill_qty)
if not self._levels[p]:
del self._levels[p]
if remaining <= 0:
break
return filled
class MatchingEngine:
"""
Price-time priority matching engine with slippage modeling.
Supports:
- Market orders (fill at best counterparty price + slippage)
- Limit orders (fill at limit price or better, if volume available)
- Stop orders (trigger to market order when stop price crossed)
- Partial fills (split across multiple price levels)
Slippage model: Market orders fill at (best_price * (1 + slippage_pct))
for buys, or (best_price * (1 - slippage_pct)) for sells.
"""
def __init__(self, slippage_pct: float = 0.0005):
"""
Args:
slippage_pct: Slippage as a fraction of price. Default 5 bps (0.05%).
This is conservative for liquid large-caps; increase for
illiquid assets or large order sizes.
"""
self.slippage_pct = slippage_pct
self._bids = OrderBookSide(is_bid=True)
self._asks = OrderBookSide(is_bid=False)
self._pending_stop_orders: list[Order] = []
self._fills: list[dict] = []
def seed_order_book(self, bids: list[tuple[float, float]], asks: list[tuple[float, float]],
timestamp: datetime) -> None:
"""
Seed the order book with a snapshot of bid/ask levels.
Replaces any existing levels.
Args:
bids: List of (price, quantity) tuples for bids.
asks: List of (price, quantity) tuples for asks.
timestamp: Snapshot timestamp.
"""
self._bids = OrderBookSide(is_bid=True)
self._asks = OrderBookSide(is_bid=False)
for price, qty in bids:
self._bids.add(price, qty, f"seed_bid_{price}", timestamp)
for price, qty in asks:
self._asks.add(price, qty, f"seed_ask_{price}", timestamp)
def update_book(self, side: str, price: float, quantity: float,
timestamp: datetime) -> None:
"""Update a single price level (add or reduce quantity)."""
book_side = self._bids if side == "bid" else self._asks
if quantity > 0:
book_side.add(price, quantity, f"update_{timestamp.isoformat()}", timestamp)
# If quantity == 0, the level is exhausted; the next match will skip it.
def process_order(self, order: Order, timestamp: datetime) -> list[dict]:
"""
Process an incoming order against the current order book.
Returns a list of fill records: [{order_id, symbol, side, quantity, price, timestamp}]
"""
fills = []
if order.order_type == OrderType.STOP:
# Stop orders are held until triggered
self._pending_stop_orders.append(order)
self._check_stop_triggers(timestamp)
return fills
# Determine the counterparty side
if order.side == OrderSide.BUY:
counterparty = self._asks
book_side = self._bids
else:
counterparty = self._bids
book_side = self._asks
# For market orders, check stop triggers first (market orders can trigger stops)
best = counterparty.best_level()
if best is None:
# No liquidity — order cannot fill
return fills
# Calculate effective price (limit price or best counterparty price)
if order.order_type == OrderType.MARKET:
# Apply slippage
if order.side == OrderSide.BUY:
effective_price = best.price * (1 + self.slippage_pct)
else:
effective_price = best.price * (1 - self.slippage_pct)
elif order.order_type == OrderType.LIMIT:
if order.side == OrderSide.BUY and best.price <= order.price:
effective_price = best.price
elif order.side == OrderSide.SELL and best.price >= order.price:
effective_price = best.price
else:
# Limit price not met — add to book
book_side.add(order.price, order.remaining_quantity, order.order_id, timestamp)
return fills
else:
effective_price = best.price
# Attempt to fill
remaining = order.remaining_quantity
while remaining > 0:
best_level = counterparty.best_level()
if best_level is None:
break
# For limit orders, check price improvement
if order.order_type == OrderType.LIMIT:
if order.side == OrderSide.BUY and best_level.price > order.price:
break
if order.side == OrderSide.SELL and best_level.price < order.price:
break
fill_qty = min(remaining, best_level.quantity)
fills.append({
"order_id": order.order_id,
"symbol": order.symbol,
"side": order.side.name,
"quantity": fill_qty,
"price": effective_price,
"timestamp": timestamp
})
remaining -= fill_qty
# Update the book (reduce consumed quantity)
# In a full implementation, you would update the level directly.
# Here we consume from the best level.
best_level.quantity -= fill_qty
return fills
def _check_stop_triggers(self, timestamp: datetime) -> None:
"""Check and trigger any stop orders whose conditions are now met."""
triggered = []
for order in self._pending_stop_orders:
if order.order_type == OrderType.STOP:
best = self._asks.best_level() if order.side == OrderSide.BUY else self._bids.best_level()
if best is None:
continue
if order.side == OrderSide.BUY and best.price >= order.stop_price:
triggered.append(order)
elif order.side == OrderSide.SELL and best.price <= order.stop_price:
triggered.append(order)
for order in triggered:
self._pending_stop_orders.remove(order)
order.order_type = OrderType.MARKET
⚠️ Engineering note: This matching engine is single-threaded and operates on in-memory state. For backtests with millions of orders, consider persisting fill state periodically and implementing a batch-matching mode that processes orders in timestamp groups. The current implementation is designed for clarity, not raw throughput.
Bringing It Together: A Complete Backtest Run
The following example demonstrates all six components working together on a simple mean-reversion strategy against synthetic US equity data.
import random
from datetime import datetime, timedelta
class SimpleBacktester:
"""
End-to-end event-driven backtester combining all components.
Strategy: Mean-reversion on a synthetic single-stock time series.
- Buy when price drops 2% below the 20-bar rolling mean.
- Sell when price rises 1% above the 20-bar rolling mean (or at stop).
- Stop-loss: 3% below entry.
"""
def __init__(self, initial_capital: float = 100_000.0,
commission: float = 0.001, slippage_pct: float = 0.0005):
self.capital = initial_capital
self.commission = commission # 10 bps per trade
self.engine = MatchingEngine(slippage_pct=slippage_pct)
self.orders: dict[str, Order] = {}
self.positions: dict[str, float] = {} # symbol -> quantity
self.equity_curve: list[dict] = []
self.cash = initial_capital
# Rolling window for mean-reversion signal
self.price_history: list[float] = []
self.window = 20
def on_bar(self, timestamp: datetime, bar: dict) -> None:
"""Process a single bar of market data."""
symbol = bar["symbol"]
close = bar["close"]
bid = bar.get("bid", close * 0.9995)
ask = bar.get("ask", close * 1.0005)
# Update order book snapshot for matching
self.engine.seed_order_book(
bids=[(bid, 5000)],
asks=[(ask, 5000)],
timestamp=timestamp
)
# Update price history for signal
self.price_history.append(close)
if len(self.price_history) > self.window + 1:
self.price_history.pop(0)
# Compute mean-reversion signal
signal = self._compute_signal(close)
# Process pending orders (check fills)
self._process_fills(timestamp)
# Generate new orders based on signal
self._generate_orders(symbol, close, signal, timestamp)
# Record equity
self._record_equity(timestamp, close)
def _compute_signal(self, price: float) -> str:
"""Compute mean-reversion signal."""
if len(self.price_history) < self.window:
return "HOLD"
rolling_mean = sum(self.price_history[-self.window:]) / self.window
deviation = (price - rolling_mean) / rolling_mean
if deviation < -0.02:
return "BUY"
elif deviation > 0.01:
return "SELL"
return "HOLD"
def _process_fills(self, timestamp: datetime) -> None:
"""Check pending orders for fills against current market."""
to_remove = []
for order_id, order in self.orders.items():
if not order.is_active():
to_remove.append(order_id)
continue
fills = self.engine.process_order(order, timestamp)
for fill in fills:
self._apply_fill(order, fill)
for oid in to_remove:
del self.orders[oid]
def _apply_fill(self, order: Order, fill: dict) -> None:
"""Apply a fill to the portfolio."""
cost = fill["quantity"] * fill["price"]
commission_cost = cost * self.commission
if order.side == OrderSide.BUY:
self.cash -= (cost + commission_cost)
self.positions[order.symbol] = self.positions.get(order.symbol, 0) + fill["quantity"]
else:
self.cash += (cost - commission_cost)
self.positions[order.symbol] = self.positions.get(order.symbol, 0) - fill["quantity"]
order.fill(fill["quantity"], fill["price"], fill["timestamp"])
def _generate_orders(self, symbol: str, price: float, signal: str,
timestamp: datetime) -> None:
"""Generate orders based on signal and current position."""
pos = self.positions.get(symbol, 0)
if signal == "BUY" and pos == 0:
order = Order(
symbol=symbol,
side=OrderSide.BUY,
order_type=OrderType.MARKET,
quantity=100,
timestamp=timestamp
)
order.submit()
self.orders[order.order_id] = order
elif signal == "SELL" and pos > 0:
order = Order(
symbol=symbol,
side=OrderSide.SELL,
order_type=OrderType.MARKET,
quantity=pos,
timestamp=timestamp
)
order.submit()
self.orders[order.order_id] = order
def _record_equity(self, timestamp: datetime, close: float) -> None:
position_value = sum(
qty * close for symbol, qty in self.positions.items()
)
equity = self.cash + position_value
self.equity_curve.append({
"timestamp": timestamp,
"equity": equity,
"cash": self.cash,
"position_value": position_value
})
def run(self, bars: list[dict]) -> dict:
"""Execute the backtest over a list of bars."""
for bar in bars:
timestamp = bar.get("timestamp", datetime.now())
self.on_bar(timestamp, bar)
return self.get_performance()
def get_performance(self) -> dict:
"""Compute performance metrics from the equity curve."""
if not self.equity_curve:
return {}
equity = [e["equity"] for e in self.equity_curve]
returns = [(equity[i] - equity[i-1]) / equity[i-1]
for i in range(1, len(equity))]
total_return = (equity[-1] - equity[0]) / equity[0]
sharpe = (sum(returns) / len(returns) / _std(returns) * (252 ** 0.5)
if _std(returns) > 0 else 0)
max_dd = _max_drawdown(equity)
return {
"initial_capital": equity[0],
"final_equity": equity[-1],
"total_return": total_return,
"annualized_return": total_return * (252 / len(equity)),
"sharpe_ratio": sharpe,
"max_drawdown": max_dd,
"num_trades": len([o for o in self.orders.values() if o.status == OrderStatus.FILLED])
}
def _std(data: list[float]) -> float:
if not data:
return 0.0
mean = sum(data) / len(data)
variance = sum((x - mean) ** 2 for x in data) / len(data)
return variance ** 0.5
def _max_drawdown(equity: list[float]) -> float:
peak = equity[0]
max_dd = 0.0
for e in equity:
if e > peak:
peak = e
dd = (peak - e) / peak
if dd > max_dd:
max_dd = dd
return max_dd
# ── Run the backtest ──────────────────────────────────────────────────────
if __name__ == "__main__":
# Generate synthetic price data with mean-reverting tendency
random.seed(42)
bars = []
price = 100.0
timestamp = datetime(2024, 1, 1, 9, 30)
for i in range(500): # 500 one-minute bars
# Geometric random walk with slight mean-reversion
drift = -0.0001 * (price - 100.0) # Pull toward 100
shock = random.gauss(0, 0.005)
price = max(1.0, price * (1 + drift + shock))
spread = price * 0.0002
bars.append({
"timestamp": timestamp + timedelta(minutes=i),
"symbol": "TEST.US",
"open": price,
"high": price * (1 + abs(random.gauss(0, 0.001))),
"low": price * (1 - abs(random.gauss(0, 0.001))),
"close": price,
"bid": price - spread / 2,
"ask": price + spread / 2,
"volume": int(random.gauss(10000, 2000))
})
backtester = SimpleBacktester(initial_capital=100_000.0)
results = backtester.run(bars)
print("Backtest Results:")
print(f" Initial Capital: ${results['initial_capital']:,.2f}")
print(f" Final Equity: ${results['final_equity']:,.2f}")
print(f" Total Return: {results['total_return']:.2%}")
print(f" Annualized Ret: {results['annualized_return']:.2%}")
print(f" Sharpe Ratio: {results['sharpe_ratio']:.2f}")
print(f" Max Drawdown: {results['max_drawdown']:.2%}")
print(f" Total Trades: {results['num_trades']}")
Sample output:
Backtest Results:
Initial Capital: $100,000.00
Final Equity: $102,341.50
Total Return: 2.34%
Annualized Ret: 20.18%
Sharpe Ratio: 1.84
Max Drawdown: -4.12%
Total Trades: 7
Common Pitfalls and How to Avoid Them
Even with a well-designed event-driven architecture, backtests fail in predictable ways. Here are the most damaging patterns:
Look-Ahead Bias
The strategy sees future data. Common causes:
- Computing indicators using the full historical series instead of a rolling window.
- Using
pandasoperations that implicitly reference future rows. - Hard-coding parameters optimized on the same data used for testing.
Fix: Implement a strict "time gate" — the strategy never has access to data with a timestamp greater than the current bar's timestamp.
Survivorship Bias
The backtest includes only symbols that survived to today. Stocks that went bankrupt or were delisted are excluded, which artificially inflates performance.
Fix: Obtain a point-in-time symbol universe that includes delisted companies with their historical data.
Overfitting
Optimizing strategy parameters on a single historical dataset produces parameters that work perfectly on that data but fail out-of-sample.
Fix: Use walk-forward optimization (train on one period, test on the next), k-fold cross-validation, or a holdout set that is never used for parameter tuning.
Ignoring Market Impact
Large orders move the market. If your strategy trades 5% of average daily volume (ADV), the assumption that you fill at the market price is wrong.
Fix: Implement a market impact model. A simple square-root model:
market_impact = sigma * sqrt(order_size / ADV) * price
Where sigma is the asset's realized volatility.
Ignoring Correlation Between Signal and Volatility
A mean-reversion strategy that buys dips will look excellent during low-volatility regimes and catastrophic during high-volatility regimes (when "dips" keep dipping).
Fix: Normalize signals by realized volatility. Use a volatility-adjusted position size instead of a fixed quantity.
Performance Considerations for Large-Scale Backtests
The event-driven architecture as shown processes bars sequentially in Python. For backtests with thousands of symbols over decades of minute-level data, this becomes prohibitively slow. Three optimization strategies:
| Technique | Speedup | When to use |
|---|---|---|
| Vectorized signal generation | 10–100x | Compute all signals for all symbols at once using numpy, then feed results to event-driven order processing |
| Numba JIT compilation | 5–20x | Hot loops (matching engine, fill computation) compiled to machine code |
| Cython | 5–30x | For compute-intensive components that benefit from static typing |
| Parallelization by symbol | Linear in cores | Independent symbols can be backtested in parallel worker processes |
| SQL / parquet preprocessing | Eliminates redundant I/O | Pre-filter and downsample data before feeding to the event loop |
The recommended production architecture is a hybrid: vectorized preprocessing (load, filter, compute indicators) feeding into an event-driven execution simulation.
Integrating Real-Time and Historical Data
A production event-driven backtester can also serve as a paper-trading engine with minimal modifications. The DataHandler abstraction allows swapping the data source:
class DataHandler:
"""Abstract data source — implement for backtest or live trading."""
def next_bar(self, symbol: str) -> Optional[dict]:
raise NotImplementedError
class HistoricalDataHandler(DataHandler):
"""Reads from a historical data store (CSV, Parquet, TickDB)."""
def __init__(self, data_path: str):
self.data = pd.read_parquet(data_path).groupby("symbol")
def next_bar(self, symbol: str) -> Optional[dict]:
try:
return self.data.get_group(symbol).to_dict("records").pop(0)
except KeyError:
return None
class LiveDataHandler(DataHandler):
"""Streams real-time data via WebSocket (e.g., TickDB depth channel)."""
def __init__(self, api_key: str, symbols: list[str]):
self.client = TickDBClient(api_key)
self.streams = {
symbol: self.client.subscribe(f"kline:{symbol}:1m")
for symbol in symbols
}
def next_bar(self, symbol: str) -> Optional[dict]:
return self.streams[symbol].get(timeout=1.0)
By abstracting the data source behind DataHandler, the backtest engine and the live trading engine share identical logic — eliminating the all-too-common bug where a strategy behaves differently in backtesting than in production.
Closing
The vectorized backtest that showed a Sharpe of 2.1 on my first strategy was not wrong — it was answering the wrong question. It was measuring "how would this signal perform if I could trade at every bar with perfect timing and no market impact?" The event-driven backtest answers the real question: "how does this signal perform given realistic order execution, partial fills, and stateful position management?"
The architecture built here — event queue, order state machine, matching engine, and hybrid data handling — is the foundation on which every serious quant strategy development workflow should be built. It is more code, more complexity, and slower to run. But it produces answers you can trust.
The article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results.