Quant Trading Systems: Architecture & Infrastructure › Strategy Engine Architecture

from abc import ABC, abstractmethod
from enum import Enum


class Signal(Enum):
    """Trading signal types."""

    LONG = 1
    NEUTRAL = 0
    SHORT = -1


class StrategyBase(ABC):
    """Abstract base class for trading strategies."""

    def __init__(self, symbols: list[str]):
        self.symbols = symbols
        self.positions: dict[str, float] = {s: 0 for s in symbols}
        self.signals: dict[str, Signal] = {s: Signal.NEUTRAL for s in symbols}

    @abstractmethod
    def on_data(self, data: dict) -> dict[str, Signal]:
        """Process new data and generate signals."""
        pass

    @abstractmethod
    def calculate_target_positions(self) -> dict[str, float]:
        """Convert signals to target position sizes."""
        pass


class MomentumStrategy(StrategyBase):
    """
    Simple momentum strategy for demonstration.
    Goes long when returns exceed threshold, short when below.
    """

    def __init__(
        self, symbols: list[str], lookback: int = 20, threshold: float = 0.02
    ):
        super().__init__(symbols)
        self.lookback = lookback
        self.threshold = threshold
        self.price_history: dict[str, list] = {s: [] for s in symbols}

    def on_data(self, data: dict) -> dict[str, Signal]:
        """Update price history and generate signals."""
        signals = {}

        for symbol in self.symbols:
            if symbol not in data:
                signals[symbol] = Signal.NEUTRAL
                continue

            # Update price history
            self.price_history[symbol].append(data[symbol])

            # Keep only lookback period
            if len(self.price_history[symbol]) > self.lookback:
                self.price_history[symbol] = self.price_history[symbol][
                    -self.lookback :
                ]

            # Calculate signal if we have enough history
            if len(self.price_history[symbol]) >= self.lookback:
                prices = np.array(self.price_history[symbol])
                momentum = (prices[-1] - prices[0]) / prices[0]

                if momentum > self.threshold:
                    signals[symbol] = Signal.LONG
                elif momentum < -self.threshold:
                    signals[symbol] = Signal.SHORT
                else:
                    signals[symbol] = Signal.NEUTRAL
            else:
                signals[symbol] = Signal.NEUTRAL

        self.signals = signals
        return signals

    def calculate_target_positions(self) -> dict[str, float]:
        """Equal weight positions based on signals."""
        n_active = sum(1 for s in self.signals.values() if s != Signal.NEUTRAL)
        if n_active == 0:
            return {s: 0 for s in self.symbols}

        weight = 1.0 / n_active
        return {
            symbol: weight * signal.value
            for symbol, signal in self.signals.items()
        }

# Simulate strategy execution
strategy = MomentumStrategy(["SPY", "QQQ"], lookback=5, threshold=0.01)

# Simulate 10 days of price data
np.random.seed(42)
spy_prices = 450 * np.cumprod(1 + np.random.randn(10) * 0.01)
qqq_prices = 380 * np.cumprod(1 + np.random.randn(10) * 0.015)

execution_log = []
for day in range(10):
    data = {"SPY": spy_prices[day], "QQQ": qqq_prices[day]}
    signals = strategy.on_data(data)
    positions = strategy.calculate_target_positions()

    log_entry = (
        f"{day + 1:3d} | {spy_prices[day]:9.2f} | {qqq_prices[day]:9.2f} | "
        f"{signals['SPY'].name:10s} | {signals['QQQ'].name:10s} | "
        f"SPY: {positions['SPY']:+.2f}, QQQ: {positions['QQQ']:+.2f}"
    )
    execution_log.append(log_entry)

import json
from pathlib import Path


@dataclass
class StrategyState:
    """Complete state snapshot for a strategy."""

    timestamp: datetime
    positions: dict[str, float]
    signals: dict[str, int]
    open_orders: list[dict]
    pnl_today: float
    pnl_total: float
    max_drawdown: float

    def to_dict(self) -> dict:
        return {
            "timestamp": self.timestamp.isoformat(),
            "positions": self.positions,
            "signals": self.signals,
            "open_orders": self.open_orders,
            "pnl_today": self.pnl_today,
            "pnl_total": self.pnl_total,
            "max_drawdown": self.max_drawdown,
        }

    @classmethod
    def from_dict(cls, data: dict) -> "StrategyState":
        data["timestamp"] = datetime.fromisoformat(data["timestamp"])
        return cls(**data)


class StateManager:
    """Manages strategy state persistence and recovery."""

    def __init__(self, state_dir: str = "./state"):
        self.state_dir = Path(state_dir)
        self.state_dir.mkdir(exist_ok=True)

    def save_state(self, strategy_id: str, state: StrategyState):
        """Persist state to disk."""
        filepath = self.state_dir / f"{strategy_id}_state.json"
        with open(filepath, "w") as f:
            json.dump(state.to_dict(), f, indent=2)

    def load_state(self, strategy_id: str) -> Optional[StrategyState]:
        """Load state from disk if it exists."""
        filepath = self.state_dir / f"{strategy_id}_state.json"
        if not filepath.exists():
            return None
        with open(filepath, "r") as f:
            return StrategyState.from_dict(json.load(f))

    def checkpoint(self, strategy_id: str, state: StrategyState):
        """
        Create a timestamped checkpoint for recovery.
        Useful for debugging and audit trails.
        """
        timestamp_str = state.timestamp.strftime("%Y%m%d_%H%M%S")
        filepath = (
            self.state_dir / f"{strategy_id}_checkpoint_{timestamp_str}.json"
        )
        with open(filepath, "w") as f:
            json.dump(state.to_dict(), f, indent=2)