Skip to main content

nautilus_polymarket/
providers.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//! Instrument provider for the Polymarket adapter.
17
18use std::{collections::HashMap, fmt::Debug, sync::Arc};
19
20use ahash::{AHashMap, AHashSet};
21use async_trait::async_trait;
22use nautilus_common::providers::{InstrumentProvider, InstrumentStore};
23use nautilus_model::{
24    identifiers::InstrumentId,
25    instruments::{Instrument, InstrumentAny},
26};
27use rust_decimal::Decimal;
28use ustr::Ustr;
29
30use crate::{
31    common::{consts::GAMMA_CONDITION_IDS_BATCH_SIZE, parse::parse_decimal_exact},
32    config::PolymarketInstrumentProviderConfig,
33    filters::InstrumentFilter,
34    http::{
35        gamma::PolymarketGammaHttpClient,
36        models::GammaTag,
37        query::{GetGammaEventsParams, GetGammaMarketsParams},
38    },
39};
40
41/// Provides Polymarket instruments via the Gamma API.
42///
43/// Wraps [`PolymarketGammaHttpClient`] with an [`InstrumentStore`] and a
44/// token_id index for resolving WebSocket asset IDs to instruments.
45///
46/// Optional [`InstrumentFilter`]s control which instruments are loaded
47/// during `load_all()`. Without filters, all active markets are fetched.
48pub struct PolymarketInstrumentProvider {
49    store: InstrumentStore,
50    http_client: PolymarketGammaHttpClient,
51    token_index: AHashMap<Ustr, InstrumentId>,
52    filters: Vec<Arc<dyn InstrumentFilter>>,
53    config: PolymarketInstrumentProviderConfig,
54}
55
56impl Debug for PolymarketInstrumentProvider {
57    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
58        f.debug_struct(stringify!(PolymarketInstrumentProvider))
59            .field("store", &self.store)
60            .field("http_client", &self.http_client)
61            .field("token_index_len", &self.token_index.len())
62            .field("filters", &self.filters)
63            .field("config", &self.config)
64            .finish()
65    }
66}
67
68impl PolymarketInstrumentProvider {
69    /// Creates a new [`PolymarketInstrumentProvider`] with an empty store and no filters.
70    #[must_use]
71    pub fn new(
72        http_client: PolymarketGammaHttpClient,
73        config: Option<PolymarketInstrumentProviderConfig>,
74    ) -> Self {
75        Self {
76            store: InstrumentStore::new(),
77            http_client,
78            token_index: AHashMap::new(),
79            filters: Vec::new(),
80            config: config.unwrap_or_default(),
81        }
82    }
83
84    /// Creates a new [`PolymarketInstrumentProvider`] with multiple filters.
85    #[must_use]
86    pub fn with_filters(
87        http_client: PolymarketGammaHttpClient,
88        config: Option<PolymarketInstrumentProviderConfig>,
89        filters: Vec<Arc<dyn InstrumentFilter>>,
90    ) -> Self {
91        Self {
92            store: InstrumentStore::new(),
93            http_client,
94            token_index: AHashMap::new(),
95            filters,
96            config: config.unwrap_or_default(),
97        }
98    }
99
100    /// Creates a new [`PolymarketInstrumentProvider`] with a single filter.
101    #[must_use]
102    pub fn with_filter(
103        http_client: PolymarketGammaHttpClient,
104        config: Option<PolymarketInstrumentProviderConfig>,
105        filter: Arc<dyn InstrumentFilter>,
106    ) -> Self {
107        Self {
108            store: InstrumentStore::new(),
109            http_client,
110            token_index: AHashMap::new(),
111            filters: vec![filter],
112            config: config.unwrap_or_default(),
113        }
114    }
115
116    /// Adds an instrument filter for subsequent `load_all()` calls.
117    pub fn add_filter(&mut self, filter: Arc<dyn InstrumentFilter>) {
118        self.filters.push(filter);
119    }
120
121    /// Clears all instrument filters, reverting to bulk load behavior.
122    pub fn clear_filters(&mut self) {
123        self.filters.clear();
124    }
125
126    /// Returns the instrument for the given token ID, if found.
127    #[must_use]
128    pub fn get_by_token_id(&self, token_id: &Ustr) -> Option<&InstrumentAny> {
129        let instrument_id = self.token_index.get(token_id)?;
130        self.store.find(instrument_id)
131    }
132
133    /// Builds a frozen snapshot mapping token IDs to instruments.
134    ///
135    /// Used to provide the WS handler task with a read-only lookup
136    /// table after instruments have been loaded.
137    #[must_use]
138    pub fn build_token_map(&self) -> AHashMap<Ustr, InstrumentAny> {
139        self.token_index
140            .iter()
141            .filter_map(|(token_id, instrument_id)| {
142                self.store
143                    .find(instrument_id)
144                    .map(|inst| (*token_id, inst.clone()))
145            })
146            .collect()
147    }
148
149    /// Loads instruments for the given slugs additively into the store.
150    ///
151    /// Unlike [`Self::load_all`], this does **not** clear existing instruments or
152    /// mark the store as initialized, allowing incremental loading of
153    /// slug-based markets alongside bulk data.
154    ///
155    /// # Errors
156    ///
157    /// Returns an error if the HTTP request or parsing fails.
158    pub async fn load_by_slugs(&mut self, slugs: Vec<String>) -> anyhow::Result<()> {
159        let instruments = self.http_client.request_instruments_by_slugs(slugs).await?;
160
161        for instrument in &instruments {
162            self.token_index.insert(
163                Ustr::from(instrument.raw_symbol().as_str()),
164                instrument.id(),
165            );
166        }
167
168        self.store.add_bulk(instruments);
169
170        Ok(())
171    }
172
173    /// Returns a clone of the configured instrument filters.
174    #[must_use]
175    pub fn filters(&self) -> Vec<Arc<dyn InstrumentFilter>> {
176        self.filters.clone()
177    }
178
179    /// Returns a reference to the underlying HTTP client.
180    #[must_use]
181    pub fn http_client(&self) -> &PolymarketGammaHttpClient {
182        &self.http_client
183    }
184
185    /// Returns the configured provider config.
186    #[must_use]
187    pub fn config(&self) -> &PolymarketInstrumentProviderConfig {
188        &self.config
189    }
190
191    /// Fetches available tags from the Gamma API.
192    pub async fn list_tags(&self) -> anyhow::Result<Vec<GammaTag>> {
193        self.http_client.request_tags().await
194    }
195
196    pub fn add_instruments(&mut self, instruments: Vec<InstrumentAny>) {
197        for inst in &instruments {
198            self.token_index
199                .insert(Ustr::from(inst.raw_symbol().as_str()), inst.id());
200        }
201        self.store.add_bulk(instruments);
202    }
203
204    /// Loads instruments for the given event slugs additively into the store.
205    ///
206    /// Unlike [`Self::load_all`], this does **not** clear existing instruments or
207    /// mark the store as initialized, allowing incremental loading of
208    /// event-scoped markets alongside bulk data.
209    pub async fn load_by_event_slugs(&mut self, slugs: Vec<String>) -> anyhow::Result<()> {
210        let instruments = self
211            .http_client
212            .request_instruments_by_event_slugs(slugs)
213            .await?;
214        self.add_instruments(instruments);
215        Ok(())
216    }
217
218    /// Loads instruments for the given Gamma series IDs additively into the store.
219    ///
220    /// Resolves each series to its active, unresolved events and loads their
221    /// markets. Unlike [`Self::load_all`], this does **not** clear existing
222    /// instruments or mark the store as initialized.
223    pub async fn load_by_series_ids(&mut self, series_ids: Vec<u64>) -> anyhow::Result<()> {
224        let instruments = self
225            .http_client
226            .request_instruments_by_event_params(series_events_params(series_ids))
227            .await?;
228        self.add_instruments(instruments);
229        Ok(())
230    }
231
232    /// Initializes the provider using its configured bootstrap scope.
233    pub async fn initialize(&mut self, reload: bool) -> anyhow::Result<()> {
234        if self.store.is_initialized() && !reload {
235            return Ok(());
236        }
237
238        let should_load_all = has_bootstrap_scope(&self.config, &self.filters);
239        let has_load_ids = self.config.has_load_ids();
240
241        if !should_load_all && !has_load_ids {
242            if self.config.log_warnings {
243                log::warn!(
244                    "No Polymarket instrument bootstrap configured: set instrument_config.load_all, instrument_config.load_ids, instrument_config.filters, instrument_config.event_slugs, instrument_config.market_slugs, instrument_config.event_slug_builder, or instrument_config.series_ids, or register an instrument filter"
245                );
246            }
247            return Ok(());
248        }
249
250        // Registered `InstrumentFilter`s take precedence over the Gamma filter map,
251        // so say when the map is being dropped rather than ignoring it silently.
252        if self.config.log_warnings
253            && !self.filters.is_empty()
254            && self.config.has_nonempty_filters()
255        {
256            log::warn!(
257                "Registered instrument filters take precedence: instrument_config.filters is ignored for this bootstrap"
258            );
259        }
260
261        if should_load_all {
262            self.load_scoped_all().await?;
263        }
264
265        if has_load_ids {
266            // Deliberately not passed `config.filters`: `load_ids` names instruments
267            // explicitly, so intersecting it with the filter map would silently drop
268            // any ID that falls outside those filters. The scopes are additive, and
269            // an explicitly requested instrument is requested unconditionally.
270            let load_ids = self.config.load_ids.clone().unwrap_or_default();
271            self.load_ids(&load_ids, None).await?;
272        }
273
274        // Registered filters take precedence over the `filters` map, so a map paired with an
275        // accept-only filter loads nothing while `should_load_all()` still reports true. Latching
276        // empty would strand the provider against a later `initialize(false)`.
277        let sourced_by_registered_filters = !has_load_ids && !self.filters.is_empty();
278
279        if sourced_by_registered_filters && self.store.count() == 0 {
280            return Ok(());
281        }
282
283        self.store.set_initialized();
284        Ok(())
285    }
286
287    async fn load_scoped_all(&mut self) -> anyhow::Result<()> {
288        let event_slugs = self.resolve_event_slugs()?;
289        let market_slugs = self
290            .config
291            .market_slugs
292            .clone()
293            .unwrap_or_default()
294            .into_iter()
295            .filter(|slug| !slug.trim().is_empty())
296            .collect::<Vec<_>>();
297        let series_ids = self.config.series_ids.clone().unwrap_or_default();
298
299        // The clearing bulk load must run before the additive scoped loads.
300        // Explicit scoping never broadens into an unfiltered full-universe fetch,
301        // but every filter-driven query is bounded, so it composes with explicit
302        // scopes instead of being silently dropped. Registered `InstrumentFilter`s
303        // count here just as the Gamma filter map does, otherwise this path would
304        // drop them while `fetch_configured_instruments` keeps them, and the
305        // bootstrap and interval-refresh universes would diverge.
306        //
307        // This deliberately does not go through `load_all`, which marks the store
308        // initialized on completion. The additive scopes below are still
309        // outstanding at this point, and a failure in one of them would otherwise
310        // leave an initialized-but-incomplete store that makes a subsequent
311        // `initialize(false)` short-circuit and skip the missing scopes for good.
312        if !self.config.has_explicit_scope()
313            || self.config.has_nonempty_filters()
314            || !self.filters.is_empty()
315        {
316            let filters = self.config.filters.clone();
317            let instruments = self.fetch_bulk_instruments(filters.as_ref()).await?;
318            self.replace_instruments(instruments);
319        }
320
321        if !series_ids.is_empty() {
322            self.load_by_series_ids(series_ids).await?;
323        }
324
325        if !event_slugs.is_empty() {
326            self.load_by_event_slugs(event_slugs).await?;
327        }
328
329        if !market_slugs.is_empty() {
330            self.load_by_slugs(market_slugs).await?;
331        }
332
333        Ok(())
334    }
335
336    fn resolve_event_slugs(&self) -> anyhow::Result<Vec<String>> {
337        if let Some(builder) = self.config.event_slug_builder.as_ref() {
338            return builder.build_event_slugs();
339        }
340
341        Ok(self
342            .config
343            .event_slugs
344            .clone()
345            .unwrap_or_default()
346            .into_iter()
347            .filter(|slug| !slug.trim().is_empty())
348            .collect())
349    }
350
351    /// Loads instruments using all configured filters, combining results from
352    /// each filter's methods that return `Some`.
353    async fn load_filtered(&self) -> anyhow::Result<Vec<InstrumentAny>> {
354        fetch_instruments(&self.http_client, &self.filters).await
355    }
356
357    /// Fetches the bulk instrument universe without mutating provider state.
358    ///
359    /// Registered [`InstrumentFilter`]s take precedence; otherwise a non-empty
360    /// Gamma filter map bounds the query, and an absent or empty map falls back
361    /// to the full universe.
362    async fn fetch_bulk_instruments(
363        &self,
364        filters: Option<&HashMap<String, String>>,
365    ) -> anyhow::Result<Vec<InstrumentAny>> {
366        if !self.filters.is_empty() {
367            return self.load_filtered().await;
368        }
369
370        match filters {
371            Some(map) if !map.is_empty() => {
372                let params = build_gamma_params_from_hashmap(map)?;
373                self.http_client.request_instruments_by_params(params).await
374            }
375            _ => self.http_client.request_instruments().await,
376        }
377    }
378
379    /// Replaces the store contents, leaving the initialized flag untouched.
380    ///
381    /// Callers that complete a full bootstrap are responsible for marking the
382    /// store initialized once every scope has succeeded.
383    fn replace_instruments(&mut self, instruments: Vec<InstrumentAny>) {
384        self.store.clear();
385        self.token_index.clear();
386        self.add_instruments(instruments);
387    }
388}
389
390/// Fetches instruments from the Gamma API, respecting any configured filters.
391pub async fn fetch_instruments(
392    http_client: &PolymarketGammaHttpClient,
393    filters: &[Arc<dyn InstrumentFilter>],
394) -> anyhow::Result<Vec<InstrumentAny>> {
395    if filters.is_empty() {
396        return http_client.request_instruments().await;
397    }
398
399    let mut instruments = Vec::new();
400
401    for filter in filters {
402        if let Some(slugs) = filter.market_slugs()
403            && !slugs.is_empty()
404        {
405            let result = http_client.request_instruments_by_slugs(slugs).await?;
406            instruments.extend(result);
407        }
408
409        if let Some(event_slugs) = filter.event_slugs()
410            && !event_slugs.is_empty()
411        {
412            let result = http_client
413                .request_instruments_by_event_slugs(event_slugs)
414                .await?;
415            instruments.extend(result);
416        }
417
418        if let Some(params) = filter.query_params() {
419            let result = http_client.request_instruments_by_params(params).await?;
420            instruments.extend(result);
421        }
422
423        if let Some(event_queries) = filter.event_queries() {
424            for (event_slug, params) in event_queries {
425                let result = http_client
426                    .request_instruments_by_event_query(&event_slug, params)
427                    .await?;
428                instruments.extend(result);
429            }
430        }
431
432        if let Some(params) = filter.event_params() {
433            let result = http_client
434                .request_instruments_by_event_params(params)
435                .await?;
436            instruments.extend(result);
437        }
438
439        if let Some(params) = filter.search_params() {
440            let result = http_client.request_instruments_by_search(params).await?;
441            instruments.extend(result);
442        }
443    }
444
445    let mut seen = AHashSet::new();
446    instruments.retain(|inst| seen.insert(inst.id()));
447    instruments.retain(|inst| filters.iter().all(|f| f.accept(inst)));
448
449    Ok(instruments)
450}
451
452/// Fetches instruments using the configured provider bootstrap scope without
453/// mutating any provider state.
454pub async fn fetch_configured_instruments(
455    http_client: &PolymarketGammaHttpClient,
456    config: &PolymarketInstrumentProviderConfig,
457    filters: &[Arc<dyn InstrumentFilter>],
458) -> anyhow::Result<Vec<InstrumentAny>> {
459    let mut instruments = Vec::new();
460
461    if has_bootstrap_scope(config, filters) {
462        let has_explicit_scope = config.has_explicit_scope();
463        let event_slugs = if let Some(builder) = config.event_slug_builder.as_ref() {
464            builder.build_event_slugs()?
465        } else {
466            config
467                .event_slugs
468                .clone()
469                .unwrap_or_default()
470                .into_iter()
471                .filter(|slug| !slug.trim().is_empty())
472                .collect::<Vec<_>>()
473        };
474
475        let market_slugs = config
476            .market_slugs
477            .clone()
478            .unwrap_or_default()
479            .into_iter()
480            .filter(|slug| !slug.trim().is_empty())
481            .collect::<Vec<_>>();
482
483        let series_ids = config.series_ids.clone().unwrap_or_default();
484
485        // Explicit scoping never broadens into an unfiltered full-universe
486        // fetch, but every filter-driven query is bounded, so it composes with
487        // explicit series and slug scopes instead of being silently dropped.
488        // This mirrors `load_scoped_all`, which the interval refresh and
489        // `request_instruments` paths must stay consistent with.
490        if !filters.is_empty() {
491            instruments.extend(fetch_instruments(http_client, filters).await?);
492        } else if let Some(map) = config.filters.as_ref().filter(|map| !map.is_empty()) {
493            let params = build_gamma_params_from_hashmap(map)?;
494            instruments.extend(http_client.request_instruments_by_params(params).await?);
495        } else if !has_explicit_scope {
496            instruments.extend(http_client.request_instruments().await?);
497        }
498
499        if !series_ids.is_empty() {
500            instruments.extend(
501                http_client
502                    .request_instruments_by_event_params(series_events_params(series_ids))
503                    .await?,
504            );
505        }
506
507        if !event_slugs.is_empty() {
508            instruments.extend(
509                http_client
510                    .request_instruments_by_event_slugs(event_slugs)
511                    .await?,
512            );
513        }
514
515        if !market_slugs.is_empty() {
516            instruments.extend(
517                http_client
518                    .request_instruments_by_slugs(market_slugs)
519                    .await?,
520            );
521        }
522    }
523
524    if config.has_load_ids() {
525        // Queried by condition ID alone. Merging `config.filters` in here would
526        // intersect an explicit scope with a filter scope, so an ID outside those
527        // filters would never load despite being named directly. Mirrors the
528        // `load_ids` call in `initialize`.
529        let condition_ids = config
530            .load_ids
531            .clone()
532            .unwrap_or_default()
533            .into_iter()
534            .filter_map(|id| extract_condition_id(&id).ok())
535            .collect::<AHashSet<_>>()
536            .into_iter()
537            .collect::<Vec<_>>();
538
539        for chunk in condition_ids.chunks(GAMMA_CONDITION_IDS_BATCH_SIZE) {
540            let params = GetGammaMarketsParams {
541                condition_ids: Some(chunk.to_vec()),
542                ..Default::default()
543            };
544            instruments.extend(http_client.request_instruments_by_params(params).await?);
545        }
546    }
547
548    // Deduplicated but NOT filtered by `accept` here. `fetch_instruments` already
549    // applies acceptance to the results of filter-driven queries, which is the only
550    // place it belongs: applying it again across the whole collection would let a
551    // registered filter reject series, slug, and ID instruments that the provider's
552    // `load_scoped_all` adds unconditionally, so a refresh would silently drop
553    // instruments the bootstrap had loaded.
554    let mut seen = AHashSet::new();
555    instruments.retain(|inst| seen.insert(inst.id()));
556    Ok(instruments)
557}
558
559// Registered filters live on the provider, not the config, so the config predicate alone cannot
560// see them. Source methods are deliberately not evaluated: they are documented as re-evaluated
561// each load cycle, so probing here would consume a batch the fetch then misses.
562fn has_bootstrap_scope(
563    config: &PolymarketInstrumentProviderConfig,
564    filters: &[Arc<dyn InstrumentFilter>],
565) -> bool {
566    config.should_load_all() || !filters.is_empty()
567}
568
569/// Builds the Gamma events query that resolves series IDs to their active,
570/// unresolved events.
571fn series_events_params(series_ids: Vec<u64>) -> GetGammaEventsParams {
572    GetGammaEventsParams {
573        series_id: Some(series_ids),
574        active: Some(true),
575        closed: Some(false),
576        ..Default::default()
577    }
578}
579
580/// Extracts the condition ID from an instrument symbol.
581///
582/// Polymarket instrument symbols follow the pattern `{condition_id}-{token_id}`.
583/// The condition_id is a hex string (e.g. `0xabc123...`) and the token_id is a
584/// large decimal number. This extracts the condition_id by splitting at the last `-`.
585pub fn extract_condition_id(instrument_id: &InstrumentId) -> anyhow::Result<String> {
586    let symbol = instrument_id.symbol.as_str();
587    symbol
588        .rfind('-')
589        .map(|idx| symbol[..idx].to_string())
590        .ok_or_else(|| {
591            anyhow::anyhow!("Cannot extract condition_id from symbol '{symbol}': no '-' separator")
592        })
593}
594
595/// Extracts the token ID from an instrument symbol.
596///
597/// Polymarket instrument symbols follow the pattern `{condition_id}-{token_id}`. This extracts the
598/// token_id by splitting at the last `-`.
599pub(crate) fn extract_token_id(instrument_id: &InstrumentId) -> anyhow::Result<String> {
600    let symbol = instrument_id.symbol.as_str();
601    symbol
602        .rsplit_once('-')
603        .map(|(_, token_id)| token_id.to_string())
604        .ok_or_else(|| {
605            anyhow::anyhow!("Cannot extract token_id from symbol '{symbol}': no '-' separator")
606        })
607}
608
609/// Builds validated market keyset parameters from string key/value filters.
610///
611/// # Errors
612///
613/// Returns an error for unknown keys, malformed values, or invalid filter combinations.
614pub fn build_gamma_params_from_hashmap(
615    map: &HashMap<String, String>,
616) -> anyhow::Result<GetGammaMarketsParams> {
617    for key in map.keys() {
618        match key.as_str() {
619            "is_active"
620            | "active"
621            | "closed"
622            | "archived"
623            | "id"
624            | "limit"
625            | "offset"
626            | "order"
627            | "ascending"
628            | "slug"
629            | "clob_token_ids"
630            | "condition_ids"
631            | "question_ids"
632            | "market_maker_address"
633            | "liquidity_num_min"
634            | "liquidity_num_max"
635            | "volume_num_min"
636            | "volume_num_max"
637            | "start_date_min"
638            | "start_date_max"
639            | "end_date_min"
640            | "end_date_max"
641            | "tag_id"
642            | "related_tags"
643            | "tag_match"
644            | "decimalized"
645            | "cyom"
646            | "rfq_enabled"
647            | "uma_resolution_status"
648            | "game_id"
649            | "sports_market_types"
650            | "include_tag"
651            | "locale"
652            | "max_markets" => {}
653            _ => anyhow::bail!("Unknown Gamma market filter key '{key}'"),
654        }
655    }
656
657    let mut params = GetGammaMarketsParams::default();
658
659    if map
660        .get("is_active")
661        .map(|value| parse_gamma_filter_bool("market", "is_active", value))
662        .transpose()?
663        .unwrap_or(false)
664    {
665        params.active = Some(true);
666        params.archived = Some(false);
667        params.closed = Some(false);
668    }
669
670    if let Some(v) = map.get("active") {
671        params.active = Some(parse_gamma_filter_bool("market", "active", v)?);
672    }
673
674    if let Some(v) = map.get("closed") {
675        params.closed = Some(parse_gamma_filter_bool("market", "closed", v)?);
676    }
677
678    if let Some(v) = map.get("archived") {
679        params.archived = Some(parse_gamma_filter_bool("market", "archived", v)?);
680    }
681
682    if let Some(v) = map.get("id") {
683        params.id = Some(parse_gamma_numeric_filter_list("market", "id", v)?);
684    }
685
686    if let Some(v) = map.get("slug") {
687        params.slug = Some(parse_gamma_filter_list("market", "slug", v)?);
688    }
689
690    if let Some(v) = map.get("tag_id") {
691        params.tag_id = Some(parse_gamma_numeric_filter_list("market", "tag_id", v)?);
692    }
693
694    if let Some(v) = map.get("condition_ids") {
695        params.condition_ids = Some(parse_gamma_filter_list("market", "condition_ids", v)?);
696    }
697
698    if let Some(v) = map.get("clob_token_ids") {
699        params.clob_token_ids = Some(parse_gamma_filter_list("market", "clob_token_ids", v)?);
700    }
701
702    if let Some(v) = map.get("question_ids") {
703        params.question_ids = Some(parse_gamma_filter_list("market", "question_ids", v)?);
704    }
705
706    if let Some(v) = map.get("market_maker_address") {
707        params.market_maker_address = Some(parse_gamma_filter_list(
708            "market",
709            "market_maker_address",
710            v,
711        )?);
712    }
713
714    if let Some(v) = map.get("liquidity_num_min") {
715        params.liquidity_num_min = Some(parse_gamma_filter_decimal(
716            "market",
717            "liquidity_num_min",
718            v,
719        )?);
720    }
721
722    if let Some(v) = map.get("liquidity_num_max") {
723        params.liquidity_num_max = Some(parse_gamma_filter_decimal(
724            "market",
725            "liquidity_num_max",
726            v,
727        )?);
728    }
729
730    if let Some(v) = map.get("volume_num_min") {
731        params.volume_num_min = Some(parse_gamma_filter_decimal("market", "volume_num_min", v)?);
732    }
733
734    if let Some(v) = map.get("volume_num_max") {
735        params.volume_num_max = Some(parse_gamma_filter_decimal("market", "volume_num_max", v)?);
736    }
737
738    if let Some(v) = map.get("order") {
739        params.order = Some(parse_gamma_filter_string("market", "order", v)?);
740    }
741
742    if let Some(v) = map.get("ascending") {
743        params.ascending = Some(parse_gamma_filter_bool("market", "ascending", v)?);
744    }
745
746    if let Some(v) = map.get("limit") {
747        params.limit = Some(parse_gamma_filter_u32("market", "limit", v)?.min(100));
748    }
749
750    if let Some(v) = map.get("offset") {
751        params.offset = Some(parse_gamma_filter_u32("market", "offset", v)?);
752    }
753
754    if let Some(v) = map.get("start_date_min") {
755        params.start_date_min = Some(parse_gamma_filter_string("market", "start_date_min", v)?);
756    }
757
758    if let Some(v) = map.get("start_date_max") {
759        params.start_date_max = Some(parse_gamma_filter_string("market", "start_date_max", v)?);
760    }
761
762    if let Some(v) = map.get("end_date_min") {
763        params.end_date_min = Some(parse_gamma_filter_string("market", "end_date_min", v)?);
764    }
765
766    if let Some(v) = map.get("end_date_max") {
767        params.end_date_max = Some(parse_gamma_filter_string("market", "end_date_max", v)?);
768    }
769
770    if let Some(v) = map.get("related_tags") {
771        params.related_tags = Some(parse_gamma_filter_bool("market", "related_tags", v)?);
772    }
773
774    if let Some(v) = map.get("tag_match") {
775        params.tag_match = Some(parse_gamma_filter_string("market", "tag_match", v)?);
776    }
777
778    if let Some(v) = map.get("decimalized") {
779        params.decimalized = Some(parse_gamma_filter_bool("market", "decimalized", v)?);
780    }
781
782    if let Some(v) = map.get("cyom") {
783        params.cyom = Some(parse_gamma_filter_bool("market", "cyom", v)?);
784    }
785
786    if let Some(v) = map.get("rfq_enabled") {
787        params.rfq_enabled = Some(parse_gamma_filter_bool("market", "rfq_enabled", v)?);
788    }
789
790    if let Some(v) = map.get("uma_resolution_status") {
791        params.uma_resolution_status = Some(parse_gamma_filter_string(
792            "market",
793            "uma_resolution_status",
794            v,
795        )?);
796    }
797
798    if let Some(v) = map.get("game_id") {
799        params.game_id = Some(parse_gamma_filter_string("market", "game_id", v)?);
800    }
801
802    if let Some(v) = map.get("sports_market_types") {
803        params.sports_market_types =
804            Some(parse_gamma_filter_list("market", "sports_market_types", v)?);
805    }
806
807    if let Some(v) = map.get("include_tag") {
808        params.include_tag = Some(parse_gamma_filter_bool("market", "include_tag", v)?);
809    }
810
811    if let Some(v) = map.get("locale") {
812        params.locale = Some(parse_gamma_filter_string("market", "locale", v)?);
813    }
814
815    if let Some(v) = map.get("max_markets") {
816        params.max_markets = Some(parse_gamma_filter_u32("market", "max_markets", v)?);
817    }
818
819    params.validate_keyset().map_err(|e| anyhow::anyhow!(e))?;
820    Ok(params)
821}
822
823/// Builds validated event keyset parameters from string key/value filters.
824///
825/// # Errors
826///
827/// Returns an error for unknown keys, malformed values, or invalid filter combinations.
828pub fn build_gamma_event_params_from_hashmap(
829    map: &HashMap<String, String>,
830) -> anyhow::Result<GetGammaEventsParams> {
831    for key in map.keys() {
832        match key.as_str() {
833            "is_active" | "active" | "closed" | "archived" | "id" | "slug" | "live"
834            | "featured" | "cyom" | "title_search" | "liquidity_min" | "liquidity_max"
835            | "volume_min" | "volume_max" | "start_date_min" | "start_date_max"
836            | "end_date_min" | "end_date_max" | "start_time_min" | "start_time_max" | "tag_id"
837            | "tag_slug" | "exclude_tag_id" | "related_tags" | "tag_match" | "series_id"
838            | "game_id" | "event_date" | "event_week" | "featured_order" | "recurrence"
839            | "created_by" | "parent_event_id" | "include_children" | "partner_slug"
840            | "include_chat" | "include_template" | "include_best_lines" | "locale" | "order"
841            | "ascending" | "limit" | "offset" | "max_events" => {}
842            _ => anyhow::bail!("Unknown Gamma event filter key '{key}'"),
843        }
844    }
845
846    let mut params = GetGammaEventsParams::default();
847
848    if map
849        .get("is_active")
850        .map(|value| parse_gamma_filter_bool("event", "is_active", value))
851        .transpose()?
852        .unwrap_or(false)
853    {
854        params.active = Some(true);
855        params.archived = Some(false);
856        params.closed = Some(false);
857    }
858
859    macro_rules! set_bool {
860        ($field:ident) => {
861            if let Some(value) = map.get(stringify!($field)) {
862                params.$field = Some(parse_gamma_filter_bool("event", stringify!($field), value)?);
863            }
864        };
865    }
866    macro_rules! set_string {
867        ($field:ident) => {
868            if let Some(value) = map.get(stringify!($field)) {
869                params.$field = Some(parse_gamma_filter_string(
870                    "event",
871                    stringify!($field),
872                    value,
873                )?);
874            }
875        };
876    }
877    macro_rules! set_decimal {
878        ($field:ident) => {
879            if let Some(value) = map.get(stringify!($field)) {
880                params.$field = Some(parse_gamma_filter_decimal(
881                    "event",
882                    stringify!($field),
883                    value,
884                )?);
885            }
886        };
887    }
888    macro_rules! set_u32 {
889        ($field:ident) => {
890            if let Some(value) = map.get(stringify!($field)) {
891                params.$field = Some(parse_gamma_filter_u32("event", stringify!($field), value)?);
892            }
893        };
894    }
895    macro_rules! set_u64 {
896        ($field:ident) => {
897            if let Some(value) = map.get(stringify!($field)) {
898                params.$field = Some(parse_gamma_filter_u64("event", stringify!($field), value)?);
899            }
900        };
901    }
902    macro_rules! set_strings {
903        ($field:ident) => {
904            if let Some(value) = map.get(stringify!($field)) {
905                params.$field = Some(parse_gamma_filter_list("event", stringify!($field), value)?);
906            }
907        };
908    }
909    macro_rules! set_u64s {
910        ($field:ident) => {
911            if let Some(value) = map.get(stringify!($field)) {
912                params.$field = Some(parse_gamma_numeric_filter_list(
913                    "event",
914                    stringify!($field),
915                    value,
916                )?);
917            }
918        };
919    }
920
921    set_bool!(active);
922    set_bool!(closed);
923    set_bool!(archived);
924    set_bool!(live);
925    set_bool!(featured);
926    set_bool!(cyom);
927    set_bool!(related_tags);
928    set_bool!(featured_order);
929    set_bool!(include_children);
930    set_bool!(include_chat);
931    set_bool!(include_template);
932    set_bool!(include_best_lines);
933    set_bool!(ascending);
934    set_strings!(slug);
935    set_strings!(created_by);
936    set_u64s!(id);
937    set_u64s!(tag_id);
938    set_u64s!(exclude_tag_id);
939    set_u64s!(series_id);
940    set_u64s!(game_id);
941    set_string!(title_search);
942    set_string!(start_date_min);
943    set_string!(start_date_max);
944    set_string!(end_date_min);
945    set_string!(end_date_max);
946    set_string!(start_time_min);
947    set_string!(start_time_max);
948    set_string!(tag_slug);
949    set_string!(tag_match);
950    set_string!(event_date);
951    set_string!(recurrence);
952    set_string!(partner_slug);
953    set_string!(locale);
954    set_string!(order);
955    set_decimal!(liquidity_min);
956    set_decimal!(liquidity_max);
957    set_decimal!(volume_min);
958    set_decimal!(volume_max);
959    set_u32!(event_week);
960    set_u32!(limit);
961    set_u32!(offset);
962    set_u32!(max_events);
963    set_u64!(parent_event_id);
964
965    params.validate_keyset().map_err(anyhow::Error::msg)?;
966    Ok(params)
967}
968
969fn parse_gamma_filter_bool(scope: &str, key: &str, value: &str) -> anyhow::Result<bool> {
970    if value.eq_ignore_ascii_case("true") {
971        Ok(true)
972    } else if value.eq_ignore_ascii_case("false") {
973        Ok(false)
974    } else {
975        anyhow::bail!("Gamma {scope} filter '{key}' must be true or false, was '{value}'")
976    }
977}
978
979fn parse_gamma_filter_u32(scope: &str, key: &str, value: &str) -> anyhow::Result<u32> {
980    value.parse::<u32>().map_err(|e| {
981        anyhow::anyhow!("Gamma {scope} filter '{key}' must be an unsigned integer: {e}")
982    })
983}
984
985fn parse_gamma_filter_u64(scope: &str, key: &str, value: &str) -> anyhow::Result<u64> {
986    value.parse::<u64>().map_err(|e| {
987        anyhow::anyhow!("Gamma {scope} filter '{key}' must be an unsigned integer: {e}")
988    })
989}
990
991fn parse_gamma_filter_decimal(scope: &str, key: &str, value: &str) -> anyhow::Result<Decimal> {
992    parse_decimal_exact(value)
993        .map_err(|e| anyhow::anyhow!("Gamma {scope} filter '{key}' must be a decimal number: {e}"))
994}
995
996fn parse_gamma_filter_list(scope: &str, key: &str, value: &str) -> anyhow::Result<Vec<String>> {
997    let values = value
998        .split(',')
999        .map(str::trim)
1000        .map(str::to_string)
1001        .collect::<Vec<_>>();
1002
1003    if values.is_empty() || values.iter().any(String::is_empty) {
1004        anyhow::bail!("Gamma {scope} filter '{key}' must contain non-empty comma-separated values")
1005    }
1006    Ok(values)
1007}
1008
1009fn parse_gamma_numeric_filter_list(
1010    scope: &str,
1011    key: &str,
1012    value: &str,
1013) -> anyhow::Result<Vec<u64>> {
1014    parse_gamma_filter_list(scope, key, value)?
1015        .into_iter()
1016        .map(|item| {
1017            item.parse::<u64>().map_err(|e| {
1018                anyhow::anyhow!(
1019                    "Gamma {scope} filter '{key}' values must be unsigned integers: {e}"
1020                )
1021            })
1022        })
1023        .collect()
1024}
1025
1026fn parse_gamma_filter_string(scope: &str, key: &str, value: &str) -> anyhow::Result<String> {
1027    if value.trim().is_empty() {
1028        anyhow::bail!("Gamma {scope} filter '{key}' cannot be empty")
1029    }
1030    Ok(value.to_string())
1031}
1032
1033/// Resolves a tag slug to a tag ID by querying the Gamma tags endpoint.
1034pub async fn resolve_tag_slug(
1035    client: &PolymarketGammaHttpClient,
1036    slug: &str,
1037) -> anyhow::Result<u64> {
1038    let tags = client.request_tags().await?;
1039    let tag_id = tags
1040        .iter()
1041        .find(|t| t.slug.as_deref() == Some(slug))
1042        .map(|t| t.id.as_str())
1043        .ok_or_else(|| anyhow::anyhow!("Tag slug '{slug}' not found"))?;
1044    tag_id
1045        .parse::<u64>()
1046        .map_err(|e| anyhow::anyhow!("Tag slug '{slug}' returned invalid ID '{tag_id}': {e}"))
1047}
1048
1049#[async_trait(?Send)]
1050impl InstrumentProvider for PolymarketInstrumentProvider {
1051    fn store(&self) -> &InstrumentStore {
1052        &self.store
1053    }
1054
1055    fn store_mut(&mut self) -> &mut InstrumentStore {
1056        &mut self.store
1057    }
1058
1059    async fn load_all(&mut self, filters: Option<&HashMap<String, String>>) -> anyhow::Result<()> {
1060        let instruments = self.fetch_bulk_instruments(filters).await?;
1061        self.replace_instruments(instruments);
1062        self.store.set_initialized();
1063
1064        Ok(())
1065    }
1066
1067    async fn load_ids(
1068        &mut self,
1069        instrument_ids: &[InstrumentId],
1070        filters: Option<&HashMap<String, String>>,
1071    ) -> anyhow::Result<()> {
1072        let missing: Vec<_> = instrument_ids
1073            .iter()
1074            .filter(|id| !self.store.contains(id))
1075            .collect();
1076
1077        if missing.is_empty() {
1078            return Ok(());
1079        }
1080
1081        // Extract unique condition IDs from instrument symbols
1082        // Symbol format: "{condition_id}-{token_id}"
1083        let mut condition_ids: Vec<String> = missing
1084            .iter()
1085            .filter_map(|id| extract_condition_id(id).ok())
1086            .collect();
1087        condition_ids.sort();
1088        condition_ids.dedup();
1089
1090        if condition_ids.is_empty() {
1091            return Ok(());
1092        }
1093
1094        let base_params = filters
1095            .map(build_gamma_params_from_hashmap)
1096            .transpose()?
1097            .unwrap_or_default();
1098
1099        for chunk in condition_ids.chunks(GAMMA_CONDITION_IDS_BATCH_SIZE) {
1100            let params = GetGammaMarketsParams {
1101                condition_ids: Some(chunk.to_vec()),
1102                ..base_params.clone()
1103            };
1104            let instruments = self
1105                .http_client
1106                .request_instruments_by_params(params)
1107                .await?;
1108            self.add_instruments(instruments);
1109        }
1110
1111        Ok(())
1112    }
1113
1114    async fn load(
1115        &mut self,
1116        instrument_id: &InstrumentId,
1117        filters: Option<&HashMap<String, String>>,
1118    ) -> anyhow::Result<()> {
1119        if self.store.contains(instrument_id) {
1120            return Ok(());
1121        }
1122
1123        // Try direct fetch via condition_id extracted from symbol
1124        if let Ok(cid) = extract_condition_id(instrument_id) {
1125            let params = GetGammaMarketsParams {
1126                condition_ids: Some(vec![cid]),
1127                ..Default::default()
1128            };
1129
1130            if let Ok(instruments) = self.http_client.request_instruments_by_params(params).await {
1131                self.add_instruments(instruments);
1132
1133                if self.store.contains(instrument_id) {
1134                    return Ok(());
1135                }
1136            }
1137        }
1138
1139        // Fallback: full load_all if not initialized. A provider with an explicit
1140        // scope is excluded: `load_all` would broaden it into the full-universe
1141        // fetch that scoping exists to avoid, and would also clear the partially
1142        // loaded store and mark it initialized, making a later `initialize(false)`
1143        // skip the scopes it still owes after a failed bootstrap.
1144        if !self.store.is_initialized() && !self.config.has_explicit_scope() {
1145            self.load_all(filters).await?;
1146        }
1147
1148        if self.store.contains(instrument_id) {
1149            Ok(())
1150        } else {
1151            anyhow::bail!("Instrument {instrument_id} not found on Polymarket")
1152        }
1153    }
1154}
1155
1156#[cfg(test)]
1157mod tests {
1158    use rstest::rstest;
1159
1160    use super::*;
1161
1162    #[rstest]
1163    #[case("0xcondition-0xtoken", Some("0xtoken"))]
1164    #[case("0xcondition-with-dash-0xtoken", Some("0xtoken"))]
1165    #[case("0xcondition", None)]
1166    fn extracts_token_id_from_instrument_symbol(
1167        #[case] symbol: &str,
1168        #[case] expected: Option<&str>,
1169    ) {
1170        let instrument_id = InstrumentId::from(format!("{symbol}.POLYMARKET").as_str());
1171        assert_eq!(extract_token_id(&instrument_id).ok().as_deref(), expected,);
1172    }
1173}