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 ahash::AHashSet;
37use dashmap::DashMap;
38use nautilus_common::{
39    cache::InstrumentLookupError,
40    clients::DataClient,
41    live::runner::get_data_event_sender,
42    messages::{
43        DataEvent,
44        data::{
45            RequestBookSnapshot, RequestCustomData, RequestInstrument, RequestInstruments,
46            RequestTrades, SubscribeBookDeltas, SubscribeBookDepth10, SubscribeCustomData,
47            SubscribeInstrument, SubscribeInstrumentClose, SubscribeInstrumentStatus,
48            SubscribeInstruments, SubscribeQuotes, SubscribeTrades, UnsubscribeBookDeltas,
49            UnsubscribeCustomData, UnsubscribeInstrument, UnsubscribeQuotes, UnsubscribeTrades,
50        },
51    },
52    msgbus::TypedHandler,
53};
54use nautilus_core::{
55    AtomicMap, AtomicSet,
56    time::{AtomicTime, get_atomic_clock_realtime},
57};
58use nautilus_live::{
59    SocketControl, SocketControlFactory,
60    task::{TaskGroup, TaskSpawner},
61};
62use nautilus_model::{
63    data::QuoteTick,
64    enums::BookType,
65    events::PositionEvent,
66    identifiers::{ClientId, InstrumentId, Venue},
67    instruments::InstrumentAny,
68    orderbook::OrderBook,
69};
70use nautilus_network::websocket::proxy::ProxyUrl;
71use parking_lot::Mutex;
72use tokio_util::sync::CancellationToken;
73use ustr::Ustr;
74
75use self::{
76    instruments::{InstrumentUpdateState, TokenMeta},
77    requests::{
78        request_book_snapshot, request_data, request_instrument, request_instruments,
79        request_trades,
80    },
81    runtime::is_instrument_expired_and_not_reported_open,
82    subscriptions::{resolve_token_id_from, sync_ws_subscription_with_terminal_async},
83};
84use crate::{
85    common::consts::POLYMARKET_VENUE,
86    config::PolymarketDataClientConfig,
87    filters::InstrumentFilter,
88    http::{
89        clob::PolymarketClobPublicClient, data_api::PolymarketDataApiHttpClient,
90        gamma::PolymarketGammaHttpClient,
91    },
92    providers::PolymarketInstrumentProvider,
93    resolve::ResolveWatchEntry,
94    rtds::{PolymarketRtdsFeed, is_supported_rtds_data_type},
95    websocket::{RTDS_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
102fn clamp_new_market_fetch_max_concurrency(value: usize) -> usize {
103    value.clamp(1, NEW_MARKET_FETCH_MAX_CONCURRENCY_CAP)
104}
105
106/// Polymarket data client for live market data streaming.
107///
108/// Integrates with the Nautilus DataEngine to provide:
109/// - Real-time order book snapshots and deltas via WebSocket
110/// - Quote ticks synthesized from book data
111/// - Trade ticks from last trade price messages
112/// - Automatic instrument discovery from the Gamma API
113#[derive(Debug)]
114pub struct PolymarketDataClient {
115    clock: &'static AtomicTime,
116    client_id: ClientId,
117    config: PolymarketDataClientConfig,
118    provider: PolymarketInstrumentProvider,
119    clob_public_client: PolymarketClobPublicClient,
120    data_api_client: PolymarketDataApiHttpClient,
121    ws_client: PolymarketMarketConnectionPool,
122    is_connected: AtomicBool,
123    cancellation_token: CancellationToken,
124    tasks: TaskGroup,
125    data_sender: tokio::sync::mpsc::UnboundedSender<DataEvent>,
126    instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
127    instrument_update_state: Arc<Mutex<InstrumentUpdateState>>,
128    token_meta: Arc<DashMap<Ustr, TokenMeta>>,
129    order_books: Arc<DashMap<InstrumentId, OrderBook>>,
130    last_quotes: Arc<DashMap<InstrumentId, QuoteTick>>,
131    active_quote_subs: Arc<AtomicSet<InstrumentId>>,
132    active_delta_subs: Arc<AtomicSet<InstrumentId>>,
133    active_trade_subs: Arc<AtomicSet<InstrumentId>>,
134    resolve_poll_watchlist: Arc<AtomicMap<String, ResolveWatchEntry>>,
135    resolve_watch_apply_mutex: Arc<Mutex<()>>,
136    pending_snapshot_after_tick_change: Arc<AtomicSet<InstrumentId>>,
137    new_market_inflight_keys: Arc<DashMap<String, ()>>,
138    new_market_fetch_semaphore: Arc<tokio::sync::Semaphore>,
139    ws_open_tokens: Arc<AtomicSet<Ustr>>,
140    ws_sub_mutex: Arc<tokio::sync::Mutex<()>>,
141    pending_auto_loads: Arc<Mutex<AHashSet<InstrumentId>>>,
142    auto_load_scheduled: Arc<AtomicBool>,
143    closed_condition_ids: Arc<Mutex<AHashSet<String>>>,
144    position_event_handler: Option<TypedHandler<PositionEvent>>,
145    rtds_feed: PolymarketRtdsFeed,
146    rtds_socket_control: Option<SocketControl>,
147    proxy_url: Option<ProxyUrl>,
148    reset_pending: bool,
149    shutdown_errors: Vec<String>,
150}
151
152impl PolymarketDataClient {
153    /// Creates a new [`PolymarketDataClient`].
154    pub fn new(
155        client_id: ClientId,
156        config: PolymarketDataClientConfig,
157        gamma_client: PolymarketGammaHttpClient,
158        clob_public_client: PolymarketClobPublicClient,
159        data_api_client: PolymarketDataApiHttpClient,
160        ws_client: PolymarketMarketConnectionPool,
161    ) -> Self {
162        Self::new_with_proxy(
163            client_id,
164            config,
165            gamma_client,
166            clob_public_client,
167            data_api_client,
168            ws_client,
169            None,
170        )
171    }
172
173    /// Creates a new data client with an optional validated proxy URL.
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    ) -> Self {
183        let clock = get_atomic_clock_realtime();
184        let data_sender = get_data_event_sender();
185        let socket_factory = SocketControlFactory::new(client_id, Some(*POLYMARKET_VENUE));
186        let ws_client = ws_client.with_socket_factory(socket_factory.clone());
187        let rtds_socket_control = Some(socket_factory.control(RTDS_STREAMS_ENDPOINT));
188        let provider =
189            PolymarketInstrumentProvider::new(gamma_client, config.instrument_config.clone());
190        let configured_fetch_max_concurrency = config.new_market_fetch_max_concurrency;
191        let fetch_max_concurrency =
192            clamp_new_market_fetch_max_concurrency(configured_fetch_max_concurrency);
193
194        if configured_fetch_max_concurrency == 0 {
195            log::warn!(
196                "PolymarketDataClientConfig.new_market_fetch_max_concurrency=0 is invalid, clamping to 1"
197            );
198        } else if configured_fetch_max_concurrency > NEW_MARKET_FETCH_MAX_CONCURRENCY_CAP {
199            log::warn!(
200                "PolymarketDataClientConfig.new_market_fetch_max_concurrency={configured_fetch_max_concurrency} exceeds cap {NEW_MARKET_FETCH_MAX_CONCURRENCY_CAP}, clamping",
201            );
202        }
203        config.new_market_fetch_max_concurrency = fetch_max_concurrency;
204
205        let rtds_url = config.rtds_url();
206        let rtds_transport_backend = config.transport_backend;
207        let rtds_data_sender = data_sender.clone();
208        let tasks = TaskGroup::new();
209        let cancellation_token = tasks.cancellation_token();
210
211        Self {
212            clock,
213            client_id,
214            config,
215            provider,
216            clob_public_client,
217            data_api_client,
218            ws_client,
219            is_connected: AtomicBool::new(false),
220            cancellation_token,
221            tasks,
222            data_sender,
223            instruments: Arc::new(AtomicMap::new()),
224            instrument_update_state: Arc::new(Mutex::new(InstrumentUpdateState::default())),
225            token_meta: Arc::new(DashMap::new()),
226            order_books: Arc::new(DashMap::new()),
227            last_quotes: Arc::new(DashMap::new()),
228            active_quote_subs: Arc::new(AtomicSet::new()),
229            active_delta_subs: Arc::new(AtomicSet::new()),
230            active_trade_subs: Arc::new(AtomicSet::new()),
231            resolve_poll_watchlist: Arc::new(AtomicMap::new()),
232            resolve_watch_apply_mutex: Arc::new(Mutex::new(())),
233            pending_snapshot_after_tick_change: Arc::new(AtomicSet::new()),
234            new_market_inflight_keys: Arc::new(DashMap::new()),
235            new_market_fetch_semaphore: Arc::new(tokio::sync::Semaphore::new(
236                fetch_max_concurrency,
237            )),
238            ws_open_tokens: Arc::new(AtomicSet::new()),
239            ws_sub_mutex: Arc::new(tokio::sync::Mutex::new(())),
240            pending_auto_loads: Arc::new(Mutex::new(AHashSet::new())),
241            auto_load_scheduled: Arc::new(AtomicBool::new(false)),
242            closed_condition_ids: Arc::new(Mutex::new(AHashSet::new())),
243            position_event_handler: None,
244            rtds_feed: PolymarketRtdsFeed::new_with_proxy_and_socket_control(
245                rtds_url,
246                rtds_transport_backend,
247                clock,
248                rtds_data_sender,
249                proxy_url.clone(),
250                rtds_socket_control.clone(),
251            ),
252            rtds_socket_control,
253            proxy_url,
254            reset_pending: false,
255            shutdown_errors: Vec::new(),
256        }
257    }
258
259    /// Returns a reference to the client configuration.
260    #[must_use]
261    pub fn config(&self) -> &PolymarketDataClientConfig {
262        &self.config
263    }
264
265    /// Returns the venue for this data client.
266    #[must_use]
267    pub fn venue(&self) -> Venue {
268        *POLYMARKET_VENUE
269    }
270
271    /// Returns a reference to the instrument provider.
272    #[must_use]
273    pub fn provider(&self) -> &PolymarketInstrumentProvider {
274        &self.provider
275    }
276
277    #[cfg(test)]
278    pub(crate) fn clob_public_client(&self) -> &PolymarketClobPublicClient {
279        &self.clob_public_client
280    }
281
282    #[cfg(test)]
283    pub(crate) fn data_api_client(&self) -> &PolymarketDataApiHttpClient {
284        &self.data_api_client
285    }
286
287    #[cfg(test)]
288    pub(crate) fn ws_client(&self) -> &PolymarketMarketConnectionPool {
289        &self.ws_client
290    }
291
292    #[cfg(test)]
293    pub(crate) fn rtds_feed(&self) -> &PolymarketRtdsFeed {
294        &self.rtds_feed
295    }
296
297    /// Adds an instrument filter on the underlying provider.
298    pub fn add_instrument_filter(&mut self, filter: Arc<dyn InstrumentFilter>) {
299        self.provider.add_filter(filter);
300    }
301
302    /// Returns `true` when the client is connected.
303    #[must_use]
304    pub fn is_connected(&self) -> bool {
305        self.is_connected.load(Ordering::Relaxed)
306    }
307
308    fn resolve_token_id(&self, instrument_id: InstrumentId) -> anyhow::Result<String> {
309        resolve_token_id_from(&self.instruments, instrument_id)
310    }
311
312    fn ensure_live_subscription_allowed(&self, instrument_id: InstrumentId) -> anyhow::Result<()> {
313        let now_ns = self.clock.get_time_ns();
314        let loaded = self.instruments.load();
315        let Some(instrument) = loaded.get(&instrument_id) else {
316            return Ok(());
317        };
318
319        if is_instrument_expired_and_not_reported_open(instrument, now_ns) {
320            anyhow::bail!(
321                "Instrument {instrument_id} is expired and no longer available for live subscription"
322            );
323        }
324
325        Ok(())
326    }
327
328    fn ensure_market_data_request_allowed(
329        &self,
330        instrument_id: InstrumentId,
331    ) -> anyhow::Result<InstrumentAny> {
332        let loaded = self.instruments.load();
333        let instrument = loaded
334            .get(&instrument_id)
335            .ok_or_else(|| InstrumentLookupError::not_found(instrument_id))?
336            .clone();
337
338        if is_instrument_expired_and_not_reported_open(&instrument, self.clock.get_time_ns()) {
339            anyhow::bail!(
340                "Instrument {instrument_id} is expired and no longer available for market data requests"
341            );
342        }
343
344        Ok(instrument)
345    }
346
347    fn add_live_subscription_intent(
348        &self,
349        instrument_id: InstrumentId,
350        subscriptions: &Arc<AtomicSet<InstrumentId>>,
351    ) -> bool {
352        self.add_live_subscription_intent_with_state(instrument_id, subscriptions, || {})
353    }
354
355    fn add_delta_subscription_intent(&self, instrument_id: InstrumentId) -> bool {
356        self.add_live_subscription_intent_with_state(instrument_id, &self.active_delta_subs, || {
357            if self.config.compute_effective_deltas {
358                self.order_books
359                    .entry(instrument_id)
360                    .or_insert_with(|| OrderBook::new(instrument_id, BookType::L2_MBP));
361            }
362        })
363    }
364
365    fn add_live_subscription_intent_with_state(
366        &self,
367        instrument_id: InstrumentId,
368        subscriptions: &Arc<AtomicSet<InstrumentId>>,
369        initialize_state: impl FnOnce(),
370    ) -> bool {
371        let Ok(condition_id) = crate::providers::extract_condition_id(&instrument_id) else {
372            subscriptions.insert(instrument_id);
373            initialize_state();
374            return true;
375        };
376        let closed = self.closed_condition_ids.lock();
377
378        if closed.contains(&condition_id) {
379            log::debug!(
380                "Ignoring live subscription for terminally closed Polymarket condition {condition_id}"
381            );
382            return false;
383        }
384
385        subscriptions.insert(instrument_id);
386        initialize_state();
387        true
388    }
389
390    // Spawns an async task that reconciles the WS subscription for
391    // `instrument_id`. The task holds `ws_sub_mutex` across the wire send so
392    // concurrent subscribe/unsubscribe calls deliver commands to the WS handler
393    // in a consistent order with the final `active_*_subs` state.
394    fn sync_ws_subscription(&self, instrument_id: InstrumentId) {
395        let token_id_str = match self.resolve_token_id(instrument_id) {
396            Ok(s) => s,
397            Err(_) => return,
398        };
399        let active_quote_subs = self.active_quote_subs.clone();
400        let active_delta_subs = self.active_delta_subs.clone();
401        let active_trade_subs = self.active_trade_subs.clone();
402        let closed_condition_ids = self.closed_condition_ids.clone();
403        let ws_open_tokens = self.ws_open_tokens.clone();
404        let ws_sub_mutex = self.ws_sub_mutex.clone();
405        let ws = self.ws_client.handle();
406
407        if let Err(e) = self.tasks.spawn(sync_ws_subscription_with_terminal_async(
408            instrument_id,
409            token_id_str,
410            active_quote_subs,
411            active_delta_subs,
412            active_trade_subs,
413            closed_condition_ids,
414            ws_open_tokens,
415            ws_sub_mutex,
416            ws,
417        )) {
418            log::debug!("Skipping Polymarket data task after shutdown began: {e}");
419        }
420    }
421}
422
423#[async_trait::async_trait(?Send)]
424impl DataClient for PolymarketDataClient {
425    fn client_id(&self) -> ClientId {
426        self.client_id
427    }
428
429    fn venue(&self) -> Option<Venue> {
430        Some(*POLYMARKET_VENUE)
431    }
432
433    fn start(&mut self) -> anyhow::Result<()> {
434        self.start_client();
435        Ok(())
436    }
437
438    fn stop(&mut self) -> anyhow::Result<()> {
439        self.stop_client();
440        Ok(())
441    }
442
443    fn reset(&mut self) -> anyhow::Result<()> {
444        self.reset_client();
445        Ok(())
446    }
447
448    fn dispose(&mut self) -> anyhow::Result<()> {
449        self.stop()
450    }
451
452    async fn connect(&mut self) -> anyhow::Result<()> {
453        self.connect_client().await
454    }
455
456    async fn disconnect(&mut self) -> anyhow::Result<()> {
457        self.disconnect_client().await
458    }
459
460    fn is_connected(&self) -> bool {
461        self.is_connected.load(Ordering::Relaxed)
462    }
463
464    fn is_disconnected(&self) -> bool {
465        !self.is_connected()
466    }
467
468    fn request_data(&self, request: RequestCustomData) -> anyhow::Result<()> {
469        request_data(self, request);
470        Ok(())
471    }
472
473    fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
474        request_instruments(self, request);
475        Ok(())
476    }
477
478    fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
479        request_instrument(self, request);
480        Ok(())
481    }
482
483    fn request_book_snapshot(&self, request: RequestBookSnapshot) -> anyhow::Result<()> {
484        request_book_snapshot(self, request)
485    }
486
487    fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
488        request_trades(self, request)
489    }
490
491    fn subscribe_instruments(&mut self, _cmd: SubscribeInstruments) -> anyhow::Result<()> {
492        log::debug!("subscribe_instruments: subscribed individually via data subscription methods");
493        Ok(())
494    }
495
496    fn subscribe_instrument(&mut self, cmd: SubscribeInstrument) -> anyhow::Result<()> {
497        log::debug!(
498            "Subscribed to instrument definition updates for {}; shared instrument sources remain active",
499            cmd.instrument_id
500        );
501
502        Ok(())
503    }
504
505    fn unsubscribe_instrument(&mut self, cmd: &UnsubscribeInstrument) -> anyhow::Result<()> {
506        log::debug!(
507            "Unsubscribed from instrument {}; shared instrument sources remain active",
508            cmd.instrument_id
509        );
510        Ok(())
511    }
512
513    fn subscribe(&mut self, cmd: SubscribeCustomData) -> anyhow::Result<()> {
514        if !is_supported_rtds_data_type(&cmd.data_type) {
515            log::debug!(
516                "Ignoring unsupported Polymarket custom data subscription: {}",
517                cmd.data_type
518            );
519            return Ok(());
520        }
521
522        log::debug!(
523            "Tracking Polymarket RTDS custom data subscription: {}",
524            cmd.data_type
525        );
526        let changed = self.rtds_feed.track_subscribe(cmd.data_type)?;
527        if !changed {
528            return Ok(());
529        }
530
531        if !self.is_connected() {
532            return Ok(());
533        }
534
535        self.rtds_feed
536            .request_reconcile(crate::rtds::ReconcileReason::DesiredChanged);
537
538        Ok(())
539    }
540
541    fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
542        if cmd.book_type != BookType::L2_MBP {
543            anyhow::bail!(
544                "Polymarket only supports L2_MBP order book deltas, received {:?}",
545                cmd.book_type
546            );
547        }
548
549        let instrument_id = cmd.instrument_id;
550        self.ensure_live_subscription_allowed(instrument_id)?;
551        let cached = self.instruments.load().contains_key(&instrument_id);
552
553        if !cached && !self.config.auto_load_missing_instruments {
554            anyhow::bail!(
555                "Instrument {instrument_id} not found, and `auto_load_missing_instruments` is disabled"
556            );
557        }
558
559        // Mark intent before routing so unsubscribe can race-safely clear it.
560        if !self.add_delta_subscription_intent(instrument_id) {
561            return Ok(());
562        }
563
564        if !cached {
565            self.queue_pending_load(instrument_id);
566            return Ok(());
567        }
568
569        self.sync_ws_subscription(instrument_id);
570        Ok(())
571    }
572
573    fn subscribe_book_depth10(&mut self, _cmd: SubscribeBookDepth10) -> anyhow::Result<()> {
574        anyhow::bail!(
575            "Polymarket does not support OrderBookDepth10 subscriptions; use managed L2_MBP order book deltas"
576        )
577    }
578
579    fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
580        let instrument_id = cmd.instrument_id;
581        self.ensure_live_subscription_allowed(instrument_id)?;
582        let cached = self.instruments.load().contains_key(&instrument_id);
583
584        if !cached && !self.config.auto_load_missing_instruments {
585            anyhow::bail!(
586                "Instrument {instrument_id} not found, and `auto_load_missing_instruments` is disabled"
587            );
588        }
589
590        if !self.add_live_subscription_intent(instrument_id, &self.active_quote_subs) {
591            return Ok(());
592        }
593
594        if !cached {
595            self.queue_pending_load(instrument_id);
596            return Ok(());
597        }
598
599        self.sync_ws_subscription(instrument_id);
600        Ok(())
601    }
602
603    fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
604        let instrument_id = cmd.instrument_id;
605        self.ensure_live_subscription_allowed(instrument_id)?;
606        let cached = self.instruments.load().contains_key(&instrument_id);
607
608        if !cached && !self.config.auto_load_missing_instruments {
609            anyhow::bail!(
610                "Instrument {instrument_id} not found, and `auto_load_missing_instruments` is disabled"
611            );
612        }
613
614        if !self.add_live_subscription_intent(instrument_id, &self.active_trade_subs) {
615            return Ok(());
616        }
617
618        if !cached {
619            self.queue_pending_load(instrument_id);
620            return Ok(());
621        }
622
623        self.sync_ws_subscription(instrument_id);
624        Ok(())
625    }
626
627    fn subscribe_instrument_status(
628        &mut self,
629        _cmd: SubscribeInstrumentStatus,
630    ) -> anyhow::Result<()> {
631        anyhow::bail!(
632            "Polymarket does not support generic instrument status subscriptions; resolution status is owned by position tracking"
633        )
634    }
635
636    fn subscribe_instrument_close(&mut self, _cmd: SubscribeInstrumentClose) -> anyhow::Result<()> {
637        anyhow::bail!(
638            "Polymarket does not support generic instrument close subscriptions; resolution close is owned by position tracking"
639        )
640    }
641
642    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
643        let instrument_id = cmd.instrument_id;
644        self.active_delta_subs.remove(&instrument_id);
645        self.pending_snapshot_after_tick_change
646            .remove(&instrument_id);
647        self.drop_pending_if_unwanted(instrument_id);
648        self.drop_local_data_state_if_unwanted(instrument_id);
649        self.sync_ws_subscription(instrument_id);
650        Ok(())
651    }
652
653    fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
654        let instrument_id = cmd.instrument_id;
655        self.active_quote_subs.remove(&instrument_id);
656        self.drop_pending_if_unwanted(instrument_id);
657        self.drop_local_data_state_if_unwanted(instrument_id);
658        self.sync_ws_subscription(instrument_id);
659        Ok(())
660    }
661
662    fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
663        let instrument_id = cmd.instrument_id;
664        self.active_trade_subs.remove(&instrument_id);
665        self.drop_pending_if_unwanted(instrument_id);
666        self.sync_ws_subscription(instrument_id);
667        Ok(())
668    }
669
670    fn unsubscribe(&mut self, cmd: &UnsubscribeCustomData) -> anyhow::Result<()> {
671        if !is_supported_rtds_data_type(&cmd.data_type) {
672            log::debug!(
673                "Ignoring unsupported Polymarket custom data unsubscription: {}",
674                cmd.data_type
675            );
676            return Ok(());
677        }
678
679        log::debug!(
680            "Tracking Polymarket RTDS custom data unsubscription: {}",
681            cmd.data_type
682        );
683        let changed = self.rtds_feed.track_unsubscribe(&cmd.data_type)?;
684        if !changed {
685            return Ok(());
686        }
687
688        if !self.is_connected() {
689            return Ok(());
690        }
691
692        self.rtds_feed
693            .request_reconcile(crate::rtds::ReconcileReason::DesiredChanged);
694
695        Ok(())
696    }
697}
698
699pub(super) fn spawn_task<F>(tasks: &TaskSpawner, future: F)
700where
701    F: Future<Output = ()> + Send + 'static,
702{
703    if let Err(e) = tasks.spawn(future) {
704        log::debug!("Skipping Polymarket data task after shutdown began: {e}");
705    }
706}
nautilus_polymarket/data/mod.rs

nautilus_polymarket/data/
mod.rs
