Add compute_effective_deltas for Polymarket

RELEASES.md

@@ -5,6 +5,7 @@ Released on TBD (UTC).
 ### Enhancements
 - Added support for `bbo-1s` and `bbo-1m` quote schemas for Databento adapter (#1990), thanks @faysou
 - Added validation for venue `book_type` configuration vs data (prevents an issue where top-of-book data is used when order book data is expected)
+- Added `compute_effective_deltas` config setting for `PolymarketDataClientConfig`, reducing snapshot size (`False` by default to maintain current behavior)
 - Standardized Betfair symbology to use hyphens instead of periods (prevents Betfair symbols being treated as composite)
 
 ### Internal Improvements

examples/live/polymarket/polymarket_market_maker.py

@@ -107,6 +107,7 @@
             passphrase=None,  # 'POLYMARKET_PASSPHRASE' env var
             instrument_provider=instrument_provider_config,
             ws_connection_delay_secs=5,
+            compute_effective_deltas=True,
         ),
     },
     exec_clients={

examples/live/polymarket/polymarket_subscriber.py

@@ -97,6 +97,7 @@
             api_secret=None,  # 'POLYMARKET_API_SECRET' env var
             passphrase=None,  # 'POLYMARKET_PASSPHRASE' env var
             instrument_provider=instrument_provider_config,
+            compute_effective_deltas=True,
         ),
     },
     timeout_connection=60.0,

nautilus_trader/adapters/polymarket/common/deltas.py

@@ -0,0 +1,211 @@
+# -------------------------------------------------------------------------------------------------
+#  Copyright (C) 2015-2024 Nautech Systems Pty Ltd. All rights reserved.
+#  https://nautechsystems.io
+#
+#  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
+#  You may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+# -------------------------------------------------------------------------------------------------
+
+from nautilus_trader.model.book import Level
+from nautilus_trader.model.book import OrderBook
+from nautilus_trader.model.data import BookOrder
+from nautilus_trader.model.data import OrderBookDelta
+from nautilus_trader.model.data import OrderBookDeltas
+from nautilus_trader.model.enums import BookAction
+from nautilus_trader.model.enums import OrderSide
+from nautilus_trader.model.enums import RecordFlag
+from nautilus_trader.model.instruments import BinaryOption
+from nautilus_trader.model.objects import Price
+
+
+def compute_effective_deltas(
+    book_old: OrderBook,
+    book_new: OrderBook,
+    instrument: BinaryOption,
+) -> OrderBookDeltas | None:
+    """
+    Compare the old and new order book states and generate a list of effective deltas.
+
+    Parameters
+    ----------
+    book_old : OrderBook
+        The previous state of the order book.
+    book_new : OrderBook
+        The new state of the order book after applying deltas.
+    instrument : BinaryOption
+        The instrument associated with the order book.
+
+    Returns
+    -------
+    OrderBookDeltas or `None`
+        A collection of deltas representing the changes between the old and new book states.
+        If no change between book states, then `None` is returned.
+
+    """
+    deltas: list[OrderBookDelta] = []
+    instrument_id = instrument.id
+    assert instrument_id == book_old.instrument_id
+    assert instrument_id == book_new.instrument_id
+    ts_event = book_new.ts_event
+    ts_init = book_new.ts_init
+
+    old_bids: dict[Price, Level] = {level.price: level for level in book_old.bids()}
+    old_asks: dict[Price, Level] = {level.price: level for level in book_old.asks()}
+
+    new_bids = book_new.bids()
+    for bid in new_bids:
+        price = bid.price
+        size = instrument.make_qty(bid.size())
+
+        if bid.price not in old_bids:
+            # New bid (ADD)
+            order = BookOrder(
+                side=OrderSide.BUY,
+                price=price,
+                size=size,
+                order_id=0,  # Not applicable for L2 data
+            )
+            deltas.append(
+                OrderBookDelta(
+                    instrument_id=instrument_id,
+                    action=BookAction.ADD,
+                    order=order,
+                    flags=0,
+                    sequence=0,
+                    ts_event=ts_event,
+                    ts_init=ts_init,
+                ),
+            )
+        elif instrument.make_qty(old_bids[bid.price].size()) != size:
+            # Updated bid (UPDATE)
+            order = BookOrder(
+                side=OrderSide.BUY,
+                price=price,
+                size=size,
+                order_id=0,  # Not applicable for L2 data
+            )
+            deltas.append(
+                OrderBookDelta(
+                    instrument_id=instrument_id,
+                    action=BookAction.UPDATE,
+                    order=order,
+                    flags=0,
+                    sequence=0,
+                    ts_event=ts_event,
+                    ts_init=ts_init,
+                ),
+            )
+
+        old_bids.pop(bid.price, None)
+
+    new_asks = book_new.asks()
+    for ask in new_asks:
+        price = ask.price
+        size = instrument.make_qty(ask.size())
+
+        if ask.price not in old_asks:
+            # New ask (ADD)
+            order = BookOrder(
+                side=OrderSide.SELL,
+                price=price,
+                size=size,
+                order_id=0,  # Not applicable for L2 data
+            )
+            deltas.append(
+                OrderBookDelta(
+                    instrument_id=instrument_id,
+                    action=BookAction.ADD,
+                    order=order,
+                    flags=0,
+                    sequence=0,
+                    ts_event=ts_event,
+                    ts_init=ts_init,
+                ),
+            )
+        elif instrument.make_qty(old_asks[ask.price].size()) != size:
+            # Updated ask (UPDATE)
+            order = BookOrder(
+                side=OrderSide.SELL,
+                price=price,
+                size=size,
+                order_id=0,  # Not applicable for L2 data
+            )
+            deltas.append(
+                OrderBookDelta(
+                    instrument_id=instrument_id,
+                    action=BookAction.UPDATE,
+                    order=order,
+                    flags=0,
+                    sequence=0,
+                    ts_event=ts_event,
+                    ts_init=ts_init,
+                ),
+            )
+        old_asks.pop(ask.price, None)
+
+    # Process remaining old bids as removals
+    for old_price, old_level in old_bids.items():
+        order = BookOrder(
+            side=OrderSide.BUY,
+            price=old_price,
+            size=instrument.make_qty(old_level.size()),
+            order_id=0,  # Not applicable for L2 data
+        )
+        deltas.append(
+            OrderBookDelta(
+                instrument_id=instrument_id,
+                action=BookAction.DELETE,
+                order=order,
+                flags=0,
+                sequence=0,
+                ts_event=ts_event,
+                ts_init=ts_init,
+            ),
+        )
+
+    # Process remaining old asks as removals
+    for old_price, old_level in old_asks.items():
+        order = BookOrder(
+            side=OrderSide.SELL,
+            price=old_price,
+            size=instrument.make_qty(old_level.size()),
+            order_id=0,  # Not applicable for L2 data
+        )
+        deltas.append(
+            OrderBookDelta(
+                instrument_id=instrument_id,
+                action=BookAction.DELETE,
+                order=order,
+                flags=0,
+                sequence=0,
+                ts_event=ts_event,
+                ts_init=ts_init,
+            ),
+        )
+
+    # Return None if there are no deltas
+    if not deltas:
+        return None
+
+    # Mark the last delta in the batch with the F_LAST flag
+    last_delta = deltas[-1]
+    last_delta = OrderBookDelta(
+        instrument_id=last_delta.instrument_id,
+        action=last_delta.action,
+        order=last_delta.order,
+        flags=RecordFlag.F_LAST,
+        sequence=last_delta.sequence,
+        ts_event=last_delta.ts_event,
+        ts_init=last_delta.ts_init,
+    )
+
+    deltas[-1] = last_delta
+
+    return OrderBookDeltas(instrument_id=instrument_id, deltas=deltas)

nautilus_trader/adapters/polymarket/config.py

@@ -54,6 +54,9 @@ class PolymarketDataClientConfig(LiveDataClientConfig, frozen=True):
         The delay (seconds) prior to main websocket connection to allow initial subscriptions to arrive.
     update_instruments_interval_mins : PositiveInt, default 60
         The interval (minutes) between updating Polymarket instruments.
+    compute_effective_deltas : bool, default False
+        If True, computes effective deltas by comparing old and new order book states,
+        reducing snapshot size. This takes ~1 millisecond, so is not recommended for latency-sensitive strategies.
 
     """
 
@@ -68,6 +71,7 @@ class PolymarketDataClientConfig(LiveDataClientConfig, frozen=True):
     base_url_ws: str | None = None
     ws_connection_delay_secs: PositiveInt = 5
     update_instrument_interval_mins: PositiveInt = 60
+    compute_effective_deltas: bool = False
 
 
 class PolymarketExecClientConfig(LiveExecClientConfig, frozen=True):

nautilus_trader/adapters/polymarket/data.py

@@ -22,6 +22,7 @@
 from py_clob_client.client import ClobClient
 
 from nautilus_trader.adapters.polymarket.common.constants import POLYMARKET_VENUE
+from nautilus_trader.adapters.polymarket.common.deltas import compute_effective_deltas
 from nautilus_trader.adapters.polymarket.common.enums import PolymarketOrderSide
 from nautilus_trader.adapters.polymarket.common.symbol import get_polymarket_instrument_id
 from nautilus_trader.adapters.polymarket.common.symbol import get_polymarket_token_id
@@ -39,8 +40,10 @@
 from nautilus_trader.common.enums import LogColor
 from nautilus_trader.core.uuid import UUID4
 from nautilus_trader.live.data_client import LiveMarketDataClient
+from nautilus_trader.model.book import OrderBook
 from nautilus_trader.model.data import BarType
 from nautilus_trader.model.data import DataType
+from nautilus_trader.model.data import OrderBookDeltas
 from nautilus_trader.model.data import QuoteTick
 from nautilus_trader.model.enums import BookType
 from nautilus_trader.model.identifiers import ClientId
@@ -118,6 +121,7 @@ def __init__(
 
         # Hot caches
         self._last_quotes: dict[InstrumentId, QuoteTick] = {}
+        self._local_books: dict[InstrumentId, OrderBook] = {}
 
     async def _connect(self) -> None:
         self._log.info("Initializing instruments...")
@@ -229,6 +233,10 @@ async def _subscribe_order_book_deltas(
             )
             return
 
+        if self._config.compute_effective_deltas:
+            local_book = OrderBook(instrument_id, book_type=BookType.L2_MBP)
+            self._local_books[instrument_id] = local_book
+
         await self._subscribe_asset_book(instrument_id)
 
     async def _subscribe_quote_ticks(self, instrument_id: InstrumentId) -> None:
@@ -394,20 +402,46 @@ def _handle_book_snapshot(
             ts_event=now_ns,
             ts_init=now_ns,
         )
-        self._handle_data(deltas)
+
+        self._handle_deltas(instrument, deltas)
 
         if instrument.id in self.subscribed_quote_ticks():
             quote = ws_message.parse_to_quote_tick(instrument=instrument, ts_init=now_ns)
             self._last_quotes[instrument.id] = quote
             self._handle_data(quote)
 
+    def _handle_deltas(self, instrument: BinaryOption, deltas: OrderBookDeltas) -> None:
+        if self._config.compute_effective_deltas:
+            # Compute effective deltas (reduce snapshot based on old and new book states),
+            # prioritizing a smaller data footprint over computational efficiency.
+            t0 = self._clock.timestamp_ns()
+            book_old = self._local_books.get(instrument.id)
+            book_new = OrderBook(instrument.id, book_type=BookType.L2_MBP)
+            book_new.apply_deltas(deltas)
+            self._local_books[instrument.id] = book_new
+            deltas = compute_effective_deltas(book_old, book_new, instrument)
+
+            interval = (self._clock.timestamp_ns() - t0) / 1_000_000
+            self._log.info(f"Computed effective deltas in {interval:.3f}ms", LogColor.BLUE)
+            # self._log.warning(book_new.pprint())  # Uncomment for development
+
+        # Check if any effective deltas remain
+        if deltas:
+            self._handle_data(deltas)
+
     def _handle_quote(
         self,
         instrument: BinaryOption,
         ws_message: PolymarketQuote,
     ) -> None:
         now_ns = self._clock.timestamp_ns()
         delta = ws_message.parse_to_delta(instrument=instrument, ts_init=now_ns)
+
+        if self._config.compute_effective_deltas:
+            local_book = self._local_books.get(instrument.id)
+            if local_book:
+                local_book.apply_delta(delta)
+
         self._handle_data(delta)
 
         if instrument.id in self.subscribed_quote_ticks():