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 instruments;
21mod lifecycle;
22mod requests;
23mod runtime;
24mod subscriptions;
25
26use std::{
27    sync::{
28        Arc, Mutex as StdMutex,
29        atomic::{AtomicBool, Ordering},
30    },
31    time::Duration,
32};
33
34use ahash::AHashSet;
35use dashmap::DashMap;
36use nautilus_common::{
37    cache::InstrumentLookupError,
38    clients::DataClient,
39    live::{get_runtime, runner::get_data_event_sender},
40    messages::{
41        DataEvent,
42        data::{
43            RequestBookSnapshot, RequestCustomData, RequestInstrument, RequestInstruments,
44            RequestTrades, SubscribeBookDeltas, SubscribeCustomData, SubscribeInstruments,
45            SubscribeQuotes, SubscribeTrades, UnsubscribeBookDeltas, UnsubscribeCustomData,
46            UnsubscribeQuotes, UnsubscribeTrades,
47        },
48    },
49    msgbus::TypedHandler,
50};
51use nautilus_core::{
52    AtomicMap, AtomicSet,
53    time::{AtomicTime, get_atomic_clock_realtime},
54};
55use nautilus_model::{
56    data::QuoteTick,
57    enums::BookType,
58    events::PositionEvent,
59    identifiers::{ClientId, InstrumentId, Venue},
60    instruments::InstrumentAny,
61    orderbook::OrderBook,
62};
63use tokio::task::JoinHandle;
64use tokio_util::sync::CancellationToken;
65use ustr::Ustr;
66
67use self::{
68    instruments::TokenMeta,
69    requests::{
70        request_book_snapshot, request_data, request_instrument, request_instruments,
71        request_trades,
72    },
73    runtime::is_instrument_expired,
74    subscriptions::{resolve_token_id_from, sync_ws_subscription_async},
75};
76use crate::{
77    common::consts::POLYMARKET_VENUE,
78    config::PolymarketDataClientConfig,
79    filters::InstrumentFilter,
80    http::{
81        clob::PolymarketClobPublicClient, data_api::PolymarketDataApiHttpClient,
82        gamma::PolymarketGammaHttpClient,
83    },
84    providers::PolymarketInstrumentProvider,
85    resolve::ResolveWatchEntry,
86    rtds::{PolymarketRtdsFeed, is_supported_rtds_data_type},
87    websocket::client::PolymarketWebSocketClient,
88};
89
90const NEW_MARKET_FETCH_MAX_CONCURRENCY_CAP: usize = 64;
91pub(super) const NEW_MARKET_EMPTY_RECHECK_MAX_ATTEMPTS: usize = 1;
92pub(super) const NEW_MARKET_EMPTY_RECHECK_DELAY: Duration = Duration::from_millis(500);
93
94fn clamp_new_market_fetch_max_concurrency(value: usize) -> usize {
95    value.clamp(1, NEW_MARKET_FETCH_MAX_CONCURRENCY_CAP)
96}
97
98/// Polymarket data client for live market data streaming.
99///
100/// Integrates with the Nautilus DataEngine to provide:
101/// - Real-time order book snapshots and deltas via WebSocket
102/// - Quote ticks synthesized from book data
103/// - Trade ticks from last trade price messages
104/// - Automatic instrument discovery from the Gamma API
105#[derive(Debug)]
106pub struct PolymarketDataClient {
107    clock: &'static AtomicTime,
108    client_id: ClientId,
109    config: PolymarketDataClientConfig,
110    provider: PolymarketInstrumentProvider,
111    clob_public_client: PolymarketClobPublicClient,
112    data_api_client: PolymarketDataApiHttpClient,
113    ws_client: PolymarketWebSocketClient,
114    is_connected: AtomicBool,
115    cancellation_token: CancellationToken,
116    tasks: Vec<JoinHandle<()>>,
117    data_sender: tokio::sync::mpsc::UnboundedSender<DataEvent>,
118    instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
119    token_meta: Arc<DashMap<Ustr, TokenMeta>>,
120    order_books: Arc<DashMap<InstrumentId, OrderBook>>,
121    last_quotes: Arc<DashMap<InstrumentId, QuoteTick>>,
122    active_quote_subs: Arc<AtomicSet<InstrumentId>>,
123    active_delta_subs: Arc<AtomicSet<InstrumentId>>,
124    active_trade_subs: Arc<AtomicSet<InstrumentId>>,
125    resolve_poll_watchlist: Arc<AtomicMap<String, ResolveWatchEntry>>,
126    resolve_watch_apply_mutex: Arc<StdMutex<()>>,
127    pending_snapshot_after_tick_change: Arc<AtomicSet<InstrumentId>>,
128    new_market_inflight_keys: Arc<DashMap<String, ()>>,
129    new_market_fetch_semaphore: Arc<tokio::sync::Semaphore>,
130    ws_open_tokens: Arc<AtomicSet<Ustr>>,
131    ws_sub_mutex: Arc<tokio::sync::Mutex<()>>,
132    pending_auto_loads: Arc<StdMutex<AHashSet<InstrumentId>>>,
133    auto_load_scheduled: Arc<AtomicBool>,
134    position_event_handler: Option<TypedHandler<PositionEvent>>,
135    rtds_feed: PolymarketRtdsFeed,
136}
137
138impl PolymarketDataClient {
139    /// Creates a new [`PolymarketDataClient`].
140    pub fn new(
141        client_id: ClientId,
142        mut config: PolymarketDataClientConfig,
143        gamma_client: PolymarketGammaHttpClient,
144        clob_public_client: PolymarketClobPublicClient,
145        data_api_client: PolymarketDataApiHttpClient,
146        ws_client: PolymarketWebSocketClient,
147    ) -> Self {
148        let clock = get_atomic_clock_realtime();
149        let data_sender = get_data_event_sender();
150        let provider =
151            PolymarketInstrumentProvider::new(gamma_client, config.instrument_config.clone());
152        let configured_fetch_max_concurrency = config.new_market_fetch_max_concurrency;
153        let fetch_max_concurrency =
154            clamp_new_market_fetch_max_concurrency(configured_fetch_max_concurrency);
155
156        if configured_fetch_max_concurrency == 0 {
157            log::warn!(
158                "PolymarketDataClientConfig.new_market_fetch_max_concurrency=0 is invalid, clamping to 1"
159            );
160        } else if configured_fetch_max_concurrency > NEW_MARKET_FETCH_MAX_CONCURRENCY_CAP {
161            log::warn!(
162                "PolymarketDataClientConfig.new_market_fetch_max_concurrency={configured_fetch_max_concurrency} exceeds cap {NEW_MARKET_FETCH_MAX_CONCURRENCY_CAP}, clamping",
163            );
164        }
165        config.new_market_fetch_max_concurrency = fetch_max_concurrency;
166
167        let rtds_url = config.rtds_url();
168        let rtds_transport_backend = config.transport_backend;
169        let rtds_data_sender = data_sender.clone();
170
171        Self {
172            clock,
173            client_id,
174            config,
175            provider,
176            clob_public_client,
177            data_api_client,
178            ws_client,
179            is_connected: AtomicBool::new(false),
180            cancellation_token: CancellationToken::new(),
181            tasks: Vec::new(),
182            data_sender,
183            instruments: Arc::new(AtomicMap::new()),
184            token_meta: Arc::new(DashMap::new()),
185            order_books: Arc::new(DashMap::new()),
186            last_quotes: Arc::new(DashMap::new()),
187            active_quote_subs: Arc::new(AtomicSet::new()),
188            active_delta_subs: Arc::new(AtomicSet::new()),
189            active_trade_subs: Arc::new(AtomicSet::new()),
190            resolve_poll_watchlist: Arc::new(AtomicMap::new()),
191            resolve_watch_apply_mutex: Arc::new(StdMutex::new(())),
192            pending_snapshot_after_tick_change: Arc::new(AtomicSet::new()),
193            new_market_inflight_keys: Arc::new(DashMap::new()),
194            new_market_fetch_semaphore: Arc::new(tokio::sync::Semaphore::new(
195                fetch_max_concurrency,
196            )),
197            ws_open_tokens: Arc::new(AtomicSet::new()),
198            ws_sub_mutex: Arc::new(tokio::sync::Mutex::new(())),
199            pending_auto_loads: Arc::new(StdMutex::new(AHashSet::new())),
200            auto_load_scheduled: Arc::new(AtomicBool::new(false)),
201            position_event_handler: None,
202            rtds_feed: PolymarketRtdsFeed::new(
203                rtds_url,
204                rtds_transport_backend,
205                clock,
206                rtds_data_sender,
207            ),
208        }
209    }
210
211    /// Returns a reference to the client configuration.
212    #[must_use]
213    pub fn config(&self) -> &PolymarketDataClientConfig {
214        &self.config
215    }
216
217    /// Returns the venue for this data client.
218    #[must_use]
219    pub fn venue(&self) -> Venue {
220        *POLYMARKET_VENUE
221    }
222
223    /// Returns a reference to the instrument provider.
224    #[must_use]
225    pub fn provider(&self) -> &PolymarketInstrumentProvider {
226        &self.provider
227    }
228
229    /// Adds an instrument filter on the underlying provider.
230    pub fn add_instrument_filter(&mut self, filter: Arc<dyn InstrumentFilter>) {
231        self.provider.add_filter(filter);
232    }
233
234    /// Returns `true` when the client is connected.
235    #[must_use]
236    pub fn is_connected(&self) -> bool {
237        self.is_connected.load(Ordering::Relaxed)
238    }
239
240    fn resolve_token_id(&self, instrument_id: InstrumentId) -> anyhow::Result<String> {
241        resolve_token_id_from(&self.instruments, instrument_id)
242    }
243
244    fn ensure_live_subscription_allowed(&self, instrument_id: InstrumentId) -> anyhow::Result<()> {
245        let now_ns = self.clock.get_time_ns();
246        let loaded = self.instruments.load();
247        let Some(instrument) = loaded.get(&instrument_id) else {
248            return Ok(());
249        };
250
251        if is_instrument_expired(instrument, now_ns) {
252            anyhow::bail!(
253                "Instrument {instrument_id} is expired and no longer available for live subscription"
254            );
255        }
256
257        Ok(())
258    }
259
260    fn ensure_market_data_request_allowed(
261        &self,
262        instrument_id: InstrumentId,
263    ) -> anyhow::Result<InstrumentAny> {
264        let loaded = self.instruments.load();
265        let instrument = loaded
266            .get(&instrument_id)
267            .ok_or_else(|| InstrumentLookupError::not_found(instrument_id))?
268            .clone();
269
270        if is_instrument_expired(&instrument, self.clock.get_time_ns()) {
271            anyhow::bail!(
272                "Instrument {instrument_id} is expired and no longer available for market data requests"
273            );
274        }
275
276        Ok(instrument)
277    }
278
279    // Spawns an async task that reconciles the WS subscription for
280    // `instrument_id`. The task holds `ws_sub_mutex` across the wire send so
281    // concurrent subscribe/unsubscribe calls deliver commands to the WS handler
282    // in a consistent order with the final `active_*_subs` state.
283    fn sync_ws_subscription(&self, instrument_id: InstrumentId) {
284        let token_id_str = match self.resolve_token_id(instrument_id) {
285            Ok(s) => s,
286            Err(_) => return,
287        };
288        let active_quote_subs = self.active_quote_subs.clone();
289        let active_delta_subs = self.active_delta_subs.clone();
290        let active_trade_subs = self.active_trade_subs.clone();
291        let ws_open_tokens = self.ws_open_tokens.clone();
292        let ws_sub_mutex = self.ws_sub_mutex.clone();
293        let ws = self.ws_client.clone_subscription_handle();
294
295        get_runtime().spawn(sync_ws_subscription_async(
296            instrument_id,
297            token_id_str,
298            active_quote_subs,
299            active_delta_subs,
300            active_trade_subs,
301            ws_open_tokens,
302            ws_sub_mutex,
303            ws,
304        ));
305    }
306}
307
308#[async_trait::async_trait(?Send)]
309impl DataClient for PolymarketDataClient {
310    fn client_id(&self) -> ClientId {
311        self.client_id
312    }
313
314    fn venue(&self) -> Option<Venue> {
315        Some(*POLYMARKET_VENUE)
316    }
317
318    fn start(&mut self) -> anyhow::Result<()> {
319        self.start_client();
320        Ok(())
321    }
322
323    fn stop(&mut self) -> anyhow::Result<()> {
324        self.stop_client();
325        Ok(())
326    }
327
328    fn reset(&mut self) -> anyhow::Result<()> {
329        self.reset_client();
330        Ok(())
331    }
332
333    fn dispose(&mut self) -> anyhow::Result<()> {
334        self.stop()
335    }
336
337    async fn connect(&mut self) -> anyhow::Result<()> {
338        self.connect_client().await
339    }
340
341    async fn disconnect(&mut self) -> anyhow::Result<()> {
342        self.disconnect_client().await
343    }
344
345    fn is_connected(&self) -> bool {
346        self.is_connected.load(Ordering::Relaxed)
347    }
348
349    fn is_disconnected(&self) -> bool {
350        !self.is_connected()
351    }
352
353    fn request_data(&self, request: RequestCustomData) -> anyhow::Result<()> {
354        request_data(self, request);
355        Ok(())
356    }
357
358    fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
359        request_instruments(self, request);
360        Ok(())
361    }
362
363    fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
364        request_instrument(self, request);
365        Ok(())
366    }
367
368    fn request_book_snapshot(&self, request: RequestBookSnapshot) -> anyhow::Result<()> {
369        request_book_snapshot(self, request)
370    }
371
372    fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
373        request_trades(self, request)
374    }
375
376    fn subscribe_instruments(&mut self, _cmd: SubscribeInstruments) -> anyhow::Result<()> {
377        log::debug!("subscribe_instruments: subscribed individually via data subscription methods");
378        Ok(())
379    }
380
381    fn subscribe(&mut self, cmd: SubscribeCustomData) -> anyhow::Result<()> {
382        if !is_supported_rtds_data_type(&cmd.data_type) {
383            log::debug!(
384                "Ignoring unsupported Polymarket custom data subscription: {}",
385                cmd.data_type
386            );
387            return Ok(());
388        }
389
390        log::debug!(
391            "Tracking Polymarket RTDS custom data subscription: {}",
392            cmd.data_type
393        );
394        let changed = self.rtds_feed.track_subscribe(cmd.data_type)?;
395        if !changed {
396            return Ok(());
397        }
398
399        if !self.is_connected() {
400            return Ok(());
401        }
402
403        self.rtds_feed.schedule_sync();
404
405        Ok(())
406    }
407
408    fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
409        if cmd.book_type != BookType::L2_MBP {
410            anyhow::bail!(
411                "Polymarket only supports L2_MBP order book deltas, received {:?}",
412                cmd.book_type
413            );
414        }
415
416        let instrument_id = cmd.instrument_id;
417        self.ensure_live_subscription_allowed(instrument_id)?;
418        let cached = self.instruments.load().contains_key(&instrument_id);
419
420        if !cached && !self.config.auto_load_missing_instruments {
421            anyhow::bail!(
422                "Instrument {instrument_id} not found, and `auto_load_missing_instruments` is disabled"
423            );
424        }
425
426        // Mark intent before routing so unsubscribe can race-safely clear it.
427        self.active_delta_subs.insert(instrument_id);
428        self.order_books
429            .entry(instrument_id)
430            .or_insert_with(|| OrderBook::new(instrument_id, BookType::L2_MBP));
431
432        if !cached {
433            self.queue_pending_load(instrument_id);
434            return Ok(());
435        }
436
437        self.sync_ws_subscription(instrument_id);
438        Ok(())
439    }
440
441    fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
442        let instrument_id = cmd.instrument_id;
443        self.ensure_live_subscription_allowed(instrument_id)?;
444        let cached = self.instruments.load().contains_key(&instrument_id);
445
446        if !cached && !self.config.auto_load_missing_instruments {
447            anyhow::bail!(
448                "Instrument {instrument_id} not found, and `auto_load_missing_instruments` is disabled"
449            );
450        }
451
452        self.active_quote_subs.insert(instrument_id);
453
454        if !cached {
455            self.queue_pending_load(instrument_id);
456            return Ok(());
457        }
458
459        self.sync_ws_subscription(instrument_id);
460        Ok(())
461    }
462
463    fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
464        let instrument_id = cmd.instrument_id;
465        self.ensure_live_subscription_allowed(instrument_id)?;
466        let cached = self.instruments.load().contains_key(&instrument_id);
467
468        if !cached && !self.config.auto_load_missing_instruments {
469            anyhow::bail!(
470                "Instrument {instrument_id} not found, and `auto_load_missing_instruments` is disabled"
471            );
472        }
473
474        self.active_trade_subs.insert(instrument_id);
475
476        if !cached {
477            self.queue_pending_load(instrument_id);
478            return Ok(());
479        }
480
481        self.sync_ws_subscription(instrument_id);
482        Ok(())
483    }
484
485    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
486        let instrument_id = cmd.instrument_id;
487        self.active_delta_subs.remove(&instrument_id);
488        self.pending_snapshot_after_tick_change
489            .remove(&instrument_id);
490        self.drop_pending_if_unwanted(instrument_id);
491        self.drop_local_book_state_if_unwanted(instrument_id);
492        self.sync_ws_subscription(instrument_id);
493        Ok(())
494    }
495
496    fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
497        let instrument_id = cmd.instrument_id;
498        self.active_quote_subs.remove(&instrument_id);
499        self.drop_pending_if_unwanted(instrument_id);
500        self.drop_local_book_state_if_unwanted(instrument_id);
501        self.sync_ws_subscription(instrument_id);
502        Ok(())
503    }
504
505    fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
506        let instrument_id = cmd.instrument_id;
507        self.active_trade_subs.remove(&instrument_id);
508        self.drop_pending_if_unwanted(instrument_id);
509        self.sync_ws_subscription(instrument_id);
510        Ok(())
511    }
512
513    fn unsubscribe(&mut self, cmd: &UnsubscribeCustomData) -> anyhow::Result<()> {
514        if !is_supported_rtds_data_type(&cmd.data_type) {
515            log::debug!(
516                "Ignoring unsupported Polymarket custom data unsubscription: {}",
517                cmd.data_type
518            );
519            return Ok(());
520        }
521
522        log::debug!(
523            "Tracking Polymarket RTDS custom data unsubscription: {}",
524            cmd.data_type
525        );
526        let changed = self.rtds_feed.track_unsubscribe(&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.schedule_sync();
536
537        Ok(())
538    }
539}
nautilus_polymarket/data/mod.rs

nautilus_polymarket/data/
mod.rs
