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 keyset constraints honored by the paginators and `load_ids` chunker:
19//!
20//! - `/markets/keyset` accepts at most 100 items per page.
21//! - `/events/keyset` accepts at most 500 items per page.
22//! - Keyset endpoints reject `offset`; the paginators apply a requested initial
23//!   offset locally for compatibility.
24//! - `next_cursor` is absent on the final page.
25//! - `condition_ids=` accepts at most 100 IDs per request, so `load_ids` for
26//!   larger sets chunks the request and unions the responses.
27
28use std::{collections::HashMap, result::Result as StdResult, sync::Arc};
29
30use nautilus_core::{
31    UnixNanos,
32    consts::NAUTILUS_USER_AGENT,
33    time::{AtomicTime, get_atomic_clock_realtime},
34};
35use nautilus_model::instruments::InstrumentAny;
36use nautilus_network::{
37    http::{HttpClient, HttpClientError, HttpResponse, Method, USER_AGENT},
38    retry::{RetryConfig, RetryManager},
39    websocket::proxy::ProxyUrl,
40};
41use serde::{Deserialize, Serialize, de::DeserializeOwned};
42use serde_json::Value;
43
44use crate::{
45    common::urls::gamma_api_url,
46    filters::set_market_closed,
47    http::{
48        error::{Error, Result},
49        models::{GammaEvent, GammaMarket, GammaTag, SearchResponse},
50        pagination::{Completion, CursorProtocol, FetchOutcome, Paginator, WindowedCollect},
51        parse::{create_instrument_from_def, parse_gamma_market},
52        query::{GetGammaEventsParams, GetGammaMarketsParams, GetSearchParams},
53        rate_limits::POLYMARKET_GAMMA_REST_QUOTA,
54    },
55};
56
57const GAMMA_MARKETS_KEYSET_PAGE_LIMIT: u32 = 100;
58const GAMMA_EVENTS_KEYSET_PAGE_LIMIT: u32 = 500;
59
60#[derive(Clone, Copy, Debug, Eq, PartialEq)]
61enum GammaStop {
62    CallerCapped,
63}
64
65/// Provides a raw HTTP client for the Polymarket Gamma API.
66///
67/// Handles HTTP transport for fetching market data from the public Gamma API.
68/// No authentication is required.
69#[derive(Debug, Clone)]
70pub struct PolymarketGammaRawHttpClient {
71    client: HttpClient,
72    base_url: String,
73}
74
75impl PolymarketGammaRawHttpClient {
76    /// Creates a new [`PolymarketGammaRawHttpClient`].
77    ///
78    /// # Errors
79    ///
80    /// Returns an error if the HTTP client cannot be created.
81    pub fn new(base_url: Option<String>, timeout_secs: u64) -> StdResult<Self, HttpClientError> {
82        Self::new_with_proxy(base_url, timeout_secs, None)
83    }
84
85    /// Creates a new raw client with an optional validated proxy URL.
86    ///
87    /// # Errors
88    ///
89    /// Returns an error if the HTTP client cannot be created.
90    pub fn new_with_proxy(
91        base_url: Option<String>,
92        timeout_secs: u64,
93        proxy_url: Option<ProxyUrl>,
94    ) -> StdResult<Self, HttpClientError> {
95        Ok(Self {
96            client: HttpClient::builder()
97                .headers(Self::default_headers())
98                .default_quota(*POLYMARKET_GAMMA_REST_QUOTA)
99                .timeout_secs(timeout_secs)
100                .maybe_proxy_url(proxy_url.map(|url| url.expose().to_string()))
101                .build()?,
102            base_url: base_url
103                .unwrap_or_else(|| gamma_api_url().to_string())
104                .trim_end_matches('/')
105                .to_string(),
106        })
107    }
108
109    fn default_headers() -> HashMap<String, String> {
110        HashMap::from([
111            (USER_AGENT.to_string(), NAUTILUS_USER_AGENT.to_string()),
112            ("Content-Type".to_string(), "application/json".to_string()),
113        ])
114    }
115
116    fn url(&self, path: &str) -> String {
117        format!("{}{path}", self.base_url)
118    }
119
120    async fn send_get<P: Serialize, T: DeserializeOwned>(
121        &self,
122        path: &str,
123        params: Option<&P>,
124    ) -> Result<T> {
125        let url = self.url(path);
126        let response = self
127            .client
128            .request_with_params(Method::GET, url, params, None, None, None, None)
129            .await
130            .map_err(Error::from_http_client)?;
131
132        decode_response(&response)
133    }
134
135    async fn send_get_query_map<T: DeserializeOwned>(
136        &self,
137        path: &str,
138        params: Option<&HashMap<String, Vec<String>>>,
139    ) -> Result<T> {
140        let url = self.url(path);
141        let response = self
142            .client
143            .request(Method::GET, url, params, None, None, None, None)
144            .await
145            .map_err(Error::from_http_client)?;
146
147        decode_response(&response)
148    }
149
150    /// Fetches markets from the Gamma API.
151    ///
152    /// Handles both bare array and `{"data": [...]}` response schemas.
153    pub async fn get_gamma_markets(
154        &self,
155        params: GetGammaMarketsParams,
156    ) -> Result<Vec<GammaMarket>> {
157        let query_params = gamma_markets_query_params(params)?;
158        let value: Value = self
159            .send_get_query_map("/markets", Some(&query_params))
160            .await?;
161
162        let array = match value {
163            Value::Array(_) => value,
164            Value::Object(ref map) if map.contains_key("data") => {
165                map.get("data").cloned().unwrap_or(Value::Array(vec![]))
166            }
167            _ => {
168                return Err(Error::decode(
169                    "Unrecognized Gamma markets response schema".to_string(),
170                ));
171            }
172        };
173
174        serde_json::from_value(array).map_err(Error::Serde)
175    }
176
177    async fn get_gamma_markets_keyset(
178        &self,
179        mut params: GetGammaMarketsParams,
180        after_cursor: Option<&str>,
181    ) -> Result<GammaMarketsKeysetResponse> {
182        params.validate_keyset().map_err(Error::decode)?;
183        params.offset = None;
184        let mut query_params = gamma_markets_query_params(params)?;
185        if let Some(after_cursor) = after_cursor {
186            query_params.insert("after_cursor".to_string(), vec![after_cursor.to_string()]);
187        }
188        self.send_get_query_map("/markets/keyset", Some(&query_params))
189            .await
190    }
191
192    /// Fetches a single market by ID from the Gamma API.
193    pub async fn get_gamma_market(&self, market_id: &str) -> Result<GammaMarket> {
194        let path = format!("/markets/{market_id}");
195        self.send_get::<(), _>(&path, None::<&()>).await
196    }
197
198    /// Fetches a market from the Gamma API `GET /markets/slug/{slug}`.
199    pub async fn get_gamma_market_by_slug(&self, slug: &str) -> Result<GammaMarket> {
200        let path = format!("/markets/slug/{slug}");
201        self.send_get::<(), _>(&path, None::<&()>).await
202    }
203
204    /// Fetches events from the Gamma API `GET /events?slug=`.
205    pub async fn get_gamma_events_by_slug(&self, slug: &str) -> Result<Vec<GammaEvent>> {
206        #[derive(Serialize)]
207        struct EventSlugParams<'a> {
208            slug: &'a str,
209        }
210        let params = EventSlugParams { slug };
211        self.send_get("/events", Some(&params)).await
212    }
213
214    /// Fetches events from the Gamma API `GET /events` with full query params.
215    pub async fn get_gamma_events(&self, params: GetGammaEventsParams) -> Result<Vec<GammaEvent>> {
216        let query_params = gamma_events_query_params(params)?;
217        self.send_get_query_map("/events", Some(&query_params))
218            .await
219    }
220
221    async fn get_gamma_events_keyset(
222        &self,
223        mut params: GetGammaEventsParams,
224        after_cursor: Option<&str>,
225    ) -> Result<GammaEventsKeysetResponse> {
226        params.validate_keyset().map_err(Error::decode)?;
227        params.offset = None;
228        let mut query_params = gamma_events_query_params(params)?;
229        if let Some(after_cursor) = after_cursor {
230            query_params.insert("after_cursor".to_string(), vec![after_cursor.to_string()]);
231        }
232        self.send_get_query_map("/events/keyset", Some(&query_params))
233            .await
234    }
235
236    /// Fetches available tags from the Gamma API `GET /tags`.
237    pub async fn get_gamma_tags(&self) -> Result<Vec<GammaTag>> {
238        self.send_get::<(), _>("/tags", None::<&()>).await
239    }
240
241    /// Searches the Gamma API via `GET /public-search`.
242    pub async fn get_public_search(&self, params: GetSearchParams) -> Result<SearchResponse> {
243        self.send_get("/public-search", Some(&params)).await
244    }
245}
246
247#[derive(Debug, Deserialize)]
248struct GammaMarketsKeysetResponse {
249    markets: Vec<GammaMarket>,
250    next_cursor: Option<String>,
251}
252
253#[derive(Debug, Deserialize)]
254struct GammaEventsKeysetResponse {
255    events: Vec<GammaEvent>,
256    next_cursor: Option<String>,
257}
258
259fn decode_response<T: DeserializeOwned>(response: &HttpResponse) -> Result<T> {
260    if response.status.is_success() {
261        serde_json::from_slice(&response.body).map_err(Error::Serde)
262    } else {
263        Err(Error::from_status_code(
264            response.status.as_u16(),
265            &response.body,
266        ))
267    }
268}
269
270fn gamma_markets_query_params(
271    params: GetGammaMarketsParams,
272) -> Result<HashMap<String, Vec<String>>> {
273    let mut scalar_params = params;
274    let id = scalar_params.id.take();
275    let slug = scalar_params.slug.take();
276    let clob_token_ids = scalar_params.clob_token_ids.take();
277    let condition_ids = scalar_params.condition_ids.take();
278    let question_ids = scalar_params.question_ids.take();
279    let market_maker_address = scalar_params.market_maker_address.take();
280    let tag_id = scalar_params.tag_id.take();
281    let sports_market_types = scalar_params.sports_market_types.take();
282    let value = serde_json::to_value(&scalar_params).map_err(Error::Serde)?;
283    let fields = value
284        .as_object()
285        .ok_or_else(|| Error::decode("Gamma markets params must encode to an object"))?;
286    let mut params = HashMap::with_capacity(fields.len());
287
288    for (key, value) in fields {
289        if let Some(value) = gamma_query_value(value)? {
290            params.insert(key.clone(), vec![value]);
291        }
292    }
293
294    insert_repeated_param(&mut params, "id", id);
295    insert_repeated_param(&mut params, "slug", slug);
296    insert_repeated_param(&mut params, "clob_token_ids", clob_token_ids);
297    insert_repeated_param(&mut params, "condition_ids", condition_ids);
298    insert_repeated_param(&mut params, "question_ids", question_ids);
299    insert_repeated_param(&mut params, "market_maker_address", market_maker_address);
300    insert_repeated_param(&mut params, "tag_id", tag_id);
301    insert_repeated_param(&mut params, "sports_market_types", sports_market_types);
302
303    Ok(params)
304}
305
306fn gamma_events_query_params(params: GetGammaEventsParams) -> Result<HashMap<String, Vec<String>>> {
307    let mut scalar_params = params;
308    let id = scalar_params.id.take();
309    let slug = scalar_params.slug.take();
310    let tag_id = scalar_params.tag_id.take();
311    let exclude_tag_id = scalar_params.exclude_tag_id.take();
312    let series_id = scalar_params.series_id.take();
313    let game_id = scalar_params.game_id.take();
314    let created_by = scalar_params.created_by.take();
315    let value = serde_json::to_value(&scalar_params).map_err(Error::Serde)?;
316    let fields = value
317        .as_object()
318        .ok_or_else(|| Error::decode("Gamma events params must encode to an object"))?;
319    let mut params = HashMap::with_capacity(fields.len());
320
321    for (key, value) in fields {
322        if let Some(value) = gamma_query_value(value)? {
323            params.insert(key.clone(), vec![value]);
324        }
325    }
326
327    insert_repeated_param(&mut params, "id", id);
328    insert_repeated_param(&mut params, "slug", slug);
329    insert_repeated_param(&mut params, "tag_id", tag_id);
330    insert_repeated_param(&mut params, "exclude_tag_id", exclude_tag_id);
331    insert_repeated_param(&mut params, "series_id", series_id);
332    insert_repeated_param(&mut params, "game_id", game_id);
333    insert_repeated_param(&mut params, "created_by", created_by);
334
335    Ok(params)
336}
337
338fn insert_repeated_param<T: ToString>(
339    params: &mut HashMap<String, Vec<String>>,
340    key: &str,
341    values: Option<Vec<T>>,
342) {
343    let Some(values) = values else {
344        return;
345    };
346
347    params.insert(
348        key.to_string(),
349        values
350            .into_iter()
351            .map(|value| value.to_string().trim().to_string())
352            .collect(),
353    );
354}
355
356fn gamma_query_value(value: &Value) -> Result<Option<String>> {
357    match value {
358        Value::Null => Ok(None),
359        Value::String(value) => Ok(Some(value.clone())),
360        Value::Bool(value) => Ok(Some(value.to_string())),
361        Value::Number(value) => Ok(Some(value.to_string())),
362        other => Err(Error::decode(format!(
363            "Unsupported Gamma query value: {other}"
364        ))),
365    }
366}
367
368fn parse_markets_to_instruments(markets: &[GammaMarket], ts_init: UnixNanos) -> Vec<InstrumentAny> {
369    let (instruments, _transient) = parse_markets_with_transient(markets, ts_init);
370    instruments
371}
372
373// Returns parsed instruments alongside condition IDs of markets still in the
374// CLOB hydration window (empty or empty-entry `clob_token_ids`), so callers
375// can retry rather than treating them as terminal.
376//
377// This is the single funnel through which live instruments reach the client caches, so Gamma's
378// `closed` state is recorded here rather than in `create_instrument_from_def`. Historical loader
379// instruments share that constructor and must not carry terminal state in `info`; they expose it
380// through `resolution_metadata` instead.
381fn parse_markets_with_transient(
382    markets: &[GammaMarket],
383    ts_init: UnixNanos,
384) -> (Vec<InstrumentAny>, Vec<String>) {
385    let mut instruments = Vec::new();
386    let mut transient = Vec::new();
387
388    for market in markets {
389        if is_transient_clob_token_ids(&market.clob_token_ids) {
390            transient.push(market.condition_id.clone());
391            continue;
392        }
393
394        match parse_gamma_market(market) {
395            Ok(defs) => {
396                for def in defs {
397                    match create_instrument_from_def(&def, ts_init) {
398                        Ok(InstrumentAny::BinaryOption(mut binary)) => {
399                            set_market_closed(&mut binary, def.closed);
400                            instruments.push(InstrumentAny::BinaryOption(binary));
401                        }
402                        Ok(other) => instruments.push(other),
403                        Err(e) => log::warn!("Failed to create instrument: {e}"),
404                    }
405                }
406            }
407            Err(e) => log::warn!("Failed to parse gamma market: {e}"),
408        }
409    }
410
411    if !transient.is_empty() {
412        log::debug!(
413            "{} market(s) without usable clob_token_ids deferred as transient (CLOB hydration)",
414            transient.len(),
415        );
416    }
417    (instruments, transient)
418}
419
420// Treats bare empty string, encoded empty array, and arrays with empty entries
421// as transient. Unparsable payloads fall through to `parse_gamma_market` so
422// real schema errors still surface.
423fn is_transient_clob_token_ids(raw: &str) -> bool {
424    if raw.is_empty() {
425        return true;
426    }
427
428    match serde_json::from_str::<Vec<String>>(raw) {
429        Ok(ids) => ids.is_empty() || ids.iter().any(|t| t.is_empty()),
430        Err(_) => false,
431    }
432}
433
434fn flatten_event_markets(events: Vec<GammaEvent>) -> Vec<GammaMarket> {
435    events
436        .into_iter()
437        .flat_map(|event| {
438            let event_game_id = event.game_id;
439            event.markets.into_iter().map(move |mut market| {
440                if market.game_id.is_none() {
441                    market.game_id.clone_from(&event_game_id);
442                }
443                market
444            })
445        })
446        .collect()
447}
448
449/// Provides a domain HTTP client for Polymarket instrument fetching.
450///
451/// Wraps [`PolymarketGammaRawHttpClient`] with instrument parsing: fetch from
452/// the Gamma API and parse into Nautilus types. Stateless with respect to
453/// instrument storage; caching is handled by the instrument provider.
454#[derive(Debug, Clone)]
455pub struct PolymarketGammaHttpClient {
456    inner: Arc<PolymarketGammaRawHttpClient>,
457    clock: &'static AtomicTime,
458    retry_manager: Arc<RetryManager<Error>>,
459}
460
461impl PolymarketGammaHttpClient {
462    /// Creates a new [`PolymarketGammaHttpClient`].
463    ///
464    /// # Errors
465    ///
466    /// Returns an error if the underlying HTTP client cannot be created.
467    pub fn new(
468        gamma_base_url: Option<String>,
469        timeout_secs: u64,
470        retry_config: RetryConfig,
471    ) -> StdResult<Self, HttpClientError> {
472        Self::new_with_proxy(gamma_base_url, timeout_secs, retry_config, None)
473    }
474
475    /// Creates a new domain client with an optional validated proxy URL.
476    ///
477    /// # Errors
478    ///
479    /// Returns an error if the underlying HTTP client cannot be created.
480    pub fn new_with_proxy(
481        gamma_base_url: Option<String>,
482        timeout_secs: u64,
483        retry_config: RetryConfig,
484        proxy_url: Option<ProxyUrl>,
485    ) -> StdResult<Self, HttpClientError> {
486        Ok(Self {
487            inner: Arc::new(PolymarketGammaRawHttpClient::new_with_proxy(
488                gamma_base_url,
489                timeout_secs,
490                proxy_url,
491            )?),
492            clock: get_atomic_clock_realtime(),
493            retry_manager: Arc::new(RetryManager::new(retry_config)),
494        })
495    }
496
497    /// Fetches markets from the Gamma API with the given base params, paginating automatically.
498    async fn fetch_gamma_markets_paginated(
499        &self,
500        base_params: GetGammaMarketsParams,
501    ) -> anyhow::Result<Vec<GammaMarket>> {
502        let page_size = base_params
503            .limit
504            .unwrap_or(GAMMA_MARKETS_KEYSET_PAGE_LIMIT)
505            .min(GAMMA_MARKETS_KEYSET_PAGE_LIMIT);
506        let protocol = CursorProtocol::<GammaStop>::gamma("Gamma market");
507        let reducer = WindowedCollect::new(
508            base_params.offset.unwrap_or(0) as usize,
509            base_params.max_markets.map(|value| value as usize),
510            GammaStop::CallerCapped,
511        );
512        let paginator = Paginator::new("Gamma market", protocol, reducer);
513        let completed = paginator
514            .run(
515                |position| {
516                    let after_cursor = position.map(|cursor| cursor.as_ref().to_string());
517                    let params = GetGammaMarketsParams {
518                        limit: Some(page_size),
519                        offset: None,
520                        ..base_params.clone()
521                    };
522                    async move {
523                        let response = self
524                            .inner
525                            .get_gamma_markets_keyset(params, after_cursor.as_deref())
526                            .await?;
527                        Ok::<_, anyhow::Error>(FetchOutcome::Page {
528                            rows: response.markets,
529                            wire: response.next_cursor,
530                        })
531                    }
532                },
533                anyhow::Error::new,
534            )
535            .await?;
536
537        match completed.completion {
538            Completion::WireExhausted | Completion::Stopped(GammaStop::CallerCapped) => {
539                Ok(completed.output)
540            }
541        }
542    }
543
544    /// Fetches all active markets from the Gamma API, paginating automatically.
545    async fn fetch_all_gamma_markets(&self) -> anyhow::Result<Vec<GammaMarket>> {
546        self.fetch_gamma_markets_paginated(GetGammaMarketsParams {
547            active: Some(true),
548            closed: Some(false),
549            ..Default::default()
550        })
551        .await
552    }
553
554    /// Fetches instruments from the Gamma API and returns Nautilus domain types.
555    ///
556    /// # Errors
557    ///
558    /// Returns an error if the HTTP request or parsing fails.
559    pub async fn request_instruments(&self) -> anyhow::Result<Vec<InstrumentAny>> {
560        let markets = self.fetch_all_gamma_markets().await?;
561        let ts_init = self.clock.get_time_ns();
562        let instruments = parse_markets_to_instruments(&markets, ts_init);
563        log::debug!("Parsed {} instruments from Gamma API", instruments.len());
564        Ok(instruments)
565    }
566
567    /// Fetches instruments for the given slugs concurrently.
568    ///
569    /// Each slug is queried individually via the Gamma API. Missing or
570    /// unparsable slugs are logged and skipped.
571    ///
572    /// # Errors
573    ///
574    /// Returns an error if all slug requests fail. Individual slug failures
575    /// are warned and skipped when at least one slug succeeds.
576    pub async fn request_instruments_by_slugs(
577        &self,
578        slugs: Vec<String>,
579    ) -> anyhow::Result<Vec<InstrumentAny>> {
580        let ts_init = self.clock.get_time_ns();
581
582        let futures = slugs.into_iter().map(|slug| {
583            let inner = Arc::clone(&self.inner);
584            async move {
585                let params = GetGammaMarketsParams {
586                    slug: Some(vec![slug.clone()]),
587                    ..Default::default()
588                };
589
590                match inner.get_gamma_markets(params).await {
591                    Ok(markets) => Some((slug, markets)),
592                    Err(e) => {
593                        log::warn!("Failed to fetch slug '{slug}': {e}");
594                        None
595                    }
596                }
597            }
598        });
599
600        let results = futures_util::future::join_all(futures).await;
601
602        let total_slugs = results.len();
603        let succeeded = results.iter().filter(|r| r.is_some()).count();
604        let mut instruments = Vec::new();
605
606        for result in results.into_iter().flatten() {
607            let (slug, markets) = result;
608            if markets.is_empty() {
609                log::debug!("No markets found for slug '{slug}'");
610                continue;
611            }
612            instruments.extend(parse_markets_to_instruments(&markets, ts_init));
613        }
614
615        if succeeded == 0 && total_slugs > 0 {
616            anyhow::bail!("All {total_slugs} slug requests failed");
617        }
618
619        log::debug!("Parsed {} instruments from slug queries", instruments.len());
620        Ok(instruments)
621    }
622
623    /// Fetches instruments for the given slugs with retry on empty results.
624    ///
625    /// Uses the client's [`RetryManager`] with exponential backoff. Gamma API
626    /// may not have indexed a newly created market yet, so empty results are
627    /// treated as retryable (indexing lag). HTTP errors are also retried per
628    /// the standard `is_retryable()` classification.
629    pub async fn request_instruments_by_slugs_with_retry(
630        &self,
631        slugs: Vec<String>,
632    ) -> anyhow::Result<Vec<InstrumentAny>> {
633        let inner = Arc::clone(&self.inner);
634        let ts_init = self.clock.get_time_ns();
635
636        self.retry_manager
637            .execute_with_retry(
638                "gamma_fetch_by_slugs",
639                || {
640                    let inner = Arc::clone(&inner);
641                    let slugs = slugs.clone();
642                    async move {
643                        let futures = slugs.into_iter().map(|slug| {
644                            let inner = Arc::clone(&inner);
645                            async move {
646                                let params = GetGammaMarketsParams {
647                                    slug: Some(vec![slug.clone()]),
648                                    ..Default::default()
649                                };
650                                inner
651                                    .get_gamma_markets(params)
652                                    .await
653                                    .map(|markets| (slug, markets))
654                            }
655                        });
656
657                        let results: Vec<_> = futures_util::future::join_all(futures)
658                            .await
659                            .into_iter()
660                            .collect::<StdResult<Vec<_>, _>>()?;
661
662                        let instruments: Vec<InstrumentAny> = results
663                            .into_iter()
664                            .flat_map(|(_, markets)| {
665                                parse_markets_to_instruments(&markets, ts_init)
666                            })
667                            .collect();
668
669                        if instruments.is_empty() {
670                            return Err(Error::transport(
671                                "Gamma returned no instruments (indexing lag)",
672                            ));
673                        }
674
675                        Ok(instruments)
676                    }
677                },
678                |e| e.is_retryable(),
679                |e| Error::transport(e.to_string()),
680            )
681            .await
682            .map_err(|e| anyhow::anyhow!("{e}"))
683    }
684
685    /// Fetches instruments from event slugs concurrently.
686    ///
687    /// Each slug queries `GET /events?slug=`, extracts the markets array from
688    /// the first matching event, and parses each market into instruments.
689    pub async fn request_instruments_by_event_slugs(
690        &self,
691        event_slugs: Vec<String>,
692    ) -> anyhow::Result<Vec<InstrumentAny>> {
693        let ts_init = self.clock.get_time_ns();
694
695        let futures = event_slugs.into_iter().map(|slug| {
696            let inner = Arc::clone(&self.inner);
697            async move {
698                match inner.get_gamma_events_by_slug(&slug).await {
699                    Ok(events) => Some((slug, events)),
700                    Err(e) => {
701                        log::warn!("Failed to fetch event slug '{slug}': {e}");
702                        None
703                    }
704                }
705            }
706        });
707
708        let results = futures_util::future::join_all(futures).await;
709
710        let total = results.len();
711        let succeeded = results.iter().filter(|r| r.is_some()).count();
712        let mut instruments = Vec::new();
713
714        for result in results.into_iter().flatten() {
715            let (slug, events) = result;
716            let markets = flatten_event_markets(events);
717            if markets.is_empty() {
718                log::warn!("No markets found in event slug '{slug}'");
719                continue;
720            }
721            instruments.extend(parse_markets_to_instruments(&markets, ts_init));
722        }
723
724        if succeeded == 0 && total > 0 {
725            anyhow::bail!("All {total} event slug requests failed");
726        }
727
728        log::debug!(
729            "Parsed {} instruments from event slug queries",
730            instruments.len()
731        );
732        Ok(instruments)
733    }
734
735    /// Fetches instruments using arbitrary Gamma API query params with auto-pagination.
736    pub async fn request_instruments_by_params(
737        &self,
738        base_params: GetGammaMarketsParams,
739    ) -> anyhow::Result<Vec<InstrumentAny>> {
740        let markets = self.fetch_gamma_markets_paginated(base_params).await?;
741        let ts_init = self.clock.get_time_ns();
742        let instruments = parse_markets_to_instruments(&markets, ts_init);
743        log::debug!("Parsed {} instruments from params query", instruments.len());
744        Ok(instruments)
745    }
746
747    /// Same as [`Self::request_instruments_by_params`] but also returns
748    /// condition IDs whose markets came back from Gamma with empty
749    /// `clob_token_ids`. Callers driving auto-load retries use the transient
750    /// list to distinguish "still hydrating in the CLOB" from "absent on the
751    /// venue".
752    pub async fn request_instruments_by_params_with_transient(
753        &self,
754        base_params: GetGammaMarketsParams,
755    ) -> anyhow::Result<(Vec<InstrumentAny>, Vec<String>)> {
756        let markets = self.fetch_gamma_markets_paginated(base_params).await?;
757        let ts_init = self.clock.get_time_ns();
758        let (instruments, transient) = parse_markets_with_transient(&markets, ts_init);
759        log::debug!(
760            "Parsed {} instruments and {} transient condition_id(s) from params query",
761            instruments.len(),
762            transient.len(),
763        );
764        Ok((instruments, transient))
765    }
766
767    /// Fetches raw Gamma markets using arbitrary query params with auto-pagination.
768    pub async fn request_markets_by_params(
769        &self,
770        base_params: GetGammaMarketsParams,
771    ) -> anyhow::Result<Vec<GammaMarket>> {
772        self.fetch_gamma_markets_paginated(base_params).await
773    }
774
775    /// Fetches instruments from an event slug with client-side sorting and limiting.
776    ///
777    /// The `/events?slug=` response already includes the full markets array,
778    /// so no second API call is needed. Sorting and truncation are applied
779    /// client-side using fields from `GetGammaMarketsParams`:
780    /// - `order`: sort field (`"liquidity"`, `"volume"`, `"volume24hr"`)
781    /// - `ascending`: sort direction (default: descending)
782    /// - `max_markets`: truncate after sorting
783    pub async fn request_instruments_by_event_query(
784        &self,
785        event_slug: &str,
786        params: GetGammaMarketsParams,
787    ) -> anyhow::Result<Vec<InstrumentAny>> {
788        let events = self.inner.get_gamma_events_by_slug(event_slug).await?;
789        let mut markets = flatten_event_markets(events);
790
791        if markets.is_empty() {
792            log::warn!("No markets found in event slug '{event_slug}'");
793            return Ok(Vec::new());
794        }
795
796        log::debug!("Event '{event_slug}' returned {} markets", markets.len());
797
798        // Client-side sort
799        if let Some(ref order_field) = params.order {
800            let ascending = params.ascending.unwrap_or(false);
801            markets.sort_by(|a, b| {
802                let cmp = match order_field.as_str() {
803                    "liquidity" => a
804                        .liquidity_num
805                        .unwrap_or(0.0)
806                        .partial_cmp(&b.liquidity_num.unwrap_or(0.0)),
807                    "volume" => a
808                        .volume_num
809                        .unwrap_or(0.0)
810                        .partial_cmp(&b.volume_num.unwrap_or(0.0)),
811                    "volume24hr" => a
812                        .volume_24hr
813                        .unwrap_or(0.0)
814                        .partial_cmp(&b.volume_24hr.unwrap_or(0.0)),
815                    "competitive" => a
816                        .competitive
817                        .unwrap_or(0.0)
818                        .partial_cmp(&b.competitive.unwrap_or(0.0)),
819                    "spread" => a
820                        .spread
821                        .unwrap_or(f64::MAX)
822                        .partial_cmp(&b.spread.unwrap_or(f64::MAX)),
823                    "best_bid" => a
824                        .best_bid
825                        .unwrap_or(0.0)
826                        .partial_cmp(&b.best_bid.unwrap_or(0.0)),
827                    "one_day_price_change" => a
828                        .one_day_price_change
829                        .unwrap_or(0.0)
830                        .partial_cmp(&b.one_day_price_change.unwrap_or(0.0)),
831                    "volume_1wk" => a
832                        .volume_1wk
833                        .unwrap_or(0.0)
834                        .partial_cmp(&b.volume_1wk.unwrap_or(0.0)),
835                    _ => None,
836                };
837                let cmp = cmp.unwrap_or(std::cmp::Ordering::Equal);
838                if ascending { cmp } else { cmp.reverse() }
839            });
840        }
841
842        // Client-side truncation
843        if let Some(cap) = params.max_markets {
844            markets.truncate(cap as usize);
845        }
846
847        let ts_init = self.clock.get_time_ns();
848        let instruments = parse_markets_to_instruments(&markets, ts_init);
849        log::debug!(
850            "Parsed {} instruments from event query '{event_slug}'",
851            instruments.len()
852        );
853        Ok(instruments)
854    }
855
856    /// Fetches events from the Gamma API with the given base params, paginating automatically.
857    async fn fetch_gamma_events_paginated(
858        &self,
859        base_params: GetGammaEventsParams,
860    ) -> anyhow::Result<Vec<GammaEvent>> {
861        let page_size = base_params
862            .limit
863            .unwrap_or(GAMMA_EVENTS_KEYSET_PAGE_LIMIT)
864            .min(GAMMA_EVENTS_KEYSET_PAGE_LIMIT);
865        let protocol = CursorProtocol::<GammaStop>::gamma("Gamma event");
866        let reducer = WindowedCollect::new(
867            base_params.offset.unwrap_or(0) as usize,
868            base_params.max_events.map(|value| value as usize),
869            GammaStop::CallerCapped,
870        );
871        let paginator = Paginator::new("Gamma event", protocol, reducer);
872        let completed = paginator
873            .run(
874                |position| {
875                    let after_cursor = position.map(|cursor| cursor.as_ref().to_string());
876                    let params = GetGammaEventsParams {
877                        limit: Some(page_size),
878                        offset: None,
879                        ..base_params.clone()
880                    };
881                    async move {
882                        let response = self
883                            .inner
884                            .get_gamma_events_keyset(params, after_cursor.as_deref())
885                            .await?;
886                        Ok::<_, anyhow::Error>(FetchOutcome::Page {
887                            rows: response.events,
888                            wire: response.next_cursor,
889                        })
890                    }
891                },
892                anyhow::Error::new,
893            )
894            .await?;
895
896        match completed.completion {
897            Completion::WireExhausted | Completion::Stopped(GammaStop::CallerCapped) => {
898                Ok(completed.output)
899            }
900        }
901    }
902
903    /// Fetches instruments from events matching full query params (paginated).
904    pub async fn request_instruments_by_event_params(
905        &self,
906        params: GetGammaEventsParams,
907    ) -> anyhow::Result<Vec<InstrumentAny>> {
908        let events = self.fetch_gamma_events_paginated(params).await?;
909        let ts_init = self.clock.get_time_ns();
910        let total_events = events.len();
911        let markets = flatten_event_markets(events);
912        let total_markets = markets.len();
913        let instruments = parse_markets_to_instruments(&markets, ts_init);
914        log::debug!(
915            "Parsed {} instruments from {total_events} events ({total_markets} markets)",
916            instruments.len(),
917        );
918        Ok(instruments)
919    }
920
921    /// Fetches raw Gamma events using arbitrary query params with auto-pagination.
922    pub async fn request_events_by_params(
923        &self,
924        params: GetGammaEventsParams,
925    ) -> anyhow::Result<Vec<GammaEvent>> {
926        self.fetch_gamma_events_paginated(params).await
927    }
928
929    /// Searches for instruments via the Gamma public search endpoint.
930    pub async fn request_instruments_by_search(
931        &self,
932        params: GetSearchParams,
933    ) -> anyhow::Result<Vec<InstrumentAny>> {
934        let response = self.inner.get_public_search(params).await?;
935        let ts_init = self.clock.get_time_ns();
936
937        let mut instruments = Vec::new();
938
939        if let Some(markets) = &response.markets {
940            instruments.extend(parse_markets_to_instruments(markets, ts_init));
941        }
942
943        if let Some(events) = &response.events {
944            let event_markets = flatten_event_markets(events.clone());
945            instruments.extend(parse_markets_to_instruments(&event_markets, ts_init));
946        }
947
948        log::debug!("Parsed {} instruments from search query", instruments.len());
949        Ok(instruments)
950    }
951
952    /// Fetches available tags from the Gamma API.
953    pub async fn request_tags(&self) -> anyhow::Result<Vec<GammaTag>> {
954        Ok(self.inner.get_gamma_tags().await?)
955    }
956
957    /// Returns a reference to the underlying raw HTTP client.
958    #[must_use]
959    pub fn inner(&self) -> &Arc<PolymarketGammaRawHttpClient> {
960        &self.inner
961    }
962}
nautilus_polymarket/http/gamma.rs

nautilus_polymarket/http/
gamma.rs
