Repository: liampauling/flumine
Pull request #552 (2022-01-19 11:57:21)

Adds an ExecutionValidation control to prevent failed requests being sent continuously


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index c333e8e..e021263 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,17 +3,6 @@
 Release History
 ---------------
 
-1.21.3 (2022-01-31)
-+++++++++++++++++++
-
-**Improvements**
-
--
-
-**Bug Fixes**
-
-- trading control exposure bug revert
-
 1.21.2 (2022-01-13)
 +++++++++++++++++++
 
diff --git a/docs/performance.md b/docs/performance.md
index 7dcfa7a..99d8027 100644
--- a/docs/performance.md
+++ b/docs/performance.md
@@ -126,14 +126,6 @@ Profiling code is always the best option for finding improvements, `cprofilev` i
 python -m cprofilev examples/backtest.py
 ```
 
-### Middleware
-
-If you don't need the simulation middleware remove it from `framework._market_middleware`, this is useful when processing markets for data collection. This can dramatically improve processing time due to the heavy functions contained in the simulation logic.
-
-### Libraries
-
-Installing betfairlightweight[speed] will have a big impact on processing speed due to the inclusion of C and Rust libraries for datetime and json decoding.
-
 ## Live
 
 For improving live trading 'Strategy' and 'cprofile' tips above will help although CPU load tends to be considerably lower compared to backtesting.
diff --git a/flumine/__version__.py b/flumine/__version__.py
index 3ab493c..0d0c49e 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.21.3"
+__version__ = "1.21.2"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/config.py b/flumine/config.py
index ad22ae7..77ce747 100644
--- a/flumine/config.py
+++ b/flumine/config.py
@@ -29,3 +29,5 @@ update_latency = 0.150
 replace_latency = 0.280
 
 order_sep = "-"  # customer_order_ref separator
+
+execution_retry_attempts = 10  # cancel attempts when the OrderStream is not connected
diff --git a/flumine/controls/tradingcontrols.py b/flumine/controls/tradingcontrols.py
index da780fa..af4884b 100644
--- a/flumine/controls/tradingcontrols.py
+++ b/flumine/controls/tradingcontrols.py
@@ -1,10 +1,12 @@
 import logging
 
 from ..clients.clients import ExchangeType
+from ..order.order import OrderStatus
 from ..order.ordertype import OrderTypes
 from ..order.orderpackage import OrderPackageType, BaseOrder
 from . import BaseControl
-from .. import utils
+from .. import utils, config
+from ..streams.orderstream import OrderStream
 
 logger = logging.getLogger(__name__)
 
@@ -68,7 +70,7 @@ class OrderValidation(BaseControl):
             size = order.order_type.size or order.order_type.bet_target_size
             if (
                 size < client.min_bet_size
-                and (order.order_type.price * size) < client.min_bet_payout
+                and (order.order_type.price - 1) * size < client.min_bet_payout
             ):
                 self._on_error(
                     order,
@@ -120,6 +122,58 @@ class MarketValidation(BaseControl):
             self._on_error(order, "Market is not open")
 
 
+class ExecutionValidation(BaseControl):
+    """
+    Validates the OrderStream is connected when executing bets
+    """
+
+    NAME = "EXECUTION_VALIDATION"
+
+    @property
+    def order_stream_connected(self):
+        for stream in self.flumine.streams:
+            if isinstance(stream, OrderStream):
+                if stream._stream._running:  # todo: add property in bflw
+                    return True
+        return False
+
+    @staticmethod
+    def failed_execution_attempts(responses):
+        return len([response for response in responses if response.status != "SUCCESS"])
+
+    def validate_order(self, order, package_type):
+        if package_type == OrderPackageType.REPLACE:
+            failed_attempts = self.failed_execution_attempts(
+                order.responses.replace_responses
+            )
+        elif package_type == OrderPackageType.UPDATE:
+            failed_attempts = self.failed_execution_attempts(
+                order.responses.update_responses
+            )
+        elif package_type == OrderPackageType.CANCEL:
+            failed_attempts = self.failed_execution_attempts(
+                order.responses.cancel_responses
+            )
+        else:
+            return
+
+        if (
+            not self.order_stream_connected
+            and failed_attempts >= config.execution_retry_attempts
+        ):
+            self._on_error(
+                order,
+                "OrderStream is not connected, execution of orders is blocked until OrderStream is reconnected",
+            )
+
+    def _validate(self, order: BaseOrder, package_type: OrderPackageType) -> None:
+        if self.flumine.BACKTEST or self.flumine.client.paper_trade:
+            return
+
+        if order.EXCHANGE == ExchangeType.BETFAIR:
+            self.validate_order(order, package_type)
+
+
 class StrategyExposure(BaseControl):
 
     """
diff --git a/flumine/markets/blotter.py b/flumine/markets/blotter.py
index 8410639..e9405d4 100644
--- a/flumine/markets/blotter.py
+++ b/flumine/markets/blotter.py
@@ -102,10 +102,6 @@ class Blotter:
                     runner.handicap,
                 ):
                     order.runner_status = runner.status
-                    order.market_type = market_book.market_definition.market_type
-                    order.each_way_divisor = (
-                        market_book.market_definition.each_way_divisor
-                    )
                     if number_of_winners > market_book.number_of_winners:
                         order.number_of_dead_heat_winners = number_of_winners
 
diff --git a/flumine/order/order.py b/flumine/order/order.py
index 2e66c0d..f449abd 100644
--- a/flumine/order/order.py
+++ b/flumine/order/order.py
@@ -76,8 +76,6 @@ class BaseOrder:
         self.lookup = self.market_id, self.selection_id, self.handicap
 
         self.runner_status = None  # RunnerBook.status
-        self.market_type = None
-        self.each_way_divisor = 1
         self.number_of_dead_heat_winners = None
         self.status = None
         self.complete = False
@@ -230,9 +228,10 @@ class BaseOrder:
 
     @property
     def elapsed_seconds(self) -> Optional[float]:
-        date_time_placed = self.responses.date_time_placed
-        if date_time_placed:
-            return (datetime.datetime.utcnow() - date_time_placed).total_seconds()
+        if self.responses.date_time_placed:
+            return (
+                datetime.datetime.utcnow() - self.responses.date_time_placed
+            ).total_seconds()
         else:
             return
 


Test patch:
diff --git a/flumine/backtest/simulated.py b/flumine/backtest/simulated.py
index 9999630..e345f24 100644
--- a/flumine/backtest/simulated.py
+++ b/flumine/backtest/simulated.py
@@ -400,59 +400,32 @@ class Simulated:
 
     @property
     def profit(self) -> float:
-        number_of_dead_heat_winners = self.order.number_of_dead_heat_winners or 1
-        if self.order.market_type == "EACH_WAY":
-            divisor = self.order.each_way_divisor
-            win = self.size_matched * (self.average_price_matched - 1)
-            place = self.size_matched * (
-                (self.average_price_matched - 1) * (1 / divisor)
+        if self.order.runner_status == "WINNER":
+            number_of_dead_heat_winners = self.order.number_of_dead_heat_winners or 1
+            profit = (self.size_matched / number_of_dead_heat_winners) * (
+                self.average_price_matched - 1
             )
-            if number_of_dead_heat_winners > 1:  # todo dead heat
-                logger.error("Each Way Dead Heat Detected but not handled")
-            if self.order.runner_status == "WINNER":
-                profit = round(win + place, 2)
-                if self.side == "BACK":
-                    return profit
-                else:
-                    return -profit
-            elif self.order.runner_status == "PLACED":
-                if self.side == "BACK":
-                    profit = place - self.size_matched
-                    return round(profit, 2)
-                else:
-                    profit = self.size_matched - place
-                    return round(profit, 2)
-            elif self.order.runner_status == "LOSER":
-                matched = round(self.size_matched * 2, 2)
-                if self.side == "BACK":
-                    return -matched
-                else:
-                    return matched
-            else:
-                return 0.0
-        else:
-            if self.order.runner_status == "WINNER":
-                profit = (self.size_matched / number_of_dead_heat_winners) * (
-                    self.average_price_matched - 1
+
+            if number_of_dead_heat_winners == 2:
+                profit = profit - (self.size_matched / number_of_dead_heat_winners)
+            elif number_of_dead_heat_winners > 2:
+                profit = profit - (
+                    self.size_matched
+                    * (number_of_dead_heat_winners - 1)
+                    / number_of_dead_heat_winners
                 )
-                if number_of_dead_heat_winners == 2:
-                    profit = profit - (self.size_matched / number_of_dead_heat_winners)
-                elif number_of_dead_heat_winners > 2:
-                    profit = profit - (
-                        self.size_matched
-                        * (number_of_dead_heat_winners - 1)
-                        / number_of_dead_heat_winners
-                    )
-                if self.side == "LAY":
-                    profit = -profit
-                return round(profit, 2)
-            elif self.order.runner_status == "LOSER":
-                if self.side == "BACK":
-                    return -self.size_matched
-                else:
-                    return self.size_matched
+
+            if self.side == "LAY":
+                profit = -profit
+
+            return round(profit, 2)
+        elif self.order.runner_status == "LOSER":
+            if self.side == "BACK":
+                return -self.size_matched
             else:
-                return 0.0
+                return self.size_matched
+        else:
+            return 0.0
 
     @property
     def status(self) -> str:
diff --git a/tests/test_blotter.py b/tests/test_blotter.py
index c63f4c6..057e65f 100644
--- a/tests/test_blotter.py
+++ b/tests/test_blotter.py
@@ -121,13 +121,6 @@ class BlotterTest(unittest.TestCase):
         self.blotter._orders = {"12345": mock_order}
         self.blotter.process_closed_market(mock_market_book)
         self.assertEqual(mock_order.runner_status, mock_runner.status)
-        self.assertEqual(
-            mock_order.market_type, mock_market_book.market_definition.market_type
-        )
-        self.assertEqual(
-            mock_order.each_way_divisor,
-            mock_market_book.market_definition.each_way_divisor,
-        )
 
     def test_process_cleared_orders(self):
         mock_cleared_orders = mock.Mock()
diff --git a/tests/test_config.py b/tests/test_config.py
index a992cf2..d9761d0 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -18,3 +18,4 @@ class ConfigTest(unittest.TestCase):
         self.assertEqual(config.update_latency, 0.150)
         self.assertEqual(config.replace_latency, 0.280)
         self.assertEqual(config.order_sep, "-")
+        self.assertEqual(config.execution_retry_attempts, 10)
diff --git a/tests/test_order.py b/tests/test_order.py
index f428080..e134e2d 100644
--- a/tests/test_order.py
+++ b/tests/test_order.py
@@ -40,8 +40,6 @@ class BaseOrderTest(unittest.TestCase):
             (self.order.market_id, self.order.selection_id, self.order.handicap),
         )
         self.assertIsNone(self.order.runner_status)
-        self.assertIsNone(self.order.market_type)
-        self.assertEqual(self.order.each_way_divisor, 1)
         self.assertIsNone(self.order.status)
         self.assertFalse(self.order.complete)
         self.assertEqual(self.order.status_log, [])
diff --git a/tests/test_simulated.py b/tests/test_simulated.py
index 7ccec47..1b8d0e3 100644
--- a/tests/test_simulated.py
+++ b/tests/test_simulated.py
@@ -738,31 +738,6 @@ class SimulatedTest(unittest.TestCase):
         self.simulated._update_matched([1234, 11.0, 50.0])
         self.assertEqual(self.simulated.profit, 87.50)
 
-    def test_profit_ew_back(self):
-        self.simulated.order.market_type = "EACH_WAY"
-        self.simulated.order.each_way_divisor = 5
-        self.assertEqual(self.simulated.profit, 0)
-        self.simulated.order.runner_status = "WINNER"
-        self.simulated._update_matched([1234, 10.0, 2.0])
-        self.assertEqual(self.simulated.profit, 21.6)
-        self.simulated.order.runner_status = "PLACED"
-        self.assertEqual(self.simulated.profit, 1.6)
-        self.simulated.order.runner_status = "LOSER"
-        self.assertEqual(self.simulated.profit, -4.0)
-
-    def test_profit_ew_lay(self):
-        self.simulated.order.side = "LAY"
-        self.simulated.order.market_type = "EACH_WAY"
-        self.simulated.order.each_way_divisor = 5
-        self.assertEqual(self.simulated.profit, 0)
-        self.simulated.order.runner_status = "WINNER"
-        self.simulated._update_matched([1234, 10.0, 2.0])
-        self.assertEqual(self.simulated.profit, -21.6)
-        self.simulated.order.runner_status = "PLACED"
-        self.assertEqual(self.simulated.profit, -1.6)
-        self.simulated.order.runner_status = "LOSER"
-        self.assertEqual(self.simulated.profit, 4.0)
-
     @mock.patch(
         "flumine.backtest.simulated.Simulated.size_remaining",
         new_callable=mock.PropertyMock,
diff --git a/tests/test_tradingcontrols.py b/tests/test_tradingcontrols.py
index 0e9fc2e..4469e65 100644
--- a/tests/test_tradingcontrols.py
+++ b/tests/test_tradingcontrols.py
@@ -1,6 +1,7 @@
 import unittest
 from unittest import mock
 
+from flumine import config
 from flumine.controls.tradingcontrols import (
     OrderValidation,
     StrategyExposure,
@@ -8,9 +9,11 @@ from flumine.controls.tradingcontrols import (
     ExchangeType,
     OrderPackageType,
     MarketValidation,
+    ExecutionValidation,
 )
 from flumine.markets.blotter import Blotter
 from flumine.order.order import OrderStatus
+from flumine.streams.orderstream import OrderStream
 
 
 class TestOrderValidation(unittest.TestCase):
@@ -347,6 +350,163 @@ class TestMarketValidation(unittest.TestCase):
         mock_on_error.assert_called()
 
 
+class TestExecutionValidation(unittest.TestCase):
+    def setUp(self):
+        self.mock_order_stream = mock.Mock(
+            spec=OrderStream, _stream=mock.Mock(_running=True)
+        )
+        self.mock_flumine = mock.Mock(
+            BACKTEST=False,
+            client=mock.Mock(paper_trade=False),
+            streams=[self.mock_order_stream],
+        )
+        self.trading_control = ExecutionValidation(self.mock_flumine)
+        self.mock_order = mock.Mock(
+            EXCHANGE=ExchangeType.BETFAIR,
+            responses=mock.Mock(
+                replace_responses=[], update_responses=[], cancel_responses=[]
+            ),
+        )
+
+    def tearDown(self) -> None:
+        config.execution_retry_attempts = 10
+
+    def test_order_stream_connected(self):
+        self.assertTrue(self.trading_control.order_stream_connected)
+        self.mock_order_stream._stream._running = False
+        self.assertFalse(self.trading_control.order_stream_connected)
+
+    def test_order_stream_missing(self):
+        self.mock_flumine.streams = []
+        self.assertFalse(self.trading_control.order_stream_connected)
+
+    def test_failed_execution_attempts(self):
+        mock_responses = [
+            mock.Mock(status="SUCCESS"),
+            mock.Mock(status="SUCCESS"),
+            mock.Mock(status="SUCCESS"),
+        ]
+        self.assertEqual(
+            self.trading_control.failed_execution_attempts(mock_responses), 0
+        )
+
+        mock_responses = [
+            mock.Mock(status="FAILURE"),
+            mock.Mock(status="SUCCESS"),
+            mock.Mock(status="TIMEOUT"),
+        ]
+        self.assertEqual(
+            self.trading_control.failed_execution_attempts(mock_responses), 2
+        )
+
+    @mock.patch("flumine.controls.tradingcontrols.ExecutionValidation._on_error")
+    def test_validate_order(self, mock_on_error):
+        self.trading_control.validate_order(self.mock_order, OrderPackageType.PLACE)
+        mock_on_error.assert_not_called()
+
+        self.trading_control.validate_order(self.mock_order, OrderPackageType.REPLACE)
+        mock_on_error.assert_not_called()
+
+        self.trading_control.validate_order(self.mock_order, OrderPackageType.UPDATE)
+        mock_on_error.assert_not_called()
+
+        self.trading_control.validate_order(self.mock_order, OrderPackageType.CANCEL)
+        mock_on_error.assert_not_called()
+
+    @mock.patch(
+        "flumine.controls.tradingcontrols.ExecutionValidation.failed_execution_attempts"
+    )
+    def test_validate_order_responses(self, mock_failed_execution_attempts):
+        self.mock_replace_responses = [mock.Mock()]
+        self.mock_order.responses.replace_responses = self.mock_replace_responses
+        self.trading_control.validate_order(self.mock_order, OrderPackageType.REPLACE)
+        mock_failed_execution_attempts.assert_called_with(self.mock_replace_responses)
+
+        mock_failed_execution_attempts.reset_mock()
+
+        self.mock_update_responses = [mock.Mock()]
+        self.mock_order.responses.update_responses = self.mock_update_responses
+        self.trading_control.validate_order(self.mock_order, OrderPackageType.UPDATE)
+        mock_failed_execution_attempts.assert_called_with(self.mock_update_responses)
+
+        mock_failed_execution_attempts.reset_mock()
+
+        self.mock_cancel_responses = [mock.Mock()]
+        self.mock_order.responses.cancel_responses = self.mock_cancel_responses
+        self.trading_control.validate_order(self.mock_order, OrderPackageType.CANCEL)
+        mock_failed_execution_attempts.assert_called_with(self.mock_cancel_responses)
+
+    @mock.patch("flumine.controls.tradingcontrols.ExecutionValidation._on_error")
+    @mock.patch(
+        "flumine.controls.tradingcontrols.ExecutionValidation.order_stream_connected",
+        new_callable=mock.PropertyMock,
+    )
+    @mock.patch(
+        "flumine.controls.tradingcontrols.ExecutionValidation.failed_execution_attempts"
+    )
+    def test_validate_error(
+        self, mock_failed_execution_attempts, mock_order_stream_connected, mock_on_error
+    ):
+        mock_failed_execution_attempts.return_value = 5
+        mock_order_stream_connected.return_value = False
+        self.trading_control.validate_order(self.mock_order, OrderPackageType.CANCEL)
+        mock_on_error.assert_not_called()
+
+        mock_failed_execution_attempts.return_value = 12
+        mock_order_stream_connected.return_value = True
+        self.trading_control.validate_order(self.mock_order, OrderPackageType.CANCEL)
+        mock_on_error.assert_not_called()
+
+        mock_failed_execution_attempts.return_value = 12
+        mock_order_stream_connected.return_value = False
+        self.trading_control.validate_order(self.mock_order, OrderPackageType.CANCEL)
+        mock_on_error.assert_called()
+
+    @mock.patch("flumine.controls.tradingcontrols.ExecutionValidation._on_error")
+    @mock.patch(
+        "flumine.controls.tradingcontrols.ExecutionValidation.order_stream_connected",
+        new_callable=mock.PropertyMock,
+    )
+    @mock.patch(
+        "flumine.controls.tradingcontrols.ExecutionValidation.failed_execution_attempts"
+    )
+    def test__validate_execution_retry_attempts(
+        self, mock_failed_execution_attempts, mock_order_stream_connected, mock_on_error
+    ):
+        mock_failed_execution_attempts.return_value = 5
+        mock_order_stream_connected.return_value = False
+        config.execution_retry_attempts = 4
+        self.trading_control.validate_order(self.mock_order, OrderPackageType.CANCEL)
+        mock_on_error.assert_called()
+
+    @mock.patch("flumine.controls.tradingcontrols.ExecutionValidation.validate_order")
+    def test__validate(self, mock_validate_order):
+        self.mock_flumine.BACKTEST = False
+        self.mock_flumine.client.paper_trade = False
+        self.trading_control._validate(self.mock_order, OrderPackageType.CANCEL)
+        mock_validate_order.assert_called()
+
+        mock_validate_order.reset_mock()
+
+        self.mock_flumine.BACKTEST = True
+        self.mock_flumine.client.paper_trade = False
+        self.trading_control._validate(self.mock_order, OrderPackageType.CANCEL)
+        mock_validate_order.assert_not_called()
+
+        mock_validate_order.reset_mock()
+
+        self.mock_flumine.BACKTEST = False
+        self.mock_flumine.client.paper_trade = True
+        self.trading_control._validate(self.mock_order, OrderPackageType.CANCEL)
+        mock_validate_order.assert_not_called()
+
+    @mock.patch("flumine.controls.tradingcontrols.ExecutionValidation.validate_order")
+    def test__validate_exchange(self, mock_validate_order):
+        self.mock_order.EXCHANGE = "OTHER"
+        self.trading_control._validate(self.mock_order, OrderPackageType.CANCEL)
+        mock_validate_order.assert_not_called()
+
+
 class TestStrategyExposure(unittest.TestCase):
     def setUp(self):
         self.market = mock.Mock()