Design Patterns for an Event-Driven Trading System

"Write once, trade anywhere" - run the same strategy code in backtest, paper, and live mode without modification.

The problem

Every algorithmic trading project runs into the same wall. A strategy validated in a backtest has to be rewritten before it can go live. The data source differs, the order execution logic differs, and position management differs. The result is a gap between backtest numbers and live performance.

Three design patterns, used together, close that gap.

Pattern	Role	Applied to
Strategy Pattern	Encapsulates strategy logic in its own class	Trading strategies (momentum, value, technical)
Abstract Factory	Abstracts the execution environment behind a factory	Broker, DataFeed, Executor
Event-Driven Architecture	Loose coupling between components	MarketEvent, SignalEvent, OrderEvent, FillEvent

Strategy Pattern: encapsulating the strategy

The Strategy Pattern separates an algorithm into its own class so it can be swapped at runtime. In a trading system the most natural target is the trading strategy itself.

Structure

text

┌─────────────────────┐
│   TradingEngine     │
│  (Context)          │
│                     │
│  strategy.on_bar()  │
│  strategy.on_tick() │
└──────┬──────────────┘
       │ uses
       ▼
┌─────────────────────┐
│  <<interface>>      │
│  Strategy           │
│                     │
│  + on_bar(data)     │
│  + on_tick(data)    │
│  + on_order(event)  │
└──────┬──────────────┘
       │ implements
       ├─────────────────┐──────────────────┐
       ▼                 ▼                  ▼
┌─────────────────┐ ┌─────────────┐ ┌──────────────────┐
│ MomentumStrategy│ │ValueStrategy│ │TechnicalStrategy │
└─────────────────┘ └─────────────┘ └──────────────────┘

The Python ABC interface

python

from abc import ABC, abstractmethod
from dataclasses import dataclass
from decimal import Decimal
 
 
class Strategy(ABC):
    """Signal generation abstraction - a strategy is a pure signal generator."""
 
    @abstractmethod
    def calculate_signals(self, event: "MarketEvent") -> "SignalEvent | None":
        """Take market data and produce a trading signal.
 
        Args:
            event: a new market data event
        Returns:
            a trading signal, or None when there is no signal
        """
        ...

The key rule is that a strategy does not need to know the portfolio state. It receives data and emits signals. Order sizing, risk management, and position updates belong to other components.

A concrete strategy

python

class MomentumStrategy(Strategy):
    """RSI-based momentum strategy."""
 
    def __init__(self, rsi_period: int = 14, oversold: float = 30, overbought: float = 70):
        self.rsi_period = rsi_period
        self.oversold = oversold
        self.overbought = overbought
 
    def calculate_signals(self, event: MarketEvent) -> SignalEvent | None:
        rsi = self._calculate_rsi(event.bars, self.rsi_period)
 
        if rsi < self.oversold:
            return SignalEvent(
                symbol=event.symbol,
                signal_type="LONG",
                strength=1.0 - (rsi / self.oversold),
            )
        elif rsi > self.overbought:
            return SignalEvent(
                symbol=event.symbol,
                signal_type="SHORT",
                strength=(rsi - self.overbought) / (100 - self.overbought),
            )
        return None

The strategy has no idea whether it is running in a backtest or in production. That is the point of the Strategy Pattern.

Abstract Factory: abstracting the execution environment

For strategy code to stay environment-independent, the components that change per environment - Broker, DataFeed, Executor - must be abstracted. The Abstract Factory pattern creates a consistent group of related objects.

Implementation per mode

text

TradingMode.LIVE     → LiveBroker,     LiveDataFeed,     LiveExecutor
TradingMode.PAPER    → PaperBroker,    PaperDataFeed,    PaperExecutor
TradingMode.BACKTEST → BacktestBroker, BacktestDataFeed, BacktestExecutor

TradingInterface

python

from abc import ABC, abstractmethod
from enum import Enum
from decimal import Decimal
 
 
class TradingMode(Enum):
    LIVE = "live"
    PAPER = "paper"
    BACKTEST = "backtest"
 
 
@dataclass
class OrderRequest:
    symbol: str
    side: str           # "buy" | "sell"
    quantity: Decimal
    order_type: str     # "market" | "limit"
    price: Decimal | None = None
 
 
@dataclass
class OrderResult:
    order_id: str
    symbol: str
    side: str
    quantity: Decimal
    filled_price: Decimal
    commission: Decimal
    timestamp: str
 
 
@dataclass
class Position:
    symbol: str
    quantity: Decimal
    avg_price: Decimal
    current_price: Decimal
    unrealized_pnl: Decimal
 
 
class TradingInterface(ABC):
    """Shared interface for live, paper, and backtest modes."""
 
    @property
    @abstractmethod
    def mode(self) -> TradingMode: ...
 
    # --- market data ---
    @abstractmethod
    async def get_current_price(self, symbol: str) -> Decimal: ...
 
    @abstractmethod
    async def get_daily_prices(
        self, symbol: str, start_date: str, end_date: str
    ) -> list[dict]: ...
 
    # --- orders ---
    @abstractmethod
    async def place_order(self, order: OrderRequest) -> OrderResult: ...
 
    @abstractmethod
    async def cancel_order(self, order_id: str) -> bool: ...
 
    @abstractmethod
    async def modify_order(
        self, order_id: str, quantity: Decimal | None, price: Decimal | None
    ) -> OrderResult: ...
 
    # --- positions and balance ---
    @abstractmethod
    async def get_positions(self) -> list[Position]: ...
 
    @abstractmethod
    async def get_balance(self) -> dict: ...
 
    @abstractmethod
    async def get_order_history(
        self, start_date: str, end_date: str
    ) -> list[dict]: ...

Switching modes with a factory

python

class TradingFactory:
    """Return the TradingInterface implementation for the execution environment."""
 
    @staticmethod
    def create(mode: TradingMode, config: dict) -> TradingInterface:
        match mode:
            case TradingMode.LIVE:
                from app.trading.live import LiveTrading
                return LiveTrading(config)
            case TradingMode.PAPER:
                from app.trading.paper import PaperTrading
                return PaperTrading(config)
            case TradingMode.BACKTEST:
                from app.trading.backtest import BacktestTrading
                return BacktestTrading(config)

Calling code only touches the factory.

python

# Read the mode from configuration and build through the factory
trading = TradingFactory.create(TradingMode.BACKTEST, config)
 
# Identical code regardless of mode
price = await trading.get_current_price("005930")
result = await trading.place_order(OrderRequest(
    symbol="005930",
    side="buy",
    quantity=Decimal("10"),
    order_type="market",
))

The agent base class

An agent takes a TradingInterface as a dependency and behaves the same in every mode.

python

class BaseAgent(ABC):
    """Base class for trading agents."""
 
    def __init__(self, trading: TradingInterface, config: dict):
        self.trading = trading  # same interface in every mode
        self.config = config
 
    @abstractmethod
    async def analyze(self, symbols: list[str]) -> list[dict]:
        """Analyze the market and produce signals."""
        ...
 
    @abstractmethod
    async def execute(self, signals: list[dict]) -> list[OrderResult]:
        """Turn signals into orders and execute them."""
        ...
 
    async def run_cycle(self, symbols: list[str]):
        """One analyze-then-execute cycle."""
        signals = await self.analyze(symbols)
        results = await self.execute(signals)
        return results

Event-Driven Architecture

An event-driven structure is what keeps coupling between components low. Every component communicates only through the event queue and never needs to know how the others are implemented.

Event flow

text

Event Queue
  │
  ├── MarketEvent   → Strategy.calculate_signals() → SignalEvent
  ├── SignalEvent    → Portfolio.handle_signal()    → OrderEvent
  ├── OrderEvent     → ExecutionHandler.execute()   → FillEvent
  └── FillEvent      → Portfolio.update_position()

Four event types

Event	Producer	Consumer	Description
`MarketEvent`	DataHandler	Strategy	New market data received
`SignalEvent`	Strategy	Portfolio	Trading signal (direction, strength)
`OrderEvent`	Portfolio	ExecutionHandler	Order request (quantity, price)
`FillEvent`	ExecutionHandler	Portfolio	Fill result, including commission

Four ABCs (Abstract Base Class)

The textbook event-driven trading system is built from these four abstract base classes.

python

from abc import ABC, abstractmethod
 
 
class DataHandler(ABC):
    """Historical and live data abstraction."""
 
    @abstractmethod
    def get_latest_bar(self, symbol: str) -> dict: ...
 
    @abstractmethod
    def update_bars(self) -> None: ...
 
 
class Strategy(ABC):
    """Signal generation abstraction."""
 
    @abstractmethod
    def calculate_signals(self, event: MarketEvent) -> SignalEvent | None: ...
 
 
class Portfolio(ABC):
    """Position and order management abstraction."""
 
    @abstractmethod
    def update_signal(self, event: SignalEvent) -> OrderEvent | None: ...
 
    @abstractmethod
    def update_fill(self, event: FillEvent) -> None: ...
 
 
class ExecutionHandler(ABC):
    """Broker connection abstraction."""
 
    @abstractmethod
    def execute_order(self, event: OrderEvent) -> FillEvent: ...

What the event structure buys you

The structure itself prevents look-ahead bias, simulates realistic order timing, and applies risk rules consistently.

Look-ahead bias prevention: the structure makes future data unreachable during a backtest. The DataHandler publishes events sequentially, so a strategy can only see data up to the current point.

Realistic timing: passing through the event queue reflects the real delay between deciding and executing. It avoids the unrealistic backtest assumption that an order fills instantly at the observed price.

Mode switching: only DataHandler and ExecutionHandler are replaced. Strategy and Portfolio are mode-agnostic.

The event loop

python

import asyncio
from collections import deque
 
 
class EventLoop:
    """Event-driven trading engine."""
 
    def __init__(
        self,
        data_handler: DataHandler,
        strategy: Strategy,
        portfolio: Portfolio,
        execution_handler: ExecutionHandler,
    ):
        self.data_handler = data_handler
        self.strategy = strategy
        self.portfolio = portfolio
        self.execution_handler = execution_handler
        self.event_queue: deque = deque()
 
    async def run(self):
        """Main event loop."""
        while True:
            # 1. receive new market data
            self.data_handler.update_bars()
            market_event = MarketEvent(...)
            self.event_queue.append(market_event)
 
            # 2. process events
            while self.event_queue:
                event = self.event_queue.popleft()
 
                if isinstance(event, MarketEvent):
                    signal = self.strategy.calculate_signals(event)
                    if signal:
                        self.event_queue.append(signal)
 
                elif isinstance(event, SignalEvent):
                    order = self.portfolio.update_signal(event)
                    if order:
                        self.event_queue.append(order)
 
                elif isinstance(event, OrderEvent):
                    fill = self.execution_handler.execute_order(event)
                    self.event_queue.append(fill)
 
                elif isinstance(event, FillEvent):
                    self.portfolio.update_fill(event)

Open source frameworks compared

Here is how existing open source frameworks implement the same patterns.

Backtrader

Backtrader takes an event-driven, object-oriented approach.

Core abstract classes:

bt.Strategy: __init__(), next(), notify_order(), notify_trade()
bt.Broker: order execution and portfolio management, with the same interface for live and backtest
bt.DataFeed: abstracts data sources such as CSV, Yahoo, pandas, and real-time streams

Mode switching: the Cerebro engine swaps only the DataFeed and Broker, so one strategy runs in both backtest and live.

python

class SmaCross(bt.Strategy):
    params = dict(fast=10, slow=30)
 
    def __init__(self):
        sma_fast = bt.ind.SMA(period=self.p.fast)
        sma_slow = bt.ind.SMA(period=self.p.slow)
        self.crossover = bt.ind.CrossOver(sma_fast, sma_slow)
 
    def next(self):
        if self.crossover > 0:
            self.buy()
        elif self.crossover < 0:
            self.sell()

Zipline

Zipline uses a functional style with an explicit context object.

Core interface:

initialize(context): initial setup
handle_data(context, data): called on every bar
context: the state-holding object

python

def initialize(context):
    context.asset = symbol("AAPL")
    context.sma_period = 20
 
def handle_data(context, data):
    prices = data.history(context.asset, "price", context.sma_period, "1d")
    if data.current(context.asset, "price") > prices.mean():
        order_target_percent(context.asset, 1.0)
    else:
        order_target_percent(context.asset, 0.0)

Freqtrade

Freqtrade is DataFrame-based and object-oriented. A strategy inherits from IStrategy (INTERFACE_VERSION = 3).

Method	Signature	Role
`populate_indicators`	`(df, metadata) -> df`	Add technical indicators
`populate_entry_trend`	`(df, metadata) -> df`	Entry signal (`enter_long=1`)
`populate_exit_trend`	`(df, metadata) -> df`	Exit signal (`exit_long=1`)

Optional callbacks:

custom_entry_price() - adjust the entry price
custom_exit() - custom exit logic
custom_stoploss() - dynamic stop loss
confirm_trade_entry() - confirm entry

python

class MomentumStrategy(IStrategy):
    INTERFACE_VERSION = 3
    timeframe = "1h"
    minimal_roi = {"0": 0.04}
    stoploss = -0.10
 
    def populate_indicators(self, dataframe, metadata):
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        return dataframe
 
    def populate_entry_trend(self, dataframe, metadata):
        dataframe.loc[dataframe["rsi"] < 30, "enter_long"] = 1
        return dataframe
 
    def populate_exit_trend(self, dataframe, metadata):
        dataframe.loc[dataframe["rsi"] > 70, "exit_long"] = 1
        return dataframe

DataProvider: self.dp.get_pair_dataframe(), self.dp.orderbook(), and friends give the same API in backtest and live.

NautilusTrader

NautilusTrader pairs a Rust core with a Python API, and every event carries a nanosecond timestamp.

Strategy ABC: on_start(), on_bar(), on_order(), on_event()
One strategy implementation: the same code is used for backtesting and live deployment
Event-driven: all data and orders flow as events
Explicit ABCs

Framework comparison

Trait	Backtrader	Zipline	Freqtrade	NautilusTrader
Paradigm	Event-driven OOP	Functional plus context	DataFrame-based OOP	Event-driven OOP
Strategy definition	Class inheritance	Function definitions	IStrategy subclass	Class inheritance
Live support	IB, Oanda	Limited	Many exchanges	Many exchanges
Mode switching	Swap DataFeed/Broker	Limited	Automatic (dry-run)	Swap DataFeed/Executor
Performance	Python (moderate)	Python (moderate)	Python (moderate)	Rust core (fast)
Assets	Equities, futures	Equities	Crypto	Varied
Python ABC usage	Implicit	Implicit	IStrategy	Explicit ABCs

Choosing an approach

For a personal project trading Korean equities, borrowing the patterns and building your own system is more practical than adopting one of these frameworks. The reasons are concrete.

Backtrader and Zipline have no integration with Korean brokerage APIs
Freqtrade is crypto-only
NautilusTrader has a steep learning curve

Implementing the combination they all share - Strategy Pattern, Abstract Factory, and event-driven architecture - yourself gives the most flexibility.

Design principles

Five rules

A strategy is a pure signal generator: it takes data and emits signals without knowing the portfolio state
Environment swaps happen in the factory: mode changes require no strategy code changes
Event-driven loose coupling: components talk only through the event queue
Anti-corruption layer: the DataHandler normalizes external API inconsistencies internally
Look-ahead bias prevention: backtests are structurally blocked from reading future data

The anti-corruption layer in detail

Every brokerage returns a different API response shape. The Korea Investment & Securities Open API (KIS) uses a field named stck_prpr for the current price; Kiwoom uses something else. The DataHandler converts these into one internal format.

python

@dataclass
class StandardBar:
    """Internal standard market data record."""
    symbol: str
    timestamp: str
    open: Decimal
    high: Decimal
    low: Decimal
    close: Decimal
    volume: int
 
 
class KISDataHandler(DataHandler):
    """Convert KIS API responses into the standard format."""
 
    def _normalize(self, raw: dict) -> StandardBar:
        return StandardBar(
            symbol=raw["mksc_shrn_iscd"],
            timestamp=raw["stck_bsop_date"],
            open=Decimal(raw["stck_oprc"]),
            high=Decimal(raw["stck_hgpr"]),
            low=Decimal(raw["stck_lwpr"]),
            close=Decimal(raw["stck_prpr"]),
            volume=int(raw["acml_vol"]),
        )

The full architecture

Summary

Combining the three patterns removes the code gap between backtest and production. The strategy only emits signals, the factory decides the execution environment, and components communicate only through the event queue. Switching modes means replacing the DataHandler and ExecutionHandler while Strategy and Portfolio stay untouched. When existing frameworks lack integration with your brokers, as with Korean equities, borrowing these three patterns beats adopting a framework wholesale. The patterns alone do not guarantee backtest fidelity, so simulation accuracy - blocking look-ahead bias and modeling realistic fills - still has to be verified separately.