nautilus_polymarket/data/
mod.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//! Live market data client implementation for the Polymarket adapter.
17
18mod auto_load;
19mod dispatch;
20mod effective_deltas;
21mod instruments;
22mod lifecycle;
23mod requests;
24mod runtime;
25mod subscriptions;
26
27#[cfg(test)]
28mod test_logging;
29
30use std::{
31    future::Future,
32    sync::{
33        Arc,
34        atomic::{AtomicBool, Ordering},
35    },
36    time::Duration,
37};
38
39use dashmap::DashMap;
40use nautilus_common::{
41    cache::InstrumentLookupError,
42    clients::{DataClient, execute_instrument_subscriptions},
43    live::{dst::time::Instant, runner::get_data_event_sender, sender::EventSender},
44    messages::{
45        DataEvent,
46        data::{
47            RequestBookSnapshot, RequestCustomData, RequestInstrument, RequestInstruments,
48            RequestTrades, SubscribeBookDeltas, SubscribeBookDepth, SubscribeCustomData,
49            SubscribeInstrument, SubscribeInstrumentClose, SubscribeInstrumentStatus,
50            SubscribeInstruments, SubscribeQuotes, SubscribeTrades, UnsubscribeBookDeltas,
51            UnsubscribeCustomData, UnsubscribeInstrument, UnsubscribeInstrumentClose,
52            UnsubscribeInstrumentStatus, UnsubscribeQuotes, UnsubscribeTrades,
53        },
54    },
55    msgbus::TypedHandler,
56};
57use nautilus_core::{
58    AtomicMap,
59    time::{AtomicTime, get_atomic_clock_realtime},
60};
61use nautilus_live::{
62    SocketControl, SocketControlFactory,
63    task::{TaskGroup, TaskSpawner},
64};
65use nautilus_model::{
66    enums::BookType,
67    events::PositionEvent,
68    identifiers::{ClientId, InstrumentId, Venue},
69    instruments::InstrumentAny,
70};
71use nautilus_network::websocket::proxy::ProxyUrl;
72use parking_lot::Mutex;
73use tokio_util::sync::CancellationToken;
74use ustr::Ustr;
75
76pub(crate) use self::subscriptions::DataSubscriptions;
77use self::{
78    instruments::InstrumentUpdateState,
79    requests::{
80        request_book_snapshot, request_data, request_instrument, request_instruments,
81        request_trades,
82    },
83    runtime::is_instrument_expired_and_not_reported_open,
84    subscriptions::{SubscriptionKind, resolve_token_id_from},
85};
86use crate::{
87    book::{BookSequenceOutcome, recovery::start_recovery},
88    common::{consts::POLYMARKET_VENUE, credential::Credential},
89    config::PolymarketDataClientConfig,
90    filters::InstrumentFilter,
91    http::{
92        clob::PolymarketClobPublicClient, data_api::PolymarketDataApiHttpClient,
93        gamma::PolymarketGammaHttpClient,
94    },
95    polybolt::{PolymarketPolyBoltFeed, is_supported_polybolt_data_type},
96    providers::PolymarketInstrumentProvider,
97    resolve::{PendingResolution, ResolveContext, ResolveWatchlist, StrictResolvedMarket},
98    websocket::{POLYBOLT_STREAMS_ENDPOINT, pool::PolymarketMarketConnectionPool},
99};
100
101const NEW_MARKET_FETCH_MAX_CONCURRENCY_CAP: usize = 64;
102pub(super) const NEW_MARKET_EMPTY_RECHECK_MAX_ATTEMPTS: usize = 1;
103pub(super) const NEW_MARKET_EMPTY_RECHECK_DELAY: Duration = Duration::from_millis(500);
104
105/// Polymarket data client for live market data streaming.
106///
107/// Integrates with the Nautilus DataEngine to provide:
108/// - Real-time order book snapshots and deltas via WebSocket
109/// - Quote ticks synthesized from book data
110/// - Trade ticks from last trade price messages
111/// - Automatic instrument discovery from the Gamma API
112#[derive(Debug)]
113pub struct PolymarketDataClient {
114    clock: &'static AtomicTime,
115    client_id: ClientId,
116    config: PolymarketDataClientConfig,
117    provider: PolymarketInstrumentProvider,
118    clob_public_client: PolymarketClobPublicClient,
119    data_api_client: PolymarketDataApiHttpClient,
120    ws_client: PolymarketMarketConnectionPool,
121    is_connected: AtomicBool,
122    cancellation_token: CancellationToken,
123    tasks: TaskGroup,
124    data_sender: EventSender<DataEvent>,
125    instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
126    subscriptions: DataSubscriptions,
127    instrument_update_state: Arc<Mutex<InstrumentUpdateState>>,
128    resolve_poll_watchlist: Arc<ResolveWatchlist>,
129    resolve_watch_apply_mutex: Arc<Mutex<()>>,
130    pending_resolutions: Arc<DashMap<String, PendingResolution>>,
131    deferred_resolutions: Arc<AtomicMap<InstrumentId, StrictResolvedMarket>>,
132    new_market_inflight_keys: Arc<DashMap<String, ()>>,
133    new_market_fetch_semaphore: Arc<tokio::sync::Semaphore>,
134    auto_load_scheduled: Arc<AtomicBool>,
135    position_event_handler: Option<TypedHandler<PositionEvent>>,
136    polybolt_feed: PolymarketPolyBoltFeed,
137    polybolt_socket_control: Option<SocketControl>,
138    polybolt_credential: Option<Credential>,
139    proxy_url: Option<ProxyUrl>,
140    reset_pending: bool,
141    shutdown_errors: Vec<String>,
142}
143
144impl PolymarketDataClient {
145    /// Creates a new [`PolymarketDataClient`].
146    ///
147    /// # Errors
148    ///
149    /// Returns an error if the configuration is invalid.
150    pub fn new(
151        client_id: ClientId,
152        config: PolymarketDataClientConfig,
153        gamma_client: PolymarketGammaHttpClient,
154        clob_public_client: PolymarketClobPublicClient,
155        data_api_client: PolymarketDataApiHttpClient,
156        ws_client: PolymarketMarketConnectionPool,
157    ) -> anyhow::Result<Self> {
158        Self::new_with_proxy(
159            client_id,
160            config,
161            gamma_client,
162            clob_public_client,
163            data_api_client,
164            ws_client,
165            None,
166        )
167    }
168
169    /// Creates a new data client with an optional validated proxy URL.
170    ///
171    /// # Errors
172    ///
173    /// Returns an error if the configuration is invalid.
174    pub fn new_with_proxy(
175        client_id: ClientId,
176        mut config: PolymarketDataClientConfig,
177        gamma_client: PolymarketGammaHttpClient,
178        clob_public_client: PolymarketClobPublicClient,
179        data_api_client: PolymarketDataApiHttpClient,
180        ws_client: PolymarketMarketConnectionPool,
181        proxy_url: Option<ProxyUrl>,
182    ) -> anyhow::Result<Self> {
183        config.validate()?;
184        let polybolt_credential = config.polybolt_credential()?;
185
186        let clock = get_atomic_clock_realtime();
187        let data_sender = get_data_event_sender();
188        let socket_factory = SocketControlFactory::new(client_id, Some(*POLYMARKET_VENUE));
189        let ws_client = ws_client
190            .with_connect_timeout(config.ws_timeout_secs)
191            .with_socket_factory(socket_factory.clone());
192        let polybolt_socket_control = Some(socket_factory.control(POLYBOLT_STREAMS_ENDPOINT));
193        let provider =
194            PolymarketInstrumentProvider::new(gamma_client, config.instrument_config.clone())?;
195        let configured_fetch_max_concurrency = config.new_market_fetch_max_concurrency;
196        let fetch_max_concurrency =
197            clamp_new_market_fetch_max_concurrency(configured_fetch_max_concurrency);
198
199        if configured_fetch_max_concurrency == 0 {
200            log::warn!(
201                "PolymarketDataClientConfig.new_market_fetch_max_concurrency=0 is invalid, clamping to 1"
202            );
203        } else if configured_fetch_max_concurrency > NEW_MARKET_FETCH_MAX_CONCURRENCY_CAP {
204            log::warn!(
205                "PolymarketDataClientConfig.new_market_fetch_max_concurrency={configured_fetch_max_concurrency} exceeds cap {NEW_MARKET_FETCH_MAX_CONCURRENCY_CAP}, clamping",
206            );
207        }
208        config.new_market_fetch_max_concurrency = fetch_max_concurrency;
209
210        let polybolt_url = config.reference_prices_url();
211        let polybolt_transport_backend = config.transport_backend;
212        let ws_timeout_secs = config.ws_timeout_secs;
213        let polybolt_data_sender = data_sender.clone();
214
215        let tasks = TaskGroup::new();
216        let cancellation_token = tasks.cancellation_token();
217
218        let instruments = Arc::new(AtomicMap::new());
219        let resolve_poll_watchlist = Arc::new(ResolveWatchlist::default());
220        let resolve_watch_apply_mutex = Arc::new(Mutex::new(()));
221        let deferred_resolutions = Arc::new(AtomicMap::new());
222
223        let subscriptions = DataSubscriptions::new(
224            instruments.clone(),
225            resolve_poll_watchlist.clone(),
226            deferred_resolutions.clone(),
227            resolve_watch_apply_mutex.clone(),
228            ws_client.handle(),
229            config.subscribe_new_markets,
230        );
231
232        Ok(Self {
233            clock,
234            client_id,
235            config,
236            provider,
237            clob_public_client,
238            data_api_client,
239            ws_client,
240            is_connected: AtomicBool::new(false),
241            cancellation_token,
242            tasks,
243            data_sender,
244            instruments,
245            subscriptions,
246            instrument_update_state: Arc::new(Mutex::new(InstrumentUpdateState::default())),
247            resolve_poll_watchlist,
248            resolve_watch_apply_mutex,
249            pending_resolutions: Arc::new(DashMap::new()),
250            deferred_resolutions,
251            new_market_inflight_keys: Arc::new(DashMap::new()),
252            new_market_fetch_semaphore: Arc::new(tokio::sync::Semaphore::new(
253                fetch_max_concurrency,
254            )),
255            auto_load_scheduled: Arc::new(AtomicBool::new(false)),
256            position_event_handler: None,
257            polybolt_feed: PolymarketPolyBoltFeed::new_with_proxy_and_socket_control(
258                polybolt_url,
259                polybolt_transport_backend,
260                clock,
261                polybolt_data_sender,
262                proxy_url.clone(),
263                polybolt_socket_control.clone(),
264                polybolt_credential.clone(),
265                ws_timeout_secs,
266            ),
267            polybolt_socket_control,
268            polybolt_credential,
269            proxy_url,
270            reset_pending: false,
271            shutdown_errors: Vec::new(),
272        })
273    }
274
275    /// Returns a reference to the client configuration.
276    #[must_use]
277    pub fn config(&self) -> &PolymarketDataClientConfig {
278        &self.config
279    }
280
281    /// Returns the venue for this data client.
282    #[must_use]
283    pub fn venue(&self) -> Venue {
284        *POLYMARKET_VENUE
285    }
286
287    /// Returns a reference to the instrument provider.
288    #[must_use]
289    pub fn provider(&self) -> &PolymarketInstrumentProvider {
290        &self.provider
291    }
292
293    #[cfg(test)]
294    pub(crate) fn clob_public_client(&self) -> &PolymarketClobPublicClient {
295        &self.clob_public_client
296    }
297
298    #[cfg(test)]
299    pub(crate) fn data_api_client(&self) -> &PolymarketDataApiHttpClient {
300        &self.data_api_client
301    }
302
303    #[cfg(test)]
304    pub(crate) fn ws_client(&self) -> &PolymarketMarketConnectionPool {
305        &self.ws_client
306    }
307
308    #[cfg(test)]
309    pub(crate) fn polybolt_feed(&self) -> &PolymarketPolyBoltFeed {
310        &self.polybolt_feed
311    }
312
313    /// Adds an instrument filter on the underlying provider.
314    pub fn add_instrument_filter(&mut self, filter: Arc<dyn InstrumentFilter>) {
315        self.provider.add_filter(filter);
316    }
317
318    /// Returns `true` when the client is connected.
319    #[must_use]
320    pub fn is_connected(&self) -> bool {
321        self.is_connected.load(Ordering::Relaxed)
322    }
323
324    fn resolve_token_id(&self, instrument_id: InstrumentId) -> anyhow::Result<String> {
325        resolve_token_id_from(&self.instruments, instrument_id)
326    }
327
328    fn resolution_context(&self) -> ResolveContext {
329        ResolveContext {
330            subscriptions: self.subscriptions.clone(),
331            clock: self.clock,
332            data_sender: self.data_sender.clone(),
333            instruments: self.instruments.clone(),
334            watchlist: self.resolve_poll_watchlist.clone(),
335            apply_mutex: self.resolve_watch_apply_mutex.clone(),
336            pending_resolutions: self.pending_resolutions.clone(),
337            deferred_resolutions: self.deferred_resolutions.clone(),
338            cancellation_token: self.cancellation_token.clone(),
339        }
340    }
341
342    fn ensure_live_subscription_allowed(&self, instrument_id: InstrumentId) -> anyhow::Result<()> {
343        let now_ns = self.clock.get_time_ns();
344        let loaded = self.instruments.load();
345        let Some(instrument) = loaded.get(&instrument_id) else {
346            return Ok(());
347        };
348
349        if is_instrument_expired_and_not_reported_open(instrument, now_ns) {
350            anyhow::bail!(
351                "Instrument {instrument_id} is expired and no longer available for live subscription"
352            );
353        }
354
355        Ok(())
356    }
357
358    fn ensure_market_data_request_allowed(
359        &self,
360        instrument_id: InstrumentId,
361    ) -> anyhow::Result<InstrumentAny> {
362        let loaded = self.instruments.load();
363        let instrument = loaded
364            .get(&instrument_id)
365            .ok_or_else(|| InstrumentLookupError::not_found(instrument_id))?
366            .clone();
367
368        if is_instrument_expired_and_not_reported_open(&instrument, self.clock.get_time_ns()) {
369            anyhow::bail!(
370                "Instrument {instrument_id} is expired and no longer available for market data requests"
371            );
372        }
373
374        Ok(instrument)
375    }
376
377    fn subscribe_resolution(
378        &self,
379        instrument_id: InstrumentId,
380        kind: SubscriptionKind,
381    ) -> anyhow::Result<()> {
382        // Expiration bounds recovery; it does not disqualify metadata needed for settlement
383        let cached = self.instruments.load().contains_key(&instrument_id);
384
385        if !cached && !self.config.auto_load_missing_instruments {
386            anyhow::bail!(
387                "Instrument {instrument_id} not found, and `auto_load_missing_instruments` is disabled"
388            );
389        }
390
391        if !self
392            .subscriptions
393            .add_resolution_subscription_intent(instrument_id, kind)
394        {
395            return Ok(());
396        }
397
398        if !cached {
399            self.queue_pending_load(instrument_id);
400            return Ok(());
401        }
402
403        self.sync_ws_subscription(instrument_id);
404        Ok(())
405    }
406
407    // Spawns an async task that reconciles the WS subscription for
408    // `instrument_id`. The task holds `ws_sub_mutex` across the wire send so
409    // concurrent subscribe/unsubscribe calls deliver commands to the WS handler
410    // in a consistent order with the final `active_*_subs` state.
411    fn sync_ws_subscription(&self, instrument_id: InstrumentId) {
412        if let Ok(token_id) = self.resolve_token_id(instrument_id) {
413            self.sync_ws_subscription_for_token(instrument_id, token_id);
414        }
415    }
416
417    fn sync_ws_subscription_for_token(&self, instrument_id: InstrumentId, token_id_str: String) {
418        let subscriptions = self.subscriptions.clone();
419        let future = async move {
420            subscriptions.reconcile(instrument_id, token_id_str).await;
421        };
422
423        if let Err(e) = self.tasks.spawn(future) {
424            log::debug!("Skipping Polymarket data task after shutdown began: {e}");
425        }
426    }
427
428    // The venue ignores a duplicate subscribe, so a token another subscription already holds
429    // sends no `book` for this one; a recovery cycle requests it instead.
430    fn sync_book_subscription(&self, instrument_id: InstrumentId) {
431        let Ok(token_id) = self.resolve_token_id(instrument_id) else {
432            return;
433        };
434
435        let spawner = match self.tasks.spawner() {
436            Ok(spawner) => spawner,
437            Err(e) => {
438                log::debug!("Skipping Polymarket data task after shutdown began: {e}");
439                return;
440            }
441        };
442
443        // Read on arrival so a subscription still queued behind the subscription lock does not
444        // look open; its `book` reaches this one.
445        let was_open = self
446            .subscriptions
447            .ws_open_tokens
448            .contains(&Ustr::from(token_id.as_str()));
449        let resolve_ctx = self.resolution_context();
450        let book_sync = self.subscriptions.book_sync.clone();
451        let snapshot_timeout = Duration::from_secs(self.config.book_snapshot_timeout_secs);
452
453        // The sync waits out any subscribe write in flight, so recovery cycles an owned token
454        let future = async move {
455            resolve_ctx
456                .sync_ws_subscription(instrument_id, token_id.clone())
457                .await;
458
459            if was_open
460                && book_sync.request_snapshot_if_subscribed(
461                    &resolve_ctx.subscriptions.active_delta_subs,
462                    instrument_id,
463                    Instant::now(),
464                ) == BookSequenceOutcome::Recover
465            {
466                log::debug!("Requesting a book snapshot for {instrument_id} on an open token");
467                start_recovery(
468                    instrument_id,
469                    Some(token_id),
470                    &resolve_ctx.subscriptions.active_delta_subs,
471                    &book_sync,
472                    &resolve_ctx.subscriptions.ws_client,
473                    snapshot_timeout,
474                    &spawner,
475                );
476            } else if !was_open {
477                resolve_ctx.subscriptions.monitor_book_snapshot(
478                    instrument_id,
479                    snapshot_timeout,
480                    &spawner,
481                );
482            }
483        };
484
485        if let Err(e) = self.tasks.spawn(future) {
486            log::debug!("Skipping Polymarket data task after shutdown began: {e}");
487        }
488    }
489}
490
491#[async_trait::async_trait(?Send)]
492impl DataClient for PolymarketDataClient {
493    fn client_id(&self) -> ClientId {
494        self.client_id
495    }
496
497    fn venue(&self) -> Option<Venue> {
498        Some(*POLYMARKET_VENUE)
499    }
500
501    fn start(&mut self) -> anyhow::Result<()> {
502        self.start_client();
503        Ok(())
504    }
505
506    fn stop(&mut self) -> anyhow::Result<()> {
507        self.stop_client();
508        Ok(())
509    }
510
511    fn reset(&mut self) -> anyhow::Result<()> {
512        self.reset_client();
513        Ok(())
514    }
515
516    fn dispose(&mut self) -> anyhow::Result<()> {
517        self.stop()
518    }
519
520    async fn connect(&mut self) -> anyhow::Result<()> {
521        self.connect_client().await
522    }
523
524    async fn disconnect(&mut self) -> anyhow::Result<()> {
525        self.disconnect_client().await
526    }
527
528    fn is_connected(&self) -> bool {
529        self.is_connected.load(Ordering::Relaxed)
530    }
531
532    fn is_disconnected(&self) -> bool {
533        !self.is_connected()
534    }
535
536    fn request_data(&self, request: RequestCustomData) -> anyhow::Result<()> {
537        request_data(self, request);
538        Ok(())
539    }
540
541    fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
542        request_instruments(self, request);
543        Ok(())
544    }
545
546    fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
547        request_instrument(self, request)
548    }
549
550    fn request_book_snapshot(&self, request: RequestBookSnapshot) -> anyhow::Result<()> {
551        request_book_snapshot(self, request)
552    }
553
554    fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
555        request_trades(self, request)
556    }
557
558    fn subscribe_instruments(&mut self, _cmd: SubscribeInstruments) -> anyhow::Result<()> {
559        log::debug!("subscribe_instruments: subscribed individually via data subscription methods");
560        Ok(())
561    }
562
563    fn subscribe_instrument(&mut self, cmd: SubscribeInstrument) -> anyhow::Result<()> {
564        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
565            log::debug!(
566                "Subscribed to instrument definition updates for {instrument_id}; shared instrument sources remain active"
567            );
568
569            Ok(())
570        })
571    }
572
573    fn unsubscribe_instrument(&mut self, cmd: &UnsubscribeInstrument) -> anyhow::Result<()> {
574        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
575            log::debug!(
576                "Unsubscribed from instrument {instrument_id}; shared instrument sources remain active"
577            );
578            Ok(())
579        })
580    }
581
582    fn subscribe(&mut self, cmd: SubscribeCustomData) -> anyhow::Result<()> {
583        if !is_supported_polybolt_data_type(&cmd.data_type) {
584            log::debug!(
585                "Ignoring unsupported Polymarket custom data subscription: {}",
586                cmd.data_type
587            );
588            return Ok(());
589        }
590
591        log::debug!(
592            "Tracking Polymarket PolyBolt custom data subscription: {}",
593            cmd.data_type
594        );
595        let changed = self.polybolt_feed.track_subscribe(cmd.data_type)?;
596        if !changed {
597            return Ok(());
598        }
599
600        if !self.is_connected() {
601            return Ok(());
602        }
603
604        self.polybolt_feed
605            .request_reconcile(crate::polybolt::ReconcileReason::DesiredChanged);
606
607        Ok(())
608    }
609
610    fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
611        if cmd.book_type != BookType::L2_MBP {
612            anyhow::bail!(
613                "Polymarket only supports L2_MBP order book deltas, received {:?}",
614                cmd.book_type
615            );
616        }
617
618        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
619            self.ensure_live_subscription_allowed(instrument_id)?;
620            let cached = self.instruments.load().contains_key(&instrument_id);
621
622            if !cached && !self.config.auto_load_missing_instruments {
623                anyhow::bail!(
624                    "Instrument {instrument_id} not found, and `auto_load_missing_instruments` is disabled"
625                );
626            }
627
628            // Mark intent before routing so unsubscribe can race-safely clear it.
629            if !self
630                .subscriptions
631                .add_delta_subscription_intent(instrument_id, self.config.compute_effective_deltas)
632            {
633                return Ok(());
634            }
635
636            if !cached {
637                self.queue_pending_load(instrument_id);
638                return Ok(());
639            }
640
641            self.sync_book_subscription(instrument_id);
642            Ok(())
643        })
644    }
645
646    fn subscribe_book_depth(&mut self, _cmd: SubscribeBookDepth) -> anyhow::Result<()> {
647        anyhow::bail!(
648            "Polymarket does not support OrderBookDepth subscriptions; use managed L2_MBP order book deltas"
649        )
650    }
651
652    fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
653        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
654            self.ensure_live_subscription_allowed(instrument_id)?;
655            let cached = self.instruments.load().contains_key(&instrument_id);
656
657            if !cached && !self.config.auto_load_missing_instruments {
658                anyhow::bail!(
659                    "Instrument {instrument_id} not found, and `auto_load_missing_instruments` is disabled"
660                );
661            }
662
663            if !self
664                .subscriptions
665                .add_live_subscription_intent(instrument_id, SubscriptionKind::Quotes)
666            {
667                return Ok(());
668            }
669
670            if !cached {
671                self.queue_pending_load(instrument_id);
672                return Ok(());
673            }
674
675            self.sync_ws_subscription(instrument_id);
676            Ok(())
677        })
678    }
679
680    fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
681        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
682            self.ensure_live_subscription_allowed(instrument_id)?;
683            let cached = self.instruments.load().contains_key(&instrument_id);
684
685            if !cached && !self.config.auto_load_missing_instruments {
686                anyhow::bail!(
687                    "Instrument {instrument_id} not found, and `auto_load_missing_instruments` is disabled"
688                );
689            }
690
691            if !self
692                .subscriptions
693                .add_live_subscription_intent(instrument_id, SubscriptionKind::Trades)
694            {
695                return Ok(());
696            }
697
698            if !cached {
699                self.queue_pending_load(instrument_id);
700                return Ok(());
701            }
702
703            self.sync_ws_subscription(instrument_id);
704            Ok(())
705        })
706    }
707
708    fn subscribe_instrument_status(
709        &mut self,
710        cmd: SubscribeInstrumentStatus,
711    ) -> anyhow::Result<()> {
712        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
713            self.subscribe_resolution(instrument_id, SubscriptionKind::Status)
714        })
715    }
716
717    fn subscribe_instrument_close(&mut self, cmd: SubscribeInstrumentClose) -> anyhow::Result<()> {
718        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
719            self.subscribe_resolution(instrument_id, SubscriptionKind::Close)
720        })
721    }
722
723    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
724        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
725            let token_id = self
726                .subscriptions
727                .unsubscribe(instrument_id, SubscriptionKind::Deltas);
728            if let Some(token_id) = token_id {
729                self.sync_ws_subscription_for_token(instrument_id, token_id);
730            }
731
732            Ok(())
733        })
734    }
735
736    fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
737        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
738            let token_id = self
739                .subscriptions
740                .unsubscribe(instrument_id, SubscriptionKind::Quotes);
741            if let Some(token_id) = token_id {
742                self.sync_ws_subscription_for_token(instrument_id, token_id);
743            }
744
745            Ok(())
746        })
747    }
748
749    fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
750        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
751            let token_id = self
752                .subscriptions
753                .unsubscribe(instrument_id, SubscriptionKind::Trades);
754            if let Some(token_id) = token_id {
755                self.sync_ws_subscription_for_token(instrument_id, token_id);
756            }
757
758            Ok(())
759        })
760    }
761
762    fn unsubscribe_instrument_status(
763        &mut self,
764        cmd: &UnsubscribeInstrumentStatus,
765    ) -> anyhow::Result<()> {
766        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
767            let token_id = self
768                .subscriptions
769                .unsubscribe(instrument_id, SubscriptionKind::Status);
770            if let Some(token_id) = token_id {
771                self.sync_ws_subscription_for_token(instrument_id, token_id);
772            }
773
774            Ok(())
775        })
776    }
777
778    fn unsubscribe_instrument_close(
779        &mut self,
780        cmd: &UnsubscribeInstrumentClose,
781    ) -> anyhow::Result<()> {
782        execute_instrument_subscriptions(&cmd.instrument_ids, |instrument_id| {
783            let token_id = self
784                .subscriptions
785                .unsubscribe(instrument_id, SubscriptionKind::Close);
786            if let Some(token_id) = token_id {
787                self.sync_ws_subscription_for_token(instrument_id, token_id);
788            }
789
790            Ok(())
791        })
792    }
793
794    fn unsubscribe(&mut self, cmd: &UnsubscribeCustomData) -> anyhow::Result<()> {
795        if !is_supported_polybolt_data_type(&cmd.data_type) {
796            log::debug!(
797                "Ignoring unsupported Polymarket custom data unsubscription: {}",
798                cmd.data_type
799            );
800            return Ok(());
801        }
802
803        log::debug!(
804            "Tracking Polymarket PolyBolt custom data unsubscription: {}",
805            cmd.data_type
806        );
807        let changed = self.polybolt_feed.track_unsubscribe(&cmd.data_type)?;
808        if !changed {
809            return Ok(());
810        }
811
812        if !self.is_connected() {
813            return Ok(());
814        }
815
816        self.polybolt_feed
817            .request_reconcile(crate::polybolt::ReconcileReason::DesiredChanged);
818
819        Ok(())
820    }
821}
822
823fn clamp_new_market_fetch_max_concurrency(value: usize) -> usize {
824    value.clamp(1, NEW_MARKET_FETCH_MAX_CONCURRENCY_CAP)
825}
826
827pub(super) fn spawn_task<F>(tasks: &TaskSpawner, future: F)
828where
829    F: Future<Output = ()> + Send + 'static,
830{
831    if let Err(e) = tasks.spawn(future) {
832        log::debug!("Skipping Polymarket data task after shutdown began: {e}");
833    }
834}
nautilus_polymarket/data/mod.rs

nautilus_polymarket/data/
mod.rs
