Repository: liampauling/flumine
Pull request #247 (2020-08-03 07:30:44)

history and version update


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index 8244bf7..9759516 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,6 +3,21 @@
 Release History
 ---------------
 
+1.10.2 (2020-08-03)
++++++++++++++++++++
+
+**Improvements**
+
+- _process_raw_data refactored to create market objects and call market.closed_market on closure
+
+**Bug Fixes**
+
+- Docs typo (thanks @petercoles)
+
+**Libraries**
+
+- betfairlightweight upgraded to 2.7.1
+
 1.10.1 (2020-07-20)
 +++++++++++++++++++
 
diff --git a/docs/quickstart.md b/docs/quickstart.md
index 03ec30e..082155d 100644
--- a/docs/quickstart.md
+++ b/docs/quickstart.md
@@ -58,7 +58,7 @@ strategy = ExampleStrategy(
     market_data_filter=streaming_market_data_filter(fields=["EX_ALL_OFFERS"])
 )
 
-flumine.add_strategy(strategy)
+framework.add_strategy(strategy)
 ```
 
 The framework can now be started:
diff --git a/examples/strategies/marketrecorder.py b/examples/strategies/marketrecorder.py
index fdc7b67..cd30016 100644
--- a/examples/strategies/marketrecorder.py
+++ b/examples/strategies/marketrecorder.py
@@ -48,13 +48,8 @@ class MarketRecorder(BaseStrategy):
                 json.dumps({"op": "mcm", "clk": None, "pt": publish_time, "mc": [data]})
                 + "\n"
             )
-        if (
-            "marketDefinition" in data
-            and data["marketDefinition"]["status"] == "CLOSED"
-        ):
-            self._on_market_closed(data)
 
-    def _on_market_closed(self, data: dict) -> None:
+    def process_closed_market(self, market, data: dict) -> None:
         market_id = data.get(self.MARKET_ID_LOOKUP)
         if market_id in self._loaded_markets:
             if self._force_update:
@@ -91,7 +86,7 @@ class MarketRecorder(BaseStrategy):
         zip_file_dir = self._zip_file(file_dir, market_id)
 
         # core load code
-        self._load(zip_file_dir, market_definition)
+        self._load(market, zip_file_dir, market_definition)
 
         # clean up
         self._clean_up()
@@ -110,7 +105,7 @@ class MarketRecorder(BaseStrategy):
             )
         return zip_file_directory
 
-    def _load(self, zip_file_dir: str, market_definition: dict) -> None:
+    def _load(self, market, zip_file_dir: str, market_definition: dict) -> None:
         pass
 
     def _clean_up(self) -> None:
@@ -154,7 +149,7 @@ class S3MarketRecorder(MarketRecorder):
         super().add()
         self.s3.head_bucket(Bucket=self._bucket)  # validate bucket/access
 
-    def _load(self, zip_file_dir: str, market_definition: dict) -> None:
+    def _load(self, market, zip_file_dir: str, market_definition: dict) -> None:
         # note this will block the main handler queue during upload
         # todo create background worker instead?
         event_type_id = (
@@ -179,3 +174,24 @@ class S3MarketRecorder(MarketRecorder):
             logger.info("%s successfully loaded to s3" % zip_file_dir)
         except (BotoCoreError, Exception) as e:
             logger.error("Error loading to s3: %s" % e)
+
+        # upload marketCatalogue data
+        if self.context.get("load_market_catalogue", True):
+            if market.market_catalogue is None:
+                logger.warning(
+                    "No marketCatalogue data available for %s" % market.market_id
+                )
+                return
+            try:
+                self.s3.put_object(
+                    Body=market.market_catalogue.json(),
+                    Bucket=self._bucket,
+                    Key=os.path.join(
+                        "marketdata", "marketCatalogue", market.market_id,
+                    ),
+                )
+                logger.info(
+                    "%s successfully loaded marketCatalogue to s3" % market.market_id
+                )
+            except (BotoCoreError, Exception) as e:
+                logger.error("Error loading to s3: %s" % e)
diff --git a/flumine/__version__.py b/flumine/__version__.py
index 5abb7ae..b5deeac 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.10.1"
+__version__ = "1.10.2"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/baseflumine.py b/flumine/baseflumine.py
index fc80fa4..e6f66ce 100644
--- a/flumine/baseflumine.py
+++ b/flumine/baseflumine.py
@@ -173,9 +173,27 @@ class BaseFlumine:
     def _process_raw_data(self, event: events.RawDataEvent) -> None:
         stream_id, publish_time, data = event.event
         for datum in data:
+            market_id = datum["id"]
+            market = self.markets.markets.get(market_id)
+            if market is None:
+                market = self._add_market(market_id, None)
+            elif market.closed:
+                self.markets.add_market(market_id, market)
+
+            if (
+                "marketDefinition" in datum
+                and datum["marketDefinition"]["status"] == "CLOSED"
+            ):
+                market.close_market()
+                _closed = True
+            else:
+                _closed = False
+
             for strategy in self.strategies:
                 if stream_id in strategy.stream_ids:
                     strategy.process_raw_data(publish_time, datum)
+                    if _closed:
+                        strategy.process_closed_market(market, datum)
 
     def _process_market_catalogues(self, event: events.MarketCatalogueEvent) -> None:
         for market_catalogue in event.event:
diff --git a/flumine/streams/historicalstream.py b/flumine/streams/historicalstream.py
index 667aa0d..8a92bfb 100644
--- a/flumine/streams/historicalstream.py
+++ b/flumine/streams/historicalstream.py
@@ -67,7 +67,7 @@ class Stream(BFLWBaseStream):
                     if cache.market_definition["inPlay"]:
                         continue
             market_books.append(
-                cache.create_resource(self.unique_id, None, self._lightweight)
+                cache.create_resource(self.unique_id, self._lightweight)
             )
         return market_books
 
@@ -100,7 +100,6 @@ class HistoricalStream(BaseStream):
         pass
 
     def create_generator(self):
-        self._listener.register_stream(0, "marketSubscription")
         stream = HistoricalGeneratorStream(
             file_path=self.market_filter, listener=self._listener
         )
diff --git a/requirements.txt b/requirements.txt
index 1a5036f..aa7c475 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,4 @@
-betfairlightweight==2.6.0
+betfairlightweight==2.7.1
 tenacity==5.0.3
 python-json-logger==0.1.11
 requests


Test patch:
diff --git a/tests/test_baseflumine.py b/tests/test_baseflumine.py
index 393a976..e653dec 100644
--- a/tests/test_baseflumine.py
+++ b/tests/test_baseflumine.py
@@ -167,10 +167,24 @@ class BaseFlumineTest(unittest.TestCase):
         mock_markets.remove_market.assert_called_with(mock_market.market_id)
         mock_middleware.remove_market.assert_called_with(mock_market)
 
-    def test__process_raw_data(self):
+    @mock.patch("flumine.baseflumine.BaseFlumine._add_market")
+    def test__process_raw_data(self, mock__add_market):
         mock_event = mock.Mock()
-        mock_event.event = (12, 12345, {})
+        mock_event.event = (12, 12345, [{"id": "1.23"}])
         self.base_flumine._process_raw_data(mock_event)
+        mock__add_market.assert_called_with("1.23", None)
+
+    @mock.patch("flumine.baseflumine.BaseFlumine._add_market")
+    def test__process_raw_data_closed(self, mock__add_market):
+        mock_event = mock.Mock()
+        mock_event.event = (
+            12,
+            12345,
+            [{"id": "1.23", "marketDefinition": {"status": "CLOSED"}}],
+        )
+        self.base_flumine._process_raw_data(mock_event)
+        mock__add_market.assert_called_with("1.23", None)
+        mock__add_market().close_market.assert_called()
 
     @mock.patch("flumine.baseflumine.events")
     @mock.patch("flumine.baseflumine.BaseFlumine.log_control")