Repository: liampauling/flumine
Pull request #439 (2021-05-21 08:08:58)

race condition where markets processed before all executable and exec…
…ution complete orders have been loaded locally

refactor:
- Start order stream first
- Wait for successful socket start
- Add market in process.py if required (previously raised warning)
- Start market streams


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index 0846035..00ab12c 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,6 +3,13 @@
 Release History
 ---------------
 
+1.18.12 (2021-05-21)
++++++++++++++++++++
+
+**Bug Fixes**
+
+- Prevent race condition on executable/execution_complete and new orders
+
 1.18.11 (2021-05-20)
 +++++++++++++++++++
 
diff --git a/flumine/__version__.py b/flumine/__version__.py
index 54f0901..1cf7955 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.18.11"
+__version__ = "1.18.12"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/baseflumine.py b/flumine/baseflumine.py
index 0beb9e5..2817b29 100644
--- a/flumine/baseflumine.py
+++ b/flumine/baseflumine.py
@@ -213,7 +213,9 @@ class BaseFlumine:
 
     def _process_current_orders(self, event: events.CurrentOrdersEvent) -> None:
         # update state
-        process_current_orders(self.markets, self.strategies, event, self.log_control)
+        process_current_orders(
+            self.markets, self.strategies, event, self.log_control, self._add_market
+        )
         for market in self.markets:
             if market.closed is False:
                 for strategy in self.strategies:
diff --git a/flumine/order/process.py b/flumine/order/process.py
index a72fcce..4f002eb 100644
--- a/flumine/order/process.py
+++ b/flumine/order/process.py
@@ -1,4 +1,5 @@
 import logging
+from typing import Optional
 
 from .. import config
 from ..markets.markets import Markets
@@ -29,7 +30,7 @@ Loop through each current order:
 
 
 def process_current_orders(
-    markets: Markets, strategies: Strategies, event, log_control
+    markets: Markets, strategies: Strategies, event, log_control, add_market
 ) -> None:
     for current_orders in event.event:
         for current_order in current_orders.orders:
@@ -48,7 +49,9 @@ def process_current_orders(
                         "customer_order_ref": current_order.customer_order_ref,
                     },
                 )
-                order = create_order_from_current(markets, strategies, current_order)
+                order = create_order_from_current(
+                    markets, strategies, current_order, add_market
+                )
                 if order:
                     logger.info(
                         "Order %s added to blotter" % current_order.bet_id,
@@ -96,23 +99,11 @@ def process_current_order(order: BaseOrder):
                 order.execution_complete()
 
 
-def create_order_from_current(markets: Markets, strategies: Strategies, current_order):
+def create_order_from_current(
+    markets: Markets, strategies: Strategies, current_order, add_market
+) -> Optional[BaseOrder]:
     strategy_name_hash = current_order.customer_order_ref[:STRATEGY_NAME_HASH_LENGTH]
     order_id = current_order.customer_order_ref[STRATEGY_NAME_HASH_LENGTH + 1 :]
-    # get market
-    market = markets.markets.get(current_order.market_id)
-    if market is None:
-        logger.warning(
-            "Market not available to create order {0}".format(order_id),
-            extra={
-                "bet_id": current_order.bet_id,
-                "market_id": current_order.market_id,
-                "customer_strategy_ref": current_order.customer_strategy_ref,
-                "customer_order_ref": current_order.customer_order_ref,
-                "strategy_name_hash": strategy_name_hash,
-            },
-        )
-        return
     # get strategy
     strategy = strategies.hashes.get(strategy_name_hash)
     if strategy is None:
@@ -127,6 +118,11 @@ def create_order_from_current(markets: Markets, strategies: Strategies, current_
             },
         )
         return
+    # get market
+    market = markets.markets.get(current_order.market_id)
+    if market is None:
+        # create market
+        market = add_market(current_order.market_id, market_book=None)
     # add trade/order
     trade = Trade(
         market.market_id, current_order.selection_id, current_order.handicap, strategy
diff --git a/flumine/streams/orderstream.py b/flumine/streams/orderstream.py
index cc800cd..62aa140 100644
--- a/flumine/streams/orderstream.py
+++ b/flumine/streams/orderstream.py
@@ -32,7 +32,6 @@ class OrderStream(BaseStream):
                 "Starting output_thread (OrderStream {0})".format(self.stream_id)
             )
             self._output_thread.start()
-            time.sleep(START_DELAY)  # allow market streams to start
 
         self._stream = self.betting_client.streaming.create_stream(
             unique_id=self.stream_id, listener=self._listener
diff --git a/flumine/streams/streams.py b/flumine/streams/streams.py
index 3ba5bdb..92a05e0 100644
--- a/flumine/streams/streams.py
+++ b/flumine/streams/streams.py
@@ -1,3 +1,4 @@
+import time
 import logging
 from typing import Union, Iterator
 
@@ -95,7 +96,7 @@ class Streams:
                 streaming_timeout=strategy.streaming_timeout,
                 conflate_ms=strategy.conflate_ms,
             )
-            self._streams.insert(0, stream)
+            self._streams.append(stream)
             return stream
 
     def add_historical_stream(
@@ -187,6 +188,11 @@ class Streams:
             logger.info("Starting streams..")
             for stream in self:
                 stream.start()
+                # wait for successful start
+                while stream.name != "SimulatedOrderStream" and (
+                    not stream._stream or not stream._stream._running
+                ):
+                    time.sleep(0.25)
 
     def stop(self) -> None:
         for stream in self:


Test patch:
diff --git a/tests/test_process.py b/tests/test_process.py
index 4ac5fbf..2a4b702 100644
--- a/tests/test_process.py
+++ b/tests/test_process.py
@@ -30,56 +30,45 @@ class BaseOrderTest(unittest.TestCase):
 
     def test_process_current_orders_with_default_sep(self):
         mock_log_control = mock.Mock()
+        mock_add_market = mock.Mock()
         market_book = mock.Mock()
         markets = Markets()
         market = Market(
             flumine=mock.Mock(), market_id="market_id", market_book=market_book
         )
-
         markets.add_market("market_id", market)
-
         strategies = Strategies()
-
         cheap_hash = create_cheap_hash("strategy_name", 13)
-
         trade = mock.Mock(market_id="market_id")
         trade.strategy.name_hash = cheap_hash
-
         current_order = mock.Mock(
             customer_order_ref=f"{cheap_hash}I123", market_id="market_id", bet_id=None
         )
-
         betfair_order = BetfairOrder(trade=trade, side="BACK", order_type=mock.Mock())
         betfair_order.id = "123"
         market.blotter = {"123": betfair_order}
-
         event = mock.Mock(event=[mock.Mock(orders=[current_order])])
-
         process_current_orders(
             markets=markets,
             strategies=strategies,
             event=event,
             log_control=mock_log_control,
+            add_market=mock_add_market,
         )
-
         self.assertEqual(current_order, betfair_order.responses.current_order)
 
     def test_create_order_from_current(self):
+        mock_add_market = mock.Mock()
         market_book = mock.Mock()
         markets = Markets()
         market = Market(
             flumine=mock.Mock(), market_id="market_id", market_book=market_book
         )
-
         markets.add_market("market_id", market)
-
         cheap_hash = create_cheap_hash("strategy_name", 13)
-
         strategy = mock.Mock(name_hash=cheap_hash)
-
         strategies = Strategies()
         strategies(strategy=strategy, client=mock.Mock())
-
         current_order = mock.Mock(
             customer_order_ref=f"{cheap_hash}I123",
             market_id="market_id",
@@ -90,11 +79,12 @@ class BaseOrderTest(unittest.TestCase):
             price_size=PriceSize(price=10.0, size=2.0),
             persistence_type="LAPSE",
         )
-
         new_order = create_order_from_current(
-            markets=markets, strategies=strategies, current_order=current_order
+            markets=markets,
+            strategies=strategies,
+            current_order=current_order,
+            add_market=mock_add_market,
         )
-
         self.assertEqual(market.blotter["123"], new_order)
         self.assertEqual(new_order.market_id, "market_id")
         self.assertEqual(new_order.selection_id, "selection_id")