binance-sdk 61.0.0

/*
 * Alpha WebSocket Market Streams
 *
 * Access Alpha market streams over WebSocket.
 *
 * The version of the OpenAPI document: 1.0.0
 *
 *
 * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
 * https://openapi-generator.tech
 * Do not edit the class manually.
 */

#![allow(unused_imports)]
use async_trait::async_trait;
use derive_builder::Builder;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::{collections::HashMap, sync::Arc};

use crate::alpha::websocket_streams::models;
use crate::common::{
    models::ParamBuildError,
    utils::replace_websocket_streams_placeholders,
    websocket::{WebsocketBase, WebsocketStream, WebsocketStreams, create_stream_handler},
};
use crate::models::StreamId;

#[async_trait]
pub trait Api: Send + Sync {
    async fn aggregate_trade_stream(
        &self,
        params: AggregateTradeStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::AggregateTradeStreamResponse>>>;
    async fn all_book_ticker_stream(
        &self,
        params: AllBookTickerStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::AllBookTickerStreamResponse>>>;
    async fn all_mini_ticker_stream(
        &self,
        params: AllMiniTickerStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::AllMiniTickerStreamResponse>>>;
    async fn all_ticker_stream(
        &self,
        params: AllTickerStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::AllTickerStreamResponse>>>;
    async fn all_tokens24h_ticker_stream(
        &self,
        params: AllTokens24hTickerStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::AllTokens24hTickerStreamResponse>>>;
    async fn book_ticker_stream(
        &self,
        params: BookTickerStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::BookTickerStreamResponse>>>;
    async fn contract_kline_stream(
        &self,
        params: ContractKlineStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::ContractKlineStreamResponse>>>;
    async fn full_depth_stream(
        &self,
        params: FullDepthStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::FullDepthStreamResponse>>>;
    async fn kline_stream(
        &self,
        params: KlineStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::KlineStreamResponse>>>;
    async fn mini_ticker_stream(
        &self,
        params: MiniTickerStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::MiniTickerStreamResponse>>>;
    async fn partial_depth_stream(
        &self,
        params: PartialDepthStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::PartialDepthStreamResponse>>>;
    async fn ticker_stream(
        &self,
        params: TickerStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::TickerStreamResponse>>>;
    async fn trade_stream(
        &self,
        params: TradeStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::TradeStreamResponse>>>;
}

pub struct ApiClient {
    websocket_streams_base: Arc<WebsocketStreams>,
}

impl ApiClient {
    pub fn new(websocket_streams_base: Arc<WebsocketStreams>) -> Self {
        Self {
            websocket_streams_base,
        }
    }
}

#[allow(non_camel_case_types)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ContractKlineStreamIntervalEnum {
    #[serde(rename = "1s")]
    Interval1s,
    #[serde(rename = "1m")]
    Interval1m,
    #[serde(rename = "5m")]
    Interval5m,
    #[serde(rename = "15m")]
    Interval15m,
    #[serde(rename = "1h")]
    Interval1h,
    #[serde(rename = "4h")]
    Interval4h,
    #[serde(rename = "1d")]
    Interval1d,
}

impl ContractKlineStreamIntervalEnum {
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Interval1s => "1s",
            Self::Interval1m => "1m",
            Self::Interval5m => "5m",
            Self::Interval15m => "15m",
            Self::Interval1h => "1h",
            Self::Interval4h => "4h",
            Self::Interval1d => "1d",
        }
    }
}

impl std::str::FromStr for ContractKlineStreamIntervalEnum {
    type Err = Box<dyn std::error::Error + Send + Sync>;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "1s" => Ok(Self::Interval1s),
            "1m" => Ok(Self::Interval1m),
            "5m" => Ok(Self::Interval5m),
            "15m" => Ok(Self::Interval15m),
            "1h" => Ok(Self::Interval1h),
            "4h" => Ok(Self::Interval4h),
            "1d" => Ok(Self::Interval1d),
            other => Err(format!("invalid ContractKlineStreamIntervalEnum: {}", other).into()),
        }
    }
}

#[allow(non_camel_case_types)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum FullDepthStreamIntervalEnum {
    #[serde(rename = "0ms")]
    Interval0ms,
    #[serde(rename = "100ms")]
    Interval100ms,
    #[serde(rename = "500ms")]
    Interval500ms,
}

impl FullDepthStreamIntervalEnum {
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Interval0ms => "0ms",
            Self::Interval100ms => "100ms",
            Self::Interval500ms => "500ms",
        }
    }
}

impl std::str::FromStr for FullDepthStreamIntervalEnum {
    type Err = Box<dyn std::error::Error + Send + Sync>;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "0ms" => Ok(Self::Interval0ms),
            "100ms" => Ok(Self::Interval100ms),
            "500ms" => Ok(Self::Interval500ms),
            other => Err(format!("invalid FullDepthStreamIntervalEnum: {}", other).into()),
        }
    }
}

#[allow(non_camel_case_types)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum KlineStreamIntervalEnum {
    #[serde(rename = "1m")]
    Interval1m,
    #[serde(rename = "3m")]
    Interval3m,
    #[serde(rename = "5m")]
    Interval5m,
    #[serde(rename = "15m")]
    Interval15m,
    #[serde(rename = "30m")]
    Interval30m,
    #[serde(rename = "1h")]
    Interval1h,
    #[serde(rename = "2h")]
    Interval2h,
    #[serde(rename = "4h")]
    Interval4h,
    #[serde(rename = "6h")]
    Interval6h,
    #[serde(rename = "8h")]
    Interval8h,
    #[serde(rename = "12h")]
    Interval12h,
    #[serde(rename = "1d")]
    Interval1d,
    #[serde(rename = "3d")]
    Interval3d,
    #[serde(rename = "1w")]
    Interval1w,
    #[serde(rename = "1M")]
    Interval1M,
}

impl KlineStreamIntervalEnum {
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Interval1m => "1m",
            Self::Interval3m => "3m",
            Self::Interval5m => "5m",
            Self::Interval15m => "15m",
            Self::Interval30m => "30m",
            Self::Interval1h => "1h",
            Self::Interval2h => "2h",
            Self::Interval4h => "4h",
            Self::Interval6h => "6h",
            Self::Interval8h => "8h",
            Self::Interval12h => "12h",
            Self::Interval1d => "1d",
            Self::Interval3d => "3d",
            Self::Interval1w => "1w",
            Self::Interval1M => "1M",
        }
    }
}

impl std::str::FromStr for KlineStreamIntervalEnum {
    type Err = Box<dyn std::error::Error + Send + Sync>;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "1m" => Ok(Self::Interval1m),
            "3m" => Ok(Self::Interval3m),
            "5m" => Ok(Self::Interval5m),
            "15m" => Ok(Self::Interval15m),
            "30m" => Ok(Self::Interval30m),
            "1h" => Ok(Self::Interval1h),
            "2h" => Ok(Self::Interval2h),
            "4h" => Ok(Self::Interval4h),
            "6h" => Ok(Self::Interval6h),
            "8h" => Ok(Self::Interval8h),
            "12h" => Ok(Self::Interval12h),
            "1d" => Ok(Self::Interval1d),
            "3d" => Ok(Self::Interval3d),
            "1w" => Ok(Self::Interval1w),
            "1M" => Ok(Self::Interval1M),
            other => Err(format!("invalid KlineStreamIntervalEnum: {}", other).into()),
        }
    }
}

#[allow(non_camel_case_types)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum PartialDepthStreamLevelsEnum {
    #[serde(rename = "5")]
    Levels5,
    #[serde(rename = "10")]
    Levels10,
    #[serde(rename = "20")]
    Levels20,
}

impl PartialDepthStreamLevelsEnum {
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Levels5 => "5",
            Self::Levels10 => "10",
            Self::Levels20 => "20",
        }
    }
}

impl std::str::FromStr for PartialDepthStreamLevelsEnum {
    type Err = Box<dyn std::error::Error + Send + Sync>;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "5" => Ok(Self::Levels5),
            "10" => Ok(Self::Levels10),
            "20" => Ok(Self::Levels20),
            other => Err(format!("invalid PartialDepthStreamLevelsEnum: {}", other).into()),
        }
    }
}

#[allow(non_camel_case_types)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum PartialDepthStreamIntervalEnum {
    #[serde(rename = "0ms")]
    Interval0ms,
    #[serde(rename = "100ms")]
    Interval100ms,
    #[serde(rename = "500ms")]
    Interval500ms,
}

impl PartialDepthStreamIntervalEnum {
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Interval0ms => "0ms",
            Self::Interval100ms => "100ms",
            Self::Interval500ms => "500ms",
        }
    }
}

impl std::str::FromStr for PartialDepthStreamIntervalEnum {
    type Err = Box<dyn std::error::Error + Send + Sync>;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "0ms" => Ok(Self::Interval0ms),
            "100ms" => Ok(Self::Interval100ms),
            "500ms" => Ok(Self::Interval500ms),
            other => Err(format!("invalid PartialDepthStreamIntervalEnum: {}", other).into()),
        }
    }
}

/// Request parameters for the [`aggregate_trade_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`aggregate_trade_stream`](#method.aggregate_trade_stream).
#[derive(Clone, Debug, Builder, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct AggregateTradeStreamParams {
    /// Symbol to subscribe, in lowercase stream format.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl AggregateTradeStreamParams {
    /// Create a builder for [`aggregate_trade_stream`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — Symbol to subscribe, in lowercase stream format.
    ///
    #[must_use]
    pub fn builder(symbol: String) -> AggregateTradeStreamParamsBuilder {
        AggregateTradeStreamParamsBuilder::default().symbol(symbol)
    }
}
/// Request parameters for the [`all_book_ticker_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`all_book_ticker_stream`](#method.all_book_ticker_stream).
#[derive(Clone, Debug, Builder, Deserialize, Default)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct AllBookTickerStreamParams {
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl AllBookTickerStreamParams {
    /// Create a builder for [`all_book_ticker_stream`].
    ///
    #[must_use]
    pub fn builder() -> AllBookTickerStreamParamsBuilder {
        AllBookTickerStreamParamsBuilder::default()
    }
}
/// Request parameters for the [`all_mini_ticker_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`all_mini_ticker_stream`](#method.all_mini_ticker_stream).
#[derive(Clone, Debug, Builder, Deserialize, Default)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct AllMiniTickerStreamParams {
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl AllMiniTickerStreamParams {
    /// Create a builder for [`all_mini_ticker_stream`].
    ///
    #[must_use]
    pub fn builder() -> AllMiniTickerStreamParamsBuilder {
        AllMiniTickerStreamParamsBuilder::default()
    }
}
/// Request parameters for the [`all_ticker_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`all_ticker_stream`](#method.all_ticker_stream).
#[derive(Clone, Debug, Builder, Deserialize, Default)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct AllTickerStreamParams {
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl AllTickerStreamParams {
    /// Create a builder for [`all_ticker_stream`].
    ///
    #[must_use]
    pub fn builder() -> AllTickerStreamParamsBuilder {
        AllTickerStreamParamsBuilder::default()
    }
}
/// Request parameters for the [`all_tokens24h_ticker_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`all_tokens24h_ticker_stream`](#method.all_tokens24h_ticker_stream).
#[derive(Clone, Debug, Builder, Deserialize, Default)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct AllTokens24hTickerStreamParams {
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl AllTokens24hTickerStreamParams {
    /// Create a builder for [`all_tokens24h_ticker_stream`].
    ///
    #[must_use]
    pub fn builder() -> AllTokens24hTickerStreamParamsBuilder {
        AllTokens24hTickerStreamParamsBuilder::default()
    }
}
/// Request parameters for the [`book_ticker_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`book_ticker_stream`](#method.book_ticker_stream).
#[derive(Clone, Debug, Builder, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct BookTickerStreamParams {
    /// Symbol to subscribe, in lowercase stream format.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl BookTickerStreamParams {
    /// Create a builder for [`book_ticker_stream`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — Symbol to subscribe, in lowercase stream format.
    ///
    #[must_use]
    pub fn builder(symbol: String) -> BookTickerStreamParamsBuilder {
        BookTickerStreamParamsBuilder::default().symbol(symbol)
    }
}
/// Request parameters for the [`contract_kline_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`contract_kline_stream`](#method.contract_kline_stream).
#[derive(Clone, Debug, Builder, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct ContractKlineStreamParams {
    /// Contract address.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "contractAddress")]
    pub contract_address: String,
    /// Chain ID.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "chainId")]
    pub chain_id: String,
    /// Kline interval.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "interval")]
    pub interval: ContractKlineStreamIntervalEnum,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl ContractKlineStreamParams {
    /// Create a builder for [`contract_kline_stream`].
    ///
    /// Required parameters:
    ///
    /// * `contract_address` — Contract address.
    /// * `chain_id` — Chain ID.
    /// * `interval` — Kline interval.
    ///
    #[must_use]
    pub fn builder(
        contract_address: String,
        chain_id: String,
        interval: ContractKlineStreamIntervalEnum,
    ) -> ContractKlineStreamParamsBuilder {
        ContractKlineStreamParamsBuilder::default()
            .contract_address(contract_address)
            .chain_id(chain_id)
            .interval(interval)
    }
}
/// Request parameters for the [`full_depth_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`full_depth_stream`](#method.full_depth_stream).
#[derive(Clone, Debug, Builder, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct FullDepthStreamParams {
    /// Symbol to subscribe, in lowercase stream format.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Update interval.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "interval")]
    pub interval: FullDepthStreamIntervalEnum,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl FullDepthStreamParams {
    /// Create a builder for [`full_depth_stream`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — Symbol to subscribe, in lowercase stream format.
    /// * `interval` — Update interval.
    ///
    #[must_use]
    pub fn builder(
        symbol: String,
        interval: FullDepthStreamIntervalEnum,
    ) -> FullDepthStreamParamsBuilder {
        FullDepthStreamParamsBuilder::default()
            .symbol(symbol)
            .interval(interval)
    }
}
/// Request parameters for the [`kline_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`kline_stream`](#method.kline_stream).
#[derive(Clone, Debug, Builder, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct KlineStreamParams {
    /// Symbol to subscribe, in lowercase stream format.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Kline interval.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "interval")]
    pub interval: KlineStreamIntervalEnum,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl KlineStreamParams {
    /// Create a builder for [`kline_stream`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — Symbol to subscribe, in lowercase stream format.
    /// * `interval` — Kline interval.
    ///
    #[must_use]
    pub fn builder(symbol: String, interval: KlineStreamIntervalEnum) -> KlineStreamParamsBuilder {
        KlineStreamParamsBuilder::default()
            .symbol(symbol)
            .interval(interval)
    }
}
/// Request parameters for the [`mini_ticker_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`mini_ticker_stream`](#method.mini_ticker_stream).
#[derive(Clone, Debug, Builder, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct MiniTickerStreamParams {
    /// Symbol to subscribe, in lowercase stream format.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl MiniTickerStreamParams {
    /// Create a builder for [`mini_ticker_stream`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — Symbol to subscribe, in lowercase stream format.
    ///
    #[must_use]
    pub fn builder(symbol: String) -> MiniTickerStreamParamsBuilder {
        MiniTickerStreamParamsBuilder::default().symbol(symbol)
    }
}
/// Request parameters for the [`partial_depth_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`partial_depth_stream`](#method.partial_depth_stream).
#[derive(Clone, Debug, Builder, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct PartialDepthStreamParams {
    /// Symbol to subscribe, in lowercase stream format.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Depth levels.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "levels")]
    pub levels: PartialDepthStreamLevelsEnum,
    /// Update interval.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "interval")]
    pub interval: PartialDepthStreamIntervalEnum,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl PartialDepthStreamParams {
    /// Create a builder for [`partial_depth_stream`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — Symbol to subscribe, in lowercase stream format.
    /// * `levels` — Depth levels.
    /// * `interval` — Update interval.
    ///
    #[must_use]
    pub fn builder(
        symbol: String,
        levels: PartialDepthStreamLevelsEnum,
        interval: PartialDepthStreamIntervalEnum,
    ) -> PartialDepthStreamParamsBuilder {
        PartialDepthStreamParamsBuilder::default()
            .symbol(symbol)
            .levels(levels)
            .interval(interval)
    }
}
/// Request parameters for the [`ticker_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`ticker_stream`](#method.ticker_stream).
#[derive(Clone, Debug, Builder, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct TickerStreamParams {
    /// Symbol to subscribe, in lowercase stream format.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl TickerStreamParams {
    /// Create a builder for [`ticker_stream`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — Symbol to subscribe, in lowercase stream format.
    ///
    #[must_use]
    pub fn builder(symbol: String) -> TickerStreamParamsBuilder {
        TickerStreamParamsBuilder::default().symbol(symbol)
    }
}
/// Request parameters for the [`trade_stream`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`trade_stream`](#method.trade_stream).
#[derive(Clone, Debug, Builder, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct TradeStreamParams {
    /// Symbol to subscribe, in lowercase stream format.
    ///
    /// This field is **required.
    #[builder(setter(into))]
    #[serde(rename = "symbol")]
    pub symbol: String,
    /// Unique WebSocket request ID.
    ///
    /// This field is **optional.
    #[builder(setter(into), default)]
    #[serde(rename = "id", default)]
    pub id: Option<String>,
}

impl TradeStreamParams {
    /// Create a builder for [`trade_stream`].
    ///
    /// Required parameters:
    ///
    /// * `symbol` — Symbol to subscribe, in lowercase stream format.
    ///
    #[must_use]
    pub fn builder(symbol: String) -> TradeStreamParamsBuilder {
        TradeStreamParamsBuilder::default().symbol(symbol)
    }
}

#[async_trait]
impl Api for ApiClient {
    async fn aggregate_trade_stream(
        &self,
        params: AggregateTradeStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::AggregateTradeStreamResponse>>> {
        let AggregateTradeStreamParams { symbol, id } = params;

        let pairs: &[(&str, Option<String>)] =
            &[("symbol", Some(symbol.clone())), ("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").cloned();

        let stream = replace_websocket_streams_placeholders("/<symbol>@aggTrade", &vars);

        Ok(
            create_stream_handler::<models::AggregateTradeStreamResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn all_book_ticker_stream(
        &self,
        params: AllBookTickerStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::AllBookTickerStreamResponse>>> {
        let AllBookTickerStreamParams { id } = params;

        let pairs: &[(&str, Option<String>)] = &[("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").cloned();

        let stream = replace_websocket_streams_placeholders("/!bookTicker", &vars);

        Ok(
            create_stream_handler::<models::AllBookTickerStreamResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn all_mini_ticker_stream(
        &self,
        params: AllMiniTickerStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::AllMiniTickerStreamResponse>>> {
        let AllMiniTickerStreamParams { id } = params;

        let pairs: &[(&str, Option<String>)] = &[("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").cloned();

        let stream = replace_websocket_streams_placeholders("/!miniTicker@arr", &vars);

        Ok(
            create_stream_handler::<models::AllMiniTickerStreamResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn all_ticker_stream(
        &self,
        params: AllTickerStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::AllTickerStreamResponse>>> {
        let AllTickerStreamParams { id } = params;

        let pairs: &[(&str, Option<String>)] = &[("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").cloned();

        let stream = replace_websocket_streams_placeholders("/!ticker@arr", &vars);

        Ok(create_stream_handler::<models::AllTickerStreamResponse>(
            WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
            stream,
            id_opt.map(|s| {
                if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                    if let Ok(n) = s.parse::<u32>() {
                        return StreamId::Number(n);
                    }
                }
                StreamId::Str(s)
            }),
            None,
        )
        .await)
    }

    async fn all_tokens24h_ticker_stream(
        &self,
        params: AllTokens24hTickerStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::AllTokens24hTickerStreamResponse>>> {
        let AllTokens24hTickerStreamParams { id } = params;

        let pairs: &[(&str, Option<String>)] = &[("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").cloned();

        let stream = replace_websocket_streams_placeholders("/came@allTokens@ticker24", &vars);

        Ok(
            create_stream_handler::<models::AllTokens24hTickerStreamResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn book_ticker_stream(
        &self,
        params: BookTickerStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::BookTickerStreamResponse>>> {
        let BookTickerStreamParams { symbol, id } = params;

        let pairs: &[(&str, Option<String>)] =
            &[("symbol", Some(symbol.clone())), ("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").cloned();

        let stream = replace_websocket_streams_placeholders("/<symbol>@bookTicker", &vars);

        Ok(create_stream_handler::<models::BookTickerStreamResponse>(
            WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
            stream,
            id_opt.map(|s| {
                if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                    if let Ok(n) = s.parse::<u32>() {
                        return StreamId::Number(n);
                    }
                }
                StreamId::Str(s)
            }),
            None,
        )
        .await)
    }

    async fn contract_kline_stream(
        &self,
        params: ContractKlineStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::ContractKlineStreamResponse>>> {
        let ContractKlineStreamParams {
            contract_address,
            chain_id,
            interval,
            id,
        } = params;

        let pairs: &[(&str, Option<String>)] = &[
            ("contractAddress", Some(contract_address.clone())),
            ("chainId", Some(chain_id.clone())),
            ("interval", Some(interval.as_str().to_string())),
            ("id", id.clone()),
        ];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").cloned();

        let stream = replace_websocket_streams_placeholders(
            "/came@<contractAddress>@<chainId>@kline_<interval>",
            &vars,
        );

        Ok(
            create_stream_handler::<models::ContractKlineStreamResponse>(
                WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
                stream,
                id_opt.map(|s| {
                    if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                        if let Ok(n) = s.parse::<u32>() {
                            return StreamId::Number(n);
                        }
                    }
                    StreamId::Str(s)
                }),
                None,
            )
            .await,
        )
    }

    async fn full_depth_stream(
        &self,
        params: FullDepthStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::FullDepthStreamResponse>>> {
        let FullDepthStreamParams {
            symbol,
            interval,
            id,
        } = params;

        let pairs: &[(&str, Option<String>)] = &[
            ("symbol", Some(symbol.clone())),
            ("interval", Some(interval.as_str().to_string())),
            ("id", id.clone()),
        ];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").cloned();

        let stream =
            replace_websocket_streams_placeholders("/<symbol>@fulldepth@<interval>", &vars);

        Ok(create_stream_handler::<models::FullDepthStreamResponse>(
            WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
            stream,
            id_opt.map(|s| {
                if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                    if let Ok(n) = s.parse::<u32>() {
                        return StreamId::Number(n);
                    }
                }
                StreamId::Str(s)
            }),
            None,
        )
        .await)
    }

    async fn kline_stream(
        &self,
        params: KlineStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::KlineStreamResponse>>> {
        let KlineStreamParams {
            symbol,
            interval,
            id,
        } = params;

        let pairs: &[(&str, Option<String>)] = &[
            ("symbol", Some(symbol.clone())),
            ("interval", Some(interval.as_str().to_string())),
            ("id", id.clone()),
        ];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").cloned();

        let stream = replace_websocket_streams_placeholders("/<symbol>@kline_<interval>", &vars);

        Ok(create_stream_handler::<models::KlineStreamResponse>(
            WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
            stream,
            id_opt.map(|s| {
                if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                    if let Ok(n) = s.parse::<u32>() {
                        return StreamId::Number(n);
                    }
                }
                StreamId::Str(s)
            }),
            None,
        )
        .await)
    }

    async fn mini_ticker_stream(
        &self,
        params: MiniTickerStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::MiniTickerStreamResponse>>> {
        let MiniTickerStreamParams { symbol, id } = params;

        let pairs: &[(&str, Option<String>)] =
            &[("symbol", Some(symbol.clone())), ("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").cloned();

        let stream = replace_websocket_streams_placeholders("/<symbol>@miniTicker", &vars);

        Ok(create_stream_handler::<models::MiniTickerStreamResponse>(
            WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
            stream,
            id_opt.map(|s| {
                if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                    if let Ok(n) = s.parse::<u32>() {
                        return StreamId::Number(n);
                    }
                }
                StreamId::Str(s)
            }),
            None,
        )
        .await)
    }

    async fn partial_depth_stream(
        &self,
        params: PartialDepthStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::PartialDepthStreamResponse>>> {
        let PartialDepthStreamParams {
            symbol,
            levels,
            interval,
            id,
        } = params;

        let pairs: &[(&str, Option<String>)] = &[
            ("symbol", Some(symbol.clone())),
            ("levels", Some(levels.as_str().to_string())),
            ("interval", Some(interval.as_str().to_string())),
            ("id", id.clone()),
        ];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").cloned();

        let stream =
            replace_websocket_streams_placeholders("/<symbol>@depth<levels>@<interval>", &vars);

        Ok(create_stream_handler::<models::PartialDepthStreamResponse>(
            WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
            stream,
            id_opt.map(|s| {
                if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                    if let Ok(n) = s.parse::<u32>() {
                        return StreamId::Number(n);
                    }
                }
                StreamId::Str(s)
            }),
            None,
        )
        .await)
    }

    async fn ticker_stream(
        &self,
        params: TickerStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::TickerStreamResponse>>> {
        let TickerStreamParams { symbol, id } = params;

        let pairs: &[(&str, Option<String>)] =
            &[("symbol", Some(symbol.clone())), ("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").cloned();

        let stream = replace_websocket_streams_placeholders("/<symbol>@ticker", &vars);

        Ok(create_stream_handler::<models::TickerStreamResponse>(
            WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
            stream,
            id_opt.map(|s| {
                if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                    if let Ok(n) = s.parse::<u32>() {
                        return StreamId::Number(n);
                    }
                }
                StreamId::Str(s)
            }),
            None,
        )
        .await)
    }

    async fn trade_stream(
        &self,
        params: TradeStreamParams,
    ) -> anyhow::Result<Arc<WebsocketStream<models::TradeStreamResponse>>> {
        let TradeStreamParams { symbol, id } = params;

        let pairs: &[(&str, Option<String>)] =
            &[("symbol", Some(symbol.clone())), ("id", id.clone())];

        let vars: HashMap<_, _> = pairs
            .iter()
            .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
            .collect();

        let id_opt: Option<String> = vars.get("id").cloned();

        let stream = replace_websocket_streams_placeholders("/<symbol>@trade", &vars);

        Ok(create_stream_handler::<models::TradeStreamResponse>(
            WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
            stream,
            id_opt.map(|s| {
                if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
                    if let Ok(n) = s.parse::<u32>() {
                        return StreamId::Number(n);
                    }
                }
                StreamId::Str(s)
            }),
            None,
        )
        .await)
    }
}

#[cfg(all(test, feature = "alpha"))]
mod tests {
    use super::*;
    use crate::TOKIO_SHARED_RT;
    use crate::{
        common::websocket::{WebsocketConnection, WebsocketHandler},
        config::ConfigurationWebsocketStreams,
    };
    use serde_json::json;
    use std::sync::atomic::{AtomicBool, Ordering};
    use tokio::task::yield_now;

    async fn make_streams_base() -> (Arc<WebsocketStreams>, Arc<WebsocketConnection>) {
        let conn = WebsocketConnection::new("test");
        let config = ConfigurationWebsocketStreams::builder()
            .build()
            .expect("Failed to build configuration");
        let streams_base = WebsocketStreams::new(config, vec![conn.clone()], vec![]);
        conn.set_handler(streams_base.clone() as Arc<dyn WebsocketHandler>)
            .await;
        (streams_base, conn)
    }

    #[test]
    fn aggregate_trade_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AggregateTradeStreamParams::builder("alpha_116usdt".to_string())
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let AggregateTradeStreamParams { symbol, id } = params.clone();

            let pairs: &[(&str, Option<String>)] =
                &[("symbol", Some(symbol.clone())), ("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@aggTrade", &vars);
            let ws_stream = api
                .aggregate_trade_stream(params)
                .await
                .expect("aggregate_trade_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn aggregate_trade_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AggregateTradeStreamParams::builder("alpha_116usdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let AggregateTradeStreamParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@aggTrade", &vars);

            let ws_stream = api.aggregate_trade_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::AggregateTradeStreamResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"aggTrade","E":1771828569861,"T":1771828569702,"a":2684294,"f":2684294,"l":2684294,"m":false,"p":"0.08530000","q":"879.24000000","s":"ALPHA_474USDT"}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn aggregate_trade_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AggregateTradeStreamParams::builder("alpha_116usdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let AggregateTradeStreamParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@aggTrade", &vars);

            let ws_stream = api.aggregate_trade_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::AggregateTradeStreamResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"aggTrade","E":1771828569861,"T":1771828569702,"a":2684294,"f":2684294,"l":2684294,"m":false,"p":"0.08530000","q":"879.24000000","s":"ALPHA_474USDT"}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn all_book_ticker_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllBookTickerStreamParams::builder()
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let AllBookTickerStreamParams { id } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!bookTicker", &vars);
            let ws_stream = api
                .all_book_ticker_stream(params)
                .await
                .expect("all_book_ticker_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn all_book_ticker_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllBookTickerStreamParams::builder().id(Some(id.clone())).build().unwrap();

            let AllBookTickerStreamParams {
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!bookTicker", &vars);

            let ws_stream = api.all_book_ticker_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::AllBookTickerStreamResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"bookTicker","E":1773108379067,"T":1773108379054,"u":6663207693,"s":"ALPHA_116USDT","b":"8.30000000","B":"0.91076900","a":"9.30000000","A":"2.00353200"}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn all_book_ticker_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllBookTickerStreamParams::builder().id(Some(id.clone())).build().unwrap();

            let AllBookTickerStreamParams {
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!bookTicker", &vars);

            let ws_stream = api.all_book_ticker_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::AllBookTickerStreamResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"bookTicker","E":1773108379067,"T":1773108379054,"u":6663207693,"s":"ALPHA_116USDT","b":"8.30000000","B":"0.91076900","a":"9.30000000","A":"2.00353200"}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn all_mini_ticker_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllMiniTickerStreamParams::builder()
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let AllMiniTickerStreamParams { id } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!miniTicker@arr", &vars);
            let ws_stream = api
                .all_mini_ticker_stream(params)
                .await
                .expect("all_mini_ticker_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn all_mini_ticker_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllMiniTickerStreamParams::builder().id(Some(id.clone())).build().unwrap();

            let AllMiniTickerStreamParams {
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!miniTicker@arr", &vars);

            let ws_stream = api.all_mini_ticker_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::AllMiniTickerStreamResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"24hrMiniTicker","E":1773109449908,"s":"ALPHA_116USDT","c":"8.40000000","o":"8.40000000","h":"8.50000000","l":"8.30000000","v":"64739.03978900","q":"543810.36005090"}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn all_mini_ticker_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllMiniTickerStreamParams::builder().id(Some(id.clone())).build().unwrap();

            let AllMiniTickerStreamParams {
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!miniTicker@arr", &vars);

            let ws_stream = api.all_mini_ticker_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::AllMiniTickerStreamResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"24hrMiniTicker","E":1773109449908,"s":"ALPHA_116USDT","c":"8.40000000","o":"8.40000000","h":"8.50000000","l":"8.30000000","v":"64739.03978900","q":"543810.36005090"}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn all_ticker_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllTickerStreamParams::builder()
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let AllTickerStreamParams { id } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!ticker@arr", &vars);
            let ws_stream = api
                .all_ticker_stream(params)
                .await
                .expect("all_ticker_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn all_ticker_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllTickerStreamParams::builder().id(Some(id.clone())).build().unwrap();

            let AllTickerStreamParams {
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!ticker@arr", &vars);

            let ws_stream = api.all_ticker_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::AllTickerStreamResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"24hrTicker","E":1773109631569,"s":"ALPHA_116USDT","p":"0.00000000","P":"0.00","w":"8.40003127","c":"8.40000000","Q":"0.49293200","o":"8.40000000","h":"8.50000000","l":"8.30000000","v":"64750.63418500","q":"543907.35169250","O":1773023220000,"C":1773109631555,"F":19847634,"L":19911287,"n":217505}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn all_ticker_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllTickerStreamParams::builder().id(Some(id.clone())).build().unwrap();

            let AllTickerStreamParams {
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/!ticker@arr", &vars);

            let ws_stream = api.all_ticker_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::AllTickerStreamResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"24hrTicker","E":1773109631569,"s":"ALPHA_116USDT","p":"0.00000000","P":"0.00","w":"8.40003127","c":"8.40000000","Q":"0.49293200","o":"8.40000000","h":"8.50000000","l":"8.30000000","v":"64750.63418500","q":"543907.35169250","O":1773023220000,"C":1773109631555,"F":19847634,"L":19911287,"n":217505}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn all_tokens24h_ticker_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllTokens24hTickerStreamParams::builder()
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let AllTokens24hTickerStreamParams { id } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/came@allTokens@ticker24", &vars);
            let ws_stream = api
                .all_tokens24h_ticker_stream(params)
                .await
                .expect("all_tokens24h_ticker_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn all_tokens24h_ticker_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllTokens24hTickerStreamParams::builder().id(Some(id.clone())).build().unwrap();

            let AllTokens24hTickerStreamParams {
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/came@allTokens@ticker24", &vars);

            let ws_stream = api.all_tokens24h_ticker_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::AllTokens24hTickerStreamResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"tickerList","d":[{"ca":"0x8fce...@56","cnt24":10426,"fdv":"72834145.18711072","hc":"6822","liq":"1442996.98151177662835","mc":"13542034.24008014","p":"0.072849536538593663","pc24":"-2.32","s":"1","t":1771825733000,"vol24":"12930712.213702157742688594285"}]}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn all_tokens24h_ticker_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = AllTokens24hTickerStreamParams::builder().id(Some(id.clone())).build().unwrap();

            let AllTokens24hTickerStreamParams {
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/came@allTokens@ticker24", &vars);

            let ws_stream = api.all_tokens24h_ticker_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::AllTokens24hTickerStreamResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"tickerList","d":[{"ca":"0x8fce...@56","cnt24":10426,"fdv":"72834145.18711072","hc":"6822","liq":"1442996.98151177662835","mc":"13542034.24008014","p":"0.072849536538593663","pc24":"-2.32","s":"1","t":1771825733000,"vol24":"12930712.213702157742688594285"}]}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn book_ticker_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = BookTickerStreamParams::builder("alpha_116usdt".to_string())
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let BookTickerStreamParams { symbol, id } = params.clone();

            let pairs: &[(&str, Option<String>)] =
                &[("symbol", Some(symbol.clone())), ("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@bookTicker", &vars);
            let ws_stream = api
                .book_ticker_stream(params)
                .await
                .expect("book_ticker_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn book_ticker_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = BookTickerStreamParams::builder("alpha_116usdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let BookTickerStreamParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@bookTicker", &vars);

            let ws_stream = api.book_ticker_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::BookTickerStreamResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"bookTicker","E":1773108379067,"T":1773108379054,"u":6663207693,"s":"ALPHA_116USDT","b":"8.30000000","B":"0.91076900","a":"9.30000000","A":"2.00353200"}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn book_ticker_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = BookTickerStreamParams::builder("alpha_116usdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let BookTickerStreamParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@bookTicker", &vars);

            let ws_stream = api.book_ticker_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::BookTickerStreamResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"bookTicker","E":1773108379067,"T":1773108379054,"u":6663207693,"s":"ALPHA_116USDT","b":"8.30000000","B":"0.91076900","a":"9.30000000","A":"2.00353200"}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn contract_kline_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = ContractKlineStreamParams::builder(
                "G7vQWurMkMMm2dU3iZpXYFTHT9Biio4F4gZCrwFpKNwG".to_string(),
                "CT_501".to_string(),
                ContractKlineStreamIntervalEnum::Interval1s,
            )
            .id(Some(id.clone()))
            .build()
            .unwrap();

            let ContractKlineStreamParams {
                contract_address,
                chain_id,
                interval,
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("contractAddress", Some(contract_address.clone())),
                ("chainId", Some(chain_id.clone())),
                ("interval", Some(interval.as_str().to_string())),
                ("id", id.clone()),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders(
                "/came@<contractAddress>@<chainId>@kline_<interval>",
                &vars,
            );
            let ws_stream = api
                .contract_kline_stream(params)
                .await
                .expect("contract_kline_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn contract_kline_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = ContractKlineStreamParams::builder("G7vQWurMkMMm2dU3iZpXYFTHT9Biio4F4gZCrwFpKNwG".to_string(),"CT_501".to_string(),ContractKlineStreamIntervalEnum::Interval1s,).id(Some(id.clone())).build().unwrap();

            let ContractKlineStreamParams {
                contract_address,chain_id,interval,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("contractAddress",
                        Some(contract_address.clone())
                ),
                ("chainId",
                        Some(chain_id.clone())
                ),
                ("interval",
                        Some(interval.as_str().to_string())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/came@<contractAddress>@<chainId>@kline_<interval>", &vars);

            let ws_stream = api.contract_kline_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::ContractKlineStreamResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"ca":"G7vQW...@CT_501","e":"kline","k":{"o":"0.15584372258086979622","c":"0.15584957424118055058","h":"0.15584957424118055058","l":"0.15584372258086979622","v":"2.8584092999332775480","ot":1771828655000,"ct":1771828656000,"i":"1s"}}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn contract_kline_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = ContractKlineStreamParams::builder("G7vQWurMkMMm2dU3iZpXYFTHT9Biio4F4gZCrwFpKNwG".to_string(),"CT_501".to_string(),ContractKlineStreamIntervalEnum::Interval1s,).id(Some(id.clone())).build().unwrap();

            let ContractKlineStreamParams {
                contract_address,chain_id,interval,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("contractAddress",
                        Some(contract_address.clone())
                ),
                ("chainId",
                        Some(chain_id.clone())
                ),
                ("interval",
                        Some(interval.as_str().to_string())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/came@<contractAddress>@<chainId>@kline_<interval>", &vars);

            let ws_stream = api.contract_kline_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::ContractKlineStreamResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"ca":"G7vQW...@CT_501","e":"kline","k":{"o":"0.15584372258086979622","c":"0.15584957424118055058","h":"0.15584957424118055058","l":"0.15584372258086979622","v":"2.8584092999332775480","ot":1771828655000,"ct":1771828656000,"i":"1s"}}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn full_depth_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = FullDepthStreamParams::builder(
                "alpha_116usdt".to_string(),
                FullDepthStreamIntervalEnum::Interval0ms,
            )
            .id(Some(id.clone()))
            .build()
            .unwrap();

            let FullDepthStreamParams {
                symbol,
                interval,
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol", Some(symbol.clone())),
                ("interval", Some(interval.as_str().to_string())),
                ("id", id.clone()),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream =
                replace_websocket_streams_placeholders("/<symbol>@fulldepth@<interval>", &vars);
            let ws_stream = api
                .full_depth_stream(params)
                .await
                .expect("full_depth_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn full_depth_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = FullDepthStreamParams::builder("alpha_116usdt".to_string(),FullDepthStreamIntervalEnum::Interval0ms,).id(Some(id.clone())).build().unwrap();

            let FullDepthStreamParams {
                symbol,interval,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("interval",
                        Some(interval.as_str().to_string())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@fulldepth@<interval>", &vars);

            let ws_stream = api.full_depth_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::FullDepthStreamResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"depthUpdate","E":1771828614381,"T":1771828614258,"U":42648947917,"u":42648948229,"pu":42648947321,"s":"ALPHA_474USDT","b":[["0.16479999","0"],["0.16480000","18.68000000"]],"a":[["0.16490000","500.00000000"],["0.16500000","0"]]}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn full_depth_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = FullDepthStreamParams::builder("alpha_116usdt".to_string(),FullDepthStreamIntervalEnum::Interval0ms,).id(Some(id.clone())).build().unwrap();

            let FullDepthStreamParams {
                symbol,interval,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("interval",
                        Some(interval.as_str().to_string())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@fulldepth@<interval>", &vars);

            let ws_stream = api.full_depth_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::FullDepthStreamResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"depthUpdate","E":1771828614381,"T":1771828614258,"U":42648947917,"u":42648948229,"pu":42648947321,"s":"ALPHA_474USDT","b":[["0.16479999","0"],["0.16480000","18.68000000"]],"a":[["0.16490000","500.00000000"],["0.16500000","0"]]}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn kline_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = KlineStreamParams::builder(
                "alpha_116usdt".to_string(),
                KlineStreamIntervalEnum::Interval1m,
            )
            .id(Some(id.clone()))
            .build()
            .unwrap();

            let KlineStreamParams {
                symbol,
                interval,
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol", Some(symbol.clone())),
                ("interval", Some(interval.as_str().to_string())),
                ("id", id.clone()),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream =
                replace_websocket_streams_placeholders("/<symbol>@kline_<interval>", &vars);
            let ws_stream = api
                .kline_stream(params)
                .await
                .expect("kline_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn kline_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = KlineStreamParams::builder("alpha_116usdt".to_string(),KlineStreamIntervalEnum::Interval1m,).id(Some(id.clone())).build().unwrap();

            let KlineStreamParams {
                symbol,interval,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("interval",
                        Some(interval.as_str().to_string())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@kline_<interval>", &vars);

            let ws_stream = api.kline_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::KlineStreamResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"kline","E":1773111055144,"s":"ALPHA_116USDT","k":{"t":1773111000000,"T":1773111059999,"s":"ALPHA_116USDT","i":"1m","f":19912696,"L":19912765,"o":"8.40000000","c":"8.40000000","h":"8.40000000","l":"8.40000000","v":"63.64876300","n":218,"x":false,"q":"534.64960920","V":"63.64876300","Q":"534.64960920","B":"0"}}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn kline_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = KlineStreamParams::builder("alpha_116usdt".to_string(),KlineStreamIntervalEnum::Interval1m,).id(Some(id.clone())).build().unwrap();

            let KlineStreamParams {
                symbol,interval,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("interval",
                        Some(interval.as_str().to_string())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@kline_<interval>", &vars);

            let ws_stream = api.kline_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::KlineStreamResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"kline","E":1773111055144,"s":"ALPHA_116USDT","k":{"t":1773111000000,"T":1773111059999,"s":"ALPHA_116USDT","i":"1m","f":19912696,"L":19912765,"o":"8.40000000","c":"8.40000000","h":"8.40000000","l":"8.40000000","v":"63.64876300","n":218,"x":false,"q":"534.64960920","V":"63.64876300","Q":"534.64960920","B":"0"}}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn mini_ticker_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = MiniTickerStreamParams::builder("alpha_116usdt".to_string())
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let MiniTickerStreamParams { symbol, id } = params.clone();

            let pairs: &[(&str, Option<String>)] =
                &[("symbol", Some(symbol.clone())), ("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@miniTicker", &vars);
            let ws_stream = api
                .mini_ticker_stream(params)
                .await
                .expect("mini_ticker_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn mini_ticker_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = MiniTickerStreamParams::builder("alpha_116usdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let MiniTickerStreamParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@miniTicker", &vars);

            let ws_stream = api.mini_ticker_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::MiniTickerStreamResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"24hrMiniTicker","E":1773109449908,"s":"ALPHA_116USDT","c":"8.40000000","o":"8.40000000","h":"8.50000000","l":"8.30000000","v":"64739.03978900","q":"543810.36005090"}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn mini_ticker_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = MiniTickerStreamParams::builder("alpha_116usdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let MiniTickerStreamParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@miniTicker", &vars);

            let ws_stream = api.mini_ticker_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::MiniTickerStreamResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"24hrMiniTicker","E":1773109449908,"s":"ALPHA_116USDT","c":"8.40000000","o":"8.40000000","h":"8.50000000","l":"8.30000000","v":"64739.03978900","q":"543810.36005090"}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn partial_depth_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = PartialDepthStreamParams::builder(
                "alpha_116usdt".to_string(),
                PartialDepthStreamLevelsEnum::Levels5,
                PartialDepthStreamIntervalEnum::Interval0ms,
            )
            .id(Some(id.clone()))
            .build()
            .unwrap();

            let PartialDepthStreamParams {
                symbol,
                levels,
                interval,
                id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol", Some(symbol.clone())),
                ("levels", Some(levels.as_str().to_string())),
                ("interval", Some(interval.as_str().to_string())),
                ("id", id.clone()),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream =
                replace_websocket_streams_placeholders("/<symbol>@depth<levels>@<interval>", &vars);
            let ws_stream = api
                .partial_depth_stream(params)
                .await
                .expect("partial_depth_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn partial_depth_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = PartialDepthStreamParams::builder("alpha_116usdt".to_string(),PartialDepthStreamLevelsEnum::Levels5,PartialDepthStreamIntervalEnum::Interval0ms,).id(Some(id.clone())).build().unwrap();

            let PartialDepthStreamParams {
                symbol,levels,interval,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("levels",
                        Some(levels.as_str().to_string())
                ),
                ("interval",
                        Some(interval.as_str().to_string())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@depth<levels>@<interval>", &vars);

            let ws_stream = api.partial_depth_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::PartialDepthStreamResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"depthUpdate","E":1773110222878,"T":1773110222767,"U":6663367135,"u":6663367147,"pu":6663366971,"s":"ALPHA_116USDT","b":[["0.16479999","0"],["0.16480000","18.68000000"]],"a":[["0.16490000","500.00000000"],["0.16500000","0"]]}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn partial_depth_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = PartialDepthStreamParams::builder("alpha_116usdt".to_string(),PartialDepthStreamLevelsEnum::Levels5,PartialDepthStreamIntervalEnum::Interval0ms,).id(Some(id.clone())).build().unwrap();

            let PartialDepthStreamParams {
                symbol,levels,interval,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("levels",
                        Some(levels.as_str().to_string())
                ),
                ("interval",
                        Some(interval.as_str().to_string())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@depth<levels>@<interval>", &vars);

            let ws_stream = api.partial_depth_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::PartialDepthStreamResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"depthUpdate","E":1773110222878,"T":1773110222767,"U":6663367135,"u":6663367147,"pu":6663366971,"s":"ALPHA_116USDT","b":[["0.16479999","0"],["0.16480000","18.68000000"]],"a":[["0.16490000","500.00000000"],["0.16500000","0"]]}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn ticker_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = TickerStreamParams::builder("alpha_116usdt".to_string())
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let TickerStreamParams { symbol, id } = params.clone();

            let pairs: &[(&str, Option<String>)] =
                &[("symbol", Some(symbol.clone())), ("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@ticker", &vars);
            let ws_stream = api
                .ticker_stream(params)
                .await
                .expect("ticker_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn ticker_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = TickerStreamParams::builder("alpha_116usdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let TickerStreamParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@ticker", &vars);

            let ws_stream = api.ticker_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::TickerStreamResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"24hrTicker","E":1773109631569,"s":"ALPHA_116USDT","p":"0.00000000","P":"0.00","w":"8.40003127","c":"8.40000000","Q":"0.49293200","o":"8.40000000","h":"8.50000000","l":"8.30000000","v":"64750.63418500","q":"543907.35169250","O":1773023220000,"C":1773109631555,"F":19847634,"L":19911287,"n":217505}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn ticker_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = TickerStreamParams::builder("alpha_116usdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let TickerStreamParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@ticker", &vars);

            let ws_stream = api.ticker_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::TickerStreamResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"24hrTicker","E":1773109631569,"s":"ALPHA_116USDT","p":"0.00000000","P":"0.00","w":"8.40003127","c":"8.40000000","Q":"0.49293200","o":"8.40000000","h":"8.50000000","l":"8.30000000","v":"64750.63418500","q":"543907.35169250","O":1773023220000,"C":1773109631555,"F":19847634,"L":19911287,"n":217505}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }

    #[test]
    fn trade_stream_should_execute_successfully() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, _) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = TradeStreamParams::builder("alpha_116usdt".to_string())
                .id(Some(id.clone()))
                .build()
                .unwrap();

            let TradeStreamParams { symbol, id } = params.clone();

            let pairs: &[(&str, Option<String>)] =
                &[("symbol", Some(symbol.clone())), ("id", id.clone())];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@trade", &vars);
            let ws_stream = api
                .trade_stream(params)
                .await
                .expect("trade_stream should return a WebsocketStream");

            assert!(
                streams_base.is_subscribed(&stream).await,
                "expected stream '{stream}' to be subscribed"
            );
            assert_eq!(ws_stream.id, Some(StreamId::Str("test-id-123".to_string())));
        });
    }

    #[test]
    fn trade_stream_should_handle_incoming_message() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = TradeStreamParams::builder("alpha_116usdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let TradeStreamParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@trade", &vars);

            let ws_stream = api.trade_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_with_message = called.clone();
            ws_stream.on_message(move |_payload: models::TradeStreamResponse| {
                called_with_message.store(true, Ordering::SeqCst);
            });

            let payload: Value = serde_json::from_str(r#"{"e":"trade","E":1773110023891,"T":1773110023877,"s":"ALPHA_116USDT","t":19911650,"p":"8.40000000","q":"0.97915700","m":false}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;
            yield_now().await;

            assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
        });
    }

    #[test]
    fn trade_stream_should_not_fire_after_unsubscribe() {
        TOKIO_SHARED_RT.block_on(async {
            let (streams_base, conn) = make_streams_base().await;
            let api = ApiClient::new(streams_base.clone());

            let id = "test-id-123".to_string();

            let params = TradeStreamParams::builder("alpha_116usdt".to_string(),).id(Some(id.clone())).build().unwrap();

            let TradeStreamParams {
                symbol,id,
            } = params.clone();

            let pairs: &[(&str, Option<String>)] = &[
                ("symbol",
                        Some(symbol.clone())
                ),
                ("id",
                        id.clone()
                ),
            ];

            let vars: HashMap<_, _> = pairs
                .iter()
                .filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
                .collect();
            let stream = replace_websocket_streams_placeholders("/<symbol>@trade", &vars);

            let ws_stream = api.trade_stream(params).await.unwrap();

            let called = Arc::new(AtomicBool::new(false));
            let called_clone = called.clone();
            ws_stream.on_message(move |_payload: models::TradeStreamResponse| {
                called_clone.store(true, Ordering::SeqCst);
            });

            assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");

            ws_stream.unsubscribe().await;

            let payload: Value = serde_json::from_str(r#"{"e":"trade","E":1773110023891,"T":1773110023877,"s":"ALPHA_116USDT","t":19911650,"p":"8.40000000","q":"0.97915700","m":false}"#).unwrap_or_else(|_| serde_json::json!({}));
            let msg = json!({
                "stream": stream,
                "data": payload,
            });

            streams_base.on_message(msg.to_string(), conn.clone()).await;

            yield_now().await;

            assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
        });
    }
}