Repository: liampauling/flumine
Pull request #231 (2020-07-13 08:15:43)

Dev


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index 8dd1db1..808029e 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,6 +3,19 @@
 Release History
 ---------------
 
+1.9.0 (2020-07-13)
++++++++++++++++++++
+
+**Improvements**
+
+- #201 requests session kept and reused to reduce latency
+- Middleware add/remove market functions added and integrated into Simulated
+- Logging improvements
+
+**Libraries**
+
+- betfairlightweight upgraded to 2.6.0
+
 1.8.2 (2020-07-06)
 +++++++++++++++++++
 
diff --git a/README.md b/README.md
index 7751340..67f776c 100644
--- a/README.md
+++ b/README.md
@@ -83,11 +83,11 @@ class ExampleStrategy(BaseStrategy):
         for order in orders:
             if order.status == OrderStatus.EXECUTABLE:
                 if order.size_remaining == 2.00:
-                    self.cancel_order(order, 0.02)  # reduce size to 1.98
+                    self.cancel_order(market, order, 0.02)  # reduce size to 1.98
                 if order.order_type.persistence_type == 'LAPSE':
-                    self.update_order(order, 'PERSIST')
+                    self.update_order(market, order, 'PERSIST')
                 if order.size_remaining > 0:
-                    self.replace_order(order, 1.02)  # move
+                    self.replace_order(market, order, 1.02)  # move
 
 
 strategy = ExampleStrategy(
diff --git a/docs/advanced.md b/docs/advanced.md
index 3683c49..a5241ef 100644
--- a/docs/advanced.md
+++ b/docs/advanced.md
@@ -158,6 +158,7 @@ Flumine will catch all errors that occur in `strategy.check_market` and `strateg
 jsonlogger is used to log extra detail, see below for a typical setup:
 
 ```python
+import time
 import logging
 from pythonjsonlogger import jsonlogger
 
diff --git a/docs/index.md b/docs/index.md
index 41b4197..1d6ee13 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -95,11 +95,11 @@ class ExampleStrategy(BaseStrategy):
         for order in orders:
             if order.status == OrderStatus.EXECUTABLE:
                 if order.size_remaining == 2.00:
-                    self.cancel_order(order, 0.02)  # reduce size to 1.98
+                    self.cancel_order(market, order, 0.02)  # reduce size to 1.98
                 if order.order_type.persistence_type == "LAPSE":
-                    self.update_order(order, "PERSIST")
+                    self.update_order(market, order, "PERSIST")
                 if order.size_remaining > 0:
-                    self.replace_order(order, 1.02)  # move
+                    self.replace_order(market, order, 1.02)  # move
 
 
 strategy = ExampleStrategy(
diff --git a/flumine/__version__.py b/flumine/__version__.py
index 9e8f1fc..6914846 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.8.2"
+__version__ = "1.9.0"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/baseflumine.py b/flumine/baseflumine.py
index 1e573da..143e872 100644
--- a/flumine/baseflumine.py
+++ b/flumine/baseflumine.py
@@ -74,6 +74,7 @@ class BaseFlumine:
         raise NotImplementedError
 
     def add_strategy(self, strategy: BaseStrategy) -> None:
+        logger.info("Adding strategy {0}".format(strategy))
         self.streams(strategy)  # create required streams
         self.strategies(strategy)  # store in strategies
         self.log_control(events.StrategyEvent(strategy))
@@ -113,7 +114,7 @@ class BaseFlumine:
 
             market = self.markets.markets.get(market_id)
             if market is None:
-                market = self._add_live_market(market_id, market_book)
+                market = self._add_market(market_id, market_book)
 
             # process market
             market(market_book)
@@ -148,14 +149,20 @@ class BaseFlumine:
         else:
             logger.warning("Empty package, not executing", extra=order_package.info)
 
-    def _add_live_market(
-        self, market_id: str, market_book: resources.MarketBook
-    ) -> Market:
+    def _add_market(self, market_id: str, market_book: resources.MarketBook) -> Market:
+        logger.info("Adding: {0} to markets".format(market_id), extra=self.info)
         market = Market(self, market_id, market_book)
         self.markets.add_market(market_id, market)
-        logger.info("Adding: {0} to markets".format(market.market_id), extra=self.info)
+        for middleware in self._market_middleware:
+            middleware.add_market(market)
         return market
 
+    def _remove_market(self, market: Market) -> None:
+        logger.info("Removing market {0}".format(market.market_id), extra=self.info)
+        for middleware in self._market_middleware:
+            middleware.remove_market(market)
+        self.markets.remove_market(market.market_id)
+
     def _process_raw_data(self, event: events.RawDataEvent) -> None:
         stream_id, publish_time, data = event.event
         for datum in data:
@@ -251,13 +258,12 @@ class BaseFlumine:
 
         # check for markets that have been closed for x seconds
         closed_markets = [
-            m.market_id
+            m
             for m in self.markets
             if m.elapsed_seconds_closed and m.elapsed_seconds_closed > 3600
         ]
-        for market_id in closed_markets:
-            logger.info("Removing market %s" % market_id)
-            self.markets.remove_market(market_id)
+        for market in closed_markets:
+            self._remove_market(market)
 
     def _process_end_flumine(self) -> None:
         for strategy in self.strategies:
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)
diff --git a/flumine/markets/markets.py b/flumine/markets/markets.py
index 2b9e24b..a05263f 100644
--- a/flumine/markets/markets.py
+++ b/flumine/markets/markets.py
@@ -8,18 +8,19 @@ class Markets:
     def __init__(self):
         self._markets = {}  # marketId: <Market>
 
-    def add_market(self, market_id: str, live_market: Market) -> None:
+    def add_market(self, market_id: str, market: Market) -> None:
         if market_id in self._markets:
             self._markets[market_id].open_market()
         else:
-            self._markets[market_id] = live_market
+            self._markets[market_id] = market
 
     def close_market(self, market_id: str) -> Market:
-        live_market = self._markets[market_id]
-        live_market.close_market()
-        return live_market
+        market = self._markets[market_id]
+        market.close_market()
+        return market
 
     def remove_market(self, market_id: str) -> None:
+        del self._markets[market_id].blotter
         del self._markets[market_id]
 
     def get_order(self, market_id: str, order_id: str) -> Optional[BetfairOrder]:
diff --git a/flumine/markets/middleware.py b/flumine/markets/middleware.py
index 0714b5b..5e27c41 100644
--- a/flumine/markets/middleware.py
+++ b/flumine/markets/middleware.py
@@ -12,6 +12,12 @@ class Middleware:
     def __call__(self, market) -> None:
         raise NotImplementedError
 
+    def add_market(self, market) -> None:
+        pass
+
+    def remove_market(self, market) -> None:
+        pass
+
 
 class SimulatedMiddleware(Middleware):
     """
@@ -34,6 +40,12 @@ class SimulatedMiddleware(Middleware):
         # process simulated orders
         self._process_simulated_orders(market, market_analytics)
 
+    def remove_market(self, market) -> None:
+        try:
+            del self.markets[market.market_id]
+        except KeyError:
+            pass
+
     @staticmethod
     def _process_simulated_orders(market, market_analytics: dict) -> None:
         for order in market.blotter.live_orders:
diff --git a/flumine/utils.py b/flumine/utils.py
index a8b56c7..8430c0e 100644
--- a/flumine/utils.py
+++ b/flumine/utils.py
@@ -2,7 +2,7 @@ import uuid
 import logging
 import hashlib
 from typing import Optional, Tuple, Callable
-from decimal import Decimal
+from decimal import Decimal, ROUND_HALF_UP
 from betfairlightweight.resources.bettingresources import MarketBook, RunnerBook
 
 from . import config
@@ -74,6 +74,19 @@ def make_prices(min_price, cutoffs):
 PRICES = make_prices(MIN_PRICE, CUTOFFS)
 
 
+def get_nearest_price(price, cutoffs=CUTOFFS):
+    if price <= MIN_PRICE:
+        return MIN_PRICE
+    if price > MAX_PRICE:
+        return MAX_PRICE
+    price = as_dec(price)
+    for cutoff, step in cutoffs:
+        if price < cutoff:
+            break
+    step = as_dec(step)
+    return float((price * step).quantize(2, ROUND_HALF_UP) / step)
+
+
 def get_price(data: list, level: int) -> Optional[float]:
     try:
         return data[level]["price"]
diff --git a/flumine/worker.py b/flumine/worker.py
index 27ee82e..3396191 100644
--- a/flumine/worker.py
+++ b/flumine/worker.py
@@ -35,15 +35,19 @@ class BackgroundWorker(threading.Thread):
         self.context = context or {}
 
     def run(self) -> None:
-        time.sleep(self.start_delay)
         logger.info(
             "BackgroundWorker {0} starting".format(self.name),
             extra={
                 "worker_name": self.name,
                 "function": self.function,
                 "context": self.context,
+                "start_delay": self.start_delay,
+                "interval": self.interval,
+                "func_args": self.func_args,
+                "func_kwargs": self.func_kwargs,
             },
         )
+        time.sleep(self.start_delay)
         while self.is_alive():
             logger.debug(
                 "BackgroundWorker {0} executing".format(self.name),
diff --git a/requirements.txt b/requirements.txt
index 46e3c2d..1a5036f 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,4 @@
-betfairlightweight==2.5.0
+betfairlightweight==2.6.0
 tenacity==5.0.3
 python-json-logger==0.1.11
 requests


Test patch:
diff --git a/flumine/backtest/backtest.py b/flumine/backtest/backtest.py
index 032696d..335ad31 100644
--- a/flumine/backtest/backtest.py
+++ b/flumine/backtest/backtest.py
@@ -75,7 +75,7 @@ class FlumineBacktest(BaseFlumine):
 
             market = self.markets.markets.get(market_id)
             if market is None:
-                market = self._add_live_market(market_id, market_book)
+                market = self._add_market(market_id, market_book)
                 self.log_control(events.MarketEvent(market))
 
             # process market
diff --git a/tests/test_baseflumine.py b/tests/test_baseflumine.py
index b06be67..88ca28d 100644
--- a/tests/test_baseflumine.py
+++ b/tests/test_baseflumine.py
@@ -127,12 +127,26 @@ class BaseFlumineTest(unittest.TestCase):
         mock_client_control.assert_called_with(mock_order_package)
 
     @mock.patch("flumine.baseflumine.Market")
-    def test__add_live_market(self, mock_market):
+    def test__add_market(self, mock_market):
+        mock_middleware = mock.Mock()
+        self.base_flumine._market_middleware = [mock_middleware]
         mock_market_book = mock.Mock()
         self.assertEqual(
-            self.base_flumine._add_live_market("1.234", mock_market_book), mock_market()
+            self.base_flumine._add_market("1.234", mock_market_book), mock_market()
         )
         self.assertEqual(len(self.base_flumine.markets._markets), 1)
+        mock_middleware.add_market.assert_called_with(mock_market())
+
+    @mock.patch("flumine.baseflumine.BaseFlumine.info")
+    def test__remove_market(self, _):
+        mock_markets = mock.Mock()
+        self.base_flumine.markets = mock_markets
+        mock_middleware = mock.Mock()
+        self.base_flumine._market_middleware = [mock_middleware]
+        mock_market = mock.Mock()
+        self.base_flumine._remove_market(mock_market)
+        mock_markets.remove_market.assert_called_with(mock_market.market_id)
+        mock_middleware.remove_market.assert_called_with(mock_market)
 
     def test__process_raw_data(self):
         mock_event = mock.Mock()
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):
diff --git a/tests/test_middleware.py b/tests/test_middleware.py
index 3d511ce..1efb229 100644
--- a/tests/test_middleware.py
+++ b/tests/test_middleware.py
@@ -17,6 +17,14 @@ class MiddlewareTest(unittest.TestCase):
         with self.assertRaises(NotImplementedError):
             self.middleware(None)
 
+    def test_add_market(self):
+        mock_market = mock.Mock()
+        self.assertIsNone(self.middleware.add_market(mock_market))
+
+    def test_remove_market(self):
+        mock_market = mock.Mock()
+        self.assertIsNone(self.middleware.remove_market(mock_market))
+
 
 class SimulatedMiddlewareTest(unittest.TestCase):
     def setUp(self) -> None:
@@ -41,6 +49,13 @@ class SimulatedMiddlewareTest(unittest.TestCase):
         self.assertEqual(mock_market.context, {"simulated": {}})
         mock__process_simulated_orders.assert_called_with(mock_market, {})
 
+    def test_remove_market(self):
+        mock_market = mock.Mock(market_id="1.23")
+        self.middleware.markets = {mock_market.market_id: []}
+        self.middleware.remove_market(mock_market)
+        self.middleware.remove_market(mock_market)
+        self.assertEqual(self.middleware.markets, {})
+
     def test__process_simulated_orders(self):
         mock_market_book = mock.Mock()
         mock_order = mock.Mock()
diff --git a/tests/test_utils.py b/tests/test_utils.py
index 20d6a79..265dc4b 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -31,7 +31,15 @@ class UtilsTest(unittest.TestCase):
     #     utils.arange()
 
     def test_make_prices(self):
-        utils.make_prices(utils.MIN_PRICE, utils.CUTOFFS)
+        prices = utils.make_prices(utils.MIN_PRICE, utils.CUTOFFS)
+        self.assertEqual(len(prices), 350)
+
+    def test_get_nearest_price(self):
+        self.assertEqual(utils.get_nearest_price(1.011), 1.01)
+        self.assertEqual(utils.get_nearest_price(0), 1.01)
+        self.assertEqual(utils.get_nearest_price(1001), 1000)
+        self.assertEqual(utils.get_nearest_price(2.01), 2.02)
+        self.assertEqual(utils.get_nearest_price(2.0099), 2.00)
 
     def test_get_price(self):
         self.assertEqual(