nautilus_polymarket/http/
gamma.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Provides the HTTP client for the Polymarket Gamma API.
17//!
18//! Gamma `/markets` server-side constraints honored by the paginator and
19//! `load_ids` chunker:
20//!
21//! - `limit` is silently capped at 100 items per page, so a larger requested
22//!   `limit` makes the "last page" check (`page_len < page_size`) trip after
23//!   page one.
24//! - `offset > 10000` is rejected with HTTP 422, so a paginator cannot walk
25//!   the full universe; callers fetching many markets must use
26//!   `condition_ids=` filtering.
27//! - `condition_ids=` accepts at most 100 IDs per request, so `load_ids` for
28//!   larger sets chunks the request and unions the responses.
29
30use std::{collections::HashMap, result::Result as StdResult, sync::Arc};
31
32use nautilus_core::{
33    UnixNanos,
34    consts::NAUTILUS_USER_AGENT,
35    time::{AtomicTime, get_atomic_clock_realtime},
36};
37use nautilus_model::instruments::InstrumentAny;
38use nautilus_network::{
39    http::{HttpClient, HttpClientError, HttpResponse, Method, USER_AGENT},
40    retry::{RetryConfig, RetryManager},
41};
42use serde::{Serialize, de::DeserializeOwned};
43use serde_json::Value;
44
45use crate::{
46    common::urls::gamma_api_url,
47    http::{
48        error::{Error, Result},
49        models::{GammaEvent, GammaMarket, GammaTag, SearchResponse},
50        parse::{create_instrument_from_def, parse_gamma_market},
51        query::{GetGammaEventsParams, GetGammaMarketsParams, GetSearchParams},
52        rate_limits::POLYMARKET_GAMMA_REST_QUOTA,
53    },
54};
55
56/// Provides a raw HTTP client for the Polymarket Gamma API.
57///
58/// Handles HTTP transport for fetching market data from the public Gamma API.
59/// No authentication is required.
60#[derive(Debug, Clone)]
61pub struct PolymarketGammaRawHttpClient {
62    client: HttpClient,
63    base_url: String,
64}
65
66impl PolymarketGammaRawHttpClient {
67    /// Creates a new [`PolymarketGammaRawHttpClient`].
68    ///
69    /// # Errors
70    ///
71    /// Returns an error if the HTTP client cannot be created.
72    pub fn new(base_url: Option<String>, timeout_secs: u64) -> StdResult<Self, HttpClientError> {
73        Ok(Self {
74            client: HttpClient::new(
75                Self::default_headers(),
76                vec![],
77                vec![],
78                Some(*POLYMARKET_GAMMA_REST_QUOTA),
79                Some(timeout_secs),
80                None,
81            )?,
82            base_url: base_url
83                .unwrap_or_else(|| gamma_api_url().to_string())
84                .trim_end_matches('/')
85                .to_string(),
86        })
87    }
88
89    fn default_headers() -> HashMap<String, String> {
90        HashMap::from([
91            (USER_AGENT.to_string(), NAUTILUS_USER_AGENT.to_string()),
92            ("Content-Type".to_string(), "application/json".to_string()),
93        ])
94    }
95
96    fn url(&self, path: &str) -> String {
97        format!("{}{path}", self.base_url)
98    }
99
100    async fn send_get<P: Serialize, T: DeserializeOwned>(
101        &self,
102        path: &str,
103        params: Option<&P>,
104    ) -> Result<T> {
105        let url = self.url(path);
106        let response = self
107            .client
108            .request_with_params(Method::GET, url, params, None, None, None, None)
109            .await
110            .map_err(Error::from_http_client)?;
111
112        decode_response(&response)
113    }
114
115    async fn send_get_query_map<T: DeserializeOwned>(
116        &self,
117        path: &str,
118        params: Option<&HashMap<String, Vec<String>>>,
119    ) -> Result<T> {
120        let url = self.url(path);
121        let response = self
122            .client
123            .request(Method::GET, url, params, None, None, None, None)
124            .await
125            .map_err(Error::from_http_client)?;
126
127        decode_response(&response)
128    }
129
130    /// Fetches markets from the Gamma API.
131    ///
132    /// Handles both bare array and `{"data": [...]}` response schemas.
133    pub async fn get_gamma_markets(
134        &self,
135        params: GetGammaMarketsParams,
136    ) -> Result<Vec<GammaMarket>> {
137        let query_params = gamma_markets_query_params(params)?;
138        let value: Value = self
139            .send_get_query_map("/markets", Some(&query_params))
140            .await?;
141
142        let array = match value {
143            Value::Array(_) => value,
144            Value::Object(ref map) if map.contains_key("data") => {
145                map.get("data").cloned().unwrap_or(Value::Array(vec![]))
146            }
147            _ => {
148                return Err(Error::decode(
149                    "Unrecognized Gamma markets response schema".to_string(),
150                ));
151            }
152        };
153
154        serde_json::from_value(array).map_err(Error::Serde)
155    }
156
157    /// Fetches a single market by ID from the Gamma API.
158    pub async fn get_gamma_market(&self, market_id: &str) -> Result<GammaMarket> {
159        let path = format!("/markets/{market_id}");
160        self.send_get::<(), _>(&path, None::<&()>).await
161    }
162
163    /// Fetches events from the Gamma API `GET /events?slug=`.
164    pub async fn get_gamma_events_by_slug(&self, slug: &str) -> Result<Vec<GammaEvent>> {
165        #[derive(Serialize)]
166        struct EventSlugParams<'a> {
167            slug: &'a str,
168        }
169        let params = EventSlugParams { slug };
170        self.send_get("/events", Some(&params)).await
171    }
172
173    /// Fetches events from the Gamma API `GET /events` with full query params.
174    pub async fn get_gamma_events(&self, params: GetGammaEventsParams) -> Result<Vec<GammaEvent>> {
175        self.send_get("/events", Some(&params)).await
176    }
177
178    /// Fetches available tags from the Gamma API `GET /tags`.
179    pub async fn get_gamma_tags(&self) -> Result<Vec<GammaTag>> {
180        self.send_get::<(), _>("/tags", None::<&()>).await
181    }
182
183    /// Searches the Gamma API via `GET /public-search`.
184    pub async fn get_public_search(&self, params: GetSearchParams) -> Result<SearchResponse> {
185        self.send_get("/public-search", Some(&params)).await
186    }
187}
188
189fn decode_response<T: DeserializeOwned>(response: &HttpResponse) -> Result<T> {
190    if response.status.is_success() {
191        serde_json::from_slice(&response.body).map_err(Error::Serde)
192    } else {
193        Err(Error::from_status_code(
194            response.status.as_u16(),
195            &response.body,
196        ))
197    }
198}
199
200fn gamma_markets_query_params(
201    params: GetGammaMarketsParams,
202) -> Result<HashMap<String, Vec<String>>> {
203    let mut scalar_params = params;
204    let clob_token_ids = scalar_params.clob_token_ids.take();
205    let condition_ids = scalar_params.condition_ids.take();
206    let question_ids = scalar_params.question_ids.take();
207    let value = serde_json::to_value(&scalar_params).map_err(Error::Serde)?;
208    let fields = value
209        .as_object()
210        .ok_or_else(|| Error::decode("Gamma markets params must encode to an object"))?;
211    let mut params = HashMap::with_capacity(fields.len());
212
213    for (key, value) in fields {
214        if let Some(value) = gamma_markets_query_value(value)? {
215            params.insert(key.clone(), vec![value]);
216        }
217    }
218
219    insert_repeated_csv_param(&mut params, "clob_token_ids", clob_token_ids);
220    insert_repeated_csv_param(&mut params, "condition_ids", condition_ids);
221    insert_repeated_csv_param(&mut params, "question_ids", question_ids);
222
223    Ok(params)
224}
225
226fn insert_repeated_csv_param(
227    params: &mut HashMap<String, Vec<String>>,
228    key: &str,
229    value: Option<String>,
230) {
231    let Some(value) = value else {
232        return;
233    };
234
235    let values: Vec<String> = value
236        .split(',')
237        .map(str::trim)
238        .filter(|item| !item.is_empty())
239        .map(str::to_string)
240        .collect();
241
242    if !values.is_empty() {
243        params.insert(key.to_string(), values);
244    }
245}
246
247fn gamma_markets_query_value(value: &Value) -> Result<Option<String>> {
248    match value {
249        Value::Null => Ok(None),
250        Value::String(value) => Ok(Some(value.clone())),
251        Value::Bool(value) => Ok(Some(value.to_string())),
252        Value::Number(value) => Ok(Some(value.to_string())),
253        other => Err(Error::decode(format!(
254            "Unsupported Gamma markets query value: {other}"
255        ))),
256    }
257}
258
259fn parse_markets_to_instruments(markets: &[GammaMarket], ts_init: UnixNanos) -> Vec<InstrumentAny> {
260    let (instruments, _transient) = parse_markets_with_transient(markets, ts_init);
261    instruments
262}
263
264// Returns parsed instruments alongside condition IDs of markets still in the
265// CLOB hydration window (empty or empty-entry `clob_token_ids`), so callers
266// can retry rather than treating them as terminal.
267fn parse_markets_with_transient(
268    markets: &[GammaMarket],
269    ts_init: UnixNanos,
270) -> (Vec<InstrumentAny>, Vec<String>) {
271    let mut instruments = Vec::new();
272    let mut transient = Vec::new();
273
274    for market in markets {
275        if is_transient_clob_token_ids(&market.clob_token_ids) {
276            transient.push(market.condition_id.clone());
277            continue;
278        }
279
280        match parse_gamma_market(market) {
281            Ok(defs) => {
282                for def in defs {
283                    match create_instrument_from_def(&def, ts_init) {
284                        Ok(instrument) => instruments.push(instrument),
285                        Err(e) => log::warn!("Failed to create instrument: {e}"),
286                    }
287                }
288            }
289            Err(e) => log::warn!("Failed to parse gamma market: {e}"),
290        }
291    }
292
293    if !transient.is_empty() {
294        log::debug!(
295            "{} market(s) without usable clob_token_ids deferred as transient (CLOB hydration)",
296            transient.len(),
297        );
298    }
299    (instruments, transient)
300}
301
302// Treats bare empty string, encoded empty array, and arrays with empty entries
303// as transient. Unparsable payloads fall through to `parse_gamma_market` so
304// real schema errors still surface.
305fn is_transient_clob_token_ids(raw: &str) -> bool {
306    if raw.is_empty() {
307        return true;
308    }
309
310    match serde_json::from_str::<Vec<String>>(raw) {
311        Ok(ids) => ids.is_empty() || ids.iter().any(|t| t.is_empty()),
312        Err(_) => false,
313    }
314}
315
316fn flatten_event_markets(events: Vec<GammaEvent>) -> Vec<GammaMarket> {
317    events
318        .into_iter()
319        .flat_map(|event| {
320            let event_game_id = event.game_id;
321            event.markets.into_iter().map(move |mut market| {
322                if market.game_id.is_none() {
323                    market.game_id = event_game_id;
324                }
325                market
326            })
327        })
328        .collect()
329}
330
331/// Provides a domain HTTP client for Polymarket instrument fetching.
332///
333/// Wraps [`PolymarketGammaRawHttpClient`] with instrument parsing: fetch from
334/// the Gamma API and parse into Nautilus types. Stateless with respect to
335/// instrument storage; caching is handled by the instrument provider.
336#[derive(Debug, Clone)]
337pub struct PolymarketGammaHttpClient {
338    inner: Arc<PolymarketGammaRawHttpClient>,
339    clock: &'static AtomicTime,
340    retry_manager: Arc<RetryManager<Error>>,
341}
342
343impl PolymarketGammaHttpClient {
344    /// Creates a new [`PolymarketGammaHttpClient`].
345    ///
346    /// # Errors
347    ///
348    /// Returns an error if the underlying HTTP client cannot be created.
349    pub fn new(
350        gamma_base_url: Option<String>,
351        timeout_secs: u64,
352        retry_config: RetryConfig,
353    ) -> StdResult<Self, HttpClientError> {
354        Ok(Self {
355            inner: Arc::new(PolymarketGammaRawHttpClient::new(
356                gamma_base_url,
357                timeout_secs,
358            )?),
359            clock: get_atomic_clock_realtime(),
360            retry_manager: Arc::new(RetryManager::new(retry_config)),
361        })
362    }
363
364    /// Fetches markets from the Gamma API with the given base params, paginating automatically.
365    async fn fetch_gamma_markets_paginated(
366        &self,
367        base_params: GetGammaMarketsParams,
368    ) -> anyhow::Result<Vec<GammaMarket>> {
369        const PAGE_LIMIT: u32 = 100;
370        let page_size = base_params.limit.unwrap_or(PAGE_LIMIT);
371        let max_markets = base_params.max_markets;
372        let mut all_markets = Vec::new();
373        let mut offset: u32 = base_params.offset.unwrap_or(0);
374        let mut page_num = 0u32;
375
376        loop {
377            let params = GetGammaMarketsParams {
378                limit: Some(page_size),
379                offset: Some(offset),
380                ..base_params.clone()
381            };
382
383            let page = self.inner.get_gamma_markets(params).await?;
384            let page_len = page.len() as u32;
385            page_num += 1;
386            all_markets.extend(page);
387
388            log::debug!(
389                "Fetched markets page {page_num}: {page_len} markets (total: {})",
390                all_markets.len(),
391            );
392
393            if let Some(cap) = max_markets
394                && all_markets.len() as u32 >= cap
395            {
396                all_markets.truncate(cap as usize);
397                break;
398            }
399
400            if page_len < page_size {
401                break;
402            }
403
404            offset += page_size;
405        }
406
407        Ok(all_markets)
408    }
409
410    /// Fetches all active markets from the Gamma API, paginating automatically.
411    async fn fetch_all_gamma_markets(&self) -> anyhow::Result<Vec<GammaMarket>> {
412        self.fetch_gamma_markets_paginated(GetGammaMarketsParams {
413            active: Some(true),
414            closed: Some(false),
415            ..Default::default()
416        })
417        .await
418    }
419
420    /// Fetches instruments from the Gamma API and returns Nautilus domain types.
421    ///
422    /// # Errors
423    ///
424    /// Returns an error if the HTTP request or parsing fails.
425    pub async fn request_instruments(&self) -> anyhow::Result<Vec<InstrumentAny>> {
426        let markets = self.fetch_all_gamma_markets().await?;
427        let ts_init = self.clock.get_time_ns();
428        let instruments = parse_markets_to_instruments(&markets, ts_init);
429        log::debug!("Parsed {} instruments from Gamma API", instruments.len());
430        Ok(instruments)
431    }
432
433    /// Fetches instruments for the given slugs concurrently.
434    ///
435    /// Each slug is queried individually via the Gamma API. Missing or
436    /// unparsable slugs are logged and skipped.
437    ///
438    /// # Errors
439    ///
440    /// Returns an error if all slug requests fail. Individual slug failures
441    /// are warned and skipped when at least one slug succeeds.
442    pub async fn request_instruments_by_slugs(
443        &self,
444        slugs: Vec<String>,
445    ) -> anyhow::Result<Vec<InstrumentAny>> {
446        let ts_init = self.clock.get_time_ns();
447
448        let futures = slugs.into_iter().map(|slug| {
449            let inner = Arc::clone(&self.inner);
450            async move {
451                let params = GetGammaMarketsParams {
452                    slug: Some(slug.clone()),
453                    ..Default::default()
454                };
455
456                match inner.get_gamma_markets(params).await {
457                    Ok(markets) => Some((slug, markets)),
458                    Err(e) => {
459                        log::warn!("Failed to fetch slug '{slug}': {e}");
460                        None
461                    }
462                }
463            }
464        });
465
466        let results = futures_util::future::join_all(futures).await;
467
468        let total_slugs = results.len();
469        let succeeded = results.iter().filter(|r| r.is_some()).count();
470        let mut instruments = Vec::new();
471
472        for result in results.into_iter().flatten() {
473            let (slug, markets) = result;
474            if markets.is_empty() {
475                log::debug!("No markets found for slug '{slug}'");
476                continue;
477            }
478            instruments.extend(parse_markets_to_instruments(&markets, ts_init));
479        }
480
481        if succeeded == 0 && total_slugs > 0 {
482            anyhow::bail!("All {total_slugs} slug requests failed");
483        }
484
485        log::debug!("Parsed {} instruments from slug queries", instruments.len());
486        Ok(instruments)
487    }
488
489    /// Fetches instruments for the given slugs with retry on empty results.
490    ///
491    /// Uses the client's [`RetryManager`] with exponential backoff. Gamma API
492    /// may not have indexed a newly created market yet, so empty results are
493    /// treated as retryable (indexing lag). HTTP errors are also retried per
494    /// the standard `is_retryable()` classification.
495    pub async fn request_instruments_by_slugs_with_retry(
496        &self,
497        slugs: Vec<String>,
498    ) -> anyhow::Result<Vec<InstrumentAny>> {
499        let inner = Arc::clone(&self.inner);
500        let ts_init = self.clock.get_time_ns();
501
502        self.retry_manager
503            .execute_with_retry(
504                "gamma_fetch_by_slugs",
505                || {
506                    let inner = Arc::clone(&inner);
507                    let slugs = slugs.clone();
508                    async move {
509                        let futures = slugs.into_iter().map(|slug| {
510                            let inner = Arc::clone(&inner);
511                            async move {
512                                let params = GetGammaMarketsParams {
513                                    slug: Some(slug.clone()),
514                                    ..Default::default()
515                                };
516                                inner
517                                    .get_gamma_markets(params)
518                                    .await
519                                    .map(|markets| (slug, markets))
520                            }
521                        });
522
523                        let results: Vec<_> = futures_util::future::join_all(futures)
524                            .await
525                            .into_iter()
526                            .collect::<StdResult<Vec<_>, _>>()?;
527
528                        let instruments: Vec<InstrumentAny> = results
529                            .into_iter()
530                            .flat_map(|(_, markets)| {
531                                parse_markets_to_instruments(&markets, ts_init)
532                            })
533                            .collect();
534
535                        if instruments.is_empty() {
536                            return Err(Error::transport(
537                                "Gamma returned no instruments (indexing lag)",
538                            ));
539                        }
540
541                        Ok(instruments)
542                    }
543                },
544                |e| e.is_retryable(),
545                Error::transport,
546            )
547            .await
548            .map_err(|e| anyhow::anyhow!("{e}"))
549    }
550
551    /// Fetches instruments from event slugs concurrently.
552    ///
553    /// Each slug queries `GET /events?slug=`, extracts the markets array from
554    /// the first matching event, and parses each market into instruments.
555    pub async fn request_instruments_by_event_slugs(
556        &self,
557        event_slugs: Vec<String>,
558    ) -> anyhow::Result<Vec<InstrumentAny>> {
559        let ts_init = self.clock.get_time_ns();
560
561        let futures = event_slugs.into_iter().map(|slug| {
562            let inner = Arc::clone(&self.inner);
563            async move {
564                match inner.get_gamma_events_by_slug(&slug).await {
565                    Ok(events) => Some((slug, events)),
566                    Err(e) => {
567                        log::warn!("Failed to fetch event slug '{slug}': {e}");
568                        None
569                    }
570                }
571            }
572        });
573
574        let results = futures_util::future::join_all(futures).await;
575
576        let total = results.len();
577        let succeeded = results.iter().filter(|r| r.is_some()).count();
578        let mut instruments = Vec::new();
579
580        for result in results.into_iter().flatten() {
581            let (slug, events) = result;
582            let markets = flatten_event_markets(events);
583            if markets.is_empty() {
584                log::warn!("No markets found in event slug '{slug}'");
585                continue;
586            }
587            instruments.extend(parse_markets_to_instruments(&markets, ts_init));
588        }
589
590        if succeeded == 0 && total > 0 {
591            anyhow::bail!("All {total} event slug requests failed");
592        }
593
594        log::debug!(
595            "Parsed {} instruments from event slug queries",
596            instruments.len()
597        );
598        Ok(instruments)
599    }
600
601    /// Fetches instruments using arbitrary Gamma API query params with auto-pagination.
602    pub async fn request_instruments_by_params(
603        &self,
604        base_params: GetGammaMarketsParams,
605    ) -> anyhow::Result<Vec<InstrumentAny>> {
606        let markets = self.fetch_gamma_markets_paginated(base_params).await?;
607        let ts_init = self.clock.get_time_ns();
608        let instruments = parse_markets_to_instruments(&markets, ts_init);
609        log::debug!("Parsed {} instruments from params query", instruments.len());
610        Ok(instruments)
611    }
612
613    /// Same as [`Self::request_instruments_by_params`] but also returns
614    /// condition IDs whose markets came back from Gamma with empty
615    /// `clob_token_ids`. Callers driving auto-load retries use the transient
616    /// list to distinguish "still hydrating in the CLOB" from "absent on the
617    /// venue".
618    pub async fn request_instruments_by_params_with_transient(
619        &self,
620        base_params: GetGammaMarketsParams,
621    ) -> anyhow::Result<(Vec<InstrumentAny>, Vec<String>)> {
622        let markets = self.fetch_gamma_markets_paginated(base_params).await?;
623        let ts_init = self.clock.get_time_ns();
624        let (instruments, transient) = parse_markets_with_transient(&markets, ts_init);
625        log::debug!(
626            "Parsed {} instruments and {} transient condition_id(s) from params query",
627            instruments.len(),
628            transient.len(),
629        );
630        Ok((instruments, transient))
631    }
632
633    /// Fetches raw Gamma markets using arbitrary query params with auto-pagination.
634    pub async fn request_markets_by_params(
635        &self,
636        base_params: GetGammaMarketsParams,
637    ) -> anyhow::Result<Vec<GammaMarket>> {
638        self.fetch_gamma_markets_paginated(base_params).await
639    }
640
641    /// Fetches instruments from an event slug with client-side sorting and limiting.
642    ///
643    /// The `/events?slug=` response already includes the full markets array,
644    /// so no second API call is needed. Sorting and truncation are applied
645    /// client-side using fields from `GetGammaMarketsParams`:
646    /// - `order`: sort field (`"liquidity"`, `"volume"`, `"volume24hr"`)
647    /// - `ascending`: sort direction (default: descending)
648    /// - `max_markets`: truncate after sorting
649    pub async fn request_instruments_by_event_query(
650        &self,
651        event_slug: &str,
652        params: GetGammaMarketsParams,
653    ) -> anyhow::Result<Vec<InstrumentAny>> {
654        let events = self.inner.get_gamma_events_by_slug(event_slug).await?;
655        let mut markets = flatten_event_markets(events);
656
657        if markets.is_empty() {
658            log::warn!("No markets found in event slug '{event_slug}'");
659            return Ok(Vec::new());
660        }
661
662        log::debug!("Event '{event_slug}' returned {} markets", markets.len());
663
664        // Client-side sort
665        if let Some(ref order_field) = params.order {
666            let ascending = params.ascending.unwrap_or(false);
667            markets.sort_by(|a, b| {
668                let cmp = match order_field.as_str() {
669                    "liquidity" => a
670                        .liquidity_num
671                        .unwrap_or(0.0)
672                        .partial_cmp(&b.liquidity_num.unwrap_or(0.0)),
673                    "volume" => a
674                        .volume_num
675                        .unwrap_or(0.0)
676                        .partial_cmp(&b.volume_num.unwrap_or(0.0)),
677                    "volume24hr" => a
678                        .volume_24hr
679                        .unwrap_or(0.0)
680                        .partial_cmp(&b.volume_24hr.unwrap_or(0.0)),
681                    "competitive" => a
682                        .competitive
683                        .unwrap_or(0.0)
684                        .partial_cmp(&b.competitive.unwrap_or(0.0)),
685                    "spread" => a
686                        .spread
687                        .unwrap_or(f64::MAX)
688                        .partial_cmp(&b.spread.unwrap_or(f64::MAX)),
689                    "best_bid" => a
690                        .best_bid
691                        .unwrap_or(0.0)
692                        .partial_cmp(&b.best_bid.unwrap_or(0.0)),
693                    "one_day_price_change" => a
694                        .one_day_price_change
695                        .unwrap_or(0.0)
696                        .partial_cmp(&b.one_day_price_change.unwrap_or(0.0)),
697                    "volume_1wk" => a
698                        .volume_1wk
699                        .unwrap_or(0.0)
700                        .partial_cmp(&b.volume_1wk.unwrap_or(0.0)),
701                    _ => None,
702                };
703                let cmp = cmp.unwrap_or(std::cmp::Ordering::Equal);
704                if ascending { cmp } else { cmp.reverse() }
705            });
706        }
707
708        // Client-side truncation
709        if let Some(cap) = params.max_markets {
710            markets.truncate(cap as usize);
711        }
712
713        let ts_init = self.clock.get_time_ns();
714        let instruments = parse_markets_to_instruments(&markets, ts_init);
715        log::debug!(
716            "Parsed {} instruments from event query '{event_slug}'",
717            instruments.len()
718        );
719        Ok(instruments)
720    }
721
722    /// Fetches events from the Gamma API with the given base params, paginating automatically.
723    async fn fetch_gamma_events_paginated(
724        &self,
725        base_params: GetGammaEventsParams,
726    ) -> anyhow::Result<Vec<GammaEvent>> {
727        const PAGE_LIMIT: u32 = 100;
728        let page_size = base_params.limit.unwrap_or(PAGE_LIMIT);
729        let max_events = base_params.max_events;
730        let mut all_events = Vec::new();
731        let mut offset: u32 = base_params.offset.unwrap_or(0);
732        let mut page_num = 0u32;
733
734        loop {
735            let params = GetGammaEventsParams {
736                limit: Some(page_size),
737                offset: Some(offset),
738                ..base_params.clone()
739            };
740
741            let page = self.inner.get_gamma_events(params).await?;
742            let page_len = page.len() as u32;
743            page_num += 1;
744            let market_count: usize = page.iter().map(|e| e.markets.len()).sum();
745            all_events.extend(page);
746
747            log::debug!(
748                "Fetched events page {page_num}: {page_len} events, {market_count} markets (total events: {})",
749                all_events.len(),
750            );
751
752            if let Some(cap) = max_events
753                && all_events.len() as u32 >= cap
754            {
755                all_events.truncate(cap as usize);
756                break;
757            }
758
759            if page_len < page_size {
760                break;
761            }
762
763            offset += page_size;
764        }
765
766        Ok(all_events)
767    }
768
769    /// Fetches instruments from events matching full query params (paginated).
770    pub async fn request_instruments_by_event_params(
771        &self,
772        params: GetGammaEventsParams,
773    ) -> anyhow::Result<Vec<InstrumentAny>> {
774        let events = self.fetch_gamma_events_paginated(params).await?;
775        let ts_init = self.clock.get_time_ns();
776        let total_events = events.len();
777        let markets = flatten_event_markets(events);
778        let total_markets = markets.len();
779        let instruments = parse_markets_to_instruments(&markets, ts_init);
780        log::debug!(
781            "Parsed {} instruments from {total_events} events ({total_markets} markets)",
782            instruments.len(),
783        );
784        Ok(instruments)
785    }
786
787    /// Searches for instruments via the Gamma public search endpoint.
788    pub async fn request_instruments_by_search(
789        &self,
790        params: GetSearchParams,
791    ) -> anyhow::Result<Vec<InstrumentAny>> {
792        let response = self.inner.get_public_search(params).await?;
793        let ts_init = self.clock.get_time_ns();
794
795        let mut instruments = Vec::new();
796
797        if let Some(markets) = &response.markets {
798            instruments.extend(parse_markets_to_instruments(markets, ts_init));
799        }
800
801        if let Some(events) = &response.events {
802            let event_markets = flatten_event_markets(events.clone());
803            instruments.extend(parse_markets_to_instruments(&event_markets, ts_init));
804        }
805
806        log::debug!("Parsed {} instruments from search query", instruments.len());
807        Ok(instruments)
808    }
809
810    /// Fetches available tags from the Gamma API.
811    pub async fn request_tags(&self) -> anyhow::Result<Vec<GammaTag>> {
812        Ok(self.inner.get_gamma_tags().await?)
813    }
814
815    /// Returns a reference to the underlying raw HTTP client.
816    #[must_use]
817    pub fn inner(&self) -> &Arc<PolymarketGammaRawHttpClient> {
818        &self.inner
819    }
820}
nautilus_polymarket/http/gamma.rs

nautilus_polymarket/http/
gamma.rs
