Update betfair tests for sequence completed messages

Author: cjdsellers

RELEASES.md

@@ -12,6 +12,7 @@ This release adds support for Python 3.13 (*not* yet compatible with free-thread
 - Added additional price and quantity precision validations for matching engine
 - Added log file rotation with additional config options `max_file_size` and `max_backup_count` (#2468), thanks @xingyanan and @twitu
 - Added `bars_timestamp_on_close` config option for `BybitDataClientConfig` (default `True` to match Nautilus conventions)
+- Added `BetfairSequenceCompleted` custom data type for Betfair to mark the completion of a sequence of messages
 - Improved robustness of in-flight order check for `LiveExecutionEngine`, once exceeded query retries will resolve submitted orders as rejected and pending orders as canceled
 - Improved logging for `BacktestNode` crashes with full stack trace and prettier config logging
 

tests/integration_tests/adapters/betfair/test_betfair_data.py

@@ -23,6 +23,7 @@
 
 from nautilus_trader.adapters.betfair.constants import BETFAIR_VENUE
 from nautilus_trader.adapters.betfair.data import BetfairDataClient
+from nautilus_trader.adapters.betfair.data_types import BetfairSequenceCompleted
 from nautilus_trader.adapters.betfair.data_types import BetfairStartingPrice
 from nautilus_trader.adapters.betfair.data_types import BetfairTicker
 from nautilus_trader.adapters.betfair.data_types import BSPOrderBookDelta
@@ -176,7 +177,12 @@ async def test_market_sub_image_market_def(data_client, mock_data_engine_process
     # Assert - expected messages
     mock_calls = mock_data_engine_process.call_args_list
     result = [type(call.args[0]).__name__ for call in mock_data_engine_process.call_args_list]
-    expected = ["BettingInstrument"] * 7 + ["InstrumentStatus"] * 7 + ["OrderBookDeltas"] * 7
+    expected = (
+        ["BettingInstrument"] * 7
+        + ["InstrumentStatus"] * 7
+        + ["OrderBookDeltas"] * 7
+        + ["BetfairSequenceCompleted"]
+    )
     assert result == expected
 
     # Assert - Check orderbook prices
@@ -219,21 +225,27 @@ def test_market_update(data_client, mock_data_engine_process):
 def test_market_update_md(data_client, mock_data_engine_process):
     data_client.on_market_update(BetfairStreaming.mcm_UPDATE_md())
     result = [type(call.args[0]).__name__ for call in mock_data_engine_process.call_args_list]
-    expected = ["BettingInstrument"] * 2 + ["InstrumentStatus"] * 2
+    expected = ["BettingInstrument"] * 2 + ["InstrumentStatus"] * 2 + ["BetfairSequenceCompleted"]
     assert result == expected
 
 
 def test_market_update_live_image(data_client, mock_data_engine_process):
     data_client.on_market_update(BetfairStreaming.mcm_live_IMAGE())
     result = [type(call.args[0]).__name__ for call in mock_data_engine_process.call_args_list]
-    expected = ["OrderBookDeltas"] + ["TradeTick"] * 13 + ["OrderBookDeltas"] + ["TradeTick"] * 17
+    expected = (
+        ["OrderBookDeltas"]
+        + ["TradeTick"] * 13
+        + ["OrderBookDeltas"]
+        + ["TradeTick"] * 17
+        + ["BetfairSequenceCompleted"]
+    )
     assert result == expected
 
 
 def test_market_update_live_update(data_client, mock_data_engine_process):
     data_client.on_market_update(BetfairStreaming.mcm_live_UPDATE())
     result = [type(call.args[0]).__name__ for call in mock_data_engine_process.call_args_list]
-    expected = ["TradeTick", "OrderBookDeltas"]
+    expected = ["TradeTick", "OrderBookDeltas", "BetfairSequenceCompleted"]
     assert result == expected
 
 
@@ -259,6 +271,7 @@ def test_market_bsp(data_client, mock_data_engine_process):
     mock_call_args = [call.args[0] for call in mock_data_engine_process.call_args_list]
     result = Counter([type(args).__name__ for args in mock_call_args])
     expected = {
+        "BetfairSequenceCompleted": 2,
         "BettingInstrument": 9,
         "TradeTick": 95,
         "OrderBookDeltas": 11,
@@ -300,6 +313,8 @@ def test_orderbook_updates(data_client, parser):
     for raw_update in BetfairStreaming.market_updates():
         line = stream_decode(raw_update)
         for update in parser.parse(mcm=line):
+            if isinstance(update, BetfairSequenceCompleted):
+                continue
             if len(order_books) > 1 and update.instrument_id != list(order_books)[1]:
                 continue
             if isinstance(update, OrderBookDeltas) and update.is_snapshot:
@@ -334,12 +349,13 @@ def test_orderbook_updates(data_client, parser):
 def test_instrument_opening_events(data_client, parser):
     updates = BetfairDataProvider.market_updates()
     messages = parser.parse(updates[0])
-    assert len(messages) == 4
+    assert len(messages) == 5
     assert isinstance(messages[0], BettingInstrument)
     assert isinstance(messages[2], InstrumentStatus)
     assert messages[2].action == MarketStatusAction.PRE_OPEN
     assert isinstance(messages[3], InstrumentStatus)
     assert messages[3].action == MarketStatusAction.PRE_OPEN
+    assert isinstance(messages[4], BetfairSequenceCompleted)
 
 
 def test_instrument_in_play_events(data_client, parser):
@@ -389,8 +405,8 @@ def test_instrument_update(data_client, cache, parser):
 def test_instrument_closing_events(data_client, parser):
     updates = BetfairDataProvider.market_updates()
     messages = parser.parse(updates[-1])
-    assert len(messages) == 6
-    ins1, ins2, status1, status2, close1, close2 = messages
+    assert len(messages) == 7
+    ins1, ins2, status1, status2, close1, close2, completed = messages
 
     # Instrument1
     assert isinstance(ins1, BettingInstrument)
@@ -408,6 +424,8 @@ def test_instrument_closing_events(data_client, parser):
     assert close2.close_price == 0.0
     assert close2.close_type == InstrumentCloseType.CONTRACT_EXPIRED
 
+    assert isinstance(completed, BetfairSequenceCompleted)
+
 
 def test_betfair_ticker(data_client, mock_data_engine_process) -> None:
     # Arrange
@@ -472,7 +490,7 @@ def test_betfair_orderbook(data_client, parser) -> None:
     # Act, Assert
     for update in BetfairDataProvider.market_updates():
         for message in parser.parse(update):
-            if isinstance(message, BettingInstrument | CustomData):
+            if isinstance(message, BetfairSequenceCompleted | BettingInstrument | CustomData):
                 continue
             if message.instrument_id not in books:
                 books[message.instrument_id] = create_betfair_order_book(

tests/integration_tests/adapters/betfair/test_betfair_parsing.py

@@ -49,6 +49,7 @@
 # fmt: off
 from nautilus_trader.adapters.betfair.common import BETFAIR_TICK_SCHEME
 from nautilus_trader.adapters.betfair.common import OrderSideParser
+from nautilus_trader.adapters.betfair.data_types import BetfairSequenceCompleted
 from nautilus_trader.adapters.betfair.data_types import BetfairStartingPrice
 from nautilus_trader.adapters.betfair.data_types import BetfairTicker
 from nautilus_trader.adapters.betfair.data_types import BSPOrderBookDelta
@@ -194,6 +195,7 @@ def test_market_definition_to_instrument_updates(self):
                 "InstrumentStatus": 7,
                 "OrderBookDeltas": 7,
                 "BettingInstrument": 7,
+                "BetfairSequenceCompleted": 1,
             },
         )
         assert counts == expected
@@ -238,10 +240,10 @@ def test_market_change_ticker(self):
     @pytest.mark.parametrize(
         ("filename", "num_msgs"),
         [
-            ("1-166564490.bz2", 2506),
-            ("1-166811431.bz2", 17852),
-            ("1-180305278.bz2", 15165),
-            ("1-206064380.bz2", 52111),
+            ("1-166564490.bz2", 4114),
+            ("1-166811431.bz2", 29209),
+            ("1-180305278.bz2", 22850),
+            ("1-206064380.bz2", 70904),
         ],
     )
     def test_parsing_streaming_file(self, filename, num_msgs):
@@ -266,6 +268,7 @@ def test_parsing_streaming_file_message_counts(self):
                 "OrderBookDeltas": 40525,
                 "BetfairTicker": 4658,
                 "TradeTick": 3487,
+                "BetfairSequenceCompleted": 18793,
                 "BettingInstrument": 260,
                 "BSPOrderBookDelta": 2824,
                 "InstrumentStatus": 260,
@@ -495,11 +498,12 @@ async def test_merge_order_book_deltas(self):
         )
         mcm = msgspec.json.decode(raw, type=MCM)
         updates = self.parser.parse(mcm)
-        assert len(updates) == 3
-        trade, ticker, deltas = updates
+        assert len(updates) == 4
+        trade, ticker, deltas, completed = updates
         assert isinstance(trade, TradeTick)
         assert isinstance(ticker, Data)
         assert isinstance(deltas, OrderBookDeltas)
+        assert isinstance(completed, BetfairSequenceCompleted)
         assert len(deltas.deltas) == 2
 
     def test_make_order_limit(self):

tests/unit_tests/persistence/test_catalog.py

@@ -54,6 +54,7 @@ def test_list_data_types(catalog_betfair: ParquetDataCatalog) -> None:
     data_types = catalog_betfair.list_data_types()
     expected = [
         "betting_instrument",
+        "custom_betfair_sequence_completed",
         "custom_betfair_ticker",
         "instrument_status",
         "order_book_delta",