Repository: liampauling/flumine
Pull request #370 (2021-01-28 08:58:45)

various optimisations (~5-10%)
- pending packages cleanup
- listener updates to skip unrequired calls
- custom generator with minimal func calls


Patch:
diff --git a/.travis.yml b/.travis.yml
index ed8017b..9e8b5c0 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -6,9 +6,6 @@ install:
   - travis_retry python setup.py install
   - pip install coveralls
 
-env:
-  - SKIP_INTEGRATION_TESTS=0
-
 jobs:
   include:
     - name: "Tests: python 3.6"
diff --git a/HISTORY.rst b/HISTORY.rst
index 1dc465d..7b52ef9 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,6 +3,15 @@
 Release History
 ---------------
 
+1.16.1 (2021-01-28)
++++++++++++++++++++
+
+**Improvements**
+
+- Various optimisations on pending_packages and low level listener updates
+- Cache stream_id when backtesting
+- Always run integrations tests (now possible with faster backtesting from bflw 2.12.0)
+
 1.16.0 (2021-01-25)
 +++++++++++++++++++
 
diff --git a/flumine/__version__.py b/flumine/__version__.py
index 48583c5..12d917b 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.16.0"
+__version__ = "1.16.1"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/strategy/strategy.py b/flumine/strategy/strategy.py
index dabf2af..65c3b25 100644
--- a/flumine/strategy/strategy.py
+++ b/flumine/strategy/strategy.py
@@ -73,6 +73,7 @@ class BaseStrategy:
 
         self._invested = {}  # {(marketId, selectionId, handicap): RunnerContext}
         self.streams = []  # list of streams strategy is subscribed
+        self.historic_stream_ids = []
         self.log_validation_failures = log_validation_failures
 
     def check_market(self, market: Market, market_book: MarketBook) -> bool:
@@ -231,7 +232,10 @@ class BaseStrategy:
 
     @property
     def stream_ids(self) -> list:
-        return [stream.stream_id for stream in self.streams]
+        if self.historic_stream_ids:
+            return self.historic_stream_ids
+        else:
+            return [stream.stream_id for stream in self.streams]
 
     @property
     def info(self) -> dict:
diff --git a/flumine/streams/historicalstream.py b/flumine/streams/historicalstream.py
index 112a6dd..0cff5ba 100644
--- a/flumine/streams/historicalstream.py
+++ b/flumine/streams/historicalstream.py
@@ -57,9 +57,9 @@ class FlumineMarketStream(MarketStream):
             if market_ids and cache.market_id not in market_ids:
                 continue
             # if market has closed send regardless
-            if cache.market_definition["status"] != "CLOSED":
+            if cache._definition_status != "CLOSED":
                 if self._listener.inplay:
-                    if not cache.market_definition["inPlay"]:
+                    if not cache._definition_in_play:
                         continue
                 elif self._listener.seconds_to_start:
                     _now = datetime.datetime.utcfromtimestamp(cache.publish_time / 1e3)
@@ -70,7 +70,7 @@ class FlumineMarketStream(MarketStream):
                     if seconds_to_start > self._listener.seconds_to_start:
                         continue
                 if self._listener.inplay is False:
-                    if cache.market_definition["inPlay"]:
+                    if cache._definition_in_play:
                         continue
             market_books.append(cache.create_resource(self.unique_id, snap=True))
         return market_books
@@ -130,8 +130,24 @@ class HistoricListener(StreamListener):
 
         # remove error handler / operation check
 
-        # skip on_change as we know it is always an update
-        self.stream.on_update(data)
+        # skip on_change / on_update as we know it is always an update
+        publish_time = data["pt"]
+        self.stream._process(data[self.stream._lookup], publish_time)
+
+
+class FlumineHistoricalGeneratorStream(HistoricalGeneratorStream):
+    """Super fast historical stream"""
+
+    def _read_loop(self) -> dict:
+        self.listener.register_stream(self.unique_id, self.operation)
+        listener_on_data = self.listener.on_data  # cache functions
+        stream_snap = self.listener.stream.snap
+        with open(self.file_path, "r") as f:
+            for update in f:
+                listener_on_data(update)
+                data = stream_snap()
+                if data:  # can return empty list
+                    yield data
 
 
 class HistoricalStream(BaseStream):
@@ -150,7 +166,7 @@ class HistoricalStream(BaseStream):
         self._listener.update_clk = (
             False  # do not update clk on updates (not required when backtesting)
         )
-        stream = HistoricalGeneratorStream(
+        stream = FlumineHistoricalGeneratorStream(
             file_path=self.market_filter,
             listener=self._listener,
             operation=self.operation,
diff --git a/flumine/streams/streams.py b/flumine/streams/streams.py
index 5f0f20c..4f50494 100644
--- a/flumine/streams/streams.py
+++ b/flumine/streams/streams.py
@@ -30,6 +30,7 @@ class Streams:
                         strategy, market, **listener_kwargs
                     )
                     strategy.streams.append(stream)
+                    strategy.historic_stream_ids.append(stream.stream_id)
         else:
             stream = self.add_stream(strategy)
             strategy.streams.append(stream)


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 2b3e4ba..d1ae483 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -37,15 +37,11 @@ jobs:
     - name: Unittest
       run: |
         python -m unittest discover -s tests
-      env:
-        SKIP_INTEGRATION_TESTS: 0
         
     - name: Coverage
       run: |
         coverage run -m unittest discover
         coverage report
-      env:
-        SKIP_INTEGRATION_TESTS: 0
 
   deploy:
     needs: test
diff --git a/flumine/backtest/backtest.py b/flumine/backtest/backtest.py
index ae1d5cb..986c4fb 100644
--- a/flumine/backtest/backtest.py
+++ b/flumine/backtest/backtest.py
@@ -6,7 +6,6 @@ from ..events import events
 from .. import config, utils
 from ..clients import ExchangeType
 from ..exceptions import RunError
-from .utils import PendingPackages
 from ..order.orderpackage import OrderPackageType
 from ..order import process
 
@@ -24,7 +23,7 @@ class FlumineBacktest(BaseFlumine):
 
     def __init__(self, client):
         super(FlumineBacktest, self).__init__(client)
-        self._pending_packages = PendingPackages()
+        self._pending_packages = []
 
     def run(self) -> None:
         if self.client.EXCHANGE != ExchangeType.SIMULATED:
@@ -62,7 +61,8 @@ class FlumineBacktest(BaseFlumine):
             config.current_time = market_book.publish_time
 
             # check if there are orders to process
-            self._check_pending_packages()
+            if self._pending_packages:
+                self._check_pending_packages()
 
             if market_book.status == "CLOSED":
                 self._process_close_market(event=events.CloseMarketEvent(market_book))
@@ -113,36 +113,39 @@ class FlumineBacktest(BaseFlumine):
         packages for later execution
         """
         for market in self.markets:
-            bet_delay = market.market_book.bet_delay
-            for order_package in market.blotter.process_orders(self.client, bet_delay):
-                self._pending_packages.append(order_package)
-
-    def _process_order_package(self, order_package) -> None:
-        """Validate trading controls and
-        then execute immediately.
-        """
-        super(FlumineBacktest, self)._process_order_package(order_package)
-        order_package.processed = True
+            if market.blotter.pending_orders:
+                bet_delay = market.market_book.bet_delay
+                for order_package in market.blotter.process_orders(
+                    self.client, bet_delay
+                ):
+                    self._pending_packages.append(order_package)
 
     def _check_pending_packages(self):
+        processed = []
         for order_package in self._pending_packages:
-            _client = order_package.client
+            client = order_package.client
             if order_package.package_type == OrderPackageType.PLACE:
                 if order_package.elapsed_seconds > (
-                    _client.execution.PLACE_LATENCY + order_package.bet_delay
+                    client.execution.PLACE_LATENCY + order_package.bet_delay
                 ):
                     self._process_order_package(order_package)
+                    processed.append(order_package)
             elif order_package.package_type == OrderPackageType.CANCEL:
-                if order_package.elapsed_seconds > _client.execution.CANCEL_LATENCY:
+                if order_package.elapsed_seconds > client.execution.CANCEL_LATENCY:
                     self._process_order_package(order_package)
+                    processed.append(order_package)
             elif order_package.package_type == OrderPackageType.UPDATE:
-                if order_package.elapsed_seconds > _client.execution.UPDATE_LATENCY:
+                if order_package.elapsed_seconds > client.execution.UPDATE_LATENCY:
                     self._process_order_package(order_package)
+                    processed.append(order_package)
             elif order_package.package_type == OrderPackageType.REPLACE:
                 if order_package.elapsed_seconds > (
-                    _client.execution.REPLACE_LATENCY + order_package.bet_delay
+                    client.execution.REPLACE_LATENCY + order_package.bet_delay
                 ):
                     self._process_order_package(order_package)
+                    processed.append(order_package)
+        for p in processed:
+            self._pending_packages.remove(p)
 
     def _monkey_patch_datetime(self):
         config.current_time = datetime.datetime.utcnow()
diff --git a/flumine/backtest/utils.py b/flumine/backtest/utils.py
index b1bb519..1ec135a 100644
--- a/flumine/backtest/utils.py
+++ b/flumine/backtest/utils.py
@@ -1,15 +1,6 @@
 import datetime
 
 
-class PendingPackages(list):
-    """List which only returns packages
-    which haven't been processed.
-    """
-
-    def __iter__(self):
-        return (x for x in list.__iter__(self) if not x.processed)
-
-
 class SimulatedPlaceResponse:
     def __init__(
         self,
diff --git a/tests/test_backtest_utils.py b/tests/test_backtest_utils.py
index dedf776..e69de29 100644
--- a/tests/test_backtest_utils.py
+++ b/tests/test_backtest_utils.py
@@ -1,22 +0,0 @@
-import unittest
-from unittest import mock
-
-from flumine.backtest import utils
-
-
-class BacktestUtilsTest(unittest.TestCase):
-    def test_pending_packages(self):
-        mock_package_one = mock.Mock()
-        mock_package_one.processed = True
-        mock_package_two = mock.Mock()
-        mock_package_two.processed = False
-        mock_package_three = mock.Mock()
-        mock_package_three.processed = False
-        pending = utils.PendingPackages()
-        pending.append(mock_package_one)
-        pending.append(mock_package_two)
-        pending.append(mock_package_three)
-        self.assertEqual(len(pending), 3)
-        self.assertEqual(len([i for i in pending]), 2)
-        mock_package_one.processed = False
-        self.assertEqual(len([i for i in pending]), 3)
diff --git a/tests/test_fluminebacktest.py b/tests/test_fluminebacktest.py
index 0c34a6d..d54e36f 100644
--- a/tests/test_fluminebacktest.py
+++ b/tests/test_fluminebacktest.py
@@ -56,6 +56,7 @@ class FlumineBacktestTest(unittest.TestCase):
         mock__process_backtest_orders,
         mock__process_market_orders,
     ):
+        self.flumine._pending_packages = [mock.Mock()]
         mock_event = mock.Mock()
         mock_market_book = mock.Mock(market_id="1.23")
         mock_market_book.runners = []
@@ -100,12 +101,6 @@ class FlumineBacktestTest(unittest.TestCase):
         )
         self.assertEqual(self.flumine._pending_packages, [mock_order_package])
 
-    @mock.patch("flumine.backtest.backtest.BaseFlumine._process_order_package")
-    def test__process_order_package(self, mock__process_order_package):
-        mock_order_package = mock.Mock(processed=False)
-        self.flumine._process_order_package(mock_order_package)
-        self.assertTrue(mock_order_package.processed)
-
     @mock.patch("flumine.backtest.backtest.FlumineBacktest._process_order_package")
     def test__check_pending_packages_place(self, mock__process_order_package):
         mock_client = mock.Mock()
diff --git a/tests/test_integration.py b/tests/test_integration.py
index 6739b00..264a28d 100644
--- a/tests/test_integration.py
+++ b/tests/test_integration.py
@@ -1,4 +1,3 @@
-import os
 import unittest
 
 from flumine import FlumineBacktest, clients, BaseStrategy, config
@@ -7,8 +6,6 @@ from flumine.order.order import OrderStatus
 from flumine.order.ordertype import LimitOrder, MarketOnCloseOrder
 from flumine.utils import get_price
 
-SKIP_INTEGRATION_TESTS = int(os.environ.get("SKIP_INTEGRATION_TESTS", 1))
-
 
 class IntegrationTest(unittest.TestCase):
     def setUp(self) -> None:
@@ -29,10 +26,6 @@ class IntegrationTest(unittest.TestCase):
         framework.add_strategy(strategy)
         framework.run()
 
-    @unittest.skipIf(
-        SKIP_INTEGRATION_TESTS,
-        "Integrations tests (set env.SKIP_INTEGRATION_TESTS = 0)",
-    )
     def test_backtest_pro(self):
         class LimitOrders(BaseStrategy):
             def check_market_book(self, market, market_book):
diff --git a/tests/test_strategy.py b/tests/test_strategy.py
index d93f835..e14c40a 100644
--- a/tests/test_strategy.py
+++ b/tests/test_strategy.py
@@ -68,6 +68,8 @@ class BaseStrategyTest(unittest.TestCase):
         self.assertEqual(self.strategy.client, self.mock_client)
         self.assertEqual(self.strategy.max_trade_count, 3)
         self.assertEqual(self.strategy.max_live_trade_count, 4)
+        self.assertEqual(self.strategy.streams, [])
+        self.assertEqual(self.strategy.historic_stream_ids, [])
 
     def test_check_market_no_subscribed(self):
         mock_market = mock.Mock()
@@ -233,6 +235,8 @@ class BaseStrategyTest(unittest.TestCase):
         mock_stream.stream_id = 321
         self.strategy.streams = [mock_stream]
         self.assertEqual(self.strategy.stream_ids, [321])
+        self.strategy.historic_stream_ids = [123]
+        self.assertEqual(self.strategy.stream_ids, [123])
 
     def test_info(self):
         self.assertEqual(
diff --git a/tests/test_streams.py b/tests/test_streams.py
index f30f869..1e4be4f 100644
--- a/tests/test_streams.py
+++ b/tests/test_streams.py
@@ -41,8 +41,7 @@ class StreamsTest(unittest.TestCase):
     @mock.patch("flumine.streams.streams.Streams.add_historical_stream")
     def test_call_backtest(self, mock_add_historical_stream):
         self.mock_flumine.BACKTEST = True
-        mock_strategy = mock.Mock()
-        mock_strategy.streams = []
+        mock_strategy = mock.Mock(streams=[], historic_stream_ids=[])
         mock_strategy.market_filter = {
             "markets": ["dubs of the mad skint and british"],
             "listener_kwargs": {"canary_yellow": True},
@@ -53,6 +52,7 @@ class StreamsTest(unittest.TestCase):
             mock_strategy, "dubs of the mad skint and british", canary_yellow=True
         )
         self.assertEqual(len(mock_strategy.streams), 1)
+        self.assertEqual(len(mock_strategy.historic_stream_ids), 1)
 
     def test_call_backtest_no_markets(self):
         self.mock_flumine.BACKTEST = True
@@ -437,7 +437,7 @@ class TestHistoricalStream(unittest.TestCase):
     def test_handle_output(self):
         self.stream.handle_output()
 
-    @mock.patch("flumine.streams.historicalstream.HistoricalGeneratorStream")
+    @mock.patch("flumine.streams.historicalstream.FlumineHistoricalGeneratorStream")
     def test_create_generator(self, mock_generator):
         generator = self.stream.create_generator()
         mock_generator.assert_called_with(
@@ -483,15 +483,15 @@ class TestFlumineMarketStream(unittest.TestCase):
             mock.Mock(inplay=True, seconds_to_start=None), 0
         )
         self.stream._caches = {
-            "1.123": mock.Mock(market_definition={"status": "OPEN", "inPlay": False})
+            "1.123": mock.Mock(_definition_status="OPEN", _definition_in_play=False),
         }
         self.assertEqual(len(self.stream.snap()), 0)
         self.stream._caches = {
-            "1.123": mock.Mock(market_definition={"status": "OPEN", "inPlay": True})
+            "1.123": mock.Mock(_definition_status="OPEN", _definition_in_play=True),
         }
         self.assertEqual(len(self.stream.snap()), 1)
         self.stream._caches = {
-            "1.123": mock.Mock(market_definition={"status": "CLOSED", "inPlay": False})
+            "1.123": mock.Mock(_definition_status="CLOSED", _definition_in_play=False),
         }
         self.assertEqual(len(self.stream.snap()), 1)
 
@@ -499,11 +499,11 @@ class TestFlumineMarketStream(unittest.TestCase):
             mock.Mock(inplay=False, seconds_to_start=None), 0
         )
         self.stream._caches = {
-            "1.123": mock.Mock(market_definition={"status": "OPEN", "inPlay": False})
+            "1.123": mock.Mock(_definition_status="OPEN", _definition_in_play=False),
         }
         self.assertEqual(len(self.stream.snap()), 1)
         self.stream._caches = {
-            "1.123": mock.Mock(market_definition={"status": "OPEN", "inPlay": True})
+            "1.123": mock.Mock(_definition_status="OPEN", _definition_in_play=True),
         }
         self.assertEqual(len(self.stream.snap()), 0)
 
@@ -587,10 +587,10 @@ class TestHistoricListener(unittest.TestCase):
         )
 
     def test_on_data(self):
-        mock_stream = mock.Mock()
+        mock_stream = mock.Mock(_lookup="mc")
         self.listener.stream = mock_stream
-        self.listener.on_data("{}")
-        mock_stream.on_update.assert_called_with({})
+        self.listener.on_data('{"pt": 123, "mc": {}}')
+        self.listener.stream._process.assert_called_with({}, 123)
         # error
         self.assertIsNone(self.listener.on_data("p"))