Merge pull request #274 from liampauling/task/269-latency-warning

#269

flumine/baseflumine.py

@@ -1,3 +1,4 @@
+import time
 import queue
 import logging
 import threading
@@ -113,6 +114,19 @@ def _add_default_workers(self) -> None:
     def _process_market_books(self, event: events.MarketBookEvent) -> None:
         for market_book in event.event:
             market_id = market_book.market_id
+
+            # check latency
+            latency = time.time() - (market_book.publish_time_epoch / 1e3)
+            if latency > 2:
+                logger.warning(
+                    "High latency between current time and MarketBook publish time",
+                    extra={
+                        "market_id": market_id,
+                        "latency": latency,
+                        "pt": market_book.publish_time,
+                    },
+                )
+
             if market_book.status == "CLOSED":
                 self.handler_queue.put(events.CloseMarketEvent(market_book))
                 continue

tests/test_baseflumine.py

@@ -92,7 +92,7 @@ def test__add_default_workers(self):
 
     def test__process_market_books(self):
         mock_event = mock.Mock()
-        mock_market_book = mock.Mock()
+        mock_market_book = mock.Mock(publish_time_epoch=123)
         mock_market_book.runners = []
         mock_event.event = [mock_market_book]
         self.base_flumine._process_market_books(mock_event)