atelier_data 0.0.15

//! Configuration for market snapshot collection pipelines.
//!
//! [`MarketSnapshotConfig`] is the top-level struct deserialized from a TOML
//! file that drives the `bybit_markets` (and future exchange) examples.
//!
//! # Example TOML
//!
//! ```toml
//! [exchange]
//! name = "bybit"
//!
//! [symbol]
//! name = "BTCUSDT"
//! sync_mode = "on_trade"
//!
//! [update_frequency]
//! value = 100
//! unit = "Millis"
//!
//! [pipeline]
//! flush_threshold = 36000
//!
//! [datatypes.orderbook]
//! enabled = true
//! depth = 50
//!
//! [datatypes.trades]
//! enabled = true
//!
//! [datatypes.liquidations]
//! enabled = true
//!
//! [datatypes.funding_rates]
//! enabled = true
//!
//! [datatypes.open_interest]
//! enabled = true
//!
//! [logs]
//! n_orderbooks = 100
//! n_trades = 10
//! n_liquidations = 1
//! n_fundings = 10
//! n_open_interests = 10
//!
//! [output]
//! dir = "datasets/collected/bybit/market_snapshots"
//! ```

use serde::Deserialize;
use std::path::{Path, PathBuf};

use crate::synchronizers::ClockMode;

// ─────────────────────────────────────────────────────────────────────────────
// Root config
// ─────────────────────────────────────────────────────────────────────────────

/// Top-level market snapshot configuration.
#[derive(Debug, Clone, Deserialize)]
pub struct MarketSnapshotConfig {
    pub exchange: ExchangeSection,
    pub symbol: SymbolSection,
    pub update_frequency: UpdateFrequency,
    pub pipeline: PipelineSection,
    pub datatypes: DataTypesSection,
    pub output: OutputSection,
    pub logs: LogsSection,
}

// ─────────────────────────────────────────────────────────────────────────────
// Sections
// ─────────────────────────────────────────────────────────────────────────────

/// `[exchange]` — identifies the target exchange.
#[derive(Debug, Clone, Deserialize)]
pub struct ExchangeSection {
    pub name: String,
}

/// `[symbol]` — the instrument to collect and how to synchronize.
#[derive(Debug, Clone, Deserialize)]
pub struct SymbolSection {
    /// Trading pair (e.g. `"BTCUSDT"`).
    pub name: String,
    /// Which event drives the grid clock.
    pub sync_mode: SyncMode,
}

/// `[update_frequency]` — grid spacing expressed as value + unit.
#[derive(Debug, Clone, Deserialize)]
pub struct UpdateFrequency {
    /// Numeric part of the frequency (e.g. `100`).
    pub value: u64,
    /// Time unit for `value`.
    pub unit: TimeUnit,
}

/// `[pipeline]` — flush cadence.
///
/// The flush interval in wall-clock time is
/// `flush_threshold × update_frequency`. For example, with a 100 ms
/// grid and `flush_threshold = 36000`, one Parquet file is written
/// every hour.
///
/// The process runs continuously until interrupted (Ctrl-C).
#[derive(Debug, Clone, Deserialize)]
pub struct PipelineSection {
    /// Flush buffered snapshots to Parquet after this many grid
    /// periods accumulate. Units: number of `update_frequency` ticks.
    pub flush_threshold: usize,
}

/// `[datatypes]` — selects which data feeds to subscribe to and collect.
///
/// Each feed is its own sub-table so exchange- or feed-specific parameters
/// can be added without polluting a flat namespace.
///
/// ```toml
/// [datatypes.orderbook]
/// enabled = true
/// depth   = 50
///
/// [datatypes.trades]
/// enabled = true
///
/// [datatypes.liquidations]
/// enabled = false
/// ```
#[derive(Debug, Clone, Deserialize)]
pub struct DataTypesSection {
    /// Orderbook feed configuration.
    #[serde(default)]
    pub orderbook: OrderbookConfig,
    /// Public trades feed.
    #[serde(default)]
    pub trades: FeedToggle,
    /// Liquidation events feed.
    #[serde(default)]
    pub liquidations: FeedToggle,
    /// Funding rate updates feed.
    #[serde(default)]
    pub funding_rates: FeedToggle,
    /// Open interest updates feed.
    #[serde(default)]
    pub open_interest: FeedToggle,
}

/// Configuration for the orderbook data feed.
///
/// Besides the `enabled` toggle shared with all feeds, orderbook has a
/// `depth` parameter controlling how many price levels to request.
#[derive(Debug, Clone, Deserialize)]
pub struct OrderbookConfig {
    /// Whether to subscribe to orderbook deltas / snapshots.
    #[serde(default)]
    pub enabled: bool,
    /// Number of price levels to request (e.g. 25, 50).
    #[serde(default = "default_ob_depth")]
    pub depth: usize,
}

impl Default for OrderbookConfig {
    fn default() -> Self {
        Self {
            enabled: false,
            depth: default_ob_depth(),
        }
    }
}

/// Generic toggle for a data feed.
///
/// Intentionally a struct (not a bare `bool`) so that feed-specific
/// parameters can be added later without a breaking schema change.
#[derive(Debug, Clone, Deserialize, Default)]
pub struct FeedToggle {
    /// Whether this feed is enabled.
    #[serde(default)]
    pub enabled: bool,
}

/// `[logs]` — per-event-type print frequency thresholds.
///
/// Each field specifies how many events of that type must accumulate
/// before a status line is printed.  Set to `0` to suppress output
/// for that event type entirely.
#[derive(Debug, Clone, Deserialize)]
pub struct LogsSection {
    /// Print a status line every N orderbook events.
    #[serde(default)]
    pub n_orderbooks: usize,
    /// Print a status line every N public trade events.
    #[serde(default)]
    pub n_trades: usize,
    /// Print a status line every N liquidation events.
    #[serde(default)]
    pub n_liquidations: usize,
    /// Print a status line every N funding rate events.
    #[serde(default)]
    pub n_fundings: usize,
    /// Print a status line every N open interest events.
    #[serde(default)]
    pub n_open_interests: usize,
}

/// `[output]` — where to write Parquet files.
#[derive(Debug, Clone, Deserialize)]
pub struct OutputSection {
    /// Directory for Parquet output, relative to the workspace root.
    pub dir: String,
}

fn default_ob_depth() -> usize {
    50
}

// ─────────────────────────────────────────────────────────────────────────────
// Enums
// ─────────────────────────────────────────────────────────────────────────────

/// Which event type drives the synchronization grid clock.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
pub enum SyncMode {
    /// Orderbook updates advance the grid.
    #[serde(rename = "on_orderbook")]
    OnOrderbook,
    /// Trade events advance the grid.
    #[serde(rename = "on_trade")]
    OnTrade,
    /// Liquidation events advance the grid.
    #[serde(rename = "on_liquidation")]
    OnLiquidation,
    /// An external wall-clock timer advances the grid.
    #[serde(rename = "on_time")]
    OnTime,
}

/// Time unit for [`UpdateFrequency`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
pub enum TimeUnit {
    Nanos,
    Micros,
    Millis,
    Secs,
}

// ─────────────────────────────────────────────────────────────────────────────
// Impl
// ─────────────────────────────────────────────────────────────────────────────

impl MarketSnapshotConfig {
    /// Load and parse a `MarketSnapshotConfig` from a TOML file.
    pub fn from_toml(path: &Path) -> anyhow::Result<Self> {
        let contents = std::fs::read_to_string(path)
            .map_err(|e| anyhow::anyhow!("failed to read {:?}: {}", path, e))?;
        let config: Self = toml::from_str(&contents)
            .map_err(|e| anyhow::anyhow!("failed to parse {:?}: {}", path, e))?;
        Ok(config)
    }

    /// Convert [`UpdateFrequency`] to a grid period in nanoseconds.
    pub fn period_ns(&self) -> u64 {
        let v = self.update_frequency.value;
        match self.update_frequency.unit {
            TimeUnit::Nanos => v,
            TimeUnit::Micros => v * 1_000,
            TimeUnit::Millis => v * 1_000_000,
            TimeUnit::Secs => v * 1_000_000_000,
        }
    }

    /// Wall-clock duration of one full flush interval in nanoseconds.
    ///
    /// `flush_interval_ns = period_ns × flush_threshold`.
    ///
    /// For a 100 ms grid with `flush_threshold = 36000` this returns
    /// 3.6 × 10¹² ns (= 1 hour).
    pub fn flush_interval_ns(&self) -> u64 {
        self.period_ns() * self.pipeline.flush_threshold as u64
    }

    /// Map [`SyncMode`] to the library's [`ClockMode`].
    pub fn clock_mode(&self) -> ClockMode {
        match self.symbol.sync_mode {
            SyncMode::OnOrderbook => ClockMode::OrderbookDriven,
            SyncMode::OnTrade => ClockMode::TradeDriven,
            SyncMode::OnLiquidation => ClockMode::LiquidationDriven,
            SyncMode::OnTime => ClockMode::ExternalClock,
        }
    }

    /// Human-readable label for the active clock mode.
    pub fn clock_mode_label(&self) -> &'static str {
        match self.symbol.sync_mode {
            SyncMode::OnOrderbook => "ClockMode::OrderbookDriven",
            SyncMode::OnTrade => "ClockMode::TradeDriven",
            SyncMode::OnLiquidation => "ClockMode::LiquidationDriven",
            SyncMode::OnTime => "ClockMode::ExternalClock",
        }
    }

    /// Build WSS subscription topics / channels for the configured exchange.
    ///
    /// Returns exchange-specific topic strings that the corresponding
    /// WSS client knows how to subscribe to.
    pub fn wss_streams(&self) -> Vec<String> {
        match self.exchange.name.to_lowercase().as_str() {
            "bybit" => self.bybit_wss_streams(),
            "coinbase" => self.coinbase_wss_channels(),
            "kraken" => self.kraken_wss_channels(),
            "binance" => self.binance_wss_streams(),
            other => {
                tracing::warn!("Unknown exchange '{}'; returning empty streams", other);
                vec![]
            }
        }
    }

    /// Bybit-specific topic construction.
    ///
    /// Funding rates and open interest share the `tickers.{symbol}` topic,
    /// so it is included if *either* is enabled.
    fn bybit_wss_streams(&self) -> Vec<String> {
        let sym = &self.symbol.name;
        let mut streams = Vec::new();

        if self.datatypes.orderbook.enabled {
            streams.push(format!(
                "orderbook.{}.{}",
                self.datatypes.orderbook.depth, sym,
            ));
        }
        if self.datatypes.trades.enabled {
            streams.push(format!("publicTrade.{}", sym));
        }
        if self.datatypes.liquidations.enabled {
            streams.push(format!("allLiquidation.{}", sym));
        }
        if self.datatypes.funding_rates.enabled || self.datatypes.open_interest.enabled {
            streams.push(format!("tickers.{}", sym));
        }

        streams
    }

    /// Coinbase-specific channel list (Advanced Trade spot).
    ///
    /// Coinbase uses separate channel names and product IDs rather than
    /// combined topic strings.  Returns just the channel names; the
    /// [`CoinbaseWssClient`] handles product_id subscription separately.
    fn coinbase_wss_channels(&self) -> Vec<String> {
        let mut channels = Vec::new();
        if self.datatypes.orderbook.enabled {
            channels.push("level2".to_string());
        }
        if self.datatypes.trades.enabled {
            channels.push("market_trades".to_string());
        }
        // Note: liquidations, funding rates, and open interest are NOT
        // available on Coinbase spot.  They require Coinbase INTX.
        channels
    }

    /// Kraken-specific channel list (WebSocket v2 spot).
    ///
    /// Kraken uses `{"method": "subscribe", "params": {"channel": ..., "symbol": [...]}}`
    /// for subscription.  Returns just the channel names; the
    /// [`KrakenWssClient`] handles symbol subscription separately.
    fn kraken_wss_channels(&self) -> Vec<String> {
        let mut channels = Vec::new();
        if self.datatypes.orderbook.enabled {
            channels.push("book".to_string());
        }
        if self.datatypes.trades.enabled {
            channels.push("trade".to_string());
        }
        // Note: liquidations, funding rates, and open interest are NOT
        // available on Kraken spot.  They require Kraken Futures.
        channels
    }

    /// Binance-specific stream construction (spot public).
    ///
    /// Binance uses lowercase symbol + stream suffix format:
    /// `<symbol>@depth@100ms`, `<symbol>@trade`.
    fn binance_wss_streams(&self) -> Vec<String> {
        let sym = self.symbol.name.to_lowercase();
        let mut streams = Vec::new();

        if self.datatypes.orderbook.enabled {
            streams.push(format!("{}@depth@100ms", sym));
        }
        if self.datatypes.trades.enabled {
            streams.push(format!("{}@trade", sym));
        }
        // Note: liquidations, funding rates, and open interest are NOT
        // available on Binance spot public streams.
        streams
    }

    /// Resolve the output directory relative to the workspace root.
    pub fn output_dir(&self) -> PathBuf {
        let manifest_dir = env!("CARGO_MANIFEST_DIR");
        let workspace_root = Path::new(manifest_dir)
            .parent()
            .expect("failed to resolve workspace root");
        workspace_root.join(&self.output.dir)
    }
}

impl std::fmt::Display for SyncMode {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::OnOrderbook => write!(f, "on_orderbook"),
            Self::OnTrade => write!(f, "on_trade"),
            Self::OnLiquidation => write!(f, "on_liquidation"),
            Self::OnTime => write!(f, "on_time"),
        }
    }
}

impl std::fmt::Display for TimeUnit {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Nanos => write!(f, "ns"),
            Self::Micros => write!(f, "µs"),
            Self::Millis => write!(f, "ms"),
            Self::Secs => write!(f, "s"),
        }
    }
}