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 credentials cannot be resolved or clients fail to construct.
131    pub fn new(
132        core: ExecutionClientCore,
133        config: PolymarketExecutionClientConfig,
134    ) -> anyhow::Result<Self> {
135        let proxy_url = config.validated_proxy_url()?;
136        config.validate_signer()?;
137        let secrets = Secrets::resolve(
138            config.private_key.clone(),
139            config.api_key.clone(),
140            config.api_secret.clone(),
141            config.passphrase.clone(),
142            config.funder.clone(),
143        )
144        .context("failed to resolve Polymarket credentials")?;
145
146        let signer_address = secrets.address.clone();
147        let maker_address = resolve_maker_address(
148            config.signature_type,
149            &signer_address,
150            secrets.funder.as_deref(),
151        )?;
152        let http_client = PolymarketClobHttpClient::new_with_proxy(
153            secrets.credential.clone(),
154            signer_address.clone(),
155            config.base_url_http.clone(),
156            config.http_timeout_secs,
157            proxy_url.clone(),
158        )
159        .map_err(|e| anyhow::anyhow!("{e}"))
160        .context("failed to create CLOB HTTP client")?;
161
162        let data_api_client = PolymarketDataApiHttpClient::new_with_proxy(
163            Some(config.data_api_url()),
164            config.http_timeout_secs,
165            proxy_url.clone(),
166        )
167        .map_err(|e| anyhow::anyhow!("{e}"))
168        .context("failed to create Data API HTTP client")?;
169
170        let order_signer = OrderSigner::new(&secrets.private_key)
171            .context("failed to create order signer")?
172            .with_signer_type(config.signer_type);
173
174        let order_builder = Arc::new(PolymarketOrderBuilder::new(
175            order_signer,
176            signer_address,
177            maker_address,
178            config.signature_type,
179        ));
180
181        let retry_config = RetryConfig {
182            max_retries: config.max_retries,
183            initial_delay_ms: config.retry_delay_initial_ms,
184            max_delay_ms: config.retry_delay_max_ms,
185            backoff_factor: 2.0,
186            jitter_ms: 1_000,
187            operation_timeout_ms: Some(config.http_timeout_secs * 1_000),
188            immediate_first: false,
189            max_elapsed_ms: Some(180_000),
190        };
191
192        let settlement = Arc::new(SettlementRegistry::new(core.account_id));
193
194        let submitter = OrderSubmitter::new(
195            http_client.clone(),
196            order_builder,
197            retry_config,
198            settlement.clone(),
199        );
200
201        let ws_client = PolymarketWebSocketClient::new_user_with_proxy(
202            config.base_url_ws.clone(),
203            secrets.credential.clone(),
204            config.transport_backend,
205            proxy_url,
206        );
207
208        let ws_client = ws_client.with_socket_control(SocketControl::new(
209            core.client_id,
210            Some(*POLYMARKET_VENUE),
211            USER_STREAMS_ENDPOINT,
212        ));
213
214        let clock = get_atomic_clock_realtime();
215        let pusd = get_pusd_currency();
216        let emitter = ExecutionEventEmitter::new(
217            clock,
218            core.trader_id,
219            core.account_id,
220            AccountType::Cash,
221            Some(pusd),
222        );
223
224        let session_tasks = TaskGroup::new();
225        let pending_tasks = TaskGroup::new();
226
227        Ok(Self {
228            core,
229            clock,
230            config,
231            emitter,
232            http_client,
233            data_api_client,
234            submitter,
235            ws_client,
236            secrets,
237            session_tasks,
238            pending_tasks,
239            shutdown_errors: Vec::new(),
240            stopping: Arc::new(AtomicBool::new(false)),
241            heartbeat_healthy: Arc::new(AtomicBool::new(true)),
242            order_event_handler: None,
243            position_event_handler: None,
244            fill_observer: None,
245            void_observer: None,
246            decline_observer: None,
247            shared_token_instruments: Arc::new(AtomicMap::new()),
248            neg_risk_index: Arc::new(AtomicMap::new()),
249            pending_submits: PendingSubmitTracker::default(),
250            pending_cancels: PendingCancelTracker::default(),
251            order_contexts: Arc::new(OrderContextRegistry::default()),
252            collateral_locks: Arc::new(Mutex::new(collateral::CollateralLocks::default())),
253            fill_tracker: Arc::new(OrderFillTrackerMap::new()),
254            settlement,
255            ws_dispatch_state: Arc::new(Mutex::new(WsDispatchState::default())),
256        })
257    }
258
259    fn check_not_faulted(&self) -> anyhow::Result<()> {
260        if let Some(reason) = self.settlement.client_fault_reason() {
261            anyhow::bail!("Polymarket execution client is faulted closed until restart: {reason}");
262        }
263
264        Ok(())
265    }
266}
267
268fn resolve_maker_address(
269    signature_type: PolymarketSignatureType,
270    signer_address: &str,
271    funder: Option<&str>,
272) -> anyhow::Result<String> {
273    let maker_address = match signature_type {
274        PolymarketSignatureType::Eoa => funder.unwrap_or(signer_address),
275        PolymarketSignatureType::PolyProxy
276        | PolymarketSignatureType::PolyGnosisSafe
277        | PolymarketSignatureType::Poly1271 => funder.ok_or_else(|| {
278            anyhow::anyhow!(
279                "Polymarket {signature_type:?} signature type requires a funder wallet address",
280            )
281        })?,
282    };
283
284    if signature_type != PolymarketSignatureType::Eoa
285        && maker_address.eq_ignore_ascii_case(signer_address)
286    {
287        anyhow::bail!(
288            "Polymarket {signature_type:?} signature type requires a funder distinct from the signing address",
289        );
290    }
291
292    Ok(maker_address.to_string())
293}
294
295#[async_trait(?Send)]
296impl ExecutionClient for PolymarketExecutionClient {
297    fn is_connected(&self) -> bool {
298        self.core.is_connected()
299            && !self.settlement.client_faulted()
300            && (!self.config.heartbeat_enabled
301                || self
302                    .heartbeat_healthy
303                    .load(std::sync::atomic::Ordering::Acquire))
304    }
305
306    fn client_id(&self) -> ClientId {
307        self.core.client_id
308    }
309
310    fn account_id(&self) -> AccountId {
311        self.core.account_id
312    }
313
314    fn venue(&self) -> Venue {
315        *POLYMARKET_VENUE
316    }
317
318    fn oms_type(&self) -> OmsType {
319        OmsType::Netting
320    }
321
322    fn get_account(&self) -> Option<AccountAny> {
323        self.core.cache().account_owned(&self.core.account_id)
324    }
325
326    fn position_reconciliation_tolerance(&self) -> Decimal {
327        crate::common::consts::POSITION_RECONCILIATION_TOLERANCE
328    }
329
330    fn allows_reconciliation_overfill(&self, order: &OrderAny, report: &FillReport) -> bool {
331        order.order_side() == OrderSide::Buy
332            && report.order_side == OrderSide::Buy
333            && !order.is_quote_quantity()
334            && order
335                .venue_order_ids()
336                .iter()
337                .any(|venue_order_id| **venue_order_id == report.venue_order_id)
338    }
339
340    // Redemption, including the venue's automatic redemption of winning tokens, removes a Data
341    // API balance without a trade, so a missing balance is not evidence of a flat position.
342    fn provides_bulk_position_coverage(&self, _instrument_id: InstrumentId) -> bool {
343        false
344    }
345
346    fn generate_account_state(
347        &self,
348        balances: Vec<AccountBalance>,
349        margins: Vec<MarginBalance>,
350        reported: bool,
351        ts_event: UnixNanos,
352        info: Option<Params>,
353    ) -> anyhow::Result<()> {
354        self.emitter
355            .emit_account_state(balances, margins, reported, ts_event, info);
356        Ok(())
357    }
358
359    fn start(&mut self) -> anyhow::Result<()> {
360        self.start_client();
361        Ok(())
362    }
363
364    fn stop(&mut self) -> anyhow::Result<()> {
365        self.stop_client();
366        Ok(())
367    }
368
369    fn reset(&mut self) -> anyhow::Result<()> {
370        self.reset_client();
371        Ok(())
372    }
373
374    fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
375        self.check_not_faulted()?;
376        self.submit_order_command(&cmd)
377    }
378
379    fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
380        self.check_not_faulted()?;
381        self.submit_order_list_command(&cmd);
382        Ok(())
383    }
384
385    fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
386        self.check_not_faulted()?;
387        self.modify_order_command(&cmd);
388        Ok(())
389    }
390
391    fn cancel_order(&self, cmd: CancelOrder) -> anyhow::Result<()> {
392        self.check_not_faulted()?;
393        self.cancel_order_command(&cmd);
394        Ok(())
395    }
396
397    fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
398        self.check_not_faulted()?;
399        self.cancel_all_orders_command(&cmd)
400    }
401
402    fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
403        self.check_not_faulted()?;
404        self.batch_cancel_orders_command(&cmd);
405        Ok(())
406    }
407
408    fn query_account(&self, cmd: QueryAccount) -> anyhow::Result<()> {
409        self.query_account_command(cmd);
410        Ok(())
411    }
412
413    fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
414        self.query_order_command(&cmd);
415        Ok(())
416    }
417
418    fn retain_unresolved_submissions(&self) -> bool {
419        true
420    }
421
422    fn register_external_order(
423        &self,
424        _client_order_id: ClientOrderId,
425        _venue_order_id: VenueOrderId,
426        _instrument_id: InstrumentId,
427        _strategy_id: StrategyId,
428        _ts_init: UnixNanos,
429    ) {
430    }
431
432    fn on_instrument(&mut self, instrument: InstrumentAny) {
433        self.on_instrument_update(&instrument);
434    }
435
436    fn calculate_commission(
437        &self,
438        instrument: &InstrumentAny,
439        last_qty: Quantity,
440        last_px: Price,
441        liquidity_side: LiquiditySide,
442    ) -> anyhow::Result<Option<Money>> {
443        self.calculate_commission_impl(instrument, last_qty, last_px, liquidity_side)
444            .map(Some)
445    }
446
447    async fn connect(&mut self) -> anyhow::Result<()> {
448        self.connect_client().await
449    }
450
451    async fn disconnect(&mut self) -> anyhow::Result<()> {
452        self.disconnect_client().await
453    }
454
455    async fn generate_order_status_report(
456        &self,
457        cmd: &GenerateOrderStatusReport,
458    ) -> anyhow::Result<Option<OrderStatusReport>> {
459        gate_report(
460            || {
461                self.settlement
462                    .ensure_resolved(cmd.instrument_id, "order status report")
463            },
464            Box::pin(self.generate_order_status_report_impl(cmd)),
465        )
466        .await
467    }
468
469    async fn generate_order_status_reports(
470        &self,
471        cmd: &GenerateOrderStatusReports,
472    ) -> anyhow::Result<Vec<OrderStatusReport>> {
473        gate_report(
474            || {
475                self.settlement
476                    .ensure_resolved(cmd.instrument_id, "order status reports")
477            },
478            self.generate_order_status_reports_impl(cmd),
479        )
480        .await
481    }
482
483    async fn generate_fill_reports(
484        &self,
485        cmd: GenerateFillReports,
486    ) -> anyhow::Result<Vec<FillReport>> {
487        let (instrument_id, venue_order_id) = (cmd.instrument_id, cmd.venue_order_id);
488        gate_report(
489            || match venue_order_id {
490                Some(venue_order_id) => self
491                    .settlement
492                    .ensure_order_resolved(&venue_order_id, "fill reports"),
493                None => self
494                    .settlement
495                    .ensure_resolved(instrument_id, "fill reports"),
496            },
497            self.generate_fill_reports_impl(cmd),
498        )
499        .await
500    }
501
502    async fn generate_position_status_reports(
503        &self,
504        cmd: &GeneratePositionStatusReports,
505    ) -> anyhow::Result<Vec<PositionStatusReport>> {
506        gate_report(
507            || {
508                self.settlement
509                    .ensure_resolved(cmd.instrument_id, "position status reports")
510            },
511            self.generate_position_status_reports_impl(cmd),
512        )
513        .await
514    }
515
516    async fn generate_mass_status(
517        &self,
518        lookback_mins: Option<u64>,
519    ) -> anyhow::Result<Option<ExecutionMassStatus>> {
520        gate_report(
521            || self.settlement.ensure_resolved(None, "mass status"),
522            self.generate_mass_status_impl(lookback_mins),
523        )
524        .await
525    }
526}
527
528/// Runs a report's venue reads between two settlement gate checks, so evidence that becomes
529/// unresolved while the reads are in flight still fails the report.
530async fn gate_report<T>(
531    gate: impl Fn() -> anyhow::Result<()>,
532    report: impl Future<Output = anyhow::Result<T>>,
533) -> anyhow::Result<T> {
534    gate()?;
535    let report = report.await?;
536    gate()?;
537    Ok(report)
538}
539
540#[cfg(test)]
541mod tests {
542    use rstest::rstest;
543
544    use super::*;
545
546    #[rstest]
547    #[tokio::test]
548    async fn test_gate_report_rechecks_settlement_after_venue_reads() {
549        let settlement = SettlementRegistry::new(AccountId::from("POLYMARKET-001"));
550
551        let result = gate_report(|| settlement.ensure_resolved(None, "mass status"), async {
552            settlement.quarantine_invalid_trade("trade-quarantined-during-read");
553            Ok(())
554        })
555        .await;
556
557        assert_eq!(
558            result.unwrap_err().to_string(),
559            "cannot generate mass status: Polymarket settlement registry holds 1 record(s) with \
560             unresolved evidence"
561        );
562    }
563
564    #[rstest]
565    #[case(PolymarketSignatureType::PolyProxy)]
566    #[case(PolymarketSignatureType::PolyGnosisSafe)]
567    #[case(PolymarketSignatureType::Poly1271)]
568    fn proxy_signature_types_require_funder(#[case] signature_type: PolymarketSignatureType) {
569        let error = resolve_maker_address(signature_type, "0xsigner", None).unwrap_err();
570
571        assert!(
572            error
573                .to_string()
574                .contains("requires a funder wallet address")
575        );
576    }
577
578    #[rstest]
579    #[case(PolymarketSignatureType::PolyProxy)]
580    #[case(PolymarketSignatureType::PolyGnosisSafe)]
581    #[case(PolymarketSignatureType::Poly1271)]
582    fn proxy_signature_types_require_distinct_funder(
583        #[case] signature_type: PolymarketSignatureType,
584    ) {
585        let error =
586            resolve_maker_address(signature_type, "0xsigner", Some("0xSIGNER")).unwrap_err();
587
588        assert!(error.to_string().contains("requires a funder distinct"));
589    }
590
591    #[rstest]
592    #[case(None, "0xsigner")]
593    #[case(Some("0xfunder"), "0xfunder")]
594    fn eoa_uses_configured_funder_or_signer(#[case] funder: Option<&str>, #[case] expected: &str) {
595        let maker_address =
596            resolve_maker_address(PolymarketSignatureType::Eoa, "0xsigner", funder).unwrap();
597
598        assert_eq!(maker_address, expected);
599    }
600}
nautilus_polymarket/execution/mod.rs

nautilus_polymarket/execution/
mod.rs
