feat: add backtesting framework with sma cross strategy

strategies/__init__.py

@@ -0,0 +1,4 @@
+from .runner import run_backtest
+from .performance import compute_performance
+
+__all__ = ["run_backtest", "compute_performance"]

strategies/broker.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+import queue
+
+from .events import OrderEvent, FillEvent, Direction
+from .data_handler import DataHandler
+
+
+class SimulatedBroker:
+    COMMISSION_RATE = 0.001
+
+    def __init__(self, data: DataHandler, events: queue.Queue):
+        self.data   = data
+        self.events = events
+
+    def execute_order(self, event: OrderEvent) -> None:
+        bar        = self.data.get_latest_bar(event.symbol)
+        fill_price = float(bar["Open"][event.symbol])
+        commission = fill_price * event.quantity * self.COMMISSION_RATE
+        self.events.put(
+            FillEvent(event.symbol, event.quantity, event.direction, fill_price, commission)
+        )

strategies/data_handler.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+import queue
+from typing import Optional
+
+import pandas as pd
+import yfinance as yf
+
+from .events import MarketEvent
+
+
+class DataHandler:
+    def __init__(self, symbols: list[str], start: str, end: str, events: queue.Queue):
+        self.symbols = symbols
+        self.events  = events
+        self.idx     = 0
+        self.continue_backtest = True
+
+        raw = yf.download(symbols, start=start, end=end, auto_adjust=True, progress=False)
+        if len(symbols) == 1:
+            raw.columns = pd.MultiIndex.from_product([raw.columns, symbols])
+        self.data  = raw
+        self.dates = self.data.index
+
+    def get_latest_bar(self, symbol: str) -> Optional[pd.Series]:
+        if self.idx == 0:
+            return None
+        return self.data.iloc[self.idx - 1]
+
+    def get_latest_bars(self, symbol: str, n: int = 1) -> pd.DataFrame:
+        start = max(0, self.idx - n)
+        return self.data.iloc[start : self.idx]
+
+    def update_bars(self):
+        if self.idx < len(self.dates):
+            self.idx += 1
+            self.events.put(MarketEvent())
+        else:
+            self.continue_backtest = False

strategies/events.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+from dataclasses import dataclass, field
+from enum import Enum, auto
+
+
+class EventType(Enum):
+    MARKET = auto()
+    SIGNAL = auto()
+    ORDER  = auto()
+    FILL   = auto()
+
+
+class Direction(Enum):
+    LONG  = "LONG"
+    SHORT = "SHORT"
+    EXIT  = "EXIT"
+
+
+@dataclass
+class Event:
+    type: EventType
+
+
+@dataclass
+class MarketEvent(Event):
+    type: EventType = field(default=EventType.MARKET, init=False)
+
+
+@dataclass
+class SignalEvent(Event):
+    symbol:    str
+    direction: Direction
+    strength:  float = 1.0
+    type: EventType = field(default=EventType.SIGNAL, init=False)
+
+
+@dataclass
+class OrderEvent(Event):
+    symbol:    str
+    quantity:  int
+    direction: Direction
+    type: EventType = field(default=EventType.ORDER, init=False)
+
+
+@dataclass
+class FillEvent(Event):
+    symbol:     str
+    quantity:   int
+    direction:  Direction
+    fill_price: float
+    commission: float
+    type: EventType = field(default=EventType.FILL, init=False)

strategies/performance.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+import numpy as np
+import pandas as pd
+
+
+def compute_performance(equity: pd.Series, initial_capital: float) -> dict:
+    returns      = equity.pct_change().dropna()
+    total_return = (equity.iloc[-1] / initial_capital) - 1
+    years        = (equity.index[-1] - equity.index[0]).days / 365.25
+    cagr         = (equity.iloc[-1] / initial_capital) ** (1 / years) - 1
+    sharpe       = returns.mean() / returns.std() * np.sqrt(252) if returns.std() > 0 else 0
+    rolling_max  = equity.cummax()
+    drawdown     = (equity - rolling_max) / rolling_max
+    max_dd       = drawdown.min()
+    volatility   = returns.std() * np.sqrt(252)
+    calmar       = cagr / abs(max_dd) if max_dd != 0 else 0
+    return {
+        "total_return": total_return,
+        "cagr":         cagr,
+        "sharpe":       sharpe,
+        "max_dd":       max_dd,
+        "volatility":   volatility,
+        "calmar":       calmar,
+        "drawdown":     drawdown,
+        "returns":      returns,
+    }

strategies/portfolio.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+import queue
+
+import pandas as pd
+
+from .events import SignalEvent, FillEvent, OrderEvent, Direction
+from .data_handler import DataHandler
+
+
+class Portfolio:
+    def __init__(
+        self,
+        data: DataHandler,
+        events: queue.Queue,
+        symbols: list[str],
+        initial_capital: float = 100_000.0,
+        position_fraction: float = 0.95,
+    ):
+        self.data      = data
+        self.events    = events
+        self.symbols   = symbols
+        self.capital   = initial_capital
+        self.fraction  = position_fraction
+        self.positions = {s: 0 for s in symbols}
+        self.equity: list[tuple] = []
+
+    def _current_price(self, symbol: str) -> float:
+        bar = self.data.get_latest_bar(symbol)
+        return float(bar["Close"][symbol])
+
+    def on_signal(self, event: SignalEvent) -> None:
+        price = self._current_price(event.symbol)
+        if event.direction == Direction.LONG:
+            qty = int((self.capital * self.fraction) / price)
+            if qty > 0:
+                self.events.put(OrderEvent(event.symbol, qty, Direction.LONG))
+        elif event.direction == Direction.EXIT:
+            qty = self.positions[event.symbol]
+            if qty > 0:
+                self.events.put(OrderEvent(event.symbol, qty, Direction.EXIT))
+
+    def on_fill(self, event: FillEvent) -> None:
+        sign = 1 if event.direction == Direction.LONG else -1
+        self.positions[event.symbol] += sign * event.quantity
+        cost = sign * event.quantity * event.fill_price + event.commission
+        self.capital -= cost
+
+    def update_equity(self) -> None:
+        if self.data.idx == 0:
+            return
+        date = self.data.dates[self.data.idx - 1]
+        mkt  = sum(self.positions[s] * self._current_price(s) for s in self.symbols)
+        self.equity.append((date, self.capital + mkt))
+
+    def equity_series(self) -> pd.Series:
+        dates, values = zip(*self.equity)
+        return pd.Series(values, index=dates, name="equity")

strategies/runner.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+import queue
+
+import pandas as pd
+
+from .events import EventType
+from .data_handler import DataHandler
+from .strategy import SMACrossStrategy
+from .portfolio import Portfolio
+from .broker import SimulatedBroker
+from .performance import compute_performance
+
+
+def run_backtest(
+    symbols: list[str],
+    start: str,
+    end: str,
+    initial_capital: float,
+    fast_window: int,
+    slow_window: int,
+) -> tuple[pd.Series, dict, list[dict]]:
+    events    = queue.Queue()
+    data      = DataHandler(symbols, start, end, events)
+    strategy  = SMACrossStrategy(symbols, events, data, fast_window, slow_window)
+    portfolio = Portfolio(data, events, symbols, initial_capital)
+    broker    = SimulatedBroker(data, events)
+
+    while data.continue_backtest:
+        data.update_bars()
+        while not events.empty():
+            event = events.get()
+            if event.type == EventType.MARKET:
+                strategy.calculate_signals(event)
+                portfolio.update_equity()
+            elif event.type == EventType.SIGNAL:
+                portfolio.on_signal(event)
+            elif event.type == EventType.ORDER:
+                broker.execute_order(event)
+            elif event.type == EventType.FILL:
+                portfolio.on_fill(event)
+
+    equity = portfolio.equity_series()
+    perf   = compute_performance(equity, initial_capital)
+    return equity, perf, strategy.trade_log

strategies/strategy.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+import queue
+from abc import ABC, abstractmethod
+
+from .events import Event, EventType, SignalEvent, Direction
+from .data_handler import DataHandler
+
+
+class Strategy(ABC):
+    @abstractmethod
+    def calculate_signals(self, event: Event) -> None: ...
+
+
+class SMACrossStrategy(Strategy):
+    def __init__(
+        self,
+        symbols: list[str],
+        events: queue.Queue,
+        data: DataHandler,
+        fast: int = 20,
+        slow: int = 50,
+    ):
+        self.symbols   = symbols
+        self.events    = events
+        self.data      = data
+        self.fast      = fast
+        self.slow      = slow
+        self.invested  = {s: False for s in symbols}
+        self.trade_log: list[dict] = []
+
+    def calculate_signals(self, event: Event) -> None:
+        if event.type != EventType.MARKET:
+            return
+        for symbol in self.symbols:
+            bars = self.data.get_latest_bars(symbol, self.slow + 1)
+            if len(bars) < self.slow:
+                continue
+            closes  = bars["Close"][symbol].values
+            fast_ma = closes[-self.fast:].mean()
+            slow_ma = closes[-self.slow:].mean()
+            date    = self.data.dates[self.data.idx - 1]
+
+            if fast_ma > slow_ma and not self.invested[symbol]:
+                self.events.put(SignalEvent(symbol, Direction.LONG))
+                self.invested[symbol] = True
+                self.trade_log.append({"date": date, "symbol": symbol, "signal": "BUY"})
+            elif fast_ma < slow_ma and self.invested[symbol]:
+                self.events.put(SignalEvent(symbol, Direction.EXIT))
+                self.invested[symbol] = False
+                self.trade_log.append({"date": date, "symbol": symbol, "signal": "SELL"})