Repository: liampauling/flumine
Pull request #444 (2021-06-02 04:57:50)

MarketRecorder race condition on file load / remove txt only if aged


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index 00582ec..1b717a7 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,6 +3,22 @@
 Release History
 ---------------
 
+1.19.2 (2021-06-03)
++++++++++++++++++++
+
+**Improvements**
+
+- Move simulated latencies to config (@lunswor)
+- Add control to validate market status  (@lunswor)
+
+**Bug Fixes**
+
+- MarketRecorder race condition on file load / remove txt only if aged
+
+**Libraries**
+
+- black upgraded to black==21.5b2
+
 1.19.1 (2021-05-27)
 +++++++++++++++++++
 
diff --git a/examples/strategies/marketrecorder.py b/examples/strategies/marketrecorder.py
index c69e0bd..0c08fdf 100644
--- a/examples/strategies/marketrecorder.py
+++ b/examples/strategies/marketrecorder.py
@@ -111,6 +111,12 @@ class MarketRecorder(BaseStrategy):
         # process compression/load in thread
         while True:
             market, file_dir, market_definition = self._queue.get(block=True)
+            # check file still exists (potential race condition)
+            if not os.path.isfile(file_dir):
+                logger.warning(
+                    "File: %s does not exist in %s" % (market.market_id, file_dir)
+                )
+                continue
             # compress file
             compress_file_dir = self._compress_file(file_dir)
             # core load code
@@ -170,11 +176,14 @@ class MarketRecorder(BaseStrategy):
                         os.remove(gz_path)
                     txt_path = os.path.join(directory, file.split(".gz")[0])
                     if os.path.exists(txt_path) and self._remove_file:
-                        logger.info(
-                            "Removing: %s, age: %ss"
-                            % (txt_path, round(seconds_since, 2))
-                        )
-                        os.remove(txt_path)
+                        file_stats = os.stat(txt_path)
+                        seconds_since = time.time() - file_stats.st_mtime
+                        if seconds_since > self._market_expiration:
+                            logger.info(
+                                "Removing: %s, age: %ss"
+                                % (txt_path, round(seconds_since, 2))
+                            )
+                            os.remove(txt_path)
 
     @staticmethod
     def _create_metadata(market_definition: dict) -> dict:
diff --git a/flumine/__version__.py b/flumine/__version__.py
index 8ec48d0..0e433c6 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.19.1"
+__version__ = "1.19.2"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/baseflumine.py b/flumine/baseflumine.py
index 2817b29..073df18 100644
--- a/flumine/baseflumine.py
+++ b/flumine/baseflumine.py
@@ -17,7 +17,11 @@ from .execution.betfairexecution import BetfairExecution
 from .execution.simulatedexecution import SimulatedExecution
 from .order.process import process_current_orders
 from .controls.clientcontrols import BaseControl, MaxTransactionCount
-from .controls.tradingcontrols import OrderValidation, StrategyExposure
+from .controls.tradingcontrols import (
+    OrderValidation,
+    StrategyExposure,
+    MarketValidation,
+)
 from .controls.loggingcontrols import LoggingControl
 from .exceptions import FlumineException
 from . import config, utils
@@ -69,6 +73,7 @@ class BaseFlumine:
         self.trading_controls = []
         # add default controls (processed in order)
         self.add_trading_control(OrderValidation)
+        self.add_trading_control(MarketValidation)
         self.add_trading_control(StrategyExposure)
         # register default client controls (processed in order)
         self.add_client_control(MaxTransactionCount)
diff --git a/flumine/config.py b/flumine/config.py
index 085a23d..db6faaf 100644
--- a/flumine/config.py
+++ b/flumine/config.py
@@ -18,3 +18,9 @@ raise_errors = False  # used for call_check_market / call_process_market_book
 max_execution_workers = 32  # max number of workers in execution thread pool
 
 async_place_orders = False  # async place orders
+
+# latencies used for backtesting
+place_latency = 0.120
+cancel_latency = 0.170
+update_latency = 0.150
+replace_latency = 0.280
diff --git a/flumine/controls/tradingcontrols.py b/flumine/controls/tradingcontrols.py
index aef96b4..95ec094 100644
--- a/flumine/controls/tradingcontrols.py
+++ b/flumine/controls/tradingcontrols.py
@@ -98,6 +98,23 @@ class OrderValidation(BaseControl):
                 )
 
 
+class MarketValidation(BaseControl):
+    """
+    Validates market is open for orders
+    """
+
+    NAME = "MARKET_VALIDATION"
+
+    def _validate(self, order: BaseOrder, package_type: OrderPackageType) -> None:
+        if order.EXCHANGE == ExchangeType.BETFAIR:
+            self._validate_betfair_market_status(order)
+
+    def _validate_betfair_market_status(self, order):
+        market = self.flumine.markets.markets.get(order.market_id)
+        if market and market.market_book.status != "OPEN":
+            self._on_error(order, "Market is not open")
+
+
 class StrategyExposure(BaseControl):
 
     """
diff --git a/flumine/execution/simulatedexecution.py b/flumine/execution/simulatedexecution.py
index 70bb195..b3c21f6 100644
--- a/flumine/execution/simulatedexecution.py
+++ b/flumine/execution/simulatedexecution.py
@@ -3,6 +3,7 @@ import requests
 from typing import Optional
 
 from .baseexecution import BaseExecution
+from .. import config
 from ..clients.clients import ExchangeType
 from ..order.orderpackage import BaseOrderPackage, OrderPackageType
 
@@ -10,10 +11,10 @@ from ..order.orderpackage import BaseOrderPackage, OrderPackageType
 class SimulatedExecution(BaseExecution):
 
     EXCHANGE = ExchangeType.SIMULATED
-    PLACE_LATENCY = 0.120
-    CANCEL_LATENCY = 0.170
-    UPDATE_LATENCY = 0.150
-    REPLACE_LATENCY = 0.280
+    PLACE_LATENCY = config.place_latency
+    CANCEL_LATENCY = config.cancel_latency
+    UPDATE_LATENCY = config.update_latency
+    REPLACE_LATENCY = config.replace_latency
 
     def handler(self, order_package: BaseOrderPackage) -> None:
         """Only uses _thread_pool if paper_trade"""


Test patch:
diff --git a/requirements-test.txt b/requirements-test.txt
index ef0c0d2..e2dff21 100644
--- a/requirements-test.txt
+++ b/requirements-test.txt
@@ -1,5 +1,5 @@
 # Tests & Linting
-black==21.5b1
+black==21.5b2
 coverage
 
 # Documentation
diff --git a/tests/test_baseflumine.py b/tests/test_baseflumine.py
index 4dd4a61..8f2f7df 100644
--- a/tests/test_baseflumine.py
+++ b/tests/test_baseflumine.py
@@ -15,7 +15,7 @@ class BaseFlumineTest(unittest.TestCase):
         self.assertFalse(self.base_flumine._running)
         self.assertEqual(self.base_flumine._market_middleware, [])
         self.assertEqual(self.base_flumine._logging_controls, [])
-        self.assertEqual(len(self.base_flumine.trading_controls), 2)
+        self.assertEqual(len(self.base_flumine.trading_controls), 3)
         self.assertEqual(self.base_flumine._workers, [])
 
     @mock.patch("flumine.baseflumine.SimulatedMiddleware")
@@ -67,7 +67,7 @@ class BaseFlumineTest(unittest.TestCase):
     def test_add_trading_control(self):
         mock_control = mock.Mock()
         self.base_flumine.add_trading_control(mock_control)
-        self.assertEqual(len(self.base_flumine.trading_controls), 3)
+        self.assertEqual(len(self.base_flumine.trading_controls), 4)
 
     def test_add_market_middleware(self):
         mock_middleware = mock.Mock()
diff --git a/tests/test_config.py b/tests/test_config.py
index 414a072..2069d73 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -12,3 +12,7 @@ class ConfigTest(unittest.TestCase):
         self.assertFalse(config.raise_errors)
         self.assertEqual(config.max_execution_workers, 32)
         self.assertFalse(config.async_place_orders)
+        self.assertEqual(config.place_latency, 0.120)
+        self.assertEqual(config.cancel_latency, 0.170)
+        self.assertEqual(config.update_latency, 0.150)
+        self.assertEqual(config.replace_latency, 0.280)
diff --git a/tests/test_execution.py b/tests/test_execution.py
index b025fb4..06a8b3e 100644
--- a/tests/test_execution.py
+++ b/tests/test_execution.py
@@ -4,6 +4,8 @@ from unittest import mock
 from unittest.mock import call
 
 from betfairlightweight import BetfairError
+
+from flumine import config
 from flumine.clients.clients import ExchangeType
 from flumine.exceptions import OrderExecutionError
 from flumine.execution.baseexecution import (
@@ -861,6 +863,10 @@ class SimulatedExecutionTest(unittest.TestCase):
 
     def test_init(self):
         self.assertEqual(self.execution.EXCHANGE, ExchangeType.SIMULATED)
+        self.assertEqual(self.execution.PLACE_LATENCY, config.place_latency)
+        self.assertEqual(self.execution.CANCEL_LATENCY, config.cancel_latency)
+        self.assertEqual(self.execution.UPDATE_LATENCY, config.update_latency)
+        self.assertEqual(self.execution.REPLACE_LATENCY, config.replace_latency)
 
     @mock.patch("flumine.execution.simulatedexecution.SimulatedExecution.execute_place")
     def test_handler_paper_trade(self, mock_execute_place):
diff --git a/tests/test_integration.py b/tests/test_integration.py
index 9157933..fd782a5 100644
--- a/tests/test_integration.py
+++ b/tests/test_integration.py
@@ -185,7 +185,7 @@ class IntegrationTest(unittest.TestCase):
             self.assertEqual(
                 round(sum([o.simulated.profit for o in limit_inplay_orders]), 2), 19.88
             )
-            self.assertEqual(len(limit_inplay_orders), 15)
+            self.assertEqual(len(limit_inplay_orders), 14)
             market_orders = market.blotter.strategy_orders(market_strategy)
             self.assertEqual(
                 round(sum([o.simulated.profit for o in market_orders]), 2), -6.68
@@ -248,7 +248,7 @@ class IntegrationTest(unittest.TestCase):
         self.assertEqual(
             round(sum([o.simulated.profit for o in limit_inplay_orders]), 2), 19.88
         )
-        self.assertEqual(len(limit_inplay_orders), 15)
+        self.assertEqual(len(limit_inplay_orders), 14)
         self.assertEqual(win_market._transaction_id, 165)
 
         # Same event
@@ -257,7 +257,7 @@ class IntegrationTest(unittest.TestCase):
         self.assertEqual(
             round(sum([o.simulated.profit for o in limit_inplay_orders]), 2), 101.44
         )
-        self.assertEqual(len(limit_inplay_orders), 88)
+        self.assertEqual(len(limit_inplay_orders), 86)
         self.assertEqual(win_market._transaction_id, 1329)
 
         place_market = framework.markets.markets["1.181223995"]
diff --git a/tests/test_tradingcontrols.py b/tests/test_tradingcontrols.py
index a84120d..55efd74 100644
--- a/tests/test_tradingcontrols.py
+++ b/tests/test_tradingcontrols.py
@@ -7,6 +7,7 @@ from flumine.controls.tradingcontrols import (
     OrderTypes,
     ExchangeType,
     OrderPackageType,
+    MarketValidation,
 )
 from flumine.markets.blotter import Blotter
 
@@ -274,6 +275,39 @@ class TestOrderValidation(unittest.TestCase):
         )
 
 
+class TestMarketValidation(unittest.TestCase):
+    def setUp(self):
+        self.mock_market = mock.Mock(market_book=mock.Mock(status="OPEN"))
+        self.mock_flumine = mock.Mock()
+        self.mock_flumine.markets.markets = {"market_id": self.mock_market}
+        self.trading_control = MarketValidation(self.mock_flumine)
+        self.mock_order = mock.Mock()
+        self.mock_order.market_id = "market_id"
+
+    @mock.patch("flumine.controls.tradingcontrols.MarketValidation._on_error")
+    def test__validate_betfair_market_status(self, mock_on_error):
+        self.trading_control._validate_betfair_market_status(self.mock_order)
+        mock_on_error.assert_not_called()
+
+    @mock.patch("flumine.controls.tradingcontrols.MarketValidation._on_error")
+    def test__validate_betfair_market_status_closed(self, mock_on_error):
+        self.mock_market.market_book.status = "CLOSED"
+        self.trading_control._validate_betfair_market_status(self.mock_order)
+        mock_on_error.assert_called()
+
+    @mock.patch("flumine.controls.tradingcontrols.MarketValidation._on_error")
+    def test__validate_betfair_market_status_suspended(self, mock_on_error):
+        self.mock_market.market_book.status = "SUSPENDED"
+        self.trading_control._validate_betfair_market_status(self.mock_order)
+        mock_on_error.assert_called()
+
+    @mock.patch("flumine.controls.tradingcontrols.MarketValidation._on_error")
+    def test__validate_betfair_market_not_found(self, mock_on_error):
+        self.mock_flumine.markets.markets = {}
+        self.trading_control._validate_betfair_market_status(self.mock_order)
+        mock_on_error.assert_not_called()
+
+
 class TestStrategyExposure(unittest.TestCase):
     def setUp(self):
         self.market = mock.Mock()