이벤트 기반 트레이딩 시스템 설계 패턴

"Write once, trade anywhere" — 전략 코드를 수정 없이 백테스트, 모의투자, 실전 모드에서 실행한다.

문제 정의

알고리즘 트레이딩 시스템을 개발하면 반드시 마주치는 문제가 있다. 백테스트로 검증한 전략을 실전에 배포할 때 코드를 다시 작성해야 한다는 것이다. 데이터 소스가 다르고, 주문 실행 로직이 다르고, 포지션 관리 방식이 다르다. 결과적으로 백테스트 결과와 실전 성과가 괴리되는 원인이 된다.

이 문제를 해결하는 핵심은 세 가지 디자인 패턴의 조합이다.

패턴	역할	적용 대상
Strategy Pattern	전략 로직을 독립된 클래스로 캡슐화	매매 전략 (모멘텀, 가치, 기술적 분석)
Abstract Factory	실행 환경을 추상화하여 팩토리로 전환	Broker, DataFeed, Executor
Event-Driven Architecture	컴포넌트 간 느슨한 결합	MarketEvent, SignalEvent, OrderEvent, FillEvent

Strategy Pattern: 전략 캡슐화

Strategy Pattern은 알고리즘(전략)을 독립된 클래스로 분리하여, 런타임에 교체 가능하게 만드는 패턴이다. 트레이딩 시스템에서 가장 자연스러운 적용 대상은 매매 전략이다.

구조

plaintext

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

Python ABC 인터페이스

python

from abc import ABC, abstractmethod
from dataclasses import dataclass
from decimal import Decimal
 
 
class Strategy(ABC):
    """시그널 생성 추상화 — 전략은 순수 시그널 생성기로만 동작한다."""
 
    @abstractmethod
    def calculate_signals(self, event: "MarketEvent") -> "SignalEvent | None":
        """시세 데이터를 받아 매매 시그널을 생성한다.
 
        Args:
            event: 새로운 시세 데이터 이벤트
        Returns:
            매매 시그널 또는 None (시그널 없음)
        """
        ...

핵심 원칙은 전략이 포트폴리오 상태를 몰라도 되는 것이다. 전략은 데이터를 받아 시그널만 생성한다. 주문 수량 결정, 리스크 관리, 포지션 업데이트는 다른 컴포넌트가 담당한다.

구체 전략 예시

python

class MomentumStrategy(Strategy):
    """RSI 기반 모멘텀 전략."""
 
    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

전략은 실행 환경(백테스트인지 실전인지)을 전혀 알지 못한다. 이것이 Strategy Pattern의 핵심이다.

Abstract Factory: 실행 환경 추상화

전략 코드가 환경에 독립적이려면, 환경별로 달라지는 컴포넌트(Broker, DataFeed, Executor)를 추상화해야 한다. Abstract Factory 패턴은 관련 객체들의 그룹을 일관되게 생성하는 역할을 한다.

모드별 구현체 매핑

plaintext

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):
    """실전/모의/백테스트 공통 인터페이스."""
 
    @property
    @abstractmethod
    def mode(self) -> TradingMode: ...
 
    # --- 시세 조회 ---
    @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]: ...
 
    # --- 주문 ---
    @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: ...
 
    # --- 포지션/잔고 ---
    @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]: ...

Factory 패턴으로 모드 전환

python

class TradingFactory:
    """실행 환경에 따라 적절한 TradingInterface 구현체 반환."""
 
    @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)

사용하는 쪽에서는 팩토리만 호출하면 된다.

python

# 설정 파일에서 모드를 읽어 팩토리로 생성
trading = TradingFactory.create(TradingMode.BACKTEST, config)
 
# 모드에 관계없이 동일한 코드
price = await trading.get_current_price("005930")
result = await trading.place_order(OrderRequest(
    symbol="005930",
    side="buy",
    quantity=Decimal("10"),
    order_type="market",
))

에이전트 기본 클래스

에이전트는 TradingInterface를 의존성으로 받아 모드에 무관하게 동작한다.

python

class BaseAgent(ABC):
    """트레이딩 에이전트 기본 클래스."""
 
    def __init__(self, trading: TradingInterface, config: dict):
        self.trading = trading  # 모드 무관 동일 인터페이스
        self.config = config
 
    @abstractmethod
    async def analyze(self, symbols: list[str]) -> list[dict]:
        """시장 분석 후 시그널을 생성한다."""
        ...
 
    @abstractmethod
    async def execute(self, signals: list[dict]) -> list[OrderResult]:
        """시그널을 주문으로 변환하여 실행한다."""
        ...
 
    async def run_cycle(self, symbols: list[str]):
        """분석 → 실행 사이클."""
        signals = await self.analyze(symbols)
        results = await self.execute(signals)
        return results

Event-Driven Architecture

이벤트 기반 아키텍처는 컴포넌트 간 결합도를 최소화하는 핵심 구조다. 각 컴포넌트는 이벤트 큐를 통해서만 통신하며, 서로의 구현을 알 필요가 없다.

이벤트 흐름

plaintext

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

4개의 이벤트 타입

이벤트	생성자	소비자	설명
`MarketEvent`	DataHandler	Strategy	새 시세 데이터 수신
`SignalEvent`	Strategy	Portfolio	매매 시그널 (방향, 강도)
`OrderEvent`	Portfolio	ExecutionHandler	주문 요청 (수량, 가격)
`FillEvent`	ExecutionHandler	Portfolio	체결 결과 (수수료 포함)

4개의 ABC (Abstract Base Class)

QuantStart에서 제시한 참조 구현은 가장 교과서적인 이벤트 기반 트레이딩 시스템 설계다.

python

from abc import ABC, abstractmethod
 
 
class DataHandler(ABC):
    """히스토리컬/라이브 데이터 추상화."""
 
    @abstractmethod
    def get_latest_bar(self, symbol: str) -> dict: ...
 
    @abstractmethod
    def update_bars(self) -> None: ...
 
 
class Strategy(ABC):
    """시그널 생성 추상화."""
 
    @abstractmethod
    def calculate_signals(self, event: MarketEvent) -> SignalEvent | None: ...
 
 
class Portfolio(ABC):
    """포지션/주문 관리 추상화."""
 
    @abstractmethod
    def update_signal(self, event: SignalEvent) -> OrderEvent | None: ...
 
    @abstractmethod
    def update_fill(self, event: FillEvent) -> None: ...
 
 
class ExecutionHandler(ABC):
    """브로커 연결 추상화."""
 
    @abstractmethod
    def execute_order(self, event: OrderEvent) -> FillEvent: ...

이벤트 기반의 장점

Look-ahead bias 방지, 현실적인 주문 타이밍 시뮬레이션, 리스크 규칙 일관 적용이 이벤트 기반 아키텍처의 핵심 장점이다.

Look-ahead bias 방지: 백테스트에서 미래 데이터에 접근할 수 없도록 구조적으로 보장한다. DataHandler가 순차적으로 이벤트를 발행하므로, 전략은 현재까지의 데이터만 볼 수 있다.

현실적 타이밍: 이벤트 큐를 거치면서 주문 실행에 지연이 발생하는 현실을 반영한다. 백테스트에서 "이 가격에 바로 체결"이라는 비현실적 가정을 피할 수 있다.

모드 전환 핵심: DataHandler와 ExecutionHandler만 교체하면 된다. Strategy와 Portfolio는 모드에 무관하다.

이벤트 루프 구현

python

import asyncio
from collections import deque
 
 
class EventLoop:
    """이벤트 기반 트레이딩 엔진."""
 
    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):
        """메인 이벤트 루프."""
        while True:
            # 1. 새 시세 데이터 수신
            self.data_handler.update_bars()
            market_event = MarketEvent(...)
            self.event_queue.append(market_event)
 
            # 2. 이벤트 처리
            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)

오픈소스 프레임워크 비교

실제 오픈소스 프레임워크들이 위 패턴을 어떻게 구현하는지 비교해보자.

Backtrader

Backtrader는 이벤트 기반 OOP 방식을 채택한다.

핵심 추상 클래스:

bt.Strategy: __init__(), next(), notify_order(), notify_trade()
bt.Broker: 주문 실행, 포트폴리오 관리 (Live/Backtest 동일 인터페이스)
bt.DataFeed: CSV, Yahoo, Pandas, 실시간 등 다양한 소스 추상화

모드 전환: Cerebro 엔진이 DataFeed와 Broker만 교체하면 동일 전략이 백테스트/라이브 모두 동작한다.

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은 함수형 + 컨텍스트 방식을 택한다.

핵심 인터페이스:

initialize(context): 초기 설정
handle_data(context, data): 매 바마다 호출
context: 상태 저장 객체

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는 DataFrame 기반 OOP 방식이다. IStrategy(INTERFACE_VERSION = 3)를 상속한다.

메서드	시그니처	역할
`populate_indicators`	`(df, metadata) -> df`	기술 지표 추가
`populate_entry_trend`	`(df, metadata) -> df`	진입 시그널 (`enter_long=1`)
`populate_exit_trend`	`(df, metadata) -> df`	청산 시그널 (`exit_long=1`)

콜백 메서드 (선택):

custom_entry_price() — 진입가 조정
custom_exit() — 커스텀 청산 로직
custom_stoploss() — 동적 스톱로스
confirm_trade_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() 등으로 백테스트/라이브 동일 API를 제공한다.

NautilusTrader

NautilusTrader는 Rust 코어 + Python API로 나노초 해상도를 지원한다.

Strategy ABC: on_start(), on_bar(), on_order(), on_event()
동일 전략 코드: "Use identical strategy implementations between backtesting and live deployments"
이벤트 기반: 모든 데이터/주문이 이벤트로 처리
명시적 ABC 사용

프레임워크 비교 요약

특성	Backtrader	Zipline	Freqtrade	NautilusTrader
패러다임	이벤트 기반 OOP	함수형 + 컨텍스트	DataFrame 기반 OOP	이벤트 기반 OOP
전략 정의	클래스 상속	함수 정의	IStrategy 상속	클래스 상속
라이브 지원	IB, Oanda	제한적	거래소 다수	거래소 다수
모드 전환	DataFeed/Broker 교체	제한적	자동 (dry-run)	DataFeed/Executor 교체
성능	Python (보통)	Python (보통)	Python (보통)	Rust 코어 (빠름)
자산	주식, 선물	주식	크립토	다양
Python ABC 사용	암묵적	암묵적	IStrategy	명시적 ABC

선택 가이드

개인 프로젝트에서 한국 주식을 다루는 경우, 위 프레임워크를 직접 쓰기보다는 패턴만 차용하여 직접 설계하는 것이 현실적이다. 이유는 다음과 같다.

Backtrader/Zipline은 한국 증권사 API 연동이 없다
Freqtrade는 크립토 전용이다
NautilusTrader는 러닝커브가 높다

대신 이들이 공통적으로 사용하는 Strategy Pattern + Abstract Factory + Event-Driven Architecture 조합을 직접 구현하는 것이 가장 유연하다.

설계 원칙

핵심 5원칙

전략은 순수 시그널 생성기: 포트폴리오 상태를 알 필요 없이, 데이터를 받아 시그널만 생성한다
환경 교체는 팩토리에서: 전략 코드 수정 없이 모드를 전환한다
이벤트 기반 느슨한 결합: 컴포넌트 간 이벤트 큐로만 통신한다
Anti-corruption Layer: DataHandler가 외부 API의 불일치를 내부에서 정규화한다
Look-ahead bias 방지: 백테스트에서 미래 데이터 접근을 구조적으로 차단한다

Anti-corruption Layer 상세

증권사마다 API 응답 형식이 다르다. KIS API는 stck_prpr(현재가)라는 필드명을 쓰고, 키움은 또 다른 이름을 쓴다. DataHandler가 이런 차이를 내부 표준 형식으로 변환한다.

python

@dataclass
class StandardBar:
    """내부 표준 시세 데이터."""
    symbol: str
    timestamp: str
    open: Decimal
    high: Decimal
    low: Decimal
    close: Decimal
    volume: int
 
 
class KISDataHandler(DataHandler):
    """KIS API 응답을 표준 형식으로 변환."""
 
    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"]),
        )

전체 아키텍처

이벤트 기반 트레이딩 시스템의 설계 패턴은 복잡해 보이지만, 핵심은 단순하다. 전략은 시그널만 생성하고, 환경은 팩토리가 결정하고, 컴포넌트는 이벤트로만 소통한다. 이 세 가지 원칙만 지키면 백테스트에서 검증한 전략을 그대로 실전에 배포할 수 있다.