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,
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 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
809pub 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
1020pub 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 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 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 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}