Repository: liampauling/flumine (python)

Issue:
Inconsistent behaviour when running framework with multiple added strategies
```py
# first strat should take longer
strategy = LowestLayer(
    market_filter={"markets": markets},
    max_order_exposure=1000,
    max_selection_exposure=105,
    context={"stake": 2},
)
# second strat should be faster
strategy2 = LowestLayer(
    market_filter={"markets": markets,
	"listener_kwargs": {"inplay": False, "seconds_to_start": 600}},
    max_order_exposure=1000,
    max_selection_exposure=105,
    context={"stake": 2},
)

# add both strats
framework.add_strategy(strategy)
framework.add_strategy(strategy2)

# run framework
start = time.process_time()
framework.run()
end = time.process_time()
```
when adding the strategies in this order it takes 20seconds (same time as only running ‘strategy’)
```py
framework.add_strategy(strategy)
framework.add_strategy(strategy2)
```
however in this order it takes 10seconds (same time as only running ‘strategy2’)
```py
framework.add_strategy(strategy2)
framework.add_strategy(strategy)
```

Potentiall issue caused by setting of "listener_kwargs", not reset in the second strategy.

Pull request:
close #478
store and check listener kwargs when creating historical streams

Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index 77c2ed5..dae9682 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,6 +3,13 @@
 Release History
 ---------------
 
+1.19.10 (2021-08-23)
++++++++++++++++++++
+
+**Bug Fixes**
+
+- #478 Listener kwargs / create bugfix
+
 1.19.9 (2021-08-16)
 +++++++++++++++++++
 
diff --git a/flumine/__version__.py b/flumine/__version__.py
index 5f845a7..ceaf308 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.19.9"
+__version__ = "1.19.10"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/streams/basestream.py b/flumine/streams/basestream.py
index a1b4a9a..8957465 100644
--- a/flumine/streams/basestream.py
+++ b/flumine/streams/basestream.py
@@ -42,6 +42,7 @@ class BaseStream(threading.Thread):
         self.event_processing = event_processing
         self.event_id = event_id
         self.operation = operation
+        self.listener_kwargs = listener_kwargs
         self._listener = self.LISTENER(
             output_queue=self._output_queue,
             max_latency=self.MAX_LATENCY,
diff --git a/flumine/streams/streams.py b/flumine/streams/streams.py
index 92a05e0..58b843a 100644
--- a/flumine/streams/streams.py
+++ b/flumine/streams/streams.py
@@ -110,6 +110,7 @@ class Streams:
             if (
                 stream.market_filter == market
                 and stream.event_processing == event_processing
+                and stream.listener_kwargs == listener_kwargs
             ):
                 return stream
         else:


Test patch:
diff --git a/tests/test_streams.py b/tests/test_streams.py
index 76dafc8..fa3ada0 100644
--- a/tests/test_streams.py
+++ b/tests/test_streams.py
@@ -221,16 +221,52 @@ class StreamsTest(unittest.TestCase):
     def test_add_historical_stream_old(self):
         self.mock_flumine.BACKTEST = True
         mock_strategy = mock.Mock()
-        mock_stream = mock.Mock(spec=streams.HistoricalStream, event_processing=False)
+        mock_stream = mock.Mock(
+            spec=streams.HistoricalStream, event_processing=False, listener_kwargs={}
+        )
         mock_stream.market_filter = "GANG"
         self.streams._streams = [mock_stream]
 
         stream = self.streams.add_historical_stream(
-            mock_strategy, "GANG", event_processing=False
+            mock_strategy, "GANG", event_processing=False, **{}
         )
         self.assertEqual(stream, mock_stream)
         self.assertEqual(len(self.streams), 1)
 
+    @mock.patch("flumine.streams.streams.get_file_md")
+    @mock.patch("flumine.streams.streams.HistoricalStream")
+    @mock.patch("flumine.streams.streams.Streams._increment_stream_id")
+    def test_add_historical_stream_kwargs(
+        self, mock_increment, mock_historical_stream_class, mock_get_file_md
+    ):
+        mock_historical_stream_class.__name__ = "test"
+        self.mock_flumine.BACKTEST = True
+        mock_strategy = mock.Mock()
+        mock_stream = mock.Mock(
+            spec=streams.HistoricalStream, event_processing=False, listener_kwargs={}
+        )
+        mock_stream.market_filter = "GANG"
+        self.streams._streams = [mock_stream]
+
+        self.streams.add_historical_stream(
+            mock_strategy, "GANG", event_processing=False, **{"inplay": True}
+        )
+        self.assertEqual(len(self.streams), 2)
+        mock_increment.assert_called_with()
+        mock_get_file_md.assert_called_with("GANG", "eventId")
+        mock_historical_stream_class.assert_called_with(
+            flumine=self.mock_flumine,
+            stream_id=mock_increment(),
+            market_filter="GANG",
+            market_data_filter=mock_strategy.market_data_filter,
+            streaming_timeout=mock_strategy.streaming_timeout,
+            conflate_ms=mock_strategy.conflate_ms,
+            output_queue=False,
+            event_processing=False,
+            event_id=mock_get_file_md(),
+            inplay=True,
+        )
+
     @mock.patch("flumine.streams.streams.get_file_md")
     @mock.patch("flumine.streams.streams.HistoricalStream")
     def test_add_historical_stream_event_processing(
@@ -333,6 +369,7 @@ class TestBaseStream(unittest.TestCase):
             event_processing=True,
             event_id="123",
             operation="test",
+            **{"calculate_market_tv": True},
         )
 
     def test_init(self):
@@ -349,6 +386,7 @@ class TestBaseStream(unittest.TestCase):
         self.assertTrue(self.stream.event_processing)
         self.assertEqual(self.stream.event_id, "123")
         self.assertEqual(self.stream.operation, "test")
+        self.assertEqual(self.stream.listener_kwargs, {"calculate_market_tv": True})
 
     def test_run(self):
         with self.assertRaises(NotImplementedError):