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