all_or_nothing transactions

flumine/execution/transaction.py

@@ -1,5 +1,6 @@
 import logging
 from collections import defaultdict
+from typing import Optional
 
 from ..order.orderpackage import OrderPackageType, BetfairOrderPackage
 from ..events import events
@@ -28,9 +29,27 @@ class Transaction:
             ..
             t.cancel_order(order)
             t.place_order(order)  # both executed on transaction __exit__
+
+    When atomic==True, orders within a transaction will
+    only be placed if the all pass validation.
+    For example, supposed we have:
+
+        with market.transaction(atomic=True) as t:
+            t.place_order(order1) # validation passes
+            t.place_order(order2) # validation does not pass and raises a ControlError
+
+    Neither order1 nor order2 will be executed, as order2 failed validation.
+
     """
 
-    def __init__(self, market, id_: int, async_place_orders: bool, client):
+    def __init__(
+        self,
+        market,
+        id_: int,
+        async_place_orders: bool,
+        client,
+        atomic: Optional[bool] = False,
+    ):
         self.market = market
         self._client = client
         self._id = id_  # unique per market only
@@ -40,6 +59,7 @@ def __init__(self, market, id_: int, async_place_orders: bool, client):
         self._pending_cancel = []  # list of (<Order>, None)
         self._pending_update = []  # list of (<Order>, None)
         self._pending_replace = []  # list of (<Order>, market_version)
+        self.atomic = atomic
 
     def place_order(
         self,
@@ -52,6 +72,7 @@ def place_order(
         if (
             execute
             and not force
+            and not self.atomic
             and self._validate_controls(order, OrderPackageType.PLACE) is False
         ):
             return False
@@ -77,8 +98,6 @@ def place_order(
         if new_trade:
             self.market.flumine.log_control(events.TradeEvent(order.trade))
         if execute:  # handles replaceOrder
-            runner_context = order.trade.strategy.get_runner_context(*order.lookup)
-            runner_context.place(order.trade.id)
             self._pending_place.append((order, market_version))
             self._pending_orders = True
         return True
@@ -94,6 +113,7 @@ def cancel_order(
             )
         if (
             not force
+            and not self.atomic
             and self._validate_controls(order, OrderPackageType.CANCEL) is False
         ):
             return False
@@ -114,6 +134,7 @@ def update_order(
             )
         if (
             not force
+            and not self.atomic
             and self._validate_controls(order, OrderPackageType.UPDATE) is False
         ):
             return False
@@ -134,6 +155,7 @@ def replace_order(
             )
         if (
             not force
+            and not self.atomic
             and self._validate_controls(order, OrderPackageType.REPLACE) is False
         ):
             return False
@@ -146,6 +168,9 @@ def replace_order(
     def execute(self) -> int:
         packages = []
         if self._pending_place:
+            for order, market_version in self._pending_place:
+                runner_context = order.trade.strategy.get_runner_context(*order.lookup)
+                runner_context.place(order.trade.id)
             packages += self._create_order_package(
                 self._pending_place,
                 OrderPackageType.PLACE,
@@ -185,15 +210,18 @@ def execute(self) -> int:
     def _validate_controls(self, order, package_type: OrderPackageType) -> bool:
         # return False on violation
         try:
-            for control in self.market.flumine.trading_controls:
-                control(order, package_type)
-            for control in self._client.trading_controls:
-                control(order, package_type)
+            self._do_validate_controls(order, package_type)
         except ControlError:
             return False
         else:
             return True
 
+    def _do_validate_controls(self, order, package_type):
+        for control in self.market.flumine.trading_controls:
+            control(order, package_type)
+        for control in self._client.trading_controls:
+            control(order, package_type)
+
     def _create_order_package(
         self, orders: list, package_type: OrderPackageType, async_: bool = False
     ) -> list:
@@ -224,5 +252,46 @@ def __enter__(self):
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):
+
         if self._pending_orders:
+
+            if self.atomic:
+                try:
+                    for order in self._pending_place:
+                        self._do_validate_controls(order, OrderPackageType.PLACE)
+                    for order in self._pending_cancel:
+                        self._do_validate_controls(order, OrderPackageType.CANCEL)
+                    for order in self._pending_update:
+                        self._do_validate_controls(order, OrderPackageType.UPDATE)
+                    for order in self._pending_replace:
+                        self._do_validate_controls(order, OrderPackageType.REPLACE)
+                except ControlError as e:
+                    if logger.isEnabledFor(logging.INFO):
+                        extra = {
+                            "market_id": self.market.market_id,
+                            "transaction_id": self._id,
+                            "client_username": self._client.username,
+                        }
+                        if self._pending_place:
+                            extra["pending_place"] = self._pending_place
+                        if self._pending_update:
+                            extra["pending_update"] = self._pending_update
+                        if self._pending_cancel:
+                            extra["pending_cancel"] = self._pending_cancel
+                        if self._pending_replace:
+                            extra["pending_replace"] = self._pending_replace
+                        logger.info(
+                            "Failed to execute transaction. Validation failed: %s"
+                            % str(e),
+                            extra=extra,
+                        )
+                    self._clear()
+                    raise
             self.execute()
+
+    def _clear(self):
+        self._pending_place.clear()
+        self._pending_update.clear()
+        self._pending_replace.clear()
+        self._pending_cancel.clear()
+        self._pending_orders = False

flumine/markets/market.py

@@ -62,7 +62,12 @@ def close_market(self) -> None:
             extra=self.info,
         )
 
-    def transaction(self, async_place_orders: bool = None, client=None) -> Transaction:
+    def transaction(
+        self,
+        async_place_orders: bool = None,
+        client=None,
+        all_or_nothing: Optional[bool] = False,
+    ) -> Transaction:
         if async_place_orders is None:
             async_place_orders = config.async_place_orders
         if client is None:
@@ -73,6 +78,7 @@ def transaction(self, async_place_orders: bool = None, client=None) -> Transacti
             id_=self._transaction_id,
             async_place_orders=async_place_orders,
             client=client,
+            atomic=all_or_nothing,
         )
 
     # order

tests/test_markets.py

@@ -160,6 +160,7 @@ def test_transaction(self, mock_transaction):
             id_=self.market._transaction_id,
             async_place_orders=False,
             client=self.market.flumine.clients.get_default(),
+            atomic=False,
         )
         self.assertEqual(transaction, mock_transaction())
 
@@ -172,6 +173,7 @@ def test_transaction_async(self, mock_transaction):
             id_=self.market._transaction_id,
             async_place_orders=True,
             client=self.market.flumine.clients.get_default(),
+            atomic=False,
         )
         self.assertEqual(transaction, mock_transaction())