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

nautilus_polymarket/data/
mod.rs
