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

nautilus_polymarket/data/
mod.rs
