Struct WssMarketClient 

pub struct WssMarketClient { /* private fields */ }

Reconnecting client for the market channel.

Implementations

impl WssMarketClient

pub fn new() -> Self

Create a new instance using the default Polymarket WSS base.

Examples found in repository

examples/wss_market.rs (line 44)

async fn main() -> Result<()> {
    let base_url =
        env::var("POLY_API_URL").unwrap_or_else(|_| "https://clob.polymarket.com".to_string());
    let clob = ClobClient::new(&base_url);

    let min_liquidity = env::var("POLY_WSS_MIN_LIQUIDITY")
        .ok()
        .and_then(|value| Decimal::from_str(&value).ok())
        .unwrap_or_else(|| Decimal::from(1_000_000));

    let params = GammaListParams {
        limit: Some(50),
        liquidity_num_min: Some(min_liquidity),
        ..Default::default()
    };

    let response = clob.get_markets(None, Some(&params)).await?;
    let market = pick_liquid_market(&response.data, min_liquidity)?;

    println!(
        "Selected market {} (liquidity={:?})",
        market.condition_id, market.liquidity_num
    );

    let asset_ids = derive_asset_ids(market).unwrap_or_else(|| Vec::new());

    if asset_ids.is_empty() {
        return Err(PolyError::validation(
            "failed to derive asset IDs for the selected market",
        ));
    }

    let mut client = WssMarketClient::new();
    client.subscribe(asset_ids.clone()).await?;

    println!("Subscribed to market channel for assets={:?}", asset_ids);

    for _ in 0..20 {
        match client.next_event().await {
            Ok(WssMarketEvent::PriceChange(change)) => {
                println!(
                    "price_change for {}: {:?}",
                    change.market, change.price_changes
                );
            }
            Ok(WssMarketEvent::Book(book)) => {
                println!(
                    "book {} bids={} asks={}",
                    book.market,
                    book.bids.len(),
                    book.asks.len()
                );
            }
            Ok(WssMarketEvent::TickSizeChange(change)) => {
                println!(
                    "tick size change {} from {} to {}",
                    change.market, change.old_tick_size, change.new_tick_size
                );
            }
            Ok(WssMarketEvent::LastTrade(trade)) => {
                println!(
                    "last_trade {} {:?}@{}",
                    trade.market, trade.side, trade.price
                );
            }
            Err(err) => {
                eprintln!("stream error: {}", err);
                break;
            }
        }
    }

    Ok(())
}

pub fn with_url(url: &str) -> Self

Create a new client against a custom endpoint (useful for tests).

pub fn stats(&self) -> WssStats

Access connection stats for observability.

pub async fn subscribe(&mut self, asset_ids: Vec<String>) -> Result<()>

Subscribe to the market channel for the provided token/market IDs.

Examples found in repository

examples/wss_market.rs (line 45)

async fn main() -> Result<()> {
    let base_url =
        env::var("POLY_API_URL").unwrap_or_else(|_| "https://clob.polymarket.com".to_string());
    let clob = ClobClient::new(&base_url);

    let min_liquidity = env::var("POLY_WSS_MIN_LIQUIDITY")
        .ok()
        .and_then(|value| Decimal::from_str(&value).ok())
        .unwrap_or_else(|| Decimal::from(1_000_000));

    let params = GammaListParams {
        limit: Some(50),
        liquidity_num_min: Some(min_liquidity),
        ..Default::default()
    };

    let response = clob.get_markets(None, Some(&params)).await?;
    let market = pick_liquid_market(&response.data, min_liquidity)?;

    println!(
        "Selected market {} (liquidity={:?})",
        market.condition_id, market.liquidity_num
    );

    let asset_ids = derive_asset_ids(market).unwrap_or_else(|| Vec::new());

    if asset_ids.is_empty() {
        return Err(PolyError::validation(
            "failed to derive asset IDs for the selected market",
        ));
    }

    let mut client = WssMarketClient::new();
    client.subscribe(asset_ids.clone()).await?;

    println!("Subscribed to market channel for assets={:?}", asset_ids);

    for _ in 0..20 {
        match client.next_event().await {
            Ok(WssMarketEvent::PriceChange(change)) => {
                println!(
                    "price_change for {}: {:?}",
                    change.market, change.price_changes
                );
            }
            Ok(WssMarketEvent::Book(book)) => {
                println!(
                    "book {} bids={} asks={}",
                    book.market,
                    book.bids.len(),
                    book.asks.len()
                );
            }
            Ok(WssMarketEvent::TickSizeChange(change)) => {
                println!(
                    "tick size change {} from {} to {}",
                    change.market, change.old_tick_size, change.new_tick_size
                );
            }
            Ok(WssMarketEvent::LastTrade(trade)) => {
                println!(
                    "last_trade {} {:?}@{}",
                    trade.market, trade.side, trade.price
                );
            }
            Err(err) => {
                eprintln!("stream error: {}", err);
                break;
            }
        }
    }

    Ok(())
}

pub async fn next_event(&mut self) -> Result<WssMarketEvent>

Read the next market channel event, reconnecting transparently when the socket drops.

Examples found in repository

examples/wss_market.rs (line 50)

async fn main() -> Result<()> {
    let base_url =
        env::var("POLY_API_URL").unwrap_or_else(|_| "https://clob.polymarket.com".to_string());
    let clob = ClobClient::new(&base_url);

    let min_liquidity = env::var("POLY_WSS_MIN_LIQUIDITY")
        .ok()
        .and_then(|value| Decimal::from_str(&value).ok())
        .unwrap_or_else(|| Decimal::from(1_000_000));

    let params = GammaListParams {
        limit: Some(50),
        liquidity_num_min: Some(min_liquidity),
        ..Default::default()
    };

    let response = clob.get_markets(None, Some(&params)).await?;
    let market = pick_liquid_market(&response.data, min_liquidity)?;

    println!(
        "Selected market {} (liquidity={:?})",
        market.condition_id, market.liquidity_num
    );

    let asset_ids = derive_asset_ids(market).unwrap_or_else(|| Vec::new());

    if asset_ids.is_empty() {
        return Err(PolyError::validation(
            "failed to derive asset IDs for the selected market",
        ));
    }

    let mut client = WssMarketClient::new();
    client.subscribe(asset_ids.clone()).await?;

    println!("Subscribed to market channel for assets={:?}", asset_ids);

    for _ in 0..20 {
        match client.next_event().await {
            Ok(WssMarketEvent::PriceChange(change)) => {
                println!(
                    "price_change for {}: {:?}",
                    change.market, change.price_changes
                );
            }
            Ok(WssMarketEvent::Book(book)) => {
                println!(
                    "book {} bids={} asks={}",
                    book.market,
                    book.bids.len(),
                    book.asks.len()
                );
            }
            Ok(WssMarketEvent::TickSizeChange(change)) => {
                println!(
                    "tick size change {} from {} to {}",
                    change.market, change.old_tick_size, change.new_tick_size
                );
            }
            Ok(WssMarketEvent::LastTrade(trade)) => {
                println!(
                    "last_trade {} {:?}@{}",
                    trade.market, trade.side, trade.price
                );
            }
            Err(err) => {
                eprintln!("stream error: {}", err);
                break;
            }
        }
    }

    Ok(())
}