ftdt-quant-lab/sim/scenario.py

"""
Scenario simulation: exchange downtime, volatility bursts, regime switches.

Models realistic exchange behaviors that affect market-making performance.
"""

from __future__ import annotations

import random
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class ExchangeDowntime:
    """Scheduled or unscheduled exchange outage."""

    start_time: float
    end_time: float
    reason: str = "scheduled_maintenance"

    def is_active(self, sim_time: float) -> bool:
        return self.start_time <= sim_time < self.end_time

    @property
    def duration_seconds(self) -> float:
        return self.end_time - self.start_time


@dataclass
class LatencySpike:
    """Temporary latency increase."""

    start_time: float
    end_time: float
    multiplier: float = 5.0  # 5x normal latency

    def is_active(self, sim_time: float) -> bool:
        return self.start_time <= sim_time < self.end_time


@dataclass
class VolatilityBurst:
    """Sudden increase in volatility."""

    start_time: float
    end_time: float
    vol_multiplier: float = 3.0

    def is_active(self, sim_time: float) -> bool:
        return self.start_time <= sim_time < self.end_time


@dataclass
class ScenarioConfig:
    """Configuration for simulation scenarios."""

    duration_seconds: float = 3600.0  # 1 hour default
    seed: int | None = None

    # Exchange behavior
    downtime_prob: float = 0.0         # probability of a downtime event
    downtime_duration_ms: int = 30000  # 30s typical
    latency_spike_prob: float = 0.05
    latency_spike_ms: int = 5000
    latency_multiplier: float = 5.0

    # Market behavior
    volatility_burst_prob: float = 0.02
    volatility_burst_ms: int = 60000
    vol_multiplier: float = 3.0

    # Trade intensity
    base_trade_rate: float = 1.0       # trades per second
    burst_trade_rate: float = 5.0      # trades per second during bursts

    # Spread behavior
    base_spread_bps: float = 1.5
    wide_spread_bps: float = 15.0


class ScenarioEngine:
    """Generates and manages simulation scenarios."""

    def __init__(self, config: ScenarioConfig | None = None, seed: int | None = None):
        self._cfg = config or ScenarioConfig()
        self._rng = random.Random(seed or self._cfg.seed)
        self._downtimes: list[ExchangeDowntime] = []
        self._latency_spikes: list[LatencySpike] = []
        self._volatility_bursts: list[VolatilityBurst] = []

    def generate(self, duration: float | None = None):
        """Pre-generate random scenarios for the simulation duration."""
        d = duration or self._cfg.duration_seconds
        self._downtimes.clear()
        self._latency_spikes.clear()
        self._volatility_bursts.clear()

        t = 0.0
        while t < d:
            t += self._rng.expovariate(1.0 / (d / 100))

            if self._rng.random() < self._cfg.downtime_prob:
                dur = self._rng.uniform(self._cfg.downtime_duration_ms / 1000 * 0.5,
                                        self._cfg.downtime_duration_ms / 1000 * 2)
                self._downtimes.append(ExchangeDowntime(t, t + dur))

            if self._rng.random() < self._cfg.latency_spike_prob:
                dur = self._rng.uniform(self._cfg.latency_spike_ms / 1000 * 0.5,
                                        self._cfg.latency_spike_ms / 1000 * 2)
                self._latency_spikes.append(
                    LatencySpike(t, t + dur, self._cfg.latency_multiplier)
                )

            if self._rng.random() < self._cfg.volatility_burst_prob:
                dur = self._rng.uniform(self._cfg.volatility_burst_ms / 1000 * 0.5,
                                        self._cfg.volatility_burst_ms / 1000 * 2)
                self._volatility_bursts.append(
                    VolatilityBurst(t, t + dur, self._cfg.vol_multiplier)
                )

    def is_exchange_up(self, sim_time: float) -> bool:
        return not any(d.is_active(sim_time) for d in self._downtimes)

    def current_latency_multiplier(self, sim_time: float) -> float:
        for spike in self._latency_spikes:
            if spike.is_active(sim_time):
                return spike.multiplier
        return 1.0

    def current_vol_multiplier(self, sim_time: float) -> float:
        for burst in self._volatility_bursts:
            if burst.is_active(sim_time):
                return burst.vol_multiplier
        return 1.0

    def current_trade_rate(self, sim_time: float) -> float:
        if self.current_vol_multiplier(sim_time) > 2.0:
            return self._cfg.burst_trade_rate
        return self._cfg.base_trade_rate

    def current_spread_bps(self, sim_time: float) -> float:
        if self.current_vol_multiplier(sim_time) > 2.0:
            return self._cfg.wide_spread_bps
        return self._cfg.base_spread_bps

    def state(self, sim_time: float) -> dict:
        return {
            "exchange_up": self.is_exchange_up(sim_time),
            "latency_multiplier": self.current_latency_multiplier(sim_time),
            "vol_multiplier": self.current_vol_multiplier(sim_time),
            "trade_rate": self.current_trade_rate(sim_time),
            "spread_bps": self.current_spread_bps(sim_time),
        }

    def active_downtime(self, sim_time: float) -> Optional[ExchangeDowntime]:
        for d in self._downtimes:
            if d.is_active(sim_time):
                return d
        return None

    @property
    def downtimes(self) -> list[ExchangeDowntime]:
        return self._downtimes

    @property
    def latency_spikes(self) -> list[LatencySpike]:
        return self._latency_spikes

    @property
    def volatility_bursts(self) -> list[VolatilityBurst]:
        return self._volatility_bursts