nautilus_polymarket/execution/
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 execution client implementation for the Polymarket adapter.
17
18pub mod order_builder;
19pub mod parse;
20
21pub(crate) mod context;
22pub(crate) mod fill_tracker;
23pub(crate) mod pending;
24pub(crate) mod reconciliation;
25pub(crate) mod settlement;
26pub(crate) mod submitter;
27pub(crate) mod types;
28
29mod cancellations;
30mod collateral;
31mod lifecycle;
32mod orders;
33mod reports;
34mod responses;
35
36use std::sync::{Arc, atomic::AtomicBool};
37
38use anyhow::Context;
39use async_trait::async_trait;
40use nautilus_common::{
41    clients::ExecutionClient,
42    messages::execution::{
43        BatchCancelOrders, CancelAllOrders, CancelOrder, GenerateFillReports,
44        GenerateOrderStatusReport, GenerateOrderStatusReports, GeneratePositionStatusReports,
45        ModifyOrder, QueryAccount, QueryOrder, SubmitOrder, SubmitOrderList,
46    },
47    msgbus::TypedHandler,
48};
49use nautilus_core::{
50    Params, UnixNanos,
51    collections::AtomicMap,
52    time::{AtomicTime, get_atomic_clock_realtime},
53};
54use nautilus_live::{ExecutionClientCore, ExecutionEventEmitter, SocketControl, task::TaskGroup};
55use nautilus_model::{
56    accounts::AccountAny,
57    enums::{AccountType, LiquiditySide, OmsType, OrderSide},
58    events::{OrderEventAny, PositionEvent},
59    identifiers::{
60        AccountId, ClientId, ClientOrderId, InstrumentId, StrategyId, Venue, VenueOrderId,
61    },
62    instruments::InstrumentAny,
63    orders::{Order, OrderAny},
64    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
65    types::{AccountBalance, MarginBalance, Money, Price, Quantity},
66};
67use nautilus_network::retry::RetryConfig;
68use parking_lot::Mutex;
69pub(crate) use responses::is_post_only_crossing;
70use rust_decimal::Decimal;
71use ustr::Ustr;
72
73pub(crate) use self::reports::get_pusd_currency;
74use self::{
75    context::OrderContextRegistry,
76    fill_tracker::OrderFillTrackerMap,
77    order_builder::PolymarketOrderBuilder,
78    pending::{PendingCancelTracker, PendingSubmitTracker},
79    settlement::SettlementRegistry,
80    submitter::OrderSubmitter,
81};
82use crate::{
83    common::{consts::POLYMARKET_VENUE, credential::Secrets, enums::PolymarketSignatureType},
84    config::PolymarketExecutionClientConfig,
85    http::{clob::PolymarketClobHttpClient, data_api::PolymarketDataApiHttpClient},
86    signing::eip712::OrderSigner,
87    websocket::{
88        USER_STREAMS_ENDPOINT, client::PolymarketWebSocketClient, dispatch::WsDispatchState,
89    },
90};
91
92/// Live execution client for the Polymarket prediction market.
93#[derive(Debug)]
94pub struct PolymarketExecutionClient {
95    core: ExecutionClientCore,
96    clock: &'static AtomicTime,
97    config: PolymarketExecutionClientConfig,
98    emitter: ExecutionEventEmitter,
99    http_client: PolymarketClobHttpClient,
100    data_api_client: PolymarketDataApiHttpClient,
101    submitter: OrderSubmitter,
102    ws_client: PolymarketWebSocketClient,
103    secrets: Secrets,
104    session_tasks: TaskGroup,
105    pending_tasks: TaskGroup,
106    shutdown_errors: Vec<String>,
107    stopping: Arc<AtomicBool>,
108    heartbeat_healthy: Arc<AtomicBool>,
109    order_event_handler: Option<TypedHandler<OrderEventAny>>,
110    position_event_handler: Option<TypedHandler<PositionEvent>>,
111    fill_observer: Option<TypedHandler<OrderEventAny>>,
112    void_observer: Option<TypedHandler<OrderEventAny>>,
113    decline_observer: Option<TypedHandler<OrderEventAny>>,
114    shared_token_instruments: Arc<AtomicMap<Ustr, InstrumentAny>>,
115    neg_risk_index: Arc<AtomicMap<InstrumentId, bool>>,
116    pending_submits: PendingSubmitTracker,
117    pending_cancels: PendingCancelTracker,
118    order_contexts: Arc<OrderContextRegistry>,
119    collateral_locks: Arc<Mutex<collateral::CollateralLocks>>,
120    fill_tracker: Arc<OrderFillTrackerMap>,
121    settlement: Arc<SettlementRegistry>,
122    ws_dispatch_state: Arc<Mutex<WsDispatchState>>,
123}
124
125impl PolymarketExecutionClient {
126    /// Creates a new [`PolymarketExecutionClient`].
127    ///
128    /// # Errors
129    ///
130    /// Returns an error if configuration validation fails, credentials cannot be resolved, or clients
131    /// fail to construct.
132    pub fn new(
133        core: ExecutionClientCore,
134        config: PolymarketExecutionClientConfig,
135    ) -> anyhow::Result<Self> {
136        config.validate()?;
137
138        let proxy_url = config.validated_proxy_url()?;
139        config.validate_signer()?;
140        let secrets = Secrets::resolve(
141            config.private_key.clone(),
142            config.api_key.clone(),
143            config.api_secret.clone(),
144            config.passphrase.clone(),
145            config.funder.clone(),
146        )
147        .context("failed to resolve Polymarket credentials")?;
148
149        let signer_address = secrets.address.clone();
150        let maker_address = resolve_maker_address(
151            config.signature_type,
152            &signer_address,
153            secrets.funder.as_deref(),
154        )?;
155        let http_client = PolymarketClobHttpClient::new_with_proxy(
156            secrets.credential.clone(),
157            signer_address.clone(),
158            config.base_url_http.clone(),
159            config.http_timeout_secs,
160            proxy_url.clone(),
161        )
162        .map_err(|e| anyhow::anyhow!("{e}"))
163        .context("failed to create CLOB HTTP client")?;
164
165        let data_api_client = PolymarketDataApiHttpClient::new_with_proxy(
166            Some(config.data_api_url()),
167            config.http_timeout_secs,
168            proxy_url.clone(),
169        )
170        .map_err(|e| anyhow::anyhow!("{e}"))
171        .context("failed to create Data API HTTP client")?;
172
173        let order_signer = OrderSigner::new(&secrets.private_key)
174            .context("failed to create order signer")?
175            .with_signer_type(config.signer_type);
176
177        let order_builder = Arc::new(PolymarketOrderBuilder::new(
178            order_signer,
179            signer_address,
180            maker_address,
181            config.signature_type,
182        ));
183
184        let retry_config = RetryConfig {
185            max_retries: config.max_retries,
186            initial_delay_ms: config.retry_delay_initial_ms,
187            max_delay_ms: config.retry_delay_max_ms,
188            backoff_factor: 2.0,
189            jitter_ms: 1_000,
190            operation_timeout_ms: Some(config.http_timeout_secs * 1_000),
191            immediate_first: false,
192            max_elapsed_ms: Some(180_000),
193        };
194
195        let settlement = Arc::new(SettlementRegistry::new(core.account_id));
196
197        let submitter = OrderSubmitter::new(
198            http_client.clone(),
199            order_builder,
200            retry_config,
201            settlement.clone(),
202        );
203
204        let ws_client = PolymarketWebSocketClient::new_user_with_proxy(
205            config.base_url_ws.clone(),
206            secrets.credential.clone(),
207            config.transport_backend,
208            proxy_url,
209        );
210
211        let ws_client = ws_client.with_socket_control(SocketControl::new(
212            core.client_id,
213            Some(*POLYMARKET_VENUE),
214            USER_STREAMS_ENDPOINT,
215        ));
216
217        let clock = get_atomic_clock_realtime();
218        let pusd = get_pusd_currency();
219        let emitter = ExecutionEventEmitter::new(
220            clock,
221            core.trader_id,
222            core.account_id,
223            AccountType::Cash,
224            Some(pusd),
225        );
226
227        let session_tasks = TaskGroup::new();
228        let pending_tasks = TaskGroup::new();
229
230        Ok(Self {
231            core,
232            clock,
233            config,
234            emitter,
235            http_client,
236            data_api_client,
237            submitter,
238            ws_client,
239            secrets,
240            session_tasks,
241            pending_tasks,
242            shutdown_errors: Vec::new(),
243            stopping: Arc::new(AtomicBool::new(false)),
244            heartbeat_healthy: Arc::new(AtomicBool::new(true)),
245            order_event_handler: None,
246            position_event_handler: None,
247            fill_observer: None,
248            void_observer: None,
249            decline_observer: None,
250            shared_token_instruments: Arc::new(AtomicMap::new()),
251            neg_risk_index: Arc::new(AtomicMap::new()),
252            pending_submits: PendingSubmitTracker::default(),
253            pending_cancels: PendingCancelTracker::default(),
254            order_contexts: Arc::new(OrderContextRegistry::default()),
255            collateral_locks: Arc::new(Mutex::new(collateral::CollateralLocks::default())),
256            fill_tracker: Arc::new(OrderFillTrackerMap::new()),
257            settlement,
258            ws_dispatch_state: Arc::new(Mutex::new(WsDispatchState::default())),
259        })
260    }
261
262    fn check_not_faulted(&self) -> anyhow::Result<()> {
263        if let Some(reason) = self.settlement.client_fault_reason() {
264            anyhow::bail!("Polymarket execution client is faulted closed until restart: {reason}");
265        }
266
267        Ok(())
268    }
269}
270
271fn resolve_maker_address(
272    signature_type: PolymarketSignatureType,
273    signer_address: &str,
274    funder: Option<&str>,
275) -> anyhow::Result<String> {
276    let maker_address = match signature_type {
277        PolymarketSignatureType::Eoa => funder.unwrap_or(signer_address),
278        PolymarketSignatureType::PolyProxy
279        | PolymarketSignatureType::PolyGnosisSafe
280        | PolymarketSignatureType::Poly1271 => funder.ok_or_else(|| {
281            anyhow::anyhow!(
282                "Polymarket {signature_type:?} signature type requires a funder wallet address",
283            )
284        })?,
285    };
286
287    if signature_type != PolymarketSignatureType::Eoa
288        && maker_address.eq_ignore_ascii_case(signer_address)
289    {
290        anyhow::bail!(
291            "Polymarket {signature_type:?} signature type requires a funder distinct from the signing address",
292        );
293    }
294
295    Ok(maker_address.to_string())
296}
297
298#[async_trait(?Send)]
299impl ExecutionClient for PolymarketExecutionClient {
300    fn is_connected(&self) -> bool {
301        self.core.is_connected()
302            && !self.settlement.client_faulted()
303            && (!self.config.heartbeat_enabled
304                || self
305                    .heartbeat_healthy
306                    .load(std::sync::atomic::Ordering::Acquire))
307    }
308
309    fn client_id(&self) -> ClientId {
310        self.core.client_id
311    }
312
313    fn account_id(&self) -> AccountId {
314        self.core.account_id
315    }
316
317    fn venue(&self) -> Venue {
318        *POLYMARKET_VENUE
319    }
320
321    fn oms_type(&self) -> OmsType {
322        OmsType::Netting
323    }
324
325    fn get_account(&self) -> Option<AccountAny> {
326        self.core.cache().account_owned(&self.core.account_id)
327    }
328
329    fn position_reconciliation_tolerance(&self) -> Decimal {
330        crate::common::consts::POSITION_RECONCILIATION_TOLERANCE
331    }
332
333    fn allows_reconciliation_overfill(&self, order: &OrderAny, report: &FillReport) -> bool {
334        order.order_side() == OrderSide::Buy
335            && report.order_side == OrderSide::Buy
336            && !order.is_quote_quantity()
337            && order
338                .venue_order_ids()
339                .iter()
340                .any(|venue_order_id| **venue_order_id == report.venue_order_id)
341    }
342
343    // Redemption, including the venue's automatic redemption of winning tokens, removes a Data
344    // API balance without a trade, so a missing balance is not evidence of a flat position.
345    fn provides_bulk_position_coverage(&self, _instrument_id: InstrumentId) -> bool {
346        false
347    }
348
349    fn generate_account_state(
350        &self,
351        balances: Vec<AccountBalance>,
352        margins: Vec<MarginBalance>,
353        reported: bool,
354        ts_event: UnixNanos,
355        info: Option<Params>,
356    ) -> anyhow::Result<()> {
357        self.emitter
358            .emit_account_state(balances, margins, reported, ts_event, info);
359        Ok(())
360    }
361
362    fn start(&mut self) -> anyhow::Result<()> {
363        self.start_client();
364        Ok(())
365    }
366
367    fn stop(&mut self) -> anyhow::Result<()> {
368        self.stop_client();
369        Ok(())
370    }
371
372    fn reset(&mut self) -> anyhow::Result<()> {
373        self.reset_client();
374        Ok(())
375    }
376
377    fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
378        self.settlement.ensure_submission_allowed()?;
379        self.submit_order_command(&cmd)
380    }
381
382    fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
383        self.settlement.ensure_submission_allowed()?;
384        self.submit_order_list_command(&cmd);
385        Ok(())
386    }
387
388    fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
389        self.check_not_faulted()?;
390
391        if let Err(e) = self.settlement.ensure_submission_allowed() {
392            self.emitter.emit_order_modify_rejected_event(
393                cmd.strategy_id,
394                cmd.instrument_id,
395                cmd.client_order_id,
396                cmd.venue_order_id,
397                &e.to_string(),
398                self.clock.get_time_ns(),
399            );
400            return Ok(());
401        }
402
403        self.modify_order_command(&cmd);
404        Ok(())
405    }
406
407    fn cancel_order(&self, cmd: CancelOrder) -> anyhow::Result<()> {
408        self.check_not_faulted()?;
409        self.cancel_order_command(&cmd);
410        Ok(())
411    }
412
413    fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
414        self.check_not_faulted()?;
415        self.cancel_all_orders_command(&cmd)
416    }
417
418    fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
419        self.check_not_faulted()?;
420        self.batch_cancel_orders_command(&cmd);
421        Ok(())
422    }
423
424    fn query_account(&self, cmd: QueryAccount) -> anyhow::Result<()> {
425        self.query_account_command(cmd);
426        Ok(())
427    }
428
429    fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
430        self.query_order_command(&cmd);
431        Ok(())
432    }
433
434    fn retain_unresolved_submissions(&self) -> bool {
435        true
436    }
437
438    fn register_external_order(
439        &self,
440        _client_order_id: ClientOrderId,
441        _venue_order_id: VenueOrderId,
442        _instrument_id: InstrumentId,
443        _strategy_id: StrategyId,
444        _ts_init: UnixNanos,
445    ) {
446    }
447
448    fn on_instrument(&mut self, instrument: InstrumentAny) {
449        self.on_instrument_update(&instrument);
450    }
451
452    fn calculate_commission(
453        &self,
454        instrument: &InstrumentAny,
455        last_qty: Quantity,
456        last_px: Price,
457        liquidity_side: LiquiditySide,
458    ) -> anyhow::Result<Option<Money>> {
459        self.calculate_commission_impl(instrument, last_qty, last_px, liquidity_side)
460            .map(Some)
461    }
462
463    async fn connect(&mut self) -> anyhow::Result<()> {
464        self.connect_client().await
465    }
466
467    async fn disconnect(&mut self) -> anyhow::Result<()> {
468        self.disconnect_client().await
469    }
470
471    async fn generate_order_status_report(
472        &self,
473        cmd: &GenerateOrderStatusReport,
474    ) -> anyhow::Result<Option<OrderStatusReport>> {
475        gate_report(
476            || {
477                self.settlement
478                    .ensure_resolved(cmd.instrument_id, "order status report")
479            },
480            Box::pin(self.generate_order_status_report_impl(cmd)),
481        )
482        .await
483    }
484
485    async fn generate_order_status_reports(
486        &self,
487        cmd: &GenerateOrderStatusReports,
488    ) -> anyhow::Result<Vec<OrderStatusReport>> {
489        gate_report(
490            || {
491                self.settlement
492                    .ensure_resolved(cmd.instrument_id, "order status reports")
493            },
494            self.generate_order_status_reports_impl(cmd),
495        )
496        .await
497    }
498
499    async fn generate_fill_reports(
500        &self,
501        cmd: GenerateFillReports,
502    ) -> anyhow::Result<Vec<FillReport>> {
503        let (instrument_id, venue_order_id) = (cmd.instrument_id, cmd.venue_order_id);
504        gate_report(
505            || match venue_order_id {
506                Some(venue_order_id) => self
507                    .settlement
508                    .ensure_order_resolved(&venue_order_id, "fill reports"),
509                None => self
510                    .settlement
511                    .ensure_resolved(instrument_id, "fill reports"),
512            },
513            self.generate_fill_reports_impl(cmd),
514        )
515        .await
516    }
517
518    async fn generate_position_status_reports(
519        &self,
520        cmd: &GeneratePositionStatusReports,
521    ) -> anyhow::Result<Vec<PositionStatusReport>> {
522        gate_report(
523            || {
524                self.settlement
525                    .ensure_resolved(cmd.instrument_id, "position status reports")
526            },
527            self.generate_position_status_reports_impl(cmd),
528        )
529        .await
530    }
531
532    async fn generate_mass_status(
533        &self,
534        lookback_mins: Option<u64>,
535    ) -> anyhow::Result<Option<ExecutionMassStatus>> {
536        gate_report(
537            || self.settlement.ensure_resolved(None, "mass status"),
538            self.generate_mass_status_impl(lookback_mins),
539        )
540        .await
541    }
542}
543
544/// Runs a report's venue reads between two settlement gate checks, so evidence that becomes
545/// unresolved while the reads are in flight still fails the report.
546async fn gate_report<T>(
547    gate: impl Fn() -> anyhow::Result<()>,
548    report: impl Future<Output = anyhow::Result<T>>,
549) -> anyhow::Result<T> {
550    gate()?;
551    let report = report.await?;
552    gate()?;
553    Ok(report)
554}
555
556#[cfg(test)]
557mod tests {
558    use rstest::rstest;
559
560    use super::*;
561
562    #[rstest]
563    #[tokio::test]
564    async fn test_gate_report_rechecks_settlement_after_venue_reads() {
565        let settlement = SettlementRegistry::new(AccountId::from("POLYMARKET-001"));
566
567        let result = gate_report(|| settlement.ensure_resolved(None, "mass status"), async {
568            settlement.quarantine_invalid_trade("trade-quarantined-during-read");
569            Ok(())
570        })
571        .await;
572
573        assert_eq!(
574            result.unwrap_err().to_string(),
575            "cannot generate mass status: Polymarket settlement registry holds 1 record(s) with \
576             unresolved evidence"
577        );
578    }
579
580    #[rstest]
581    #[case(PolymarketSignatureType::PolyProxy)]
582    #[case(PolymarketSignatureType::PolyGnosisSafe)]
583    #[case(PolymarketSignatureType::Poly1271)]
584    fn proxy_signature_types_require_funder(#[case] signature_type: PolymarketSignatureType) {
585        let error = resolve_maker_address(signature_type, "0xsigner", None).unwrap_err();
586
587        assert!(
588            error
589                .to_string()
590                .contains("requires a funder wallet address")
591        );
592    }
593
594    #[rstest]
595    #[case(PolymarketSignatureType::PolyProxy)]
596    #[case(PolymarketSignatureType::PolyGnosisSafe)]
597    #[case(PolymarketSignatureType::Poly1271)]
598    fn proxy_signature_types_require_distinct_funder(
599        #[case] signature_type: PolymarketSignatureType,
600    ) {
601        let error =
602            resolve_maker_address(signature_type, "0xsigner", Some("0xSIGNER")).unwrap_err();
603
604        assert!(error.to_string().contains("requires a funder distinct"));
605    }
606
607    #[rstest]
608    #[case(None, "0xsigner")]
609    #[case(Some("0xfunder"), "0xfunder")]
610    fn eoa_uses_configured_funder_or_signer(#[case] funder: Option<&str>, #[case] expected: &str) {
611        let maker_address =
612            resolve_maker_address(PolymarketSignatureType::Eoa, "0xsigner", funder).unwrap();
613
614        assert_eq!(maker_address, expected);
615    }
616
617    #[rstest]
618    fn test_execution_client_rejects_duplicate_config_before_setup() {
619        use std::{cell::RefCell, rc::Rc};
620
621        use nautilus_common::{cache::Cache, config::ConfigError};
622        use nautilus_model::{
623            enums::{AccountType, OmsType},
624            identifiers::{TraderId, Venue},
625        };
626
627        use crate::config::PolymarketInstrumentProviderConfig;
628
629        let core = ExecutionClientCore::new(
630            TraderId::from("TESTER-001"),
631            ClientId::from("POLYMARKET"),
632            Venue::from("POLYMARKET"),
633            OmsType::Netting,
634            AccountId::from("POLYMARKET-001"),
635            AccountType::Cash,
636            None,
637            Rc::new(RefCell::new(Cache::default())),
638        );
639
640        let config = PolymarketExecutionClientConfig {
641            instrument_config: Some(PolymarketInstrumentProviderConfig {
642                series_ids: Some(vec![1, 1]),
643                ..Default::default()
644            }),
645            ..Default::default()
646        };
647
648        let error = PolymarketExecutionClient::new(core, config)
649            .expect_err("duplicate config must fail before credentials or client setup");
650        assert_eq!(
651            error.downcast_ref::<ConfigError>(),
652            Some(&ConfigError::duplicate("series_ids", None))
653        );
654    }
655}
nautilus_polymarket/execution/mod.rs

nautilus_polymarket/execution/
mod.rs
