nautilus_polymarket/websocket/
parse.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//! Parse functions for converting Polymarket WebSocket messages to Nautilus data types.
17
18use std::str::FromStr;
19
20use nautilus_core::{
21    UnixNanos,
22    correctness::{CorrectnessError, CorrectnessResult},
23    datetime::NANOSECONDS_IN_MILLISECOND,
24};
25use nautilus_model::{
26    data::{BookOrder, OrderBookDelta, OrderBookDeltas, QuoteTick, TradeTick},
27    enums::{AggressorSide, BookAction, OrderSide, RecordFlag},
28    identifiers::InstrumentId,
29    types::{Price, Quantity},
30};
31use rust_decimal::Decimal;
32
33use super::messages::{PolymarketBookSnapshot, PolymarketQuote, PolymarketQuotes, PolymarketTrade};
34use crate::common::{enums::PolymarketOrderSide, parse::determine_trade_id};
35
36/// Parses a millisecond epoch timestamp string into [`UnixNanos`].
37pub fn parse_timestamp_ms(ts: &str) -> anyhow::Result<UnixNanos> {
38    let ms: u64 = ts
39        .parse()
40        .map_err(|e| anyhow::anyhow!("Invalid timestamp '{ts}': {e}"))?;
41    let ns = ms
42        .checked_mul(NANOSECONDS_IN_MILLISECOND)
43        .ok_or_else(|| anyhow::anyhow!("Timestamp overflow for '{ts}'"))?;
44    Ok(UnixNanos::from(ns))
45}
46
47pub(crate) fn parse_price(s: &str, precision: u8) -> CorrectnessResult<Price> {
48    let value = Decimal::from_str(s).map_err(|e| CorrectnessError::PredicateViolation {
49        message: format!("Invalid price '{s}': {e}"),
50    })?;
51    Price::from_decimal_dp(value, precision)
52}
53
54pub(crate) fn parse_quantity(s: &str, precision: u8) -> CorrectnessResult<Quantity> {
55    let value = Decimal::from_str(s).map_err(|e| CorrectnessError::PredicateViolation {
56        message: format!("Invalid quantity '{s}': {e}"),
57    })?;
58    Quantity::from_decimal_dp(value, precision)
59}
60
61/// Parses a book snapshot into [`OrderBookDeltas`] (CLEAR + ADD).
62pub fn parse_book_snapshot(
63    snap: &PolymarketBookSnapshot,
64    instrument_id: InstrumentId,
65    price_precision: u8,
66    size_precision: u8,
67    ts_init: UnixNanos,
68) -> anyhow::Result<OrderBookDeltas> {
69    let ts_event = parse_timestamp_ms(&snap.timestamp)?;
70
71    let bids_len = snap.bids.len();
72    let asks_len = snap.asks.len();
73
74    if bids_len == 0 && asks_len == 0 {
75        anyhow::bail!("Empty book snapshot for {instrument_id}");
76    }
77
78    let total = bids_len + asks_len;
79    let mut deltas = Vec::with_capacity(total + 1);
80
81    // Every snapshot delta (including the opening CLEAR) carries F_SNAPSHOT so
82    // downstream consumers can recognise the rebuild; F_LAST closes the batch
83    // on the final delta. `OrderBookDelta::clear` already sets F_SNAPSHOT.
84    let snapshot_flag = RecordFlag::F_SNAPSHOT as u8;
85    deltas.push(OrderBookDelta::clear(instrument_id, 0, ts_event, ts_init));
86
87    let mut count = 0;
88
89    for level in &snap.bids {
90        count += 1;
91        let price = parse_price(&level.price, price_precision)?;
92        let size = parse_quantity(&level.size, size_precision)?;
93        let order = BookOrder::new(OrderSide::Buy, price, size, 0);
94
95        let mut flags = snapshot_flag;
96        if count == total {
97            flags |= RecordFlag::F_LAST as u8;
98        }
99
100        deltas.push(OrderBookDelta::new_checked(
101            instrument_id,
102            BookAction::Add,
103            order,
104            flags,
105            0,
106            ts_event,
107            ts_init,
108        )?);
109    }
110
111    for level in &snap.asks {
112        count += 1;
113        let price = parse_price(&level.price, price_precision)?;
114        let size = parse_quantity(&level.size, size_precision)?;
115        let order = BookOrder::new(OrderSide::Sell, price, size, 0);
116
117        let mut flags = snapshot_flag;
118        if count == total {
119            flags |= RecordFlag::F_LAST as u8;
120        }
121
122        deltas.push(OrderBookDelta::new_checked(
123            instrument_id,
124            BookAction::Add,
125            order,
126            flags,
127            0,
128            ts_event,
129            ts_init,
130        )?);
131    }
132
133    Ok(OrderBookDeltas::new(instrument_id, deltas))
134}
135
136/// Parses price change quotes into incremental [`OrderBookDeltas`].
137pub fn parse_book_deltas(
138    quotes: &PolymarketQuotes,
139    instrument_id: InstrumentId,
140    price_precision: u8,
141    size_precision: u8,
142    ts_init: UnixNanos,
143) -> anyhow::Result<OrderBookDeltas> {
144    let ts_event = parse_timestamp_ms(&quotes.timestamp)?;
145
146    let total = quotes.price_changes.len();
147    let mut deltas = Vec::with_capacity(total);
148
149    for (idx, change) in quotes.price_changes.iter().enumerate() {
150        let price = parse_price(&change.price, price_precision)?;
151        let size = parse_quantity(&change.size, size_precision)?;
152        let side = match change.side {
153            PolymarketOrderSide::Buy => OrderSide::Buy,
154            PolymarketOrderSide::Sell => OrderSide::Sell,
155        };
156
157        let (action, order_size) = if size.is_zero() {
158            (BookAction::Delete, Quantity::zero(size_precision))
159        } else {
160            (BookAction::Update, size)
161        };
162
163        let order = BookOrder::new(side, price, order_size, 0);
164        let flags = if idx == total - 1 {
165            RecordFlag::F_LAST as u8
166        } else {
167            0
168        };
169
170        deltas.push(OrderBookDelta::new_checked(
171            instrument_id,
172            action,
173            order,
174            flags,
175            0,
176            ts_event,
177            ts_init,
178        )?);
179    }
180
181    Ok(OrderBookDeltas::new(instrument_id, deltas))
182}
183
184/// Parses a trade message into a [`TradeTick`].
185pub fn parse_trade_tick(
186    trade: &PolymarketTrade,
187    instrument_id: InstrumentId,
188    price_precision: u8,
189    size_precision: u8,
190    ts_init: UnixNanos,
191) -> anyhow::Result<TradeTick> {
192    let price = parse_price(&trade.price, price_precision)?;
193    let size = parse_quantity(&trade.size, size_precision)?;
194    let aggressor_side = match trade.side {
195        PolymarketOrderSide::Buy => AggressorSide::Buyer,
196        PolymarketOrderSide::Sell => AggressorSide::Seller,
197    };
198    let ts_event = parse_timestamp_ms(&trade.timestamp)?;
199
200    let trade_id = determine_trade_id(
201        &trade.asset_id,
202        trade.side,
203        &trade.price,
204        &trade.size,
205        &trade.timestamp,
206    );
207
208    TradeTick::new_checked(
209        instrument_id,
210        price,
211        size,
212        aggressor_side,
213        trade_id,
214        ts_event,
215        ts_init,
216    )
217}
218
219/// Extracts a top-of-book [`QuoteTick`] from a book snapshot.
220///
221/// Returns `None` if either side is empty.
222///
223/// # Panics
224///
225/// Cannot panic: `.expect()` calls are guarded by the empty-side
226/// early return above.
227pub fn parse_quote_from_snapshot(
228    snap: &PolymarketBookSnapshot,
229    instrument_id: InstrumentId,
230    price_precision: u8,
231    size_precision: u8,
232    ts_init: UnixNanos,
233) -> anyhow::Result<Option<QuoteTick>> {
234    if snap.bids.is_empty() || snap.asks.is_empty() {
235        return Ok(None);
236    }
237
238    let ts_event = parse_timestamp_ms(&snap.timestamp)?;
239
240    // Polymarket sends bids ascending and asks descending, so best-of-book is last
241    let best_bid = snap.bids.last().expect("bids not empty");
242    let best_ask = snap.asks.last().expect("asks not empty");
243
244    let bid_price = parse_price(&best_bid.price, price_precision)?;
245    let ask_price = parse_price(&best_ask.price, price_precision)?;
246    let bid_size = parse_quantity(&best_bid.size, size_precision)?;
247    let ask_size = parse_quantity(&best_ask.size, size_precision)?;
248
249    Ok(Some(QuoteTick::new_checked(
250        instrument_id,
251        bid_price,
252        ask_price,
253        bid_size,
254        ask_size,
255        ts_event,
256        ts_init,
257    )?))
258}
259
260/// Parses a quote tick from a price change message using its best_bid/best_ask fields.
261///
262/// Returns `None` when either best_bid or best_ask is absent (empty book side).
263/// When `last_quote` is provided the opposite side's size is carried forward
264/// instead of being set to zero, matching the Python adapter's behavior.
265pub fn parse_quote_from_price_change(
266    quote: &PolymarketQuote,
267    instrument_id: InstrumentId,
268    price_precision: u8,
269    size_precision: u8,
270    last_quote: Option<&QuoteTick>,
271    ts_event: UnixNanos,
272    ts_init: UnixNanos,
273) -> anyhow::Result<Option<QuoteTick>> {
274    let (Some(best_bid), Some(best_ask)) = (&quote.best_bid, &quote.best_ask) else {
275        return Ok(None);
276    };
277    let bid_price = parse_price(best_bid, price_precision)?;
278    let ask_price = parse_price(best_ask, price_precision)?;
279    let changed_price = parse_price(&quote.price, price_precision)?;
280
281    let size = parse_quantity(&quote.size, size_precision)?;
282    let zero = || Quantity::zero(size_precision);
283
284    // Only use the changed level's size when it matches the best price,
285    // otherwise preserve the previous quote's size for that side
286    let (bid_size, ask_size) = match quote.side {
287        PolymarketOrderSide::Buy => {
288            let bid_size = if changed_price == bid_price {
289                size
290            } else {
291                last_quote.map_or_else(zero, |q| q.bid_size)
292            };
293            let ask_size = last_quote.map_or_else(zero, |q| q.ask_size);
294            (bid_size, ask_size)
295        }
296        PolymarketOrderSide::Sell => {
297            let ask_size = if changed_price == ask_price {
298                size
299            } else {
300                last_quote.map_or_else(zero, |q| q.ask_size)
301            };
302            let bid_size = last_quote.map_or_else(zero, |q| q.bid_size);
303            (bid_size, ask_size)
304        }
305    };
306
307    Ok(Some(QuoteTick::new_checked(
308        instrument_id,
309        bid_price,
310        ask_price,
311        bid_size,
312        ask_size,
313        ts_event,
314        ts_init,
315    )?))
316}
317
318#[cfg(test)]
319mod tests {
320    use nautilus_core::UnixNanos;
321    use nautilus_model::instruments::{Instrument, InstrumentAny};
322    use rstest::rstest;
323
324    use super::*;
325    use crate::http::parse::{create_instrument_from_def, parse_gamma_market};
326
327    fn load<T: serde::de::DeserializeOwned>(filename: &str) -> T {
328        let content =
329            std::fs::read_to_string(format!("test_data/{filename}")).expect("test data missing");
330        serde_json::from_str(&content).expect("parse failed")
331    }
332
333    fn test_instrument() -> InstrumentAny {
334        let market: crate::http::models::GammaMarket = load("gamma_market.json");
335        let defs = parse_gamma_market(&market).unwrap();
336        create_instrument_from_def(&defs[0], UnixNanos::from(1_000_000_000u64)).unwrap()
337    }
338
339    #[rstest]
340    fn test_parse_timestamp_ms() {
341        let ns = parse_timestamp_ms("1703875200000").unwrap();
342        assert_eq!(ns, UnixNanos::from(1_703_875_200_000_000_000u64));
343    }
344
345    #[rstest]
346    fn test_parse_timestamp_ms_invalid() {
347        assert!(parse_timestamp_ms("not_a_number").is_err());
348    }
349
350    #[rstest]
351    fn test_parse_book_snapshot() {
352        let snap: PolymarketBookSnapshot = load("ws_book_snapshot.json");
353        let instrument = test_instrument();
354        let ts_init = UnixNanos::from(1_000_000_000u64);
355
356        let deltas = parse_book_snapshot(
357            &snap,
358            instrument.id(),
359            instrument.price_precision(),
360            instrument.size_precision(),
361            ts_init,
362        )
363        .unwrap();
364
365        // CLEAR + 3 bids + 3 asks = 7 deltas
366        assert_eq!(deltas.deltas.len(), 7);
367        assert_eq!(deltas.deltas[0].action, BookAction::Clear);
368        assert_eq!(deltas.deltas[1].action, BookAction::Add);
369        assert_eq!(deltas.deltas[1].order.side, OrderSide::Buy);
370        assert_eq!(deltas.deltas[4].action, BookAction::Add);
371        assert_eq!(deltas.deltas[4].order.side, OrderSide::Sell);
372
373        // Every snapshot delta carries F_SNAPSHOT
374        for delta in &deltas.deltas {
375            assert_ne!(delta.flags & RecordFlag::F_SNAPSHOT as u8, 0);
376        }
377
378        // Exactly one delta carries F_LAST, and it must be the last one
379        let f_last_count = deltas
380            .deltas
381            .iter()
382            .filter(|d| d.flags & RecordFlag::F_LAST as u8 != 0)
383            .count();
384        assert_eq!(f_last_count, 1);
385        assert_ne!(
386            deltas.deltas.last().unwrap().flags & RecordFlag::F_LAST as u8,
387            0
388        );
389    }
390
391    #[rstest]
392    fn test_parse_book_deltas() {
393        let quotes: PolymarketQuotes = load("ws_quotes.json");
394        let instrument = test_instrument();
395        let ts_init = UnixNanos::from(1_000_000_000u64);
396
397        let deltas = parse_book_deltas(
398            &quotes,
399            instrument.id(),
400            instrument.price_precision(),
401            instrument.size_precision(),
402            ts_init,
403        )
404        .unwrap();
405
406        assert_eq!(deltas.deltas.len(), 2);
407
408        // Exactly one delta carries F_LAST, and it must be the last one
409        let f_last_count = deltas
410            .deltas
411            .iter()
412            .filter(|d| d.flags & RecordFlag::F_LAST as u8 != 0)
413            .count();
414        assert_eq!(f_last_count, 1);
415        assert_ne!(
416            deltas.deltas.last().unwrap().flags & RecordFlag::F_LAST as u8,
417            0
418        );
419    }
420
421    #[rstest]
422    fn test_parse_book_deltas_zero_size_is_delete() {
423        let mut quotes: PolymarketQuotes = load("ws_quotes.json");
424        quotes.price_changes[0].size = "0".to_string();
425        let instrument = test_instrument();
426        let ts_init = UnixNanos::from(1_000_000_000u64);
427
428        let deltas = parse_book_deltas(
429            &quotes,
430            instrument.id(),
431            instrument.price_precision(),
432            instrument.size_precision(),
433            ts_init,
434        )
435        .unwrap();
436
437        assert_eq!(deltas.deltas[0].action, BookAction::Delete);
438    }
439
440    #[rstest]
441    fn test_parse_trade_tick() {
442        let trade: PolymarketTrade = load("ws_last_trade.json");
443        let instrument = test_instrument();
444        let ts_init = UnixNanos::from(1_000_000_000u64);
445
446        let tick = parse_trade_tick(
447            &trade,
448            instrument.id(),
449            instrument.price_precision(),
450            instrument.size_precision(),
451            ts_init,
452        )
453        .unwrap();
454
455        assert_eq!(tick.instrument_id, instrument.id());
456        assert_eq!(tick.aggressor_side, AggressorSide::Buyer);
457        assert_eq!(tick.ts_event, UnixNanos::from(1_703_875_202_000_000_000u64));
458    }
459
460    #[rstest]
461    fn test_parse_trade_tick_deterministic_id() {
462        let trade: PolymarketTrade = load("ws_last_trade.json");
463        let instrument = test_instrument();
464        let ts_init = UnixNanos::from(1_000_000_000u64);
465
466        let tick1 = parse_trade_tick(
467            &trade,
468            instrument.id(),
469            instrument.price_precision(),
470            instrument.size_precision(),
471            ts_init,
472        )
473        .unwrap();
474        let tick2 = parse_trade_tick(
475            &trade,
476            instrument.id(),
477            instrument.price_precision(),
478            instrument.size_precision(),
479            ts_init,
480        )
481        .unwrap();
482
483        assert_eq!(tick1.trade_id, tick2.trade_id);
484    }
485
486    #[rstest]
487    fn test_parse_quote_from_snapshot() {
488        let snap: PolymarketBookSnapshot = load("ws_book_snapshot.json");
489        let instrument = test_instrument();
490        let ts_init = UnixNanos::from(1_000_000_000u64);
491
492        let quote = parse_quote_from_snapshot(
493            &snap,
494            instrument.id(),
495            instrument.price_precision(),
496            instrument.size_precision(),
497            ts_init,
498        )
499        .unwrap()
500        .unwrap();
501
502        assert_eq!(quote.instrument_id, instrument.id());
503        assert_eq!(quote.bid_price, Price::from("0.50"));
504        assert_eq!(quote.ask_price, Price::from("0.51"));
505        assert_eq!(
506            quote.ts_event,
507            UnixNanos::from(1_703_875_200_000_000_000u64)
508        );
509    }
510
511    #[rstest]
512    fn test_parse_quote_from_snapshot_empty_side_returns_none() {
513        let mut snap: PolymarketBookSnapshot = load("ws_book_snapshot.json");
514        snap.bids.clear();
515        let instrument = test_instrument();
516        let ts_init = UnixNanos::from(1_000_000_000u64);
517
518        let result = parse_quote_from_snapshot(
519            &snap,
520            instrument.id(),
521            instrument.price_precision(),
522            instrument.size_precision(),
523            ts_init,
524        )
525        .unwrap();
526
527        assert!(result.is_none());
528    }
529
530    #[rstest]
531    fn test_parse_quote_from_price_change() {
532        let quotes: PolymarketQuotes = load("ws_quotes.json");
533        let instrument = test_instrument();
534        let ts_event = parse_timestamp_ms(&quotes.timestamp).unwrap();
535        let ts_init = UnixNanos::from(1_000_000_000u64);
536
537        let quote = parse_quote_from_price_change(
538            &quotes.price_changes[0],
539            instrument.id(),
540            instrument.price_precision(),
541            instrument.size_precision(),
542            None,
543            ts_event,
544            ts_init,
545        )
546        .unwrap()
547        .expect("quote should be Some when best_bid/best_ask present");
548
549        assert_eq!(quote.instrument_id, instrument.id());
550    }
551}
nautilus_polymarket/websocket/parse.rs

nautilus_polymarket/websocket/
parse.rs
