Repository: liampauling/flumine
Pull request #234 (2020-07-13 12:44:38)

Task/184 execution retry


Patch:
diff --git a/flumine/controls/loggingcontrols.py b/flumine/controls/loggingcontrols.py
index ec686bf..e49521e 100644
--- a/flumine/controls/loggingcontrols.py
+++ b/flumine/controls/loggingcontrols.py
@@ -65,7 +65,7 @@ class LoggingControl(Thread):
             self._process_cleared_orders(event)
 
         elif event.EVENT_TYPE == EventType.CLEARED_ORDERS_META:
-            self._process_cleared_orders_meta(event)  # todo
+            self._process_cleared_orders_meta(event)
 
         elif event.EVENT_TYPE == EventType.CLEARED_MARKETS:
             self._process_cleared_markets(event)
diff --git a/flumine/execution/baseexecution.py b/flumine/execution/baseexecution.py
index 88f92fa..f55cfaa 100644
--- a/flumine/execution/baseexecution.py
+++ b/flumine/execution/baseexecution.py
@@ -67,14 +67,20 @@ class BaseExecution:
                 return self._sessions.pop()
             except IndexError:
                 continue
+        self._sessions_created += 1
         logger.info(
             "New requests.Session created",
             extra={"sessions_created": self._sessions_created},
         )
         return requests.Session()
 
-    def _return_http_session(self, http_session: requests.Session) -> None:
-        if len(self._sessions) < self._max_workers:
+    def _return_http_session(
+        self, http_session: requests.Session, err: bool = False
+    ) -> None:
+        if err or len(self._sessions) >= self._max_workers:
+            http_session.close()
+            del http_session
+        else:
             self._sessions.append(http_session)
 
     def _order_logger(
diff --git a/flumine/execution/betfairexecution.py b/flumine/execution/betfairexecution.py
index 2e33a59..f3da633 100644
--- a/flumine/execution/betfairexecution.py
+++ b/flumine/execution/betfairexecution.py
@@ -185,7 +185,7 @@ class BetfairExecution(BaseExecution):
             session=session,
         )
 
-    def _execution_helper(  # todo retry!
+    def _execution_helper(
         self,
         trading_function: Callable,
         order_package: BaseOrderPackage,
@@ -202,8 +202,13 @@ class BetfairExecution(BaseExecution):
                         "response": e,
                         "order_package": order_package.info,
                     },
+                    exc_info=True,
                 )
-                return  # session not returned on purpose
+                if order_package.retry():
+                    self.handler_queue.put(order_package)
+
+                self._return_http_session(http_session, err=True)
+                return
             logger.info(
                 "execute_%s" % trading_function.__name__,
                 extra={
diff --git a/flumine/order/orderpackage.py b/flumine/order/orderpackage.py
index 53de53f..8eb8b1a 100644
--- a/flumine/order/orderpackage.py
+++ b/flumine/order/orderpackage.py
@@ -1,4 +1,5 @@
 import uuid
+import time
 from enum import Enum
 from typing import Iterator, Optional
 from betfairlightweight.metadata import order_limits
@@ -17,14 +18,11 @@ class OrderPackageType(Enum):
 
 
 class BaseOrderPackage(BaseEvent):
-
     """
     Data structure for multiple orders,
     temporary to allow execution
     """
 
-    # todo client/retry
-
     EVENT_TYPE = EventType.ORDER_PACKAGE
     QUEUE_TYPE = QueueType.HANDLER
     EXCHANGE = None
@@ -48,6 +46,16 @@ class BaseOrderPackage(BaseEvent):
         self.async_ = async_
         self.customer_strategy_ref = config.hostname
         self.processed = False  # used for simulated execution
+        self._retry = True
+        self._max_retries = 3  # will retry 3 times
+        self._retry_count = 0
+
+    def retry(self):
+        if self._retry and self._retry_count < self._max_retries:
+            time.sleep(self._retry_count)  # back-off
+            self._retry_count += 1
+            return True
+        return False
 
     @property
     def place_instructions(self) -> dict:
@@ -84,6 +92,8 @@ class BaseOrderPackage(BaseEvent):
             "customer_strategy_ref": self.customer_strategy_ref,
             "bet_delay": self.bet_delay,
             "market_version": self.market_version,
+            "retry": self._retry,
+            "retry_count": self._retry_count,
         }
 
     @property


Test patch:
diff --git a/tests/test_execution.py b/tests/test_execution.py
index f97518a..32ed8bb 100644
--- a/tests/test_execution.py
+++ b/tests/test_execution.py
@@ -104,13 +104,26 @@ class BaseExecutionTest(unittest.TestCase):
         self.assertEqual(self.execution._get_http_session(), 2)
         self.assertEqual(self.execution._get_http_session(), 1)
         self.assertEqual(self.execution._get_http_session(), mock_requests.Session())
+        self.assertEqual(self.execution._sessions_created, 1)
 
     def test__return_http_session(self):
-        self.execution._return_http_session(1)
-        self.assertEqual(self.execution._sessions, [1])
-        self.execution._return_http_session(2)
-        self.execution._return_http_session(3)
-        self.assertEqual(self.execution._sessions, [1, 2])
+        mock_session = mock.Mock()
+        self.execution._return_http_session(mock_session)
+        self.assertEqual(self.execution._sessions, [mock_session])
+        self.execution._return_http_session(mock_session)
+        self.execution._return_http_session(mock_session)
+        self.assertEqual(self.execution._sessions, [mock_session, mock_session])
+
+    def test__return_http_session_close(self):
+        self.execution._sessions = [1, 2]
+        mock_session = mock.Mock()
+        self.execution._return_http_session(mock_session)
+        mock_session.close.assert_called_with()
+
+    def test__return_http_session_err_close(self):
+        mock_session = mock.Mock()
+        self.execution._return_http_session(mock_session, err=True)
+        mock_session.close.assert_called_with()
 
     @mock.patch("flumine.execution.baseexecution.OrderEvent")
     def test__order_logger_place(self, mock_order_event):
@@ -658,23 +671,51 @@ class BetfairExecutionTest(unittest.TestCase):
         mock_trading_function.assert_not_called()
         mock__return_http_session.assert_called_with(mock_session)
 
+    @mock.patch("flumine.execution.betfairexecution.BetfairExecution.handler_queue")
+    @mock.patch(
+        "flumine.execution.betfairexecution.BetfairExecution._return_http_session"
+    )
+    def test__execution_helper_error(
+        self, mock__return_http_session, mock_handler_queue
+    ):
+        mock_trading_function = mock.Mock()
+        mock_trading_function.__name__ = "test"
+        mock_trading_function.side_effect = BetfairError()
+        mock_session = mock.Mock()
+        mock_order_package = mock.Mock()
+        mock_order_package.info = {}
+        mock_order_package.retry.return_value = True
+        self.assertIsNone(
+            self.execution._execution_helper(
+                mock_trading_function, mock_order_package, mock_session
+            )
+        )
+        mock_trading_function.assert_called_with(mock_order_package, mock_session)
+        mock__return_http_session.assert_called_with(mock_session, err=True)
+        mock_handler_queue.put.assert_called_with(mock_order_package)
+
+    @mock.patch("flumine.execution.betfairexecution.BetfairExecution.handler_queue")
     @mock.patch(
         "flumine.execution.betfairexecution.BetfairExecution._return_http_session"
     )
-    def test__execution_helper_error(self, mock__return_http_session):
+    def test__execution_helper_error_no_retry(
+        self, mock__return_http_session, mock_handler_queue
+    ):
         mock_trading_function = mock.Mock()
         mock_trading_function.__name__ = "test"
         mock_trading_function.side_effect = BetfairError()
         mock_session = mock.Mock()
         mock_order_package = mock.Mock()
         mock_order_package.info = {}
+        mock_order_package.retry.return_value = False
         self.assertIsNone(
             self.execution._execution_helper(
                 mock_trading_function, mock_order_package, mock_session
             )
         )
         mock_trading_function.assert_called_with(mock_order_package, mock_session)
-        mock__return_http_session.assert_not_called()
+        mock__return_http_session.assert_called_with(mock_session, err=True)
+        mock_handler_queue.put.assert_not_called()
 
 
 class SimulatedExecutionTest(unittest.TestCase):
diff --git a/tests/test_orderpackage.py b/tests/test_orderpackage.py
index cf210bb..72ba4b2 100644
--- a/tests/test_orderpackage.py
+++ b/tests/test_orderpackage.py
@@ -38,6 +38,25 @@ class OrderPackageTest(unittest.TestCase):
         self.assertIsNone(self.order_package.EXCHANGE)
         self.assertFalse(self.order_package.async_)
         self.assertFalse(self.order_package.processed)
+        self.assertTrue(self.order_package._retry)
+        self.assertEqual(self.order_package._max_retries, 3)
+        self.assertEqual(self.order_package._retry_count, 0)
+
+    def test_retry(self):
+        self.assertTrue(self.order_package.retry())
+        self.assertEqual(self.order_package._retry_count, 1)
+
+    @mock.patch("flumine.order.orderpackage.time")
+    def test_retry_false_count(self, mock_time):
+        self.assertTrue(self.order_package.retry())
+        self.assertEqual(self.order_package._retry_count, 1)
+        self.assertTrue(self.order_package.retry())
+        self.assertEqual(self.order_package._retry_count, 2)
+        self.assertTrue(self.order_package.retry())
+        self.assertEqual(self.order_package._retry_count, 3)
+        self.assertFalse(self.order_package.retry())
+        self.assertEqual(self.order_package._retry_count, 3)
+        mock_time.sleep.assert_called()
 
     def test_place_instructions(self):
         with self.assertRaises(NotImplementedError):
@@ -80,6 +99,8 @@ class OrderPackageTest(unittest.TestCase):
                 "customer_strategy_ref": self.order_package.customer_strategy_ref,
                 "bet_delay": self.order_package.bet_delay,
                 "market_version": self.order_package.market_version,
+                "retry": self.order_package._retry,
+                "retry_count": self.order_package._retry_count,
             },
         )