Repository: liampauling/flumine
Pull request #562 (2022-02-17 13:38:13)

blotter lookups updated to lists (order_status)
OrderStreams customer_strategy_refs fix when None provided
betfairlightweight upgraded to 2.16.1


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index 9b1fbe7..80ac61e 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,6 +3,22 @@
 Release History
 ---------------
 
+1.21.6 (2022-02-18)
++++++++++++++++++++
+
+**Improvements**
+
+- blotter lookups updated to lists (order_status)
+
+**Bug Fixes**
+
+- OrderStreams customer_strategy_refs fix when None provided
+- Handle null market_filter when creating streams
+
+**Libraries**
+
+- betfairlightweight upgraded to 2.16.1
+
 1.21.5 (2022-02-14)
 +++++++++++++++++++
 
diff --git a/flumine/__version__.py b/flumine/__version__.py
index 5aae9a9..b745569 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.21.5"
+__version__ = "1.21.6"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/markets/blotter.py b/flumine/markets/blotter.py
index 8410639..eb78816 100644
--- a/flumine/markets/blotter.py
+++ b/flumine/markets/blotter.py
@@ -1,5 +1,6 @@
 import logging
-from typing import Iterable, Optional
+import warnings
+from typing import Iterable, Optional, List
 from collections import defaultdict
 
 from ..order.ordertype import OrderTypes
@@ -56,13 +57,20 @@ class Blotter:
     def strategy_orders(
         self,
         strategy,
-        order_status: Optional[OrderStatus] = None,
+        order_status: Optional[List[OrderStatus]] = None,
         matched_only: Optional[bool] = None,
     ) -> list:
         """Returns all orders related to a strategy."""
         orders = self._strategy_orders[strategy]
         if order_status:
-            orders = [o for o in orders if o.status == order_status]
+            if isinstance(order_status, OrderStatus):
+                warnings.warn(
+                    "`order_status` to accept List[OrderStatus] only from v1.22.0",
+                    DeprecationWarning,
+                )
+                orders = [o for o in orders if o.status == order_status]
+            else:
+                orders = [o for o in orders if o.status in order_status]
         if matched_only:
             orders = [o for o in orders if o.size_matched > 0]
         return orders
@@ -72,13 +80,20 @@ class Blotter:
         strategy,
         selection_id: int,
         handicap: float = 0,
-        order_status: Optional[OrderStatus] = None,
+        order_status: Optional[List[OrderStatus]] = None,
         matched_only: Optional[bool] = None,
     ) -> list:
         """Returns all orders related to a strategy selection."""
         orders = self._strategy_selection_orders[(strategy, selection_id, handicap)]
         if order_status:
-            orders = [o for o in orders if o.status == order_status]
+            if isinstance(order_status, OrderStatus):
+                warnings.warn(
+                    "`order_status` to accept List[OrderStatus] only from v1.22.0",
+                    DeprecationWarning,
+                )
+                orders = [o for o in orders if o.status == order_status]
+            else:
+                orders = [o for o in orders if o.status in order_status]
         if matched_only:
             orders = [o for o in orders if o.size_matched > 0]
         return orders
diff --git a/flumine/streams/datastream.py b/flumine/streams/datastream.py
index 6199a32..f1cc0e1 100644
--- a/flumine/streams/datastream.py
+++ b/flumine/streams/datastream.py
@@ -1,4 +1,5 @@
 import logging
+from typing import Optional
 from tenacity import retry
 from betfairlightweight import StreamListener, filters, BetfairError
 from betfairlightweight.streaming.stream import BaseStream as BFBaseStream
@@ -28,7 +29,7 @@ class FlumineListener(StreamListener):
 
 
 class FlumineStream(BFBaseStream):
-    def on_process(self, caches: list) -> None:
+    def on_process(self, caches: list, publish_time: Optional[int] = None) -> None:
         output = RawDataEvent(caches)
         self.output_queue.put(output)
 
@@ -172,7 +173,9 @@ class OrderDataStream(DataStream):
         try:
             self.stream_id = self._stream.subscribe_to_orders(
                 order_filter=filters.streaming_order_filter(
-                    customer_strategy_refs=[config.customer_strategy_ref],
+                    customer_strategy_refs=[config.customer_strategy_ref]
+                    if config.customer_strategy_ref
+                    else None,
                     partition_matched_by_strategy_ref=True,
                     include_overall_position=False,
                 ),
diff --git a/flumine/streams/orderstream.py b/flumine/streams/orderstream.py
index 74a184f..ed05931 100644
--- a/flumine/streams/orderstream.py
+++ b/flumine/streams/orderstream.py
@@ -39,7 +39,9 @@ class OrderStream(BaseStream):
         try:
             self.stream_id = self._stream.subscribe_to_orders(
                 order_filter=filters.streaming_order_filter(
-                    customer_strategy_refs=[config.customer_strategy_ref],
+                    customer_strategy_refs=[config.customer_strategy_ref]
+                    if config.customer_strategy_ref
+                    else None,
                     partition_matched_by_strategy_ref=True,
                     include_overall_position=False,
                 ),
diff --git a/flumine/streams/streams.py b/flumine/streams/streams.py
index f325795..cf1a345 100644
--- a/flumine/streams/streams.py
+++ b/flumine/streams/streams.py
@@ -79,7 +79,7 @@ class Streams:
     """ market data """
 
     def add_stream(self, strategy: BaseStrategy) -> None:
-        if isinstance(strategy.market_filter, dict):
+        if isinstance(strategy.market_filter, dict) or strategy.market_filter is None:
             market_filters = [strategy.market_filter]
         else:
             market_filters = strategy.market_filter
diff --git a/requirements.txt b/requirements.txt
index 58b39f9..bd428f3 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,4 @@
-betfairlightweight==2.16.0
+betfairlightweight==2.16.1
 tenacity>=7.0.0,<=8.0.1
 python-json-logger==2.0.2
 requests


Test patch:
diff --git a/tests/test_blotter.py b/tests/test_blotter.py
index c63f4c6..318fdae 100644
--- a/tests/test_blotter.py
+++ b/tests/test_blotter.py
@@ -57,13 +57,13 @@ class BlotterTest(unittest.TestCase):
         )
         self.assertEqual(
             self.blotter.strategy_orders(
-                69, order_status=OrderStatus.EXECUTION_COMPLETE
+                69, order_status=[OrderStatus.EXECUTION_COMPLETE]
             ),
             [mock_order_two, mock_order_three],
         )
         self.assertEqual(
             self.blotter.strategy_orders(
-                69, order_status=OrderStatus.EXECUTION_COMPLETE, matched_only=True
+                69, order_status=[OrderStatus.EXECUTION_COMPLETE], matched_only=True
             ),
             [mock_order_three],
         )
@@ -91,13 +91,17 @@ class BlotterTest(unittest.TestCase):
         )
         self.assertEqual(
             self.blotter.strategy_selection_orders(
-                69, 2, 3, order_status=OrderStatus.EXECUTION_COMPLETE
+                69, 2, 3, order_status=[OrderStatus.EXECUTION_COMPLETE]
             ),
             [mock_order_two, mock_order_three],
         )
         self.assertEqual(
             self.blotter.strategy_selection_orders(
-                69, 2, 3, order_status=OrderStatus.EXECUTION_COMPLETE, matched_only=True
+                69,
+                2,
+                3,
+                order_status=[OrderStatus.EXECUTION_COMPLETE],
+                matched_only=True,
             ),
             [mock_order_three],
         )
diff --git a/tests/test_streams.py b/tests/test_streams.py
index db21bc4..418d387 100644
--- a/tests/test_streams.py
+++ b/tests/test_streams.py
@@ -193,6 +193,24 @@ class StreamsTest(unittest.TestCase):
             conflate_ms=mock_strategy.conflate_ms,
         )
 
+    @mock.patch("flumine.streams.streams.Streams._increment_stream_id")
+    def test_add_stream_none(self, mock_increment):
+        mock_strategy = mock.Mock(market_filter=None)
+        mock_stream_class = mock.Mock()
+        mock_strategy.stream_class = mock_stream_class
+
+        self.streams.add_stream(mock_strategy)
+        self.assertEqual(len(self.streams), 1)
+        mock_increment.assert_called_with()
+        mock_strategy.stream_class.assert_called_with(
+            flumine=self.mock_flumine,
+            stream_id=mock_increment(),
+            market_filter=mock_strategy.market_filter,
+            market_data_filter=mock_strategy.market_data_filter,
+            streaming_timeout=mock_strategy.streaming_timeout,
+            conflate_ms=mock_strategy.conflate_ms,
+        )
+
     @mock.patch("flumine.streams.streams.Streams._increment_stream_id")
     def test_add_stream_multi(self, mock_increment):
         mock_strategy = mock.Mock(