1use 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
41pub 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 #[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 #[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 #[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 pub fn add_filter(&mut self, filter: Arc<dyn InstrumentFilter>) {
118 self.filters.push(filter);
119 }
120
121 pub fn clear_filters(&mut self) {
123 self.filters.clear();
124 }
125
126 #[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 #[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 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 #[must_use]
175 pub fn filters(&self) -> Vec<Arc<dyn InstrumentFilter>> {
176 self.filters.clone()
177 }
178
179 #[must_use]
181 pub fn http_client(&self) -> &PolymarketGammaHttpClient {
182 &self.http_client
183 }
184
185 #[must_use]
187 pub fn config(&self) -> &PolymarketInstrumentProviderConfig {
188 &self.config
189 }
190
191 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 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 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 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 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 let load_ids = self.config.load_ids.clone().unwrap_or_default();
271 self.load_ids(&load_ids, None).await?;
272 }
273
274 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 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 async fn load_filtered(&self) -> anyhow::Result<Vec<InstrumentAny>> {
354 fetch_instruments(&self.http_client, &self.filters).await
355 }
356
357 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 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
390pub 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
452pub 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 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 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 let mut seen = AHashSet::new();
555 instruments.retain(|inst| seen.insert(inst.id()));
556 Ok(instruments)
557}
558
559fn has_bootstrap_scope(
563 config: &PolymarketInstrumentProviderConfig,
564 filters: &[Arc<dyn InstrumentFilter>],
565) -> bool {
566 config.should_load_all() || !filters.is_empty()
567}
568
569fn 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
580pub 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
595pub(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
609pub 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
823pub 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
1033pub 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 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 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 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}