Repository: liampauling/flumine
Pull request #336 (2020-11-26 11:30:20)

`_process` updated to not call `on_process` which reduces some functi…
…on calls (5-10% improvement per market)


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index 12b7c01..0251eca 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,17 +3,6 @@
 Release History
 ---------------
 
-1.14.12 (2020-11-28)
-+++++++++++++++++++
-
-**Improvements**
-
-- 'on_process' function optimised
-
-**Libraries**
-
-- betfairlightweight upgraded to 2.10.2
-
 1.14.11 (2020-11-25)
 +++++++++++++++++++
 
diff --git a/flumine/__version__.py b/flumine/__version__.py
index 4101cdd..cac8f74 100644
--- a/flumine/__version__.py
+++ b/flumine/__version__.py
@@ -1,6 +1,6 @@
 __title__ = "flumine"
 __description__ = "Betfair trading framework"
 __url__ = "https://github.com/liampauling/flumine"
-__version__ = "1.14.12"
+__version__ = "1.14.11"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/streams/historicalstream.py b/flumine/streams/historicalstream.py
index 8706def..5817aed 100644
--- a/flumine/streams/historicalstream.py
+++ b/flumine/streams/historicalstream.py
@@ -2,6 +2,7 @@ import logging
 import datetime
 from betfairlightweight.streaming import StreamListener, HistoricalGeneratorStream
 from betfairlightweight.streaming.stream import MarketStream
+from betfairlightweight.streaming.cache import MarketBookCache
 from betfairlightweight.resources.baseresource import BaseResource
 
 from .basestream import BaseStream
@@ -14,8 +15,39 @@ class FlumineMarketStream(MarketStream):
     Custom bflw stream to speed up processing
     by limiting to inplay/not inplay or limited
     seconds to start.
+    `_process` updated to not call `on_process`
+    which reduces some function calls.
     """
 
+    def _process(self, data: list, publish_time: int) -> bool:
+        for market_book in data:
+            market_id = market_book["id"]
+            full_image = market_book.get("img", False)
+            market_book_cache = self._caches.get(market_id)
+
+            if (
+                full_image or market_book_cache is None
+            ):  # historic data does not contain img
+                if "marketDefinition" not in market_book:
+                    logger.error(
+                        "[%s: %s]: Unable to add %s to cache due to marketDefinition "
+                        "not being present (make sure EX_MARKET_DEF is requested)"
+                        % (self, self.unique_id, market_id)
+                    )
+                    continue
+                market_book_cache = MarketBookCache(
+                    publish_time=publish_time, **market_book
+                )
+                self._caches[market_id] = market_book_cache
+                logger.info(
+                    "[%s: %s]: %s added, %s markets in cache"
+                    % (self, self.unique_id, market_id, len(self._caches))
+                )
+
+            market_book_cache.update_cache(market_book, publish_time)
+            self._updates_processed += 1
+        return False
+
     def snap(self, market_ids: list = None) -> list:
         market_books = []
         for cache in list(self._caches.values()):
diff --git a/requirements.txt b/requirements.txt
index 706d632..f22e786 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,4 @@
-betfairlightweight==2.10.2
+betfairlightweight==2.10.1
 tenacity==5.0.3
 python-json-logger==2.0.1
 requests


Test patch:
diff --git a/tests/test_streams.py b/tests/test_streams.py
index a8aa313..f305541 100644
--- a/tests/test_streams.py
+++ b/tests/test_streams.py
@@ -457,6 +457,23 @@ class TestFlumineMarketStream(unittest.TestCase):
         self.assertEqual(self.stream._listener, self.listener)
         self.assertEqual(self.stream._lookup, "mc")
 
+    @mock.patch("flumine.streams.historicalstream.MarketBookCache")
+    def test__process(self, mock_cache):
+        self.assertFalse(
+            self.stream._process(
+                [{"id": "1.23", "img": {1: 2}, "marketDefinition": {"runners": []}}],
+                12345,
+            )
+        )
+        self.assertEqual(len(self.stream._caches), 1)
+        self.assertEqual(self.stream._updates_processed, 1)
+        mock_cache.assert_called_with(
+            publish_time=12345, id="1.23", img={1: 2}, marketDefinition={"runners": []}
+        )
+        mock_cache().update_cache.assert_called_with(
+            {"id": "1.23", "img": {1: 2}, "marketDefinition": {"runners": []}}, 12345
+        )
+
     def test_snap_inplay(self):
         # inPlay
         self.stream = historicalstream.FlumineMarketStream(