Repository: liampauling/flumine
Pull request #182 (2020-05-29 13:31:24)

remove unnecessary market lookup


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index 7e34b00..0b2f24c 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,6 +3,15 @@
 Release History
 ---------------
 
+1.5.6 (2020-06-01)
++++++++++++++++++++
+
+**Improvements**
+
+- #186 Error handling when calling strategy functions
+- Start delay bumped on workers and name changed
+- Minor typos / cleanups
+
 1.5.5 (2020-05-29)
 +++++++++++++++++++
 
diff --git a/flumine/__version__.py b/flumine/__version__.py
index 3a260db..91a53dc 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.5.5"
+__version__ = "1.5.6"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/baseflumine.py b/flumine/baseflumine.py
index 9332acd..fa526df 100644
--- a/flumine/baseflumine.py
+++ b/flumine/baseflumine.py
@@ -17,7 +17,7 @@ from .order.process import process_current_orders
 from .controls.clientcontrols import BaseControl, MaxOrderCount
 from .controls.tradingcontrols import OrderValidation, StrategyExposure
 from .controls.loggingcontrols import LoggingControl
-from . import config
+from . import config, utils
 
 
 logger = logging.getLogger(__name__)
@@ -122,8 +122,10 @@ class BaseFlumine:
                 middleware(market)  # todo err handling?
 
             for strategy in self.strategies:
-                if strategy.check_market(market, market_book):
-                    strategy.process_market_book(market, market_book)
+                if utils.call_check_market(strategy.check_market, market, market_book):
+                    utils.call_process_market_book(
+                        strategy.process_market_book, market, market_book
+                    )
 
             self._process_market_orders(market)
 
diff --git a/flumine/config.py b/flumine/config.py
index b202c36..122ba06 100644
--- a/flumine/config.py
+++ b/flumine/config.py
@@ -10,3 +10,5 @@ hostname = socket.gethostname()[
 process_id = os.getpid()  # process id of app
 
 current_time = None  # used for backtesting
+
+raise_errors = False  # used for call_check_market / call_process_market_book
diff --git a/flumine/execution/betfairexecution.py b/flumine/execution/betfairexecution.py
index 327899f..566e68a 100644
--- a/flumine/execution/betfairexecution.py
+++ b/flumine/execution/betfairexecution.py
@@ -139,8 +139,7 @@ class BetfairExecution(BaseExecution):
                         OrderPackageType.REPLACE,
                     )
                     # add to blotter
-                    market = self.markets.markets[order.market_id]
-                    market.place_order(replacement_order, execute=False)
+                    order_package.market.place_order(replacement_order, execute=False)
                     replacement_order.executable()
                 elif instruction_report.place_instruction_reports.status == "FAILURE":
                     pass  # todo
diff --git a/flumine/flumine.py b/flumine/flumine.py
index 9fb5af0..9902be5 100644
--- a/flumine/flumine.py
+++ b/flumine/flumine.py
@@ -54,17 +54,14 @@ class Flumine(BaseFlumine):
 
     def _add_default_workers(self):
         self.add_worker(
-            worker.BackgroundWorker(
-                self, function=worker.keep_alive, interval=1200, name="keep_alive",
-            )
+            worker.BackgroundWorker(self, function=worker.keep_alive, interval=1200)
         )
         self.add_worker(
             worker.BackgroundWorker(
                 self,
                 function=worker.poll_account_balance,
                 interval=60,
-                start_delay=5,  # wait for login
-                name="poll_account_balance",
+                start_delay=10,  # wait for login
             )
         )
         self.add_worker(
@@ -72,8 +69,7 @@ class Flumine(BaseFlumine):
                 self,
                 function=worker.poll_market_catalogue,
                 interval=60,
-                start_delay=5,  # wait for streams to populate
-                name="poll_market_catalogue",
+                start_delay=10,  # wait for streams to populate
             )
         )
         self.add_worker(
@@ -81,7 +77,7 @@ class Flumine(BaseFlumine):
                 self,
                 function=worker.poll_cleared_orders,
                 interval=10,  # restart
-                name="poll_cleared_orders",
+                start_delay=10,  # wait for login
             )
         )
 
diff --git a/flumine/streams/orderstream.py b/flumine/streams/orderstream.py
index f228e43..5e65888 100644
--- a/flumine/streams/orderstream.py
+++ b/flumine/streams/orderstream.py
@@ -33,12 +33,12 @@ class OrderStream(BaseStream):
             )
             self._stream.start()
         except BetfairError:
-            logger.error("MarketStream run error", exc_info=True)
+            logger.error("OrderStream run error", exc_info=True)
             raise
         except Exception:
-            logger.critical("MarketStream run error", exc_info=True)
+            logger.critical("OrderStream run error", exc_info=True)
             raise
-        logger.info("Stopped MarketStream {0}".format(self.stream_id))
+        logger.info("Stopped OrderStream {0}".format(self.stream_id))
 
     def handle_output(self) -> None:
         """Handles output from stream.
diff --git a/flumine/utils.py b/flumine/utils.py
index f920b3b..869cd6b 100644
--- a/flumine/utils.py
+++ b/flumine/utils.py
@@ -1,10 +1,13 @@
 import uuid
 import logging
 import hashlib
-from typing import Optional, Tuple
+from typing import Optional, Tuple, Callable
 from decimal import Decimal
 from betfairlightweight.resources.bettingresources import RunnerBook
 
+from . import config
+from .exceptions import FlumineException
+
 logger = logging.getLogger(__name__)
 
 CUTOFFS = (
@@ -139,3 +142,44 @@ def wap(matched: list) -> Tuple[float, float]:
         return 0, 0
     else:
         return round(b, 2), round(a / b, 2)
+
+
+def call_check_market(strategy_check_market: Callable, market, market_book) -> bool:
+    try:
+        return strategy_check_market(market, market_book)
+    except FlumineException as e:
+        logger.error(
+            "FlumineException %s in strategy_check_market %s %s"
+            % (e, strategy_check_market, market.market_id),
+            exc_info=True,
+        )
+    except Exception as e:
+        logger.critical(
+            "Unknown error %s in strategy_check_market %s %s"
+            % (e, strategy_check_market, market.market_id),
+            exc_info=True,
+        )
+        if config.raise_errors:
+            raise
+    return False
+
+
+def call_process_market_book(
+    strategy_process_market_book: Callable, market, market_book
+) -> None:
+    try:
+        strategy_process_market_book(market, market_book)
+    except FlumineException as e:
+        logger.error(
+            "FlumineException %s in strategy_process_market_book %s %s"
+            % (e, strategy_process_market_book, market.market_id),
+            exc_info=True,
+        )
+    except Exception as e:
+        logger.critical(
+            "Unknown error %s in strategy_process_market_book %s %s"
+            % (e, strategy_process_market_book, market.market_id),
+            exc_info=True,
+        )
+        if config.raise_errors:
+            raise
diff --git a/flumine/worker.py b/flumine/worker.py
index 10473ab..bc4c6b8 100644
--- a/flumine/worker.py
+++ b/flumine/worker.py
@@ -1,7 +1,7 @@
 import time
 import threading
 import logging
-import queue
+from typing import Callable
 from betfairlightweight import BetfairError, filters
 
 from . import config
@@ -15,15 +15,17 @@ class BackgroundWorker(threading.Thread):
     def __init__(
         self,
         flumine,
-        function,
+        function: Callable,
         interval: int,
         func_args: tuple = None,
         func_kwargs: dict = None,
         start_delay: int = 0,
         context: dict = None,
+        name: str = None,
         **kwargs
     ):
-        threading.Thread.__init__(self, daemon=True, **kwargs)
+        name = name or function.__name__
+        threading.Thread.__init__(self, daemon=True, name=name, **kwargs)
         self.flumine = flumine
         self.function = function
         self.interval = interval
@@ -37,7 +39,7 @@ class BackgroundWorker(threading.Thread):
         logger.info(
             "BackgroundWorker {0} starting".format(self.name),
             extra={
-                "worker_name": self.name,
+                "name": self.name,
                 "function": self.function,
                 "context": self.context,
             },
@@ -46,7 +48,7 @@ class BackgroundWorker(threading.Thread):
             logger.debug(
                 "BackgroundWorker {0} executing".format(self.name),
                 extra={
-                    "worker_name": self.name,
+                    "name": self.name,
                     "function": self.function,
                     "context": self.context,
                 },
@@ -59,7 +61,7 @@ class BackgroundWorker(threading.Thread):
                 logger.error(
                     "Error in BackgroundWorker {0}: {1}".format(self.name, e),
                     extra={
-                        "worker_name": self.name,
+                        "name": self.name,
                         "function": self.function,
                         "context": self.context,
                     },


Test patch:
diff --git a/flumine/backtest/backtest.py b/flumine/backtest/backtest.py
index 223304d..2c7fdd3 100644
--- a/flumine/backtest/backtest.py
+++ b/flumine/backtest/backtest.py
@@ -3,7 +3,7 @@ import datetime
 
 from ..baseflumine import BaseFlumine
 from ..events import events
-from .. import config
+from .. import config, utils
 from ..clients import ExchangeType
 from ..exceptions import RunError
 from .utils import PendingPackages
@@ -87,8 +87,10 @@ class FlumineBacktest(BaseFlumine):
                 middleware(market)  # todo err handling?
 
             for strategy in self.strategies:
-                if strategy.check_market(market, market_book):
-                    strategy.process_market_book(market, market_book)
+                if utils.call_check_market(strategy.check_market, market, market_book):
+                    utils.call_process_market_book(
+                        strategy.process_market_book, market, market_book
+                    )
 
             # process current orders
             blotter = market.blotter
diff --git a/tests/test_execution.py b/tests/test_execution.py
index 98f2263..cb3feba 100644
--- a/tests/test_execution.py
+++ b/tests/test_execution.py
@@ -494,8 +494,6 @@ class BetfairExecutionTest(unittest.TestCase):
     def test_execute_replace_success(
         self, mock__execution_helper, mock_replace, mock__order_logger
     ):
-        mock_market = mock.Mock()
-        self.mock_flumine.markets.markets = {"1.234": mock_market}
         mock_session = mock.Mock()
         mock_order = mock.Mock()
         mock_order.market_id = "1.234"
@@ -520,7 +518,9 @@ class BetfairExecutionTest(unittest.TestCase):
 
         replacement_order = mock_order.trade.create_order_replacement()
         replacement_order.executable.assert_called_with()
-        mock_market.place_order.assert_called_with(replacement_order, execute=False)
+        mock_order_package.market.place_order.assert_called_with(
+            replacement_order, execute=False
+        )
         mock__order_logger.assert_called_with(
             replacement_order,
             mock_instruction_report.place_instruction_reports,
diff --git a/tests/test_utils.py b/tests/test_utils.py
index ef4f2d7..0fae954 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -1,10 +1,14 @@
+import logging
 import unittest
 from unittest import mock
 
-from flumine import utils
+from flumine import utils, FlumineException
 
 
 class UtilsTest(unittest.TestCase):
+    def setUp(self) -> None:
+        logging.disable(logging.CRITICAL)
+
     def test_create_short_uuid(self):
         self.assertTrue(utils.create_short_uuid())
 
@@ -103,3 +107,67 @@ class UtilsTest(unittest.TestCase):
         self.assertEqual(utils.wap([(1.5, 100), (1.6, 100)]), (200, 1.55))
         self.assertEqual(utils.wap([]), (0, 0))
         self.assertEqual(utils.wap([(1.5, 0)]), (0, 0))
+
+    def test_call_check_market(self):
+        mock_strategy_check = mock.Mock()
+        mock_market = mock.Mock()
+        mock_market_book = mock.Mock()
+        utils.call_check_market(mock_strategy_check, mock_market, mock_market_book)
+        mock_strategy_check.assert_called_with(mock_market, mock_market_book)
+
+    def test_call_check_market_flumine_error(self):
+        mock_strategy_check = mock.Mock(side_effect=FlumineException)
+        mock_market = mock.Mock()
+        mock_market_book = mock.Mock()
+        self.assertFalse(
+            utils.call_check_market(mock_strategy_check, mock_market, mock_market_book)
+        )
+        mock_strategy_check.assert_called_with(mock_market, mock_market_book)
+
+    def test_call_check_market_error(self):
+        mock_strategy_check = mock.Mock(side_effect=ValueError)
+        mock_market = mock.Mock()
+        mock_market_book = mock.Mock()
+        self.assertFalse(
+            utils.call_check_market(mock_strategy_check, mock_market, mock_market_book)
+        )
+        mock_strategy_check.assert_called_with(mock_market, mock_market_book)
+
+    @mock.patch("flumine.utils.config")
+    def test_call_check_market_raise(self, mock_config):
+        mock_config.raise_errors = True
+        mock_strategy_check = mock.Mock(side_effect=ValueError)
+        mock_market = mock.Mock()
+        mock_market_book = mock.Mock()
+        with self.assertRaises(ValueError):
+            utils.call_check_market(mock_strategy_check, mock_market, mock_market_book)
+
+    def test_call_process_market_book(self):
+        mock_strategy = mock.Mock()
+        mock_market = mock.Mock()
+        mock_market_book = mock.Mock()
+        utils.call_process_market_book(mock_strategy, mock_market, mock_market_book)
+        mock_strategy.assert_called_with(mock_market, mock_market_book)
+
+    def test_call_process_market_book_flumine_error(self):
+        mock_strategy = mock.Mock(side_effect=FlumineException)
+        mock_market = mock.Mock()
+        mock_market_book = mock.Mock()
+        utils.call_process_market_book(mock_strategy, mock_market, mock_market_book)
+        mock_strategy.assert_called_with(mock_market, mock_market_book)
+
+    def test_call_process_market_book_error(self):
+        mock_strategy = mock.Mock(side_effect=ZeroDivisionError)
+        mock_market = mock.Mock()
+        mock_market_book = mock.Mock()
+        utils.call_process_market_book(mock_strategy, mock_market, mock_market_book)
+        mock_strategy.assert_called_with(mock_market, mock_market_book)
+
+    @mock.patch("flumine.utils.config")
+    def test_call_process_market_book_raise(self, mock_config):
+        mock_config.raise_errors = True
+        mock_strategy = mock.Mock(side_effect=ZeroDivisionError)
+        mock_market = mock.Mock()
+        mock_market_book = mock.Mock()
+        with self.assertRaises(ZeroDivisionError):
+            utils.call_process_market_book(mock_strategy, mock_market, mock_market_book)
diff --git a/tests/test_worker.py b/tests/test_worker.py
index bab95e7..1120df8 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -8,7 +8,7 @@ from flumine import worker
 
 class BackgroundWorkerTest(unittest.TestCase):
     def setUp(self):
-        self.mock_function = mock.Mock()
+        self.mock_function = mock.Mock(__name__="test")
         self.mock_flumine = mock.Mock()
         self.worker = worker.BackgroundWorker(
             self.mock_flumine,
@@ -28,6 +28,7 @@ class BackgroundWorkerTest(unittest.TestCase):
         self.assertEqual(self.worker.func_kwargs, {"hello": "world"})
         self.assertEqual(self.worker.start_delay, 5)
         self.assertEqual(self.worker.context, {1: 2})
+        self.assertEqual(self.worker.name, "test")
 
     # def test_run(self):
     #     self.worker.run()