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 identity;
22pub(crate) mod order_fill_tracker;
23pub(crate) mod pending;
24pub(crate) mod reconciliation;
25pub(crate) mod submitter;
26pub(crate) mod types;
27
28mod cancellations;
29mod lifecycle;
30mod orders;
31mod reports;
32mod responses;
33
34use std::sync::{Arc, atomic::AtomicBool};
35
36use anyhow::Context;
37use async_trait::async_trait;
38use nautilus_common::{
39    clients::ExecutionClient,
40    messages::execution::{
41        BatchCancelOrders, CancelAllOrders, CancelOrder, GenerateFillReports,
42        GenerateOrderStatusReport, GenerateOrderStatusReports, GeneratePositionStatusReports,
43        ModifyOrder, QueryAccount, QueryOrder, SubmitOrder, SubmitOrderList,
44    },
45    msgbus::TypedHandler,
46};
47use nautilus_core::{
48    Params, UnixNanos,
49    collections::AtomicMap,
50    time::{AtomicTime, get_atomic_clock_realtime},
51};
52use nautilus_live::{ExecutionClientCore, ExecutionEventEmitter, SocketControl, task::TaskGroup};
53use nautilus_model::{
54    accounts::AccountAny,
55    enums::{AccountType, LiquiditySide, OmsType},
56    events::{OrderEventAny, PositionEvent},
57    identifiers::{
58        AccountId, ClientId, ClientOrderId, InstrumentId, StrategyId, Venue, VenueOrderId,
59    },
60    instruments::InstrumentAny,
61    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
62    types::{AccountBalance, MarginBalance, Money, Price, Quantity},
63};
64use nautilus_network::retry::RetryConfig;
65use parking_lot::Mutex;
66pub(crate) use responses::is_post_only_crossing;
67use rust_decimal::Decimal;
68use ustr::Ustr;
69
70pub(crate) use self::reports::get_pusd_currency;
71use self::{
72    identity::OrderIdentityRegistry,
73    order_builder::PolymarketOrderBuilder,
74    order_fill_tracker::OrderFillTrackerMap,
75    pending::{PendingCancelTracker, PendingSubmitTracker},
76    submitter::OrderSubmitter,
77};
78use crate::{
79    common::{consts::POLYMARKET_VENUE, credential::Secrets, enums::SignatureType},
80    config::PolymarketExecutionClientConfig,
81    http::{clob::PolymarketClobHttpClient, data_api::PolymarketDataApiHttpClient},
82    signing::eip712::OrderSigner,
83    websocket::{
84        USER_STREAMS_ENDPOINT, client::PolymarketWebSocketClient, dispatch::WsDispatchState,
85    },
86};
87
88/// Live execution client for the Polymarket prediction market.
89#[derive(Debug)]
90pub struct PolymarketExecutionClient {
91    core: ExecutionClientCore,
92    clock: &'static AtomicTime,
93    config: PolymarketExecutionClientConfig,
94    emitter: ExecutionEventEmitter,
95    http_client: PolymarketClobHttpClient,
96    data_api_client: PolymarketDataApiHttpClient,
97    submitter: OrderSubmitter,
98    ws_client: PolymarketWebSocketClient,
99    secrets: Secrets,
100    session_tasks: TaskGroup,
101    pending_tasks: TaskGroup,
102    shutdown_errors: Vec<String>,
103    stopping: Arc<AtomicBool>,
104    heartbeat_healthy: Arc<AtomicBool>,
105    order_event_handler: Option<TypedHandler<OrderEventAny>>,
106    position_event_handler: Option<TypedHandler<PositionEvent>>,
107    shared_token_instruments: Arc<AtomicMap<Ustr, InstrumentAny>>,
108    neg_risk_index: Arc<AtomicMap<InstrumentId, bool>>,
109    pending_submits: PendingSubmitTracker,
110    pending_cancels: PendingCancelTracker,
111    order_identities: Arc<OrderIdentityRegistry>,
112    fill_tracker: Arc<OrderFillTrackerMap>,
113    ws_dispatch_state: Arc<Mutex<WsDispatchState>>,
114}
115
116impl PolymarketExecutionClient {
117    /// Creates a new [`PolymarketExecutionClient`].
118    ///
119    /// # Errors
120    ///
121    /// Returns an error if credentials cannot be resolved or clients fail to construct.
122    pub fn new(
123        core: ExecutionClientCore,
124        config: PolymarketExecutionClientConfig,
125    ) -> anyhow::Result<Self> {
126        let proxy_url = config.validated_proxy_url()?;
127        let secrets = Secrets::resolve(
128            config.private_key.as_deref(),
129            config.api_key.clone(),
130            config.api_secret.clone(),
131            config.passphrase.clone(),
132            config.funder.clone(),
133        )
134        .context("failed to resolve Polymarket credentials")?;
135
136        let signer_address = secrets.address.clone();
137        let maker_address = resolve_maker_address(
138            config.signature_type,
139            &signer_address,
140            secrets.funder.as_deref(),
141        )?;
142        let http_client = PolymarketClobHttpClient::new_with_proxy(
143            secrets.credential.clone(),
144            signer_address.clone(),
145            config.base_url_http.clone(),
146            config.http_timeout_secs,
147            proxy_url.clone(),
148        )
149        .map_err(|e| anyhow::anyhow!("{e}"))
150        .context("failed to create CLOB HTTP client")?;
151
152        let data_api_client = PolymarketDataApiHttpClient::new_with_proxy(
153            Some(config.data_api_url()),
154            config.http_timeout_secs,
155            proxy_url.clone(),
156        )
157        .map_err(|e| anyhow::anyhow!("{e}"))
158        .context("failed to create Data API HTTP client")?;
159
160        let order_signer =
161            OrderSigner::new(&secrets.private_key).context("failed to create order signer")?;
162        let order_builder = Arc::new(PolymarketOrderBuilder::new(
163            order_signer,
164            signer_address,
165            maker_address,
166            config.signature_type,
167        ));
168
169        let retry_config = RetryConfig {
170            max_retries: config.max_retries,
171            initial_delay_ms: config.retry_delay_initial_ms,
172            max_delay_ms: config.retry_delay_max_ms,
173            backoff_factor: 2.0,
174            jitter_ms: 1_000,
175            operation_timeout_ms: Some(config.http_timeout_secs * 1_000),
176            immediate_first: false,
177            max_elapsed_ms: Some(180_000),
178        };
179        let submitter = OrderSubmitter::new(http_client.clone(), order_builder, retry_config);
180
181        let ws_client = PolymarketWebSocketClient::new_user_with_proxy(
182            config.base_url_ws.clone(),
183            secrets.credential.clone(),
184            config.transport_backend,
185            proxy_url,
186        );
187
188        let ws_client = ws_client.with_socket_control(SocketControl::new(
189            core.client_id,
190            Some(*POLYMARKET_VENUE),
191            USER_STREAMS_ENDPOINT,
192        ));
193
194        let clock = get_atomic_clock_realtime();
195        let pusd = get_pusd_currency();
196        let emitter = ExecutionEventEmitter::new(
197            clock,
198            core.trader_id,
199            core.account_id,
200            AccountType::Cash,
201            Some(pusd),
202        );
203
204        let session_tasks = TaskGroup::new();
205        let pending_tasks = TaskGroup::new();
206
207        Ok(Self {
208            core,
209            clock,
210            config,
211            emitter,
212            http_client,
213            data_api_client,
214            submitter,
215            ws_client,
216            secrets,
217            session_tasks,
218            pending_tasks,
219            shutdown_errors: Vec::new(),
220            stopping: Arc::new(AtomicBool::new(false)),
221            heartbeat_healthy: Arc::new(AtomicBool::new(true)),
222            order_event_handler: None,
223            position_event_handler: None,
224            shared_token_instruments: Arc::new(AtomicMap::new()),
225            neg_risk_index: Arc::new(AtomicMap::new()),
226            pending_submits: PendingSubmitTracker::default(),
227            pending_cancels: PendingCancelTracker::default(),
228            order_identities: Arc::new(OrderIdentityRegistry::default()),
229            fill_tracker: Arc::new(OrderFillTrackerMap::new()),
230            ws_dispatch_state: Arc::new(Mutex::new(WsDispatchState::default())),
231        })
232    }
233}
234
235fn resolve_maker_address(
236    signature_type: SignatureType,
237    signer_address: &str,
238    funder: Option<&str>,
239) -> anyhow::Result<String> {
240    let maker_address = match signature_type {
241        SignatureType::Eoa => funder.unwrap_or(signer_address),
242        SignatureType::PolyProxy | SignatureType::PolyGnosisSafe | SignatureType::Poly1271 => {
243            funder.ok_or_else(|| {
244                anyhow::anyhow!(
245                    "Polymarket {signature_type:?} signature type requires a funder wallet address",
246                )
247            })?
248        }
249    };
250
251    if signature_type != SignatureType::Eoa && maker_address.eq_ignore_ascii_case(signer_address) {
252        anyhow::bail!(
253            "Polymarket {signature_type:?} signature type requires a funder distinct from the signing address",
254        );
255    }
256
257    Ok(maker_address.to_string())
258}
259
260#[async_trait(?Send)]
261impl ExecutionClient for PolymarketExecutionClient {
262    fn is_connected(&self) -> bool {
263        self.core.is_connected()
264            && (!self.config.heartbeat_enabled
265                || self
266                    .heartbeat_healthy
267                    .load(std::sync::atomic::Ordering::Acquire))
268    }
269
270    fn client_id(&self) -> ClientId {
271        self.core.client_id
272    }
273
274    fn account_id(&self) -> AccountId {
275        self.core.account_id
276    }
277
278    fn venue(&self) -> Venue {
279        *POLYMARKET_VENUE
280    }
281
282    fn oms_type(&self) -> OmsType {
283        OmsType::Netting
284    }
285
286    fn get_account(&self) -> Option<AccountAny> {
287        self.core.cache().account_owned(&self.core.account_id)
288    }
289
290    fn position_reconciliation_tolerance(&self) -> Decimal {
291        crate::common::consts::POSITION_RECONCILIATION_TOLERANCE
292    }
293
294    fn generate_account_state(
295        &self,
296        balances: Vec<AccountBalance>,
297        margins: Vec<MarginBalance>,
298        reported: bool,
299        ts_event: UnixNanos,
300        info: Option<Params>,
301    ) -> anyhow::Result<()> {
302        self.emitter
303            .emit_account_state(balances, margins, reported, ts_event, info);
304        Ok(())
305    }
306
307    fn start(&mut self) -> anyhow::Result<()> {
308        self.start_client();
309        Ok(())
310    }
311
312    fn stop(&mut self) -> anyhow::Result<()> {
313        self.stop_client();
314        Ok(())
315    }
316
317    fn reset(&mut self) -> anyhow::Result<()> {
318        self.reset_client();
319        Ok(())
320    }
321
322    fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
323        self.submit_order_command(&cmd)
324    }
325
326    fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
327        self.submit_order_list_command(&cmd);
328        Ok(())
329    }
330
331    fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
332        self.modify_order_command(&cmd);
333        Ok(())
334    }
335
336    fn cancel_order(&self, cmd: CancelOrder) -> anyhow::Result<()> {
337        self.cancel_order_command(&cmd);
338        Ok(())
339    }
340
341    fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
342        self.cancel_all_orders_command(&cmd)
343    }
344
345    fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
346        self.batch_cancel_orders_command(&cmd);
347        Ok(())
348    }
349
350    fn query_account(&self, cmd: QueryAccount) -> anyhow::Result<()> {
351        self.query_account_command(cmd);
352        Ok(())
353    }
354
355    fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
356        self.query_order_command(&cmd);
357        Ok(())
358    }
359
360    fn register_external_order(
361        &self,
362        _client_order_id: ClientOrderId,
363        _venue_order_id: VenueOrderId,
364        _instrument_id: InstrumentId,
365        _strategy_id: StrategyId,
366        _ts_init: UnixNanos,
367    ) {
368    }
369
370    fn on_instrument(&mut self, instrument: InstrumentAny) {
371        self.on_instrument_update(&instrument);
372    }
373
374    fn calculate_commission(
375        &self,
376        instrument: &InstrumentAny,
377        last_qty: Quantity,
378        last_px: Price,
379        liquidity_side: LiquiditySide,
380    ) -> anyhow::Result<Option<Money>> {
381        self.calculate_commission_impl(instrument, last_qty, last_px, liquidity_side)
382            .map(Some)
383    }
384
385    async fn connect(&mut self) -> anyhow::Result<()> {
386        self.connect_client().await
387    }
388
389    async fn disconnect(&mut self) -> anyhow::Result<()> {
390        self.disconnect_client().await
391    }
392
393    async fn generate_order_status_report(
394        &self,
395        cmd: &GenerateOrderStatusReport,
396    ) -> anyhow::Result<Option<OrderStatusReport>> {
397        self.generate_order_status_report_impl(cmd).await
398    }
399
400    async fn generate_order_status_reports(
401        &self,
402        cmd: &GenerateOrderStatusReports,
403    ) -> anyhow::Result<Vec<OrderStatusReport>> {
404        self.generate_order_status_reports_impl(cmd).await
405    }
406
407    async fn generate_fill_reports(
408        &self,
409        cmd: GenerateFillReports,
410    ) -> anyhow::Result<Vec<FillReport>> {
411        self.generate_fill_reports_impl(cmd).await
412    }
413
414    async fn generate_position_status_reports(
415        &self,
416        cmd: &GeneratePositionStatusReports,
417    ) -> anyhow::Result<Vec<PositionStatusReport>> {
418        self.generate_position_status_reports_impl(cmd).await
419    }
420
421    async fn generate_mass_status(
422        &self,
423        lookback_mins: Option<u64>,
424    ) -> anyhow::Result<Option<ExecutionMassStatus>> {
425        self.generate_mass_status_impl(lookback_mins).await
426    }
427}
428
429#[cfg(test)]
430mod tests {
431    use rstest::rstest;
432
433    use super::*;
434
435    #[rstest]
436    #[case(SignatureType::PolyProxy)]
437    #[case(SignatureType::PolyGnosisSafe)]
438    #[case(SignatureType::Poly1271)]
439    fn proxy_signature_types_require_funder(#[case] signature_type: SignatureType) {
440        let error = resolve_maker_address(signature_type, "0xsigner", None).unwrap_err();
441
442        assert!(
443            error
444                .to_string()
445                .contains("requires a funder wallet address")
446        );
447    }
448
449    #[rstest]
450    #[case(SignatureType::PolyProxy)]
451    #[case(SignatureType::PolyGnosisSafe)]
452    #[case(SignatureType::Poly1271)]
453    fn proxy_signature_types_require_distinct_funder(#[case] signature_type: SignatureType) {
454        let error =
455            resolve_maker_address(signature_type, "0xsigner", Some("0xSIGNER")).unwrap_err();
456
457        assert!(error.to_string().contains("requires a funder distinct"));
458    }
459
460    #[rstest]
461    #[case(None, "0xsigner")]
462    #[case(Some("0xfunder"), "0xfunder")]
463    fn eoa_uses_configured_funder_or_signer(#[case] funder: Option<&str>, #[case] expected: &str) {
464        let maker_address = resolve_maker_address(SignatureType::Eoa, "0xsigner", funder).unwrap();
465
466        assert_eq!(maker_address, expected);
467    }
468}
nautilus_polymarket/execution/mod.rs

nautilus_polymarket/execution/
mod.rs
