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,
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/// Builds validated market keyset parameters from string key/value filters.
596///
597/// # Errors
598///
599/// Returns an error for unknown keys, malformed values, or invalid filter combinations.
600pub fn build_gamma_params_from_hashmap(
601    map: &HashMap<String, String>,
602) -> anyhow::Result<GetGammaMarketsParams> {
603    for key in map.keys() {
604        match key.as_str() {
605            "is_active"
606            | "active"
607            | "closed"
608            | "archived"
609            | "id"
610            | "limit"
611            | "offset"
612            | "order"
613            | "ascending"
614            | "slug"
615            | "clob_token_ids"
616            | "condition_ids"
617            | "question_ids"
618            | "market_maker_address"
619            | "liquidity_num_min"
620            | "liquidity_num_max"
621            | "volume_num_min"
622            | "volume_num_max"
623            | "start_date_min"
624            | "start_date_max"
625            | "end_date_min"
626            | "end_date_max"
627            | "tag_id"
628            | "related_tags"
629            | "tag_match"
630            | "decimalized"
631            | "cyom"
632            | "rfq_enabled"
633            | "uma_resolution_status"
634            | "game_id"
635            | "sports_market_types"
636            | "include_tag"
637            | "locale"
638            | "max_markets" => {}
639            _ => anyhow::bail!("Unknown Gamma market filter key '{key}'"),
640        }
641    }
642
643    let mut params = GetGammaMarketsParams::default();
644
645    if map
646        .get("is_active")
647        .map(|value| parse_gamma_filter_bool("market", "is_active", value))
648        .transpose()?
649        .unwrap_or(false)
650    {
651        params.active = Some(true);
652        params.archived = Some(false);
653        params.closed = Some(false);
654    }
655
656    if let Some(v) = map.get("active") {
657        params.active = Some(parse_gamma_filter_bool("market", "active", v)?);
658    }
659
660    if let Some(v) = map.get("closed") {
661        params.closed = Some(parse_gamma_filter_bool("market", "closed", v)?);
662    }
663
664    if let Some(v) = map.get("archived") {
665        params.archived = Some(parse_gamma_filter_bool("market", "archived", v)?);
666    }
667
668    if let Some(v) = map.get("id") {
669        params.id = Some(parse_gamma_numeric_filter_list("market", "id", v)?);
670    }
671
672    if let Some(v) = map.get("slug") {
673        params.slug = Some(parse_gamma_filter_list("market", "slug", v)?);
674    }
675
676    if let Some(v) = map.get("tag_id") {
677        params.tag_id = Some(parse_gamma_numeric_filter_list("market", "tag_id", v)?);
678    }
679
680    if let Some(v) = map.get("condition_ids") {
681        params.condition_ids = Some(parse_gamma_filter_list("market", "condition_ids", v)?);
682    }
683
684    if let Some(v) = map.get("clob_token_ids") {
685        params.clob_token_ids = Some(parse_gamma_filter_list("market", "clob_token_ids", v)?);
686    }
687
688    if let Some(v) = map.get("question_ids") {
689        params.question_ids = Some(parse_gamma_filter_list("market", "question_ids", v)?);
690    }
691
692    if let Some(v) = map.get("market_maker_address") {
693        params.market_maker_address = Some(parse_gamma_filter_list(
694            "market",
695            "market_maker_address",
696            v,
697        )?);
698    }
699
700    if let Some(v) = map.get("liquidity_num_min") {
701        params.liquidity_num_min = Some(parse_gamma_filter_decimal(
702            "market",
703            "liquidity_num_min",
704            v,
705        )?);
706    }
707
708    if let Some(v) = map.get("liquidity_num_max") {
709        params.liquidity_num_max = Some(parse_gamma_filter_decimal(
710            "market",
711            "liquidity_num_max",
712            v,
713        )?);
714    }
715
716    if let Some(v) = map.get("volume_num_min") {
717        params.volume_num_min = Some(parse_gamma_filter_decimal("market", "volume_num_min", v)?);
718    }
719
720    if let Some(v) = map.get("volume_num_max") {
721        params.volume_num_max = Some(parse_gamma_filter_decimal("market", "volume_num_max", v)?);
722    }
723
724    if let Some(v) = map.get("order") {
725        params.order = Some(parse_gamma_filter_string("market", "order", v)?);
726    }
727
728    if let Some(v) = map.get("ascending") {
729        params.ascending = Some(parse_gamma_filter_bool("market", "ascending", v)?);
730    }
731
732    if let Some(v) = map.get("limit") {
733        params.limit = Some(parse_gamma_filter_u32("market", "limit", v)?.min(100));
734    }
735
736    if let Some(v) = map.get("offset") {
737        params.offset = Some(parse_gamma_filter_u32("market", "offset", v)?);
738    }
739
740    if let Some(v) = map.get("start_date_min") {
741        params.start_date_min = Some(parse_gamma_filter_string("market", "start_date_min", v)?);
742    }
743
744    if let Some(v) = map.get("start_date_max") {
745        params.start_date_max = Some(parse_gamma_filter_string("market", "start_date_max", v)?);
746    }
747
748    if let Some(v) = map.get("end_date_min") {
749        params.end_date_min = Some(parse_gamma_filter_string("market", "end_date_min", v)?);
750    }
751
752    if let Some(v) = map.get("end_date_max") {
753        params.end_date_max = Some(parse_gamma_filter_string("market", "end_date_max", v)?);
754    }
755
756    if let Some(v) = map.get("related_tags") {
757        params.related_tags = Some(parse_gamma_filter_bool("market", "related_tags", v)?);
758    }
759
760    if let Some(v) = map.get("tag_match") {
761        params.tag_match = Some(parse_gamma_filter_string("market", "tag_match", v)?);
762    }
763
764    if let Some(v) = map.get("decimalized") {
765        params.decimalized = Some(parse_gamma_filter_bool("market", "decimalized", v)?);
766    }
767
768    if let Some(v) = map.get("cyom") {
769        params.cyom = Some(parse_gamma_filter_bool("market", "cyom", v)?);
770    }
771
772    if let Some(v) = map.get("rfq_enabled") {
773        params.rfq_enabled = Some(parse_gamma_filter_bool("market", "rfq_enabled", v)?);
774    }
775
776    if let Some(v) = map.get("uma_resolution_status") {
777        params.uma_resolution_status = Some(parse_gamma_filter_string(
778            "market",
779            "uma_resolution_status",
780            v,
781        )?);
782    }
783
784    if let Some(v) = map.get("game_id") {
785        params.game_id = Some(parse_gamma_filter_string("market", "game_id", v)?);
786    }
787
788    if let Some(v) = map.get("sports_market_types") {
789        params.sports_market_types =
790            Some(parse_gamma_filter_list("market", "sports_market_types", v)?);
791    }
792
793    if let Some(v) = map.get("include_tag") {
794        params.include_tag = Some(parse_gamma_filter_bool("market", "include_tag", v)?);
795    }
796
797    if let Some(v) = map.get("locale") {
798        params.locale = Some(parse_gamma_filter_string("market", "locale", v)?);
799    }
800
801    if let Some(v) = map.get("max_markets") {
802        params.max_markets = Some(parse_gamma_filter_u32("market", "max_markets", v)?);
803    }
804
805    params.validate_keyset().map_err(|e| anyhow::anyhow!(e))?;
806    Ok(params)
807}
808
809/// Builds validated event keyset parameters from string key/value filters.
810///
811/// # Errors
812///
813/// Returns an error for unknown keys, malformed values, or invalid filter combinations.
814pub fn build_gamma_event_params_from_hashmap(
815    map: &HashMap<String, String>,
816) -> anyhow::Result<GetGammaEventsParams> {
817    for key in map.keys() {
818        match key.as_str() {
819            "is_active" | "active" | "closed" | "archived" | "id" | "slug" | "live"
820            | "featured" | "cyom" | "title_search" | "liquidity_min" | "liquidity_max"
821            | "volume_min" | "volume_max" | "start_date_min" | "start_date_max"
822            | "end_date_min" | "end_date_max" | "start_time_min" | "start_time_max" | "tag_id"
823            | "tag_slug" | "exclude_tag_id" | "related_tags" | "tag_match" | "series_id"
824            | "game_id" | "event_date" | "event_week" | "featured_order" | "recurrence"
825            | "created_by" | "parent_event_id" | "include_children" | "partner_slug"
826            | "include_chat" | "include_template" | "include_best_lines" | "locale" | "order"
827            | "ascending" | "limit" | "offset" | "max_events" => {}
828            _ => anyhow::bail!("Unknown Gamma event filter key '{key}'"),
829        }
830    }
831
832    let mut params = GetGammaEventsParams::default();
833
834    if map
835        .get("is_active")
836        .map(|value| parse_gamma_filter_bool("event", "is_active", value))
837        .transpose()?
838        .unwrap_or(false)
839    {
840        params.active = Some(true);
841        params.archived = Some(false);
842        params.closed = Some(false);
843    }
844
845    macro_rules! set_bool {
846        ($field:ident) => {
847            if let Some(value) = map.get(stringify!($field)) {
848                params.$field = Some(parse_gamma_filter_bool("event", stringify!($field), value)?);
849            }
850        };
851    }
852    macro_rules! set_string {
853        ($field:ident) => {
854            if let Some(value) = map.get(stringify!($field)) {
855                params.$field = Some(parse_gamma_filter_string(
856                    "event",
857                    stringify!($field),
858                    value,
859                )?);
860            }
861        };
862    }
863    macro_rules! set_decimal {
864        ($field:ident) => {
865            if let Some(value) = map.get(stringify!($field)) {
866                params.$field = Some(parse_gamma_filter_decimal(
867                    "event",
868                    stringify!($field),
869                    value,
870                )?);
871            }
872        };
873    }
874    macro_rules! set_u32 {
875        ($field:ident) => {
876            if let Some(value) = map.get(stringify!($field)) {
877                params.$field = Some(parse_gamma_filter_u32("event", stringify!($field), value)?);
878            }
879        };
880    }
881    macro_rules! set_u64 {
882        ($field:ident) => {
883            if let Some(value) = map.get(stringify!($field)) {
884                params.$field = Some(parse_gamma_filter_u64("event", stringify!($field), value)?);
885            }
886        };
887    }
888    macro_rules! set_strings {
889        ($field:ident) => {
890            if let Some(value) = map.get(stringify!($field)) {
891                params.$field = Some(parse_gamma_filter_list("event", stringify!($field), value)?);
892            }
893        };
894    }
895    macro_rules! set_u64s {
896        ($field:ident) => {
897            if let Some(value) = map.get(stringify!($field)) {
898                params.$field = Some(parse_gamma_numeric_filter_list(
899                    "event",
900                    stringify!($field),
901                    value,
902                )?);
903            }
904        };
905    }
906
907    set_bool!(active);
908    set_bool!(closed);
909    set_bool!(archived);
910    set_bool!(live);
911    set_bool!(featured);
912    set_bool!(cyom);
913    set_bool!(related_tags);
914    set_bool!(featured_order);
915    set_bool!(include_children);
916    set_bool!(include_chat);
917    set_bool!(include_template);
918    set_bool!(include_best_lines);
919    set_bool!(ascending);
920    set_strings!(slug);
921    set_strings!(created_by);
922    set_u64s!(id);
923    set_u64s!(tag_id);
924    set_u64s!(exclude_tag_id);
925    set_u64s!(series_id);
926    set_u64s!(game_id);
927    set_string!(title_search);
928    set_string!(start_date_min);
929    set_string!(start_date_max);
930    set_string!(end_date_min);
931    set_string!(end_date_max);
932    set_string!(start_time_min);
933    set_string!(start_time_max);
934    set_string!(tag_slug);
935    set_string!(tag_match);
936    set_string!(event_date);
937    set_string!(recurrence);
938    set_string!(partner_slug);
939    set_string!(locale);
940    set_string!(order);
941    set_decimal!(liquidity_min);
942    set_decimal!(liquidity_max);
943    set_decimal!(volume_min);
944    set_decimal!(volume_max);
945    set_u32!(event_week);
946    set_u32!(limit);
947    set_u32!(offset);
948    set_u32!(max_events);
949    set_u64!(parent_event_id);
950
951    params.validate_keyset().map_err(anyhow::Error::msg)?;
952    Ok(params)
953}
954
955fn parse_gamma_filter_bool(scope: &str, key: &str, value: &str) -> anyhow::Result<bool> {
956    if value.eq_ignore_ascii_case("true") {
957        Ok(true)
958    } else if value.eq_ignore_ascii_case("false") {
959        Ok(false)
960    } else {
961        anyhow::bail!("Gamma {scope} filter '{key}' must be true or false, was '{value}'")
962    }
963}
964
965fn parse_gamma_filter_u32(scope: &str, key: &str, value: &str) -> anyhow::Result<u32> {
966    value.parse::<u32>().map_err(|e| {
967        anyhow::anyhow!("Gamma {scope} filter '{key}' must be an unsigned integer: {e}")
968    })
969}
970
971fn parse_gamma_filter_u64(scope: &str, key: &str, value: &str) -> anyhow::Result<u64> {
972    value.parse::<u64>().map_err(|e| {
973        anyhow::anyhow!("Gamma {scope} filter '{key}' must be an unsigned integer: {e}")
974    })
975}
976
977fn parse_gamma_filter_decimal(scope: &str, key: &str, value: &str) -> anyhow::Result<Decimal> {
978    value
979        .parse::<Decimal>()
980        .map_err(|e| anyhow::anyhow!("Gamma {scope} filter '{key}' must be a decimal number: {e}"))
981}
982
983fn parse_gamma_filter_list(scope: &str, key: &str, value: &str) -> anyhow::Result<Vec<String>> {
984    let values = value
985        .split(',')
986        .map(str::trim)
987        .map(str::to_string)
988        .collect::<Vec<_>>();
989
990    if values.is_empty() || values.iter().any(String::is_empty) {
991        anyhow::bail!("Gamma {scope} filter '{key}' must contain non-empty comma-separated values")
992    }
993    Ok(values)
994}
995
996fn parse_gamma_numeric_filter_list(
997    scope: &str,
998    key: &str,
999    value: &str,
1000) -> anyhow::Result<Vec<u64>> {
1001    parse_gamma_filter_list(scope, key, value)?
1002        .into_iter()
1003        .map(|item| {
1004            item.parse::<u64>().map_err(|e| {
1005                anyhow::anyhow!(
1006                    "Gamma {scope} filter '{key}' values must be unsigned integers: {e}"
1007                )
1008            })
1009        })
1010        .collect()
1011}
1012
1013fn parse_gamma_filter_string(scope: &str, key: &str, value: &str) -> anyhow::Result<String> {
1014    if value.trim().is_empty() {
1015        anyhow::bail!("Gamma {scope} filter '{key}' cannot be empty")
1016    }
1017    Ok(value.to_string())
1018}
1019
1020/// Resolves a tag slug to a tag ID by querying the Gamma tags endpoint.
1021pub async fn resolve_tag_slug(
1022    client: &PolymarketGammaHttpClient,
1023    slug: &str,
1024) -> anyhow::Result<u64> {
1025    let tags = client.request_tags().await?;
1026    let tag_id = tags
1027        .iter()
1028        .find(|t| t.slug.as_deref() == Some(slug))
1029        .map(|t| t.id.as_str())
1030        .ok_or_else(|| anyhow::anyhow!("Tag slug '{slug}' not found"))?;
1031    tag_id
1032        .parse::<u64>()
1033        .map_err(|e| anyhow::anyhow!("Tag slug '{slug}' returned invalid ID '{tag_id}': {e}"))
1034}
1035
1036#[async_trait(?Send)]
1037impl InstrumentProvider for PolymarketInstrumentProvider {
1038    fn store(&self) -> &InstrumentStore {
1039        &self.store
1040    }
1041
1042    fn store_mut(&mut self) -> &mut InstrumentStore {
1043        &mut self.store
1044    }
1045
1046    async fn load_all(&mut self, filters: Option<&HashMap<String, String>>) -> anyhow::Result<()> {
1047        let instruments = self.fetch_bulk_instruments(filters).await?;
1048        self.replace_instruments(instruments);
1049        self.store.set_initialized();
1050
1051        Ok(())
1052    }
1053
1054    async fn load_ids(
1055        &mut self,
1056        instrument_ids: &[InstrumentId],
1057        filters: Option<&HashMap<String, String>>,
1058    ) -> anyhow::Result<()> {
1059        let missing: Vec<_> = instrument_ids
1060            .iter()
1061            .filter(|id| !self.store.contains(id))
1062            .collect();
1063
1064        if missing.is_empty() {
1065            return Ok(());
1066        }
1067
1068        // Extract unique condition IDs from instrument symbols
1069        // Symbol format: "{condition_id}-{token_id}"
1070        let mut condition_ids: Vec<String> = missing
1071            .iter()
1072            .filter_map(|id| extract_condition_id(id).ok())
1073            .collect();
1074        condition_ids.sort();
1075        condition_ids.dedup();
1076
1077        if condition_ids.is_empty() {
1078            return Ok(());
1079        }
1080
1081        let base_params = filters
1082            .map(build_gamma_params_from_hashmap)
1083            .transpose()?
1084            .unwrap_or_default();
1085
1086        for chunk in condition_ids.chunks(GAMMA_CONDITION_IDS_BATCH_SIZE) {
1087            let params = GetGammaMarketsParams {
1088                condition_ids: Some(chunk.to_vec()),
1089                ..base_params.clone()
1090            };
1091            let instruments = self
1092                .http_client
1093                .request_instruments_by_params(params)
1094                .await?;
1095            self.add_instruments(instruments);
1096        }
1097
1098        Ok(())
1099    }
1100
1101    async fn load(
1102        &mut self,
1103        instrument_id: &InstrumentId,
1104        filters: Option<&HashMap<String, String>>,
1105    ) -> anyhow::Result<()> {
1106        if self.store.contains(instrument_id) {
1107            return Ok(());
1108        }
1109
1110        // Try direct fetch via condition_id extracted from symbol
1111        if let Ok(cid) = extract_condition_id(instrument_id) {
1112            let params = GetGammaMarketsParams {
1113                condition_ids: Some(vec![cid]),
1114                ..Default::default()
1115            };
1116
1117            if let Ok(instruments) = self.http_client.request_instruments_by_params(params).await {
1118                self.add_instruments(instruments);
1119
1120                if self.store.contains(instrument_id) {
1121                    return Ok(());
1122                }
1123            }
1124        }
1125
1126        // Fallback: full load_all if not initialized. A provider with an explicit
1127        // scope is excluded: `load_all` would broaden it into the full-universe
1128        // fetch that scoping exists to avoid, and would also clear the partially
1129        // loaded store and mark it initialized, making a later `initialize(false)`
1130        // skip the scopes it still owes after a failed bootstrap.
1131        if !self.store.is_initialized() && !self.config.has_explicit_scope() {
1132            self.load_all(filters).await?;
1133        }
1134
1135        if self.store.contains(instrument_id) {
1136            Ok(())
1137        } else {
1138            anyhow::bail!("Instrument {instrument_id} not found on Polymarket")
1139        }
1140    }
1141}