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