ticksupply 0.2.1

//! catalog — Exchanges and instruments.
//!
//! Datastreams live under their own module, [`crate::resources::datastreams`].
//!
//! # Examples
//!
//! ```no_run
//! # async fn example() -> ticksupply::Result<()> {
//! let client = ticksupply::Client::new()?;
//!
//! // All exchanges.
//! for ex in client.exchanges().list().await? {
//!     println!("{}: {}", ex.code, ex.display_name);
//! }
//!
//! // Filtered instrument list (paginated).
//! let page = client.exchanges()
//!     .list_instruments("binance")
//!     .search("BTC")
//!     .limit(10)
//!     .send().await?;
//!
//! // Datastreams on one instrument.
//! let streams = client.exchanges()
//!     .list_datastreams("binance", "BTCUSDT")
//!     .stream_type("trade")
//!     .send().await?;
//! # let _ = (page, streams);
//! # Ok(()) }
//! ```

use futures::Stream;
use serde::Deserialize;

use crate::client::Client;
use crate::error::Result;
use crate::http::{send, RequestOpts};
use crate::pagination::Page;

/// A public exchange.
#[derive(Debug, Clone, Deserialize)]
pub struct Exchange {
    /// Canonical short code (e.g. `"binance"`, `"okx_spot"`).
    pub code: String,
    /// Human-friendly name.
    pub display_name: String,
}

/// A trading instrument (symbol) on an exchange.
#[derive(Debug, Clone, Deserialize)]
pub struct Instrument {
    /// Exchange-native symbol (e.g. `"BTCUSDT"`).
    pub symbol: String,
    /// Base asset, when known.
    #[serde(default)]
    pub base: Option<String>,
    /// Quote asset, when known.
    #[serde(default)]
    pub quote: Option<String>,
    /// Instrument type (e.g. `"spot"`, `"perpetual"`), when reported.
    #[serde(default)]
    pub instrument_type: Option<String>,
}

/// Minimal identifying info for a data stream.
#[derive(Debug, Clone, Deserialize)]
pub struct DatastreamInfo {
    /// Numeric datastream identifier used when creating subscriptions / exports.
    pub datastream_id: i64,
    /// Exchange code.
    pub exchange: String,
    /// Instrument symbol.
    pub instrument: String,
    /// Stream type code.
    pub stream_type: String,
    /// Wire format identifier.
    pub wire_format: String,
}

/// Accessor for catalog endpoints.
pub struct ExchangesResource<'a> {
    pub(crate) client: &'a Client,
}

impl<'a> ExchangesResource<'a> {
    /// Lists all public exchanges.
    ///
    /// # Errors
    ///
    /// - [`crate::Error::Authentication`] if the API key is invalid.
    /// - [`crate::Error::Network`] on transport failure.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let exchanges = ticksupply::Client::new()?.exchanges().list().await?;
    /// # let _ = exchanges;
    /// # Ok(()) }
    /// ```
    pub async fn list(&self) -> Result<Vec<Exchange>> {
        send::<_, ()>(
            self.client,
            reqwest::Method::GET,
            "/exchanges",
            None,
            None,
            RequestOpts::default(),
        )
        .await
    }

    /// Returns a builder for listing instruments on an exchange.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// let page = client.exchanges()
    ///     .list_instruments("binance")
    ///     .search("BTC")
    ///     .limit(10)
    ///     .send().await?;
    /// # let _ = page;
    /// # Ok(()) }
    /// ```
    pub fn list_instruments(&self, exchange: impl Into<String>) -> ListInstrumentsRequest<'a> {
        ListInstrumentsRequest {
            client: self.client,
            exchange: exchange.into(),
            search: None,
            base: None,
            quote: None,
            instrument_type: None,
            limit: None,
            page_token: None,
        }
    }

    /// Returns a builder for listing datastreams on a specific
    /// exchange + instrument.
    ///
    /// Equivalent to Python's `client.exchanges.list_datastreams(exchange, instrument)`.
    /// Hits the path-param route `GET /v1/exchanges/{exchange}/instruments/{instrument}/datastreams`;
    /// use [`crate::Client::datastreams`] for the flat, query-filter route.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// let page = client.exchanges()
    ///     .list_datastreams("binance", "BTCUSDT")
    ///     .stream_type("trade")
    ///     .send().await?;
    /// # let _ = page;
    /// # Ok(()) }
    /// ```
    pub fn list_datastreams(
        &self,
        exchange: impl Into<String>,
        instrument: impl Into<String>,
    ) -> ListExchangeDatastreamsRequest<'a> {
        ListExchangeDatastreamsRequest {
            client: self.client,
            exchange: exchange.into(),
            instrument: instrument.into(),
            stream_type: None,
            wire_format: None,
            limit: None,
            page_token: None,
        }
    }
}

/// Builder for `GET /v1/exchanges/{code}/instruments`.
pub struct ListInstrumentsRequest<'a> {
    client: &'a Client,
    exchange: String,
    search: Option<String>,
    base: Option<String>,
    quote: Option<String>,
    instrument_type: Option<String>,
    limit: Option<u32>,
    page_token: Option<String>,
}

impl<'a> ListInstrumentsRequest<'a> {
    /// Filters results by a case-insensitive symbol substring match.
    pub fn search(mut self, s: impl Into<String>) -> Self {
        self.search = Some(s.into());
        self
    }

    /// Filters results by base asset (e.g. `"BTC"`).
    pub fn base(mut self, b: impl Into<String>) -> Self {
        self.base = Some(b.into());
        self
    }

    /// Filters results by quote asset (e.g. `"USDT"`).
    pub fn quote(mut self, q: impl Into<String>) -> Self {
        self.quote = Some(q.into());
        self
    }

    /// Filters results by instrument type (e.g. `"spot"`, `"perpetual"`,
    /// `"futures"`, `"option"`).
    pub fn instrument_type(mut self, t: impl Into<String>) -> Self {
        self.instrument_type = Some(t.into());
        self
    }

    /// Sets the maximum results per page (default 100, max 1000).
    pub fn limit(mut self, n: u32) -> Self {
        self.limit = Some(n);
        self
    }

    /// Sets the page token returned by a prior response.
    pub fn page_token(mut self, t: impl Into<String>) -> Self {
        self.page_token = Some(t.into());
        self
    }

    fn query(&self) -> Vec<(&'static str, String)> {
        let mut q = Vec::new();
        if let Some(s) = &self.search {
            q.push(("search", s.clone()));
        }
        if let Some(s) = &self.base {
            q.push(("base", s.clone()));
        }
        if let Some(s) = &self.quote {
            q.push(("quote", s.clone()));
        }
        if let Some(s) = &self.instrument_type {
            q.push(("instrument_type", s.clone()));
        }
        if let Some(n) = self.limit {
            q.push(("limit", n.to_string()));
        }
        if let Some(t) = &self.page_token {
            q.push(("page_token", t.clone()));
        }
        q
    }

    /// Fetches a single page of results.
    ///
    /// # Errors
    ///
    /// - [`crate::Error::NotFound`] if the exchange code is unknown.
    /// - [`crate::Error::Authentication`] on invalid credentials.
    /// - [`crate::Error::Validation`] if filter parameters are invalid.
    /// - [`crate::Error::Network`] on transport failure.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// let page = client.exchanges()
    ///     .list_instruments("binance")
    ///     .search("BTC")
    ///     .send().await?;
    /// # let _ = page;
    /// # Ok(()) }
    /// ```
    pub async fn send(self) -> Result<Page<Instrument>> {
        let path = format!("/exchanges/{}/instruments", self.exchange);
        let q = self.query();
        send::<_, ()>(
            self.client,
            reqwest::Method::GET,
            &path,
            Some(q.as_slice()),
            None,
            RequestOpts::default(),
        )
        .await
    }

    /// Auto-paginates across all matching pages, yielding each instrument.
    ///
    /// Each yielded item is a [`Result`] that surfaces the same errors as
    /// [`Self::send`] if a page fetch fails.
    ///
    /// Streaming always starts from the first page; any `page_token`
    /// previously set on the builder is ignored. Filters and `limit` are
    /// preserved across pages.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// use futures::StreamExt;
    /// let client = ticksupply::Client::new()?;
    /// let mut s = Box::pin(client.exchanges().list_instruments("binance").stream());
    /// while let Some(inst) = s.next().await {
    ///     let _inst = inst?;
    /// }
    /// # Ok(()) }
    /// ```
    pub fn stream(self) -> impl Stream<Item = Result<Instrument>> + 'a {
        let Self {
            client,
            exchange,
            search,
            base,
            quote,
            instrument_type,
            limit,
            ..
        } = self;
        async_stream::try_stream! {
            let mut page_token: Option<String> = None;
            loop {
                let req = ListInstrumentsRequest {
                    client,
                    exchange: exchange.clone(),
                    search: search.clone(),
                    base: base.clone(),
                    quote: quote.clone(),
                    instrument_type: instrument_type.clone(),
                    limit,
                    page_token: page_token.clone(),
                };
                let page = req.send().await?;
                for item in page.items { yield item; }
                match page.next_page_token {
                    Some(t) => page_token = Some(t),
                    None => break,
                }
            }
        }
    }
}

/// Builder for `GET /v1/exchanges/{exchange}/instruments/{instrument}/datastreams`.
pub struct ListExchangeDatastreamsRequest<'a> {
    client: &'a Client,
    exchange: String,
    instrument: String,
    stream_type: Option<String>,
    wire_format: Option<String>,
    limit: Option<u32>,
    page_token: Option<String>,
}

impl<'a> ListExchangeDatastreamsRequest<'a> {
    /// Filters results by stream type code (e.g. `"trade"`, `"depth"`, `"ticker"`).
    pub fn stream_type(mut self, t: impl Into<String>) -> Self {
        self.stream_type = Some(t.into());
        self
    }

    /// Filters results by wire format identifier (e.g. `"json"`).
    pub fn wire_format(mut self, w: impl Into<String>) -> Self {
        self.wire_format = Some(w.into());
        self
    }

    /// Sets the maximum results per page (default 100, max 1000).
    pub fn limit(mut self, n: u32) -> Self {
        self.limit = Some(n);
        self
    }

    /// Sets the page token returned by a prior response.
    pub fn page_token(mut self, t: impl Into<String>) -> Self {
        self.page_token = Some(t.into());
        self
    }

    fn query(&self) -> Vec<(&'static str, String)> {
        let mut q = Vec::new();
        if let Some(s) = &self.stream_type {
            q.push(("stream_type", s.clone()));
        }
        if let Some(s) = &self.wire_format {
            q.push(("wire_format", s.clone()));
        }
        if let Some(n) = self.limit {
            q.push(("limit", n.to_string()));
        }
        if let Some(t) = &self.page_token {
            q.push(("page_token", t.clone()));
        }
        q
    }

    /// Fetches a single page of results.
    ///
    /// # Errors
    ///
    /// - [`crate::Error::NotFound`] if the exchange or instrument is unknown.
    /// - [`crate::Error::Authentication`] on invalid credentials.
    /// - [`crate::Error::Validation`] if filter parameters are invalid.
    /// - [`crate::Error::Network`] on transport failure.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// let page = client.exchanges()
    ///     .list_datastreams("binance", "BTCUSDT")
    ///     .stream_type("trade")
    ///     .send().await?;
    /// # let _ = page;
    /// # Ok(()) }
    /// ```
    pub async fn send(self) -> Result<Page<DatastreamInfo>> {
        let path = format!(
            "/exchanges/{}/instruments/{}/datastreams",
            self.exchange, self.instrument
        );
        let q = self.query();
        send::<_, ()>(
            self.client,
            reqwest::Method::GET,
            &path,
            Some(q.as_slice()),
            None,
            RequestOpts::default(),
        )
        .await
    }

    /// Auto-paginates across all matching pages, yielding each datastream.
    ///
    /// Equivalent to iterating Python's `client.exchanges.list_datastreams(...)`
    /// across pages. Each yielded item is a [`Result`] that surfaces the same
    /// errors as [`Self::send`] if a page fetch fails.
    ///
    /// Streaming always starts from the first page; any `page_token`
    /// previously set on the builder is ignored. Filters and `limit` are
    /// preserved across pages.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// use futures::StreamExt;
    /// let client = ticksupply::Client::new()?;
    /// let mut s = Box::pin(
    ///     client.exchanges().list_datastreams("binance", "BTCUSDT").stream()
    /// );
    /// while let Some(ds) = s.next().await {
    ///     let _ds = ds?;
    /// }
    /// # Ok(()) }
    /// ```
    pub fn stream(self) -> impl Stream<Item = Result<DatastreamInfo>> + 'a {
        let Self {
            client,
            exchange,
            instrument,
            stream_type,
            wire_format,
            limit,
            ..
        } = self;
        async_stream::try_stream! {
            let mut page_token: Option<String> = None;
            loop {
                let req = ListExchangeDatastreamsRequest {
                    client,
                    exchange: exchange.clone(),
                    instrument: instrument.clone(),
                    stream_type: stream_type.clone(),
                    wire_format: wire_format.clone(),
                    limit,
                    page_token: page_token.clone(),
                };
                let page = req.send().await?;
                for item in page.items { yield item; }
                match page.next_page_token {
                    Some(t) => page_token = Some(t),
                    None => break,
                }
            }
        }
    }
}