Repository: liampauling/flumine (python)

Issue:
Execution - keep alive
All new execution requests use a new requests.Session, this results in the https round trip.

Pull request:
closes #201
http session used and kept after use (0.2-0.3s -> <0.2s)
removed on execution error

Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index 3690679..3956923 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -8,6 +8,7 @@ Release History
 
 **Improvements**
 
+- #201 requests session kept and reused to reduce latency
 - Logging improvements
 
 **Bug Fixes**
diff --git a/flumine/execution/baseexecution.py b/flumine/execution/baseexecution.py
index 1fdc56b..88f92fa 100644
--- a/flumine/execution/baseexecution.py
+++ b/flumine/execution/baseexecution.py
@@ -17,14 +17,17 @@ class BaseExecution:
 
     def __init__(self, flumine, max_workers=MAX_WORKERS):
         self.flumine = flumine
-        self._thread_pool = ThreadPoolExecutor(max_workers=max_workers)
+        self._max_workers = max_workers
+        self._thread_pool = ThreadPoolExecutor(max_workers=self._max_workers)
         self._bet_id = BET_ID_START
+        self._sessions = []
+        self._sessions_created = 0
 
     def handler(self, order_package: BaseOrderPackage):
         """ Handles order_package, capable of place, cancel,
         replace and update.
         """
-        http_session = requests.Session()  # todo keep
+        http_session = self._get_http_session()
         if order_package.package_type == OrderPackageType.PLACE:
             func = self.execute_place
         elif order_package.package_type == OrderPackageType.CANCEL:
@@ -58,6 +61,22 @@ class BaseExecution:
     ) -> None:
         raise NotImplementedError
 
+    def _get_http_session(self) -> requests.Session:
+        while self._sessions:
+            try:
+                return self._sessions.pop()
+            except IndexError:
+                continue
+        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:
+            self._sessions.append(http_session)
+
     def _order_logger(
         self, order: BaseOrder, instruction_report, package_type: OrderPackageType
     ):
diff --git a/flumine/execution/betfairexecution.py b/flumine/execution/betfairexecution.py
index c1335f9..2e33a59 100644
--- a/flumine/execution/betfairexecution.py
+++ b/flumine/execution/betfairexecution.py
@@ -203,11 +203,18 @@ class BetfairExecution(BaseExecution):
                         "order_package": order_package.info,
                     },
                 )
-                return
+                return  # session not returned on purpose
             logger.info(
                 "execute_%s" % trading_function.__name__,
-                extra={**order_package.info, **{"response": response._data}},
+                extra={
+                    "trading_function": trading_function.__name__,
+                    "elapsed_time": response.elapsed_time,
+                    "response": response._data,
+                    "order_package": order_package.info,
+                },
             )
+            self._return_http_session(http_session)
             return response
         else:
             logger.warning("Empty package, not executing", extra=order_package.info)
+            self._return_http_session(http_session)


Test patch:
diff --git a/tests/test_execution.py b/tests/test_execution.py
index 96a3176..f97518a 100644
--- a/tests/test_execution.py
+++ b/tests/test_execution.py
@@ -2,6 +2,7 @@ import unittest
 from unittest import mock
 from unittest.mock import call
 
+from betfairlightweight import BetfairError
 from flumine.clients.clients import ExchangeType
 from flumine.exceptions import OrderExecutionError
 from flumine.execution.baseexecution import BaseExecution, OrderPackageType
@@ -12,61 +13,68 @@ from flumine.execution.simulatedexecution import SimulatedExecution
 class BaseExecutionTest(unittest.TestCase):
     def setUp(self) -> None:
         self.mock_flumine = mock.Mock()
-        self.execution = BaseExecution(self.mock_flumine)
+        self.execution = BaseExecution(self.mock_flumine, max_workers=2)
 
     def test_init(self):
         self.assertEqual(self.execution.flumine, self.mock_flumine)
+        self.assertEqual(self.execution._max_workers, 2)
         self.assertIsNotNone(self.execution._thread_pool)
         self.assertIsNone(self.execution.EXCHANGE)
         self.assertEqual(self.execution._bet_id, 100000000000)
+        self.assertEqual(self.execution._sessions, [])
+        self.assertEqual(self.execution._sessions_created, 0)
 
-    @mock.patch("flumine.execution.baseexecution.requests")
+    @mock.patch("flumine.execution.baseexecution.BaseExecution._get_http_session")
     @mock.patch("flumine.execution.baseexecution.BaseExecution.execute_place")
-    def test_handler_place(self, mock_execute_place, mock_requests):
+    def test_handler_place(self, mock_execute_place, mock__get_http_session):
         mock_order_package = mock.Mock()
         mock_order_package.package_type = OrderPackageType.PLACE
         mock_thread_pool = mock.Mock()
         self.execution._thread_pool = mock_thread_pool
         self.execution.handler(mock_order_package)
         mock_thread_pool.submit.assert_called_with(
-            mock_execute_place, mock_order_package, mock_requests.Session()
+            mock_execute_place, mock_order_package, mock__get_http_session()
         )
+        mock__get_http_session.assert_called_with()
 
-    @mock.patch("flumine.execution.baseexecution.requests")
+    @mock.patch("flumine.execution.baseexecution.BaseExecution._get_http_session")
     @mock.patch("flumine.execution.baseexecution.BaseExecution.execute_cancel")
-    def test_handler_cancel(self, mock_execute_cancel, mock_requests):
+    def test_handler_cancel(self, mock_execute_cancel, mock__get_http_session):
         mock_order_package = mock.Mock()
         mock_order_package.package_type = OrderPackageType.PLACE.CANCEL
         mock_thread_pool = mock.Mock()
         self.execution._thread_pool = mock_thread_pool
         self.execution.handler(mock_order_package)
         mock_thread_pool.submit.assert_called_with(
-            mock_execute_cancel, mock_order_package, mock_requests.Session()
+            mock_execute_cancel, mock_order_package, mock__get_http_session()
         )
+        mock__get_http_session.assert_called_with()
 
-    @mock.patch("flumine.execution.baseexecution.requests")
+    @mock.patch("flumine.execution.baseexecution.BaseExecution._get_http_session")
     @mock.patch("flumine.execution.baseexecution.BaseExecution.execute_replace")
-    def test_handler_replace(self, mock_execute_replace, mock_requests):
+    def test_handler_replace(self, mock_execute_replace, mock__get_http_session):
         mock_order_package = mock.Mock()
         mock_order_package.package_type = OrderPackageType.REPLACE
         mock_thread_pool = mock.Mock()
         self.execution._thread_pool = mock_thread_pool
         self.execution.handler(mock_order_package)
         mock_thread_pool.submit.assert_called_with(
-            mock_execute_replace, mock_order_package, mock_requests.Session()
+            mock_execute_replace, mock_order_package, mock__get_http_session()
         )
+        mock__get_http_session.assert_called_with()
 
-    @mock.patch("flumine.execution.baseexecution.requests")
+    @mock.patch("flumine.execution.baseexecution.BaseExecution._get_http_session")
     @mock.patch("flumine.execution.baseexecution.BaseExecution.execute_update")
-    def test_handler_update(self, mock_execute_update, mock_requests):
+    def test_handler_update(self, mock_execute_update, mock__get_http_session):
         mock_order_package = mock.Mock()
         mock_order_package.package_type = OrderPackageType.UPDATE
         mock_thread_pool = mock.Mock()
         self.execution._thread_pool = mock_thread_pool
         self.execution.handler(mock_order_package)
         mock_thread_pool.submit.assert_called_with(
-            mock_execute_update, mock_order_package, mock_requests.Session()
+            mock_execute_update, mock_order_package, mock__get_http_session()
         )
+        mock__get_http_session.assert_called_with()
 
     def test_handler_unknown(self):
         mock_order_package = mock.Mock()
@@ -90,6 +98,20 @@ class BaseExecutionTest(unittest.TestCase):
         with self.assertRaises(NotImplementedError):
             self.execution.execute_place(None, None)
 
+    @mock.patch("flumine.execution.baseexecution.requests")
+    def test__get_http_session(self, mock_requests):
+        self.execution._sessions = [1, 2]
+        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())
+
+    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.patch("flumine.execution.baseexecution.OrderEvent")
     def test__order_logger_place(self, mock_order_event):
         mock_order = mock.Mock()
@@ -606,7 +628,10 @@ class BetfairExecutionTest(unittest.TestCase):
             mock_order_package.client.betting_client.betting.replace_orders(),
         )
 
-    def test__execution_helper(self):
+    @mock.patch(
+        "flumine.execution.betfairexecution.BetfairExecution._return_http_session"
+    )
+    def test__execution_helper(self, mock__return_http_session):
         mock_trading_function = mock.Mock()
         mock_trading_function.__name__ = "test"
         mock_session = mock.Mock()
@@ -616,6 +641,40 @@ class BetfairExecutionTest(unittest.TestCase):
             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)
+
+    @mock.patch(
+        "flumine.execution.betfairexecution.BetfairExecution._return_http_session"
+    )
+    def test__execution_helper_empty(self, mock__return_http_session):
+        mock_trading_function = mock.Mock()
+        mock_trading_function.__name__ = "test"
+        mock_session = mock.Mock()
+        mock_order_package = mock.Mock(orders=[])
+        mock_order_package.info = {}
+        self.execution._execution_helper(
+            mock_trading_function, mock_order_package, mock_session
+        )
+        mock_trading_function.assert_not_called()
+        mock__return_http_session.assert_called_with(mock_session)
+
+    @mock.patch(
+        "flumine.execution.betfairexecution.BetfairExecution._return_http_session"
+    )
+    def test__execution_helper_error(self, mock__return_http_session):
+        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 = {}
+        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()
 
 
 class SimulatedExecutionTest(unittest.TestCase):