Merge pull request #374 from liampauling/task/373-market-cache-refactor

Task/373 market cache refactor

.travis.yml

@@ -8,10 +8,6 @@ install:
 
 jobs:
   include:
-    - name: "Tests: python 3.5"
-      python: 3.5
-      script:
-        - coverage run --source=betfairlightweight setup.py test
     - name: "Tests: python 3.6"
       python: 3.6
       script:

HISTORY.rst

@@ -8,11 +8,16 @@ Release History
 
 **Improvements**
 
+- #373 Streaming refactor (2-3x+ speed improvement across the board)
 - #369 Output streaming matched backs/lays
 - #370 Session timeout updated to 24hrs for international exchange
 - License update
 - Removed build.sh
 
+**Dependencies**
+
+- py3.5 testing removed
+
 2.11.1 (2020-12-26)
 +++++++++++++++++++
 

appveyor.yml

@@ -2,9 +2,6 @@ build: off
 
 environment:
   matrix:
-    - PYTHON: "C:\\Python35"
-      PYTHON_VERSION: "3.5.x"
-
     - PYTHON: "C:\\Python36"
       PYTHON_VERSION: "3.6.x"
 

betfairlightweight/endpoints/streaming.py

@@ -54,6 +54,7 @@ def create_historical_stream(
         file_path: str = None,
         listener: BaseListener = None,
         operation: str = "marketSubscription",
+        unique_id: int = 0,
     ) -> HistoricalStream:
         """
         Uses streaming listener/cache to parse betfair
@@ -63,17 +64,19 @@ def create_historical_stream(
         :param str file_path: Path to historic betfair file
         :param BaseListener listener: Listener object
         :param str operation: Operation type
+        :param int unique_id: Stream id (added to updates)
 
         :rtype: HistoricalStream
         """
         listener = listener if listener else BaseListener()
-        return HistoricalStream(file_path, listener, operation)
+        return HistoricalStream(file_path, listener, operation, unique_id)
 
     @staticmethod
     def create_historical_generator_stream(
         file_path: str = None,
         listener: BaseListener = None,
         operation: str = "marketSubscription",
+        unique_id: int = 0,
     ) -> HistoricalGeneratorStream:
         """
         Uses generator listener/cache to parse betfair
@@ -83,8 +86,9 @@ def create_historical_generator_stream(
         :param str file_path: Path to historic betfair file
         :param BaseListener listener: Listener object
         :param str operation: Operation type
+        :param int unique_id: Stream id (added to updates)
 
         :rtype: HistoricalGeneratorStream
         """
         listener = listener if listener else StreamListener()
-        return HistoricalGeneratorStream(file_path, listener, operation)
+        return HistoricalGeneratorStream(file_path, listener, operation, unique_id)

betfairlightweight/resources/__init__.py

@@ -12,6 +12,7 @@
     VenueResult,
     MarketCatalogue,
     MarketBook,
+    RunnerBook,
     Match,
     CurrentOrders,
     ClearedOrders,

betfairlightweight/resources/streamingresources.py

@@ -195,7 +195,7 @@ def __init__(self, **kwargs):
         self.race_runners = [RaceChange(**runner) for runner in kwargs.get("rrc") or []]
 
 
-class RaceProgress(BaseResource):
+class RaceProgress:
     """
     :type publish_time: int
     :type feed_time: int
@@ -209,9 +209,8 @@ class RaceProgress(BaseResource):
     """
 
     def __init__(self, **kwargs):
-        super(RaceProgress, self).__init__(**kwargs)
         self.feed_time_epoch = kwargs.get("ft")
-        self.feed_time = self.strip_datetime(kwargs.get("ft"))
+        self.feed_time = BaseResource.strip_datetime(kwargs.get("ft"))
         self.gate_name = kwargs.get("g")
         self.sectional_time = kwargs.get("st")
         self.running_time = kwargs.get("rt")
@@ -221,7 +220,7 @@ def __init__(self, **kwargs):
         self.jumps = kwargs.get("J")
 
 
-class RaceChange(BaseResource):
+class RaceChange:
     """
     :type publish_time: int
     :type feed_time: int
@@ -235,9 +234,8 @@ class RaceChange(BaseResource):
     """
 
     def __init__(self, **kwargs):
-        super(RaceChange, self).__init__(**kwargs)
         self.feed_time_epoch = kwargs.get("ft")
-        self.feed_time = self.strip_datetime(kwargs.get("ft"))
+        self.feed_time = BaseResource.strip_datetime(kwargs.get("ft"))
         self.selection_id = kwargs.get("id")
         self.lat = kwargs.get("lat")
         self.long = kwargs.get("long")

betfairlightweight/streaming/betfairstream.py

@@ -127,7 +127,7 @@ def subscribe_to_markets(
         }
         if initial_clk and clk:
             # if resubscribe only update unique_id
-            self.listener.stream_unique_id = unique_id
+            self.listener.update_unique_id(unique_id)
         else:
             self.listener.register_stream(unique_id, "marketSubscription")
         self._send(message)
@@ -166,7 +166,7 @@ def subscribe_to_orders(
         }
         if initial_clk and clk:
             # if resubscribe only update unique_id
-            self.listener.stream_unique_id = unique_id
+            self.listener.update_unique_id(unique_id)
         else:
             self.listener.register_stream(unique_id, "orderSubscription")
         self._send(message)
@@ -295,15 +295,19 @@ class HistoricalStream:
     historical data.
     """
 
-    def __init__(self, file_path: str, listener: BaseListener, operation: str):
+    def __init__(
+        self, file_path: str, listener: BaseListener, operation: str, unique_id: int
+    ):
         """
         :param str file_path: Directory of betfair data
         :param BaseListener listener: Listener object
         :param str operation: Operation type
+        :param int unique_id: Stream id (added to updates)
         """
         self.file_path = file_path
         self.listener = listener
         self.operation = operation
+        self.unique_id = unique_id
         self._running = False
 
     def start(self) -> None:
@@ -314,7 +318,7 @@ def stop(self) -> None:
         self._running = False
 
     def _read_loop(self) -> None:
-        self.listener.register_stream(0, self.operation)
+        self.listener.register_stream(self.unique_id, self.operation)
         with open(self.file_path, "r") as f:
             for update in f:
                 if self.listener.on_data(update) is False:
@@ -338,7 +342,7 @@ def get_generator(self):
 
     def _read_loop(self) -> dict:
         self._running = True
-        self.listener.register_stream(0, self.operation)
+        self.listener.register_stream(self.unique_id, self.operation)
         with open(self.file_path, "r") as f:
             for update in f:
                 if self.listener.on_data(update) is False: