1use std::{any::Any, cell::RefCell, rc::Rc, sync::Arc};
19
20use nautilus_common::{
21 cache::CacheView,
22 clients::{DataClient, ExecutionClient},
23 clock::Clock,
24 factories::{ClientConfig, DataClientFactory, ExecutionClientFactory},
25};
26use nautilus_live::ExecutionClientCore;
27use nautilus_model::{
28 enums::{AccountType, OmsType},
29 identifiers::{ClientId, TraderId},
30};
31use nautilus_network::retry::RetryConfig;
32#[cfg(test)]
33use nautilus_network::websocket::proxy::ProxyUrl;
34
35use crate::{
36 common::consts::{POLYMARKET, POLYMARKET_VENUE},
37 config::{PolymarketDataClientConfig, PolymarketExecutionClientConfig},
38 data::PolymarketDataClient,
39 execution::PolymarketExecutionClient,
40 http::{
41 clob::PolymarketClobPublicClient, data_api::PolymarketDataApiHttpClient,
42 gamma::PolymarketGammaHttpClient,
43 },
44 websocket::pool::PolymarketMarketConnectionPool,
45};
46
47impl ClientConfig for PolymarketDataClientConfig {
48 fn as_any(&self) -> &dyn Any {
49 self
50 }
51}
52
53#[cfg_attr(
55 feature = "python",
56 pyo3::pyclass(module = "nautilus_trader.adapters.polymarket", from_py_object)
57)]
58#[cfg_attr(
59 feature = "python",
60 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.polymarket")
61)]
62#[derive(Debug, Clone)]
63pub struct PolymarketDataClientFactory;
64
65impl DataClientFactory for PolymarketDataClientFactory {
66 fn create(
67 &self,
68 name: &str,
69 config: &dyn ClientConfig,
70 _cache: CacheView,
71 _clock: Rc<RefCell<dyn Clock>>,
72 ) -> anyhow::Result<Box<dyn DataClient>> {
73 let polymarket_config = config
74 .as_any()
75 .downcast_ref::<PolymarketDataClientConfig>()
76 .ok_or_else(|| {
77 anyhow::anyhow!(
78 "Invalid config type for PolymarketDataClientFactory. Expected PolymarketDataClientConfig, was {config:?}",
79 )
80 })?;
81
82 Ok(Box::new(Self::create_client(name, polymarket_config)?))
83 }
84
85 fn name(&self) -> &'static str {
86 POLYMARKET
87 }
88
89 fn config_type(&self) -> &'static str {
90 "PolymarketDataClientConfig"
91 }
92}
93
94impl PolymarketDataClientFactory {
95 fn create_client(
96 name: &str,
97 polymarket_config: &PolymarketDataClientConfig,
98 ) -> anyhow::Result<PolymarketDataClient> {
99 polymarket_config.validate()?;
100
101 let client_id = ClientId::from(name);
102 let proxy_url = polymarket_config.validated_proxy_url()?;
103 polymarket_config.polybolt_credential()?;
105
106 let gamma_client = PolymarketGammaHttpClient::new_with_proxy(
107 Some(polymarket_config.gamma_url()),
108 polymarket_config.http_timeout_secs,
109 RetryConfig {
110 max_retries: 10,
111 initial_delay_ms: 5_000,
112 max_delay_ms: 30_000,
113 backoff_factor: 1.5,
114 jitter_ms: 2_000,
115 operation_timeout_ms: Some(30_000),
116 immediate_first: true,
117 max_elapsed_ms: Some(300_000),
118 },
119 proxy_url.clone(),
120 )?;
121
122 let clob_public_client = PolymarketClobPublicClient::new_with_proxy(
123 polymarket_config.base_url_http.clone(),
124 polymarket_config.http_timeout_secs,
125 proxy_url.clone(),
126 )?;
127
128 let data_api_client = PolymarketDataApiHttpClient::new_with_proxy(
129 Some(polymarket_config.data_api_url()),
130 polymarket_config.http_timeout_secs,
131 proxy_url.clone(),
132 )?;
133
134 let ws_client = PolymarketMarketConnectionPool::new_with_proxy(
135 polymarket_config.base_url_ws.clone(),
136 polymarket_config.subscribe_new_markets,
137 polymarket_config.transport_backend,
138 polymarket_config.ws_max_subscriptions,
139 proxy_url.clone(),
140 );
141
142 let mut client = PolymarketDataClient::new_with_proxy(
143 client_id,
144 polymarket_config.clone(),
145 gamma_client,
146 clob_public_client,
147 data_api_client,
148 ws_client,
149 proxy_url,
150 )?;
151
152 for filter in &polymarket_config.filters {
153 client.add_instrument_filter(Arc::clone(filter));
154 }
155
156 Ok(client)
157 }
158}
159
160impl ClientConfig for PolymarketExecutionClientConfig {
161 fn as_any(&self) -> &dyn Any {
162 self
163 }
164}
165
166#[cfg_attr(
168 feature = "python",
169 pyo3::pyclass(module = "nautilus_trader.adapters.polymarket", from_py_object)
170)]
171#[cfg_attr(
172 feature = "python",
173 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.polymarket")
174)]
175#[derive(Debug, Clone)]
176pub struct PolymarketExecutionClientFactory;
177
178impl ExecutionClientFactory for PolymarketExecutionClientFactory {
179 fn create(
180 &self,
181 trader_id: TraderId,
182 name: &str,
183 config: &dyn ClientConfig,
184 cache: CacheView,
185 _clock: Rc<RefCell<dyn Clock>>,
186 ) -> anyhow::Result<Box<dyn ExecutionClient>> {
187 let polymarket_config = config
188 .as_any()
189 .downcast_ref::<PolymarketExecutionClientConfig>()
190 .ok_or_else(|| {
191 anyhow::anyhow!(
192 "Invalid config type for PolymarketExecutionClientFactory. Expected PolymarketExecutionClientConfig, was {config:?}",
193 )
194 })?
195 .clone();
196
197 let oms_type = OmsType::Netting;
198 let account_type = AccountType::Cash;
199
200 let client_id = ClientId::from(name);
201 let core = ExecutionClientCore::new(
202 trader_id,
203 client_id,
204 *POLYMARKET_VENUE,
205 oms_type,
206 polymarket_config.account_id,
207 account_type,
208 None, cache,
210 );
211
212 let client = PolymarketExecutionClient::new(core, polymarket_config)?;
213
214 Ok(Box::new(client))
215 }
216
217 fn name(&self) -> &'static str {
218 POLYMARKET
219 }
220
221 fn config_type(&self) -> &'static str {
222 "PolymarketExecutionClientConfig"
223 }
224}
225
226#[cfg(test)]
227pub(crate) async fn spawn_rejecting_proxy(
228 connection_count: usize,
229) -> (
230 std::net::SocketAddr,
231 std::sync::Arc<tokio::sync::Mutex<Vec<String>>>,
232) {
233 use tokio::io::{AsyncReadExt, AsyncWriteExt};
234
235 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
236 let addr = listener.local_addr().unwrap();
237 let requests = std::sync::Arc::new(tokio::sync::Mutex::new(Vec::new()));
238 let captured = std::sync::Arc::clone(&requests);
239
240 tokio::spawn(async move {
241 for _ in 0..connection_count {
242 let (mut stream, _) = listener.accept().await.unwrap();
243 let mut request = Vec::new();
244 let mut chunk = [0u8; 1024];
245 loop {
246 let read = stream.read(&mut chunk).await.unwrap();
247 if read == 0 {
248 break;
249 }
250 request.extend_from_slice(&chunk[..read]);
251 if request.windows(4).any(|window| window == b"\r\n\r\n") {
252 break;
253 }
254 }
255 captured
256 .lock()
257 .await
258 .push(String::from_utf8(request).unwrap());
259 stream
260 .write_all(
261 b"HTTP/1.1 407 Proxy Authentication Required\r\nContent-Length: 0\r\n\r\n",
262 )
263 .await
264 .unwrap();
265 }
266 });
267
268 (addr, requests)
269}
270
271#[cfg(test)]
272mod tests {
273 use std::{cell::RefCell, rc::Rc};
274
275 use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
276 use nautilus_common::{
277 cache::Cache,
278 clock::VirtualClock,
279 config::ConfigError,
280 factories::{ClientConfig, DataClientFactory, ExecutionClientFactory},
281 live::runner::replace_data_event_sender,
282 messages::DataEvent,
283 };
284 use rstest::rstest;
285
286 use super::*;
287 use crate::{
288 common::credential::Credential,
289 config::{PolymarketDataClientConfig, PolymarketExecutionClientConfig},
290 http::clob::PolymarketClobHttpClient,
291 };
292
293 #[derive(Debug)]
294 struct WrongConfig;
295
296 impl ClientConfig for WrongConfig {
297 fn as_any(&self) -> &dyn std::any::Any {
298 self
299 }
300 }
301
302 #[rstest]
303 fn test_polymarket_data_client_factory_creation() {
304 let factory = PolymarketDataClientFactory;
305 assert_eq!(factory.name(), POLYMARKET);
306 assert_eq!(factory.config_type(), "PolymarketDataClientConfig");
307 }
308
309 #[rstest]
310 fn test_polymarket_data_client_config_implements_client_config() {
311 let config = PolymarketDataClientConfig::default();
312 let boxed_config: Box<dyn ClientConfig> = Box::new(config);
313 let downcasted = boxed_config
314 .as_any()
315 .downcast_ref::<PolymarketDataClientConfig>();
316 assert!(downcasted.is_some());
317 }
318
319 #[rstest]
320 fn test_polymarket_data_client_factory_rejects_wrong_config_type() {
321 let factory = PolymarketDataClientFactory;
322 let wrong_config = WrongConfig;
323 let cache = Rc::new(RefCell::new(Cache::default()));
324 let clock = Rc::new(RefCell::new(VirtualClock::new()));
325
326 let result = factory.create(POLYMARKET, &wrong_config, cache.into(), clock);
327 assert!(result.is_err());
328 assert!(
329 result
330 .err()
331 .unwrap()
332 .to_string()
333 .contains("Invalid config type")
334 );
335 }
336
337 #[rstest]
338 fn test_polymarket_execution_client_factory_creation() {
339 let factory = PolymarketExecutionClientFactory;
340 assert_eq!(factory.name(), POLYMARKET);
341 assert_eq!(factory.config_type(), "PolymarketExecutionClientConfig");
342 }
343
344 #[rstest]
345 fn test_polymarket_exec_client_config_implements_client_config() {
346 let config = PolymarketExecutionClientConfig::default();
347 let boxed_config: Box<dyn ClientConfig> = Box::new(config);
348 let downcasted = boxed_config
349 .as_any()
350 .downcast_ref::<PolymarketExecutionClientConfig>();
351 assert!(downcasted.is_some());
352 }
353
354 #[rstest]
355 fn test_polymarket_execution_client_factory_rejects_wrong_config_type() {
356 let factory = PolymarketExecutionClientFactory;
357 let wrong_config = WrongConfig;
358 let cache = Rc::new(RefCell::new(Cache::default()));
359
360 let result = factory.create(
361 TraderId::from("TRADER-001"),
362 POLYMARKET,
363 &wrong_config,
364 cache.into(),
365 Rc::new(RefCell::new(VirtualClock::new())),
366 );
367 assert!(result.is_err());
368 assert!(
369 result
370 .err()
371 .unwrap()
372 .to_string()
373 .contains("Invalid config type")
374 );
375 }
376
377 #[rstest]
378 fn data_factory_invalid_proxy_error_redacts_credentials() {
379 const SECRET: &str = "data-factory-proxy-secret";
380 let factory = PolymarketDataClientFactory;
381 let config = PolymarketDataClientConfig {
382 proxy_url: Some(format!("http://proxy-user:{SECRET}@[::1").into()),
383 ..PolymarketDataClientConfig::default()
384 };
385 let cache = Rc::new(RefCell::new(Cache::default()));
386 let clock = Rc::new(RefCell::new(VirtualClock::new()));
387 let Err(e) = factory.create(POLYMARKET, &config, cache.into(), clock) else {
388 panic!("malformed proxy URL should fail");
389 };
390
391 assert!(!e.to_string().contains(SECRET));
392 }
393
394 #[rstest]
395 fn execution_factory_invalid_proxy_error_redacts_credentials() {
396 const SECRET: &str = "execution-factory-proxy-secret";
397 let factory = PolymarketExecutionClientFactory;
398 let config = PolymarketExecutionClientConfig {
399 proxy_url: Some(format!("http://proxy-user:{SECRET}@[::1").into()),
400 ..PolymarketExecutionClientConfig::default()
401 };
402 let cache = Rc::new(RefCell::new(Cache::default()));
403 let Err(e) = factory.create(
404 TraderId::from("TRADER-001"),
405 POLYMARKET,
406 &config,
407 cache.into(),
408 Rc::new(RefCell::new(VirtualClock::new())),
409 ) else {
410 panic!("malformed proxy URL should fail");
411 };
412
413 assert!(!e.to_string().contains(SECRET));
414 }
415
416 #[rstest]
417 #[tokio::test]
418 async fn data_factory_propagates_configured_proxy() {
419 const USERNAME: &str = "proxytest";
420 const SECRET: &str = "http-client-proxy-secret";
421 let (proxy_addr, requests) = spawn_rejecting_proxy(3).await;
422 let proxy_url = format!("http://{USERNAME}:{SECRET}@{proxy_addr}");
423 let (data_tx, _data_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
424 replace_data_event_sender(data_tx);
425 let config = PolymarketDataClientConfig {
426 base_url_http: Some("https://clob-public.fixture".to_string()),
427 base_url_ws: Some("wss://market.fixture/ws".to_string()),
428 base_url_gamma: Some("https://gamma.fixture".to_string()),
429 base_url_data_api: Some("https://data.fixture".to_string()),
430 base_url_reference_prices: Some("wss://polybolt.fixture/ws".to_string()),
431 proxy_url: Some(proxy_url.clone().into()),
432 http_timeout_secs: 2,
433 ..PolymarketDataClientConfig::default()
434 };
435 let client = PolymarketDataClientFactory::create_client(POLYMARKET, &config).unwrap();
436
437 let errors = [
438 client
439 .provider()
440 .http_client()
441 .request_tags()
442 .await
443 .unwrap_err()
444 .to_string(),
445 client
446 .clob_public_client()
447 .get_book("public-token")
448 .await
449 .unwrap_err()
450 .to_string(),
451 client
452 .data_api_client()
453 .get_positions("0x0000000000000000000000000000000000000002")
454 .await
455 .unwrap_err()
456 .to_string(),
457 ];
458 let configured_proxy = client.config().validated_proxy_url().unwrap().unwrap();
459
460 assert_eq!(client.ws_client().proxy_url().unwrap().expose(), proxy_url);
461 assert_eq!(
462 client.polybolt_feed().proxy_url().unwrap().expose(),
463 proxy_url
464 );
465 assert_eq!(configured_proxy.expose(), proxy_url);
466
467 let requests = requests.lock().await;
468 let request_lines = requests
469 .iter()
470 .map(|request| request.lines().next().unwrap().to_string())
471 .collect::<Vec<_>>();
472 let expected_auth = format!("Basic {}", BASE64.encode(format!("{USERNAME}:{SECRET}")));
473
474 assert_eq!(
475 request_lines,
476 [
477 "CONNECT gamma.fixture:443 HTTP/1.1",
478 "CONNECT clob-public.fixture:443 HTTP/1.1",
479 "CONNECT data.fixture:443 HTTP/1.1",
480 ]
481 );
482
483 for request in requests.iter() {
484 let auth = request
485 .lines()
486 .find_map(|line| {
487 let (name, value) = line.split_once(':')?;
488 name.eq_ignore_ascii_case("proxy-authorization")
489 .then_some(value.trim())
490 })
491 .expect("Proxy-Authorization header");
492 assert_eq!(auth, expected_auth);
493 }
494
495 for error in errors {
496 assert!(!error.contains(SECRET));
497 assert!(!error.contains(&BASE64.encode(SECRET)));
498 assert!(!error.contains(&expected_auth));
499 }
500 }
501
502 #[rstest]
503 #[tokio::test]
504 async fn authenticated_clob_http_client_uses_configured_proxy() {
505 const USERNAME: &str = "proxytest";
506 const SECRET: &str = "authenticated-clob-proxy-secret";
507 let (proxy_addr, requests) = spawn_rejecting_proxy(1).await;
508 let proxy_url =
509 ProxyUrl::parse(format!("http://{USERNAME}:{SECRET}@{proxy_addr}")).unwrap();
510 let credential = Credential::new(
511 "fixture-key".into(),
512 "Zml4dHVyZQ==".into(),
513 "fixture-passphrase".into(),
514 )
515 .unwrap();
516 let clob_auth = PolymarketClobHttpClient::new_with_proxy(
517 credential,
518 "0x0000000000000000000000000000000000000001".to_string(),
519 Some("https://clob-auth.fixture".to_string()),
520 2,
521 Some(proxy_url),
522 )
523 .unwrap();
524
525 let error = clob_auth
526 .get_book("auth-token")
527 .await
528 .unwrap_err()
529 .to_string();
530 let requests = requests.lock().await;
531 let request = requests.first().expect("captured CONNECT request");
532 let request_line = request.lines().next().unwrap();
533 let expected_auth = format!("Basic {}", BASE64.encode(format!("{USERNAME}:{SECRET}")));
534 let auth = request
535 .lines()
536 .find_map(|line| {
537 let (name, value) = line.split_once(':')?;
538 name.eq_ignore_ascii_case("proxy-authorization")
539 .then_some(value.trim())
540 })
541 .expect("Proxy-Authorization header");
542
543 assert_eq!(request_line, "CONNECT clob-auth.fixture:443 HTTP/1.1");
544 assert_eq!(auth, expected_auth);
545 assert!(!error.contains(SECRET));
546 assert!(!error.contains(&BASE64.encode(SECRET)));
547 assert!(!error.contains(&expected_auth));
548 }
549
550 #[rstest]
551 fn test_factory_rejects_duplicate_selections_before_setup() {
552 let config = PolymarketDataClientConfig {
553 instrument_config: Some(crate::config::PolymarketInstrumentProviderConfig {
554 event_slugs: Some(vec!["event-one".to_string(), "event-one".to_string()]),
555 ..Default::default()
556 }),
557 ..Default::default()
558 };
559
560 let cache = Rc::new(RefCell::new(Cache::default()));
561 let clock = Rc::new(RefCell::new(VirtualClock::new()));
562 let error = PolymarketDataClientFactory
563 .create("POLYMARKET", &config, cache.into(), clock)
564 .err()
565 .expect("duplicate config must fail before client setup");
566
567 assert_eq!(
568 error.downcast_ref::<ConfigError>(),
569 Some(&ConfigError::duplicate("event_slugs", None))
570 );
571 }
572}