Event-Driven Engine in Python
A backtesting architecture that processes market data and trading logic sequentially through a chronological event queue, faithfully replicating the information constraints of live trading.
Definition
An Event-Driven Backtesting Engine simulates the temporal mechanics of live trading by processing a chronological stream of discrete events — market data ticks, bar closes, order submissions, fill confirmations, and portfolio updates — one at a time through a central event queue. Each event triggers specific handler functions that can only access information available at that precise moment in time, making lookahead bias architecturally impossible by design. This stands in direct contrast to vectorized backtesting, where the entire dataset is visible simultaneously. Event-driven engines are the institutional standard for production-grade strategy validation because they accurately model order latency, partial fills, slippage, and path-dependent logic such as stop losses, trailing stops, and position pyramiding.
Why It Matters in Backtesting
The gap between a vectorized backtest and an event-driven backtest is the gap between a hypothesis and a deployable strategy. Vectorized backtests cannot model: stop losses triggered intrabar, order queue position and partial fills, capital constraint violations, margin calls during drawdowns, or the impact of your own orders on market prices. Any strategy that uses stop losses, has position sizing rules that depend on current capital, or trades instruments with meaningful market impact must be validated in an event-driven engine before any live deployment decision. Valetha's core backtesting infrastructure is built entirely on an event-driven architecture precisely to eliminate this gap between simulation and reality.
Python Implementation
import numpy as np
import pandas as pd
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
class EventType(Enum):
MARKET = "MARKET"
SIGNAL = "SIGNAL"
ORDER = "ORDER"
FILL = "FILL"
@dataclass
class MarketEvent:
type: EventType = EventType.MARKET
timestamp: pd.Timestamp = None
ticker: str = ""
open: float = 0.0
high: float = 0.0
low: float = 0.0
close: float = 0.0
volume: float = 0.0
@dataclass
class SignalEvent:
type: EventType = EventType.SIGNAL
timestamp: pd.Timestamp = None
ticker: str = ""
direction: int = 0 # +1 long, -1 short, 0 flat
strength: float = 1.0
@dataclass
class FillEvent:
type: EventType = EventType.FILL
timestamp: pd.Timestamp = None
ticker: str = ""
quantity: float = 0.0
fill_price: float = 0.0
commission: float = 0.0
direction: int = 0
class SimpleEventDrivenBacktest:
"""
Minimal event-driven backtesting engine demonstrating chronological
event processing with strict information barriers.
"""
def __init__(self, ohlcv_df: pd.DataFrame, initial_capital: float = 100000.0,
commission_bps: float = 5.0):
self.data = ohlcv_df
self.capital = initial_capital
self.commission_bps = commission_bps
self.positions: dict = {}
self.equity_curve: list = []
self.fills: list = []
def _generate_signal(self, event: MarketEvent) -> Optional[SignalEvent]:
"""Strategy logic: override this method to implement custom signals."""
raise NotImplementedError("Implement strategy signal logic here.")
def _execute_order(self, signal: SignalEvent, market: MarketEvent) -> FillEvent:
"""Simulates order execution at next bar open with commission."""
fill_price = market.open # Execute at NEXT bar open (no lookahead)
commission = abs(signal.strength) * fill_price * self.commission_bps / 10000
return FillEvent(
type=EventType.FILL, timestamp=market.timestamp,
ticker=signal.ticker, quantity=signal.strength,
fill_price=fill_price, commission=commission,
direction=signal.direction
)
def run(self) -> dict:
pending_signal: Optional[SignalEvent] = None
for i, (timestamp, row) in enumerate(self.data.iterrows()):
market_event = MarketEvent(
timestamp=timestamp, ticker="ASSET",
open=row["Open"], high=row["High"],
low=row["Low"], close=row["Close"], volume=row["Volume"]
)
# Execute pending signal from PREVIOUS bar at current bar open
if pending_signal is not None:
fill = self._execute_order(pending_signal, market_event)
self.fills.append(fill)
self.positions[fill.ticker] = self.positions.get(fill.ticker, 0) + fill.quantity
self.capital -= fill.commission
pending_signal = None
# Generate signal from CURRENT bar close (will execute next bar)
try:
pending_signal = self._generate_signal(market_event)
except NotImplementedError:
pass
position_value = sum(qty * market_event.close
for ticker, qty in self.positions.items())
self.equity_curve.append({"timestamp": timestamp,
"equity": self.capital + position_value})
equity_df = pd.DataFrame(self.equity_curve).set_index("timestamp")
returns = equity_df["equity"].pct_change().dropna()
return {
"equity_curve": equity_df,
"returns": returns,
"total_fills": len(self.fills),
"final_capital": equity_df["equity"].iloc[-1]
}Test this in a live environment
Stop running Jupyter notebooks locally. Paste this Event-Driven Engine code directly into Valetha's Strategy Lab and run a full historical backtest in seconds.
Open the Python Strategy Lab