Repository: liampauling/flumine
Pull request #391 (2021-03-08 15:12:43)

refactor closure worker to check all closed markets requiring clearing
removes the '7' limit and updates on a closure


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index 989af21..8c33548 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,25 +3,13 @@
 Release History
 ---------------
 
-1.17.9 (2021-03-09)
-+++++++++++++++++++
-
-**Improvements**
-
-- Remove session close in execution when removing stale sessions (very slow)
-- Refactor closure worker to check all closed markets requiring clearing
-
-**Libraries**
-
-- betfairlightweight upgraded to 2.12.1
-
 1.17.8 (2021-03-08)
 +++++++++++++++++++
 
 **Improvements**
 
 - Allow kwargs to be passed to `trade.create_order`
-- Correct handling off completed offset orders
+- Correct handling off completed offset orderrs
 
 **Bug Fixes**
 
diff --git a/flumine/__version__.py b/flumine/__version__.py
index 9cde18a..66cc33b 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.17.9"
+__version__ = "1.17.8"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/baseflumine.py b/flumine/baseflumine.py
index c9c8768..37ea2e1 100644
--- a/flumine/baseflumine.py
+++ b/flumine/baseflumine.py
@@ -39,7 +39,6 @@ class BaseFlumine:
 
         # queues
         self.handler_queue = queue.Queue()
-        self.cleared_market_queue = queue.Queue()
 
         # all markets
         self.markets = Markets()
@@ -256,8 +255,6 @@ class BaseFlumine:
         if recorder is False:
             if self.BACKTEST or self.client.paper_trade:
                 self._process_cleared_orders(events.ClearedOrdersEvent(market))
-            else:
-                self.cleared_market_queue.put(market_id)
         self.log_control(event)
         logger.info("Market closed", extra={"market_id": market_id, **self.info})
 
diff --git a/flumine/execution/baseexecution.py b/flumine/execution/baseexecution.py
index ca2b68e..d0b21b7 100644
--- a/flumine/execution/baseexecution.py
+++ b/flumine/execution/baseexecution.py
@@ -97,7 +97,7 @@ class BaseExecution:
     ) -> None:
         if err or len(self._sessions) >= self._max_workers:
             logger.info(
-                "Deleting requests.Session",
+                "Closing and deleting requests.Session",
                 extra={
                     "sessions_created": self._sessions_created,
                     "session": http_session,
@@ -107,6 +107,7 @@ class BaseExecution:
                     "err": err,
                 },
             )
+            http_session.close()
             del http_session
         else:
             http_session.time_returned = time.time()
diff --git a/flumine/execution/betfairexecution.py b/flumine/execution/betfairexecution.py
index fde4654..da55414 100644
--- a/flumine/execution/betfairexecution.py
+++ b/flumine/execution/betfairexecution.py
@@ -224,7 +224,6 @@ class BetfairExecution(BaseExecution):
                 "High latency between current time and OrderPackage creation time, it is likely that the thread pool is currently exhausted",
                 extra={
                     "trading_function": trading_function.__name__,
-                    "session": http_session,
                     "latency": round(order_package.elapsed_seconds, 3),
                     "order_package": order_package.info,
                 },
diff --git a/flumine/flumine.py b/flumine/flumine.py
index cbfd5a9..1590da9 100644
--- a/flumine/flumine.py
+++ b/flumine/flumine.py
@@ -77,8 +77,8 @@ class Flumine(BaseFlumine):
         self.add_worker(
             worker.BackgroundWorker(
                 self,
-                function=worker.poll_cleared_orders,
-                interval=10,  # restart
+                function=worker.poll_market_closure,
+                interval=60,
                 start_delay=10,  # wait for login
             )
         )
diff --git a/flumine/markets/market.py b/flumine/markets/market.py
index 1ce3e88..d0a62ee 100644
--- a/flumine/markets/market.py
+++ b/flumine/markets/market.py
@@ -32,6 +32,8 @@ class Market:
         self.market_book = market_book
         self.market_catalogue = market_catalogue
         self.update_market_catalogue = True
+        self.orders_cleared = False
+        self.market_cleared = False
         self.context = {"simulated": {}}  # data store (raceCard / scores etc)
         self.blotter = Blotter(market_id)
         self._transaction_id = 0
@@ -43,6 +45,8 @@ class Market:
 
     def open_market(self) -> None:
         self.closed = False
+        self.orders_cleared = False
+        self.market_cleared = False
         logger.info(
             "Market {0} opened".format(self.market_id),
             extra=self.info,
@@ -171,4 +175,7 @@ class Market:
             "country_code": self.country_code,
             "venue": self.venue,
             "race_type": self.race_type,
+            "orders_cleared": self.orders_cleared,
+            "market_cleared": self.market_cleared,
+            "closed": self.closed,
         }
diff --git a/flumine/worker.py b/flumine/worker.py
index ae591c6..33c6be4 100644
--- a/flumine/worker.py
+++ b/flumine/worker.py
@@ -155,28 +155,23 @@ def poll_account_balance(context: dict, flumine) -> None:
         flumine.log_control(events.BalanceEvent(client))
 
 
-def poll_cleared_orders(context: dict, flumine) -> None:
+def poll_market_closure(context: dict, flumine) -> None:
     client = flumine.client
-    processed_orders, processed_markets, market_cache = [], [], []
-    while True:
-        market_id = flumine.cleared_market_queue.get()
-        if market_id not in processed_orders:
-            if _get_cleared_orders(flumine, client.betting_client, market_id):
-                processed_orders.append(market_id)
-            else:
-                time.sleep(10)
-                flumine.cleared_market_queue.put(market_id)  # try again
-
-        if market_id not in processed_markets:
-            time.sleep(32)  # takes ~30s for orders to be aggregated to market level
-            if _get_cleared_market(flumine, client.betting_client, market_id):
-                processed_markets.append(market_id)
-            else:
-                if market_cache.count(market_id) > 7:  # give up after 7 attempts
-                    market_cache = [m for m in market_cache if m != market_id]
-                else:
-                    market_cache.append(market_id)
-                    flumine.cleared_market_queue.put(market_id)  # try again
+    if client.paper_trade:
+        return
+    markets = [
+        market
+        for market in list(flumine.markets.markets.values())
+        if market.closed
+        and (market.orders_cleared is False or market.market_cleared is False)
+    ]
+    for market in markets:
+        if market.orders_cleared is False:
+            if _get_cleared_orders(flumine, client.betting_client, market.market_id):
+                market.orders_cleared = True
+        if market.market_cleared is False:
+            if _get_cleared_market(flumine, client.betting_client, market.market_id):
+                market.market_cleared = True
 
 
 def _get_cleared_orders(flumine, betting_client, market_id: str) -> bool:
diff --git a/requirements.txt b/requirements.txt
index aee936f..6841d11 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,4 @@
-betfairlightweight==2.12.1
+betfairlightweight==2.12.0
 tenacity==5.0.3
 python-json-logger==2.0.1
 requests


Test patch:
diff --git a/tests/test_baseflumine.py b/tests/test_baseflumine.py
index 32c9c2a..74c3440 100644
--- a/tests/test_baseflumine.py
+++ b/tests/test_baseflumine.py
@@ -208,9 +208,6 @@ class BaseFlumineTest(unittest.TestCase):
             mock_market, mock_market_book
         )
         mock_log_control.assert_called_with(mock_event)
-        self.assertEqual(
-            self.base_flumine.cleared_market_queue.get(), mock_market_book.market_id
-        )
 
     @mock.patch("flumine.baseflumine.BaseFlumine.info")
     @mock.patch("flumine.baseflumine.BaseFlumine.log_control")
diff --git a/tests/test_execution.py b/tests/test_execution.py
index abc8ecb..7763d65 100644
--- a/tests/test_execution.py
+++ b/tests/test_execution.py
@@ -139,14 +139,16 @@ class BaseExecutionTest(unittest.TestCase):
         self.assertEqual(self.execution._sessions, [mock_session, mock_session])
         self.assertGreater(mock_session.time_returned, 0)
 
-    def test__return_http_session_returned(self):
+    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):
diff --git a/tests/test_flumine.py b/tests/test_flumine.py
index 7022306..b6d4e4d 100644
--- a/tests/test_flumine.py
+++ b/tests/test_flumine.py
@@ -84,8 +84,8 @@ class FlumineTest(unittest.TestCase):
                 ),
                 mock.call(
                     self.flumine,
-                    function=worker.poll_cleared_orders,
-                    interval=10,
+                    function=worker.poll_market_closure,
+                    interval=60,
                     start_delay=10,
                 ),
             ],
diff --git a/tests/test_markets.py b/tests/test_markets.py
index 6346885..7c57433 100644
--- a/tests/test_markets.py
+++ b/tests/test_markets.py
@@ -109,6 +109,8 @@ class MarketTest(unittest.TestCase):
         self.assertEqual(self.market.flumine, self.mock_flumine)
         self.assertEqual(self.market.market_id, "1.234")
         self.assertFalse(self.market.closed)
+        self.assertFalse(self.market.orders_cleared)
+        self.assertFalse(self.market.market_cleared)
         self.assertIsNone(self.market.date_time_closed)
         self.assertEqual(self.market.market_book, self.mock_market_book)
         self.assertEqual(self.market.market_catalogue, self.mock_market_catalogue)
@@ -122,8 +124,13 @@ class MarketTest(unittest.TestCase):
         self.assertTrue(self.market.update_market_catalogue)
 
     def test_open_market(self):
+        self.market.closed = True
+        self.market.orders_cleared = True
+        self.market.market_cleared = True
         self.market.open_market()
         self.assertFalse(self.market.closed)
+        self.assertFalse(self.market.orders_cleared)
+        self.assertFalse(self.market.market_cleared)
 
     def test_close_market(self):
         self.market.close_market()
@@ -321,5 +328,8 @@ class MarketTest(unittest.TestCase):
                 "country_code": self.market.country_code,
                 "venue": self.market.venue,
                 "race_type": self.market.race_type,
+                "orders_cleared": self.market.orders_cleared,
+                "market_cleared": self.market.market_cleared,
+                "closed": self.market.closed,
             },
         )
diff --git a/tests/test_worker.py b/tests/test_worker.py
index 6fe26d7..d4621dd 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -127,14 +127,35 @@ class WorkersTest(unittest.TestCase):
             mock_events.BalanceEvent(mock_flumine.client.account_funds)
         )
 
-    # @mock.patch("flumine.worker._get_cleared_market")
-    # @mock.patch("flumine.worker._get_cleared_orders")
-    # def test_poll_cleared_orders(self, mock__get_cleared_orders, mock__get_cleared_market):
-    #     mock_flumine = mock.Mock()
-    #     mock_client = mock.Mock()
-    #
-    #     worker.poll_cleared_orders(mock_flumine, mock_client)
-    #
+    @mock.patch("flumine.worker._get_cleared_market")
+    @mock.patch("flumine.worker._get_cleared_orders")
+    def test_poll_market_closure(
+        self, mock__get_cleared_orders, mock__get_cleared_market
+    ):
+        mock_client = mock.Mock(paper_trade=False)
+        mock_flumine = mock.Mock(client=mock_client)
+        market_one = mock.Mock(closed=False)
+        market_two = mock.Mock(closed=True, orders_cleared=True, market_cleared=True)
+        market_three = mock.Mock(
+            closed=True, orders_cleared=False, market_cleared=False
+        )
+        mock_flumine.markets.markets = {
+            1: market_one,
+            2: market_two,
+            3: market_three,
+        }
+        worker.poll_market_closure({}, mock_flumine)
+        mock__get_cleared_orders.assert_called_with(
+            mock_flumine, mock_client.betting_client, market_three.market_id
+        )
+        mock__get_cleared_market.assert_called_with(
+            mock_flumine, mock_client.betting_client, market_three.market_id
+        )
+
+    def test_poll_market_closure_paper(self):
+        mock_client = mock.Mock(paper_trade=True)
+        mock_flumine = mock.Mock(client=mock_client)
+        worker.poll_market_closure({}, mock_flumine)
 
     @mock.patch("flumine.worker.config")
     @mock.patch("flumine.worker.events")