Repository: liampauling/flumine (python)

Issue:
Backtesting with multiple s3 filepaths results in RequestTimeTooSkewed error
Even when using context manager e.g. 
`with mock.patch("builtins.open", smart_open.open):
    with framework.simulated_datetime.real_time():
        framework.add_strategy(strategy)
    framework.run()`
the backtest results in errors. The problem stems from the _readloop() function in FlumineHistoricalGeneratorStream using the SimulatedDateTime object when it tries to open the file

Pull request:
reset real datetime added to allow s3 download (RequestTimeTooSkewed)
close #519

Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index 728993f..ff573cf 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,6 +3,21 @@
 Release History
 ---------------
 
+1.20.8 (2021-11-01)
++++++++++++++++++++
+
+**Improvements**
+
+- Handle list of market filters in strategy
+
+**Bug Fixes**
+
+- #519 reset real datetime added to allow s3 download (RequestTimeTooSkewed)
+
+**Libraries**
+
+- black upgraded to 21.10b0
+
 1.20.7 (2021-10-25)
 +++++++++++++++++++
 
diff --git a/docs/strategies.md b/docs/strategies.md
index 1ab0ed1..044cb65 100644
--- a/docs/strategies.md
+++ b/docs/strategies.md
@@ -6,7 +6,7 @@ The base strategy class should be used for all strategies and contains the follo
 
 ### Parameters
 
-- `market_filter` Streaming market filter required
+- `market_filter` Streaming market filter or list of filters required
 - `market_data_filter` Streaming market data filter required
 - `streaming_timeout` Streaming timeout, will call snap() on cache every x seconds
 - `conflate_ms` Streaming conflate
diff --git a/flumine/__version__.py b/flumine/__version__.py
index 441f4ed..243ac63 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.20.7"
+__version__ = "1.20.8"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/controls/loggingcontrols.py b/flumine/controls/loggingcontrols.py
index d08b223..2c84b9e 100644
--- a/flumine/controls/loggingcontrols.py
+++ b/flumine/controls/loggingcontrols.py
@@ -34,7 +34,7 @@ class LoggingControl(Thread):
                     self.process_event(event)
                 except Exception as e:
                     logger.critical(
-                        "{0} exception raised in {0}".format(e, self.NAME),
+                        "{0} exception raised in {1}".format(e, self.NAME),
                         exc_info=True,
                         extra={"event": event},
                     )
diff --git a/flumine/strategy/strategy.py b/flumine/strategy/strategy.py
index 4925bd4..f487495 100644
--- a/flumine/strategy/strategy.py
+++ b/flumine/strategy/strategy.py
@@ -1,5 +1,5 @@
 import logging
-from typing import Type, Iterator
+from typing import Type, Iterator, Union
 from betfairlightweight import filters
 from betfairlightweight.resources import MarketBook
 
@@ -38,7 +38,7 @@ class BaseStrategy:
 
     def __init__(
         self,
-        market_filter: dict,
+        market_filter: Union[dict, list],
         market_data_filter: dict = None,
         streaming_timeout: float = None,
         conflate_ms: int = None,
@@ -53,7 +53,7 @@ class BaseStrategy:
         multi_order_trades: bool = False,
     ):
         """
-        :param market_filter: Streaming market filter
+        :param market_filter: Streaming market filter dict or list of market filters
         :param market_data_filter: Streaming market data filter
         :param streaming_timeout: Streaming timeout in seconds, will call snap() on cache
         :param conflate_ms: Streaming conflation
diff --git a/flumine/streams/streams.py b/flumine/streams/streams.py
index 26c89d7..33b912c 100644
--- a/flumine/streams/streams.py
+++ b/flumine/streams/streams.py
@@ -76,8 +76,7 @@ class Streams:
             elif events:
                 raise NotImplementedError()
         else:
-            stream = self.add_stream(strategy)
-            strategy.streams.append(stream)
+            self.add_stream(strategy)
 
     def add_client(self, client: BaseClient) -> None:
         if client.order_stream:
@@ -88,38 +87,44 @@ class Streams:
 
     """ market data """
 
-    def add_stream(self, strategy: BaseStrategy) -> Union[MarketStream, DataStream]:
-        for stream in self:  # check if market stream already exists
-            if (
-                isinstance(stream, strategy.stream_class)
-                and stream.market_filter == strategy.market_filter
-                and stream.market_data_filter == strategy.market_data_filter
-                and stream.streaming_timeout == strategy.streaming_timeout
-                and stream.conflate_ms == strategy.conflate_ms
-            ):
+    def add_stream(self, strategy: BaseStrategy) -> None:
+        if isinstance(strategy.market_filter, dict):
+            market_filters = [strategy.market_filter]
+        else:
+            market_filters = strategy.market_filter
+        for market_filter in market_filters:
+            for stream in self:  # check if market stream already exists
+                if (
+                    isinstance(stream, strategy.stream_class)
+                    and stream.market_filter == market_filter
+                    and stream.market_data_filter == strategy.market_data_filter
+                    and stream.streaming_timeout == strategy.streaming_timeout
+                    and stream.conflate_ms == strategy.conflate_ms
+                ):
+                    logger.info(
+                        "Using {0} ({1}) for strategy {2}".format(
+                            strategy.stream_class, stream.stream_id, strategy
+                        )
+                    )
+                    strategy.streams.append(stream)
+                    break
+            else:  # nope? lets create a new one
+                stream_id = self._increment_stream_id()
                 logger.info(
-                    "Using {0} ({1}) for strategy {2}".format(
-                        strategy.stream_class, stream.stream_id, strategy
+                    "Creating new {0} ({1}) for strategy {2}".format(
+                        strategy.stream_class, stream_id, strategy
                     )
                 )
-                return stream
-        else:  # nope? lets create a new one
-            stream_id = self._increment_stream_id()
-            logger.info(
-                "Creating new {0} ({1}) for strategy {2}".format(
-                    strategy.stream_class, stream_id, strategy
+                stream = strategy.stream_class(
+                    flumine=self.flumine,
+                    stream_id=stream_id,
+                    market_filter=market_filter,
+                    market_data_filter=strategy.market_data_filter,
+                    streaming_timeout=strategy.streaming_timeout,
+                    conflate_ms=strategy.conflate_ms,
                 )
-            )
-            stream = strategy.stream_class(
-                flumine=self.flumine,
-                stream_id=stream_id,
-                market_filter=strategy.market_filter,
-                market_data_filter=strategy.market_data_filter,
-                streaming_timeout=strategy.streaming_timeout,
-                conflate_ms=strategy.conflate_ms,
-            )
-            self._streams.append(stream)
-            return stream
+                self._streams.append(stream)
+                strategy.streams.append(stream)
 
     def add_historical_stream(
         self,


Test patch:
diff --git a/flumine/backtest/backtest.py b/flumine/backtest/backtest.py
index 38441c2..8f34813 100644
--- a/flumine/backtest/backtest.py
+++ b/flumine/backtest/backtest.py
@@ -55,6 +55,7 @@ class FlumineBacktest(BaseFlumine):
                                 "markets": [s.market_filter for s in streams],
                             },
                         )
+                        self.simulated_datetime.reset_real_datetime()
                         # create cycles
                         cycles = []  # [[epoch, [MarketBook], gen], ..]
                         for stream in streams:
@@ -90,6 +91,7 @@ class FlumineBacktest(BaseFlumine):
                                 ),
                                 extra={"market": stream.market_filter},
                             )
+                            self.simulated_datetime.reset_real_datetime()
                             stream_gen = stream.create_generator()
                             for event in stream_gen():
                                 self._process_market_books(
diff --git a/flumine/backtest/utils.py b/flumine/backtest/utils.py
index b3941c3..21fef75 100644
--- a/flumine/backtest/utils.py
+++ b/flumine/backtest/utils.py
@@ -17,6 +17,9 @@ class SimulatedDateTime:
     def __call__(self, pt: datetime.datetime):
         config.current_time = pt
 
+    def reset_real_datetime(self):
+        config.current_time = self._real_datetime.utcnow()
+
     @contextmanager
     def real_time(self):
         datetime.datetime = self._real_datetime
diff --git a/requirements-test.txt b/requirements-test.txt
index f1710c0..ef646f0 100644
--- a/requirements-test.txt
+++ b/requirements-test.txt
@@ -1,5 +1,5 @@
 # Tests & Linting
-black==21.9b0
+black==21.10b0
 coverage
 
 # Documentation
diff --git a/tests/test_backtest_utils.py b/tests/test_backtest_utils.py
index ab04b35..4a08ec5 100644
--- a/tests/test_backtest_utils.py
+++ b/tests/test_backtest_utils.py
@@ -19,6 +19,14 @@ class SimulatedDateTimeTest(unittest.TestCase):
     def test_init(self):
         self.assertIsNone(self.s._real_datetime)
 
+    @mock.patch("flumine.backtest.utils.config")
+    def test_reset_real_datetime(self, mock_config):
+        mock_real_datetime = mock.Mock()
+        self.s._real_datetime = mock_real_datetime
+        self.s.reset_real_datetime()
+        mock_real_datetime.utcnow.assert_called()
+        self.assertEqual(mock_config.current_time, mock_real_datetime.utcnow())
+
     @mock.patch("flumine.backtest.utils.config")
     def test_call(self, mock_config):
         mock_dt = mock.Mock()
diff --git a/tests/test_baseflumine.py b/tests/test_baseflumine.py
index 653565a..5e05723 100644
--- a/tests/test_baseflumine.py
+++ b/tests/test_baseflumine.py
@@ -43,7 +43,7 @@ class BaseFlumineTest(unittest.TestCase):
     @mock.patch("flumine.baseflumine.events")
     @mock.patch("flumine.baseflumine.BaseFlumine.log_control")
     def test_add_strategy(self, mock_log_control, mock_events):
-        mock_strategy = mock.Mock()
+        mock_strategy = mock.Mock(market_filter={})
         mock_client = mock.Mock()
         self.base_flumine.add_strategy(mock_strategy, mock_client)
         self.assertEqual(len(self.base_flumine.strategies), 1)
diff --git a/tests/test_streams.py b/tests/test_streams.py
index af5e90f..c42ec1d 100644
--- a/tests/test_streams.py
+++ b/tests/test_streams.py
@@ -25,17 +25,13 @@ class StreamsTest(unittest.TestCase):
     def test_call(self, mock_add_stream):
         mock_strategy = mock.Mock(streams=[], raw_data=False)
         self.streams(mock_strategy)
-
         mock_add_stream.assert_called_with(mock_strategy)
-        self.assertEqual(len(mock_strategy.streams), 1)
 
     @mock.patch("flumine.streams.streams.Streams.add_stream")
     def test_call_data_stream(self, mock_add_stream):
         mock_strategy = mock.Mock(streams=[], stream_class=datastream.DataStream)
         self.streams(mock_strategy)
-
         mock_add_stream.assert_called_with(mock_strategy)
-        self.assertEqual(len(mock_strategy.streams), 1)
 
     @mock.patch("flumine.streams.streams.get_file_md")
     @mock.patch("flumine.streams.streams.Streams.add_historical_stream")
@@ -181,7 +177,7 @@ class StreamsTest(unittest.TestCase):
 
     @mock.patch("flumine.streams.streams.Streams._increment_stream_id")
     def test_add_stream_new(self, mock_increment):
-        mock_strategy = mock.Mock()
+        mock_strategy = mock.Mock(market_filter={})
         mock_stream_class = mock.Mock()
         mock_strategy.stream_class = mock_stream_class
 
@@ -197,26 +193,33 @@ class StreamsTest(unittest.TestCase):
             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(
+            market_filter=[{1: 2}, {3: 4}], stream_class=streams.MarketStream
+        )
+        self.streams.add_stream(mock_strategy)
+        self.assertEqual(len(self.streams), 2)
+        mock_increment.assert_called_with()
+
     @mock.patch("flumine.streams.streams.Streams._increment_stream_id")
     def test_add_stream_old(self, mock_increment):
-        mock_strategy = mock.Mock()
-        mock_strategy.market_filter = 1
+        mock_strategy = mock.Mock(market_filter={})
         mock_strategy.market_data_filter = 2
         mock_strategy.streaming_timeout = 3
         mock_strategy.conflate_ms = 4
         mock_strategy.stream_class = streams.MarketStream
 
         stream = mock.Mock(spec=streams.MarketStream)
-        stream.market_filter = 1
+        stream.market_filter = {}
         stream.market_data_filter = 2
         stream.streaming_timeout = 3
         stream.conflate_ms = 4
         stream.stream_id = 1001
         self.streams._streams = [stream]
 
-        new_stream = self.streams.add_stream(mock_strategy)
+        self.streams.add_stream(mock_strategy)
         self.assertEqual(len(self.streams), 1)
-        self.assertEqual(stream, new_stream)
         mock_increment.assert_not_called()
 
     @mock.patch("flumine.streams.streams.get_file_md")