Repository: liampauling/flumine
Pull request #437 (2021-05-17 14:20:50)

Add order stream
- start delay
- delta snap of 5s
- add market streams to front of list (start first)
should prevent hanging orders / duplicate execution on restart


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index 4dc5d93..0846035 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,6 +3,18 @@
 Release History
 ---------------
 
+1.18.11 (2021-05-20)
++++++++++++++++++++
+
+**Improvements**
+
+- Market recorder refactored to have a single processor thread to remove blocking
+
+**Bug Fixes**
+
+- Add order stream start delay and snap pickup
+- Missing update current order
+
 1.18.10 (2021-05-17)
 +++++++++++++++++++
 
diff --git a/examples/middleware/marketcatalogue.py b/examples/middleware/marketcatalogue.py
index 4e5d5c3..1fbab2a 100644
--- a/examples/middleware/marketcatalogue.py
+++ b/examples/middleware/marketcatalogue.py
@@ -10,9 +10,6 @@ MARKET_CATALOGUE_PATH = ""  # update to correct path
 # Will read and parse the market_catalogue file and add to the market object when backtesting
 # Usage framework.add_middleware(MarketCatalogueMiddleware())
 class MarketCatalogueMiddleware(Middleware):
-    def __call__(self, market) -> None:
-        pass
-
     def add_market(self, market) -> None:
         catalogue_file_path = os.path.join(MARKET_CATALOGUE_PATH, market.market_id)
         if os.path.exists(catalogue_file_path):
diff --git a/examples/middleware/orders.py b/examples/middleware/orders.py
new file mode 100644
index 0000000..f1e82f6
--- /dev/null
+++ b/examples/middleware/orders.py
@@ -0,0 +1,83 @@
+import logging
+from flumine import config
+from flumine.utils import STRATEGY_NAME_HASH_LENGTH
+from flumine.markets.middleware import Middleware
+from flumine.order.trade import Trade
+
+logger = logging.getLogger(__name__)
+
+
+class OrdersMiddleware(Middleware):
+    """
+    Middleware to add execution complete orders
+    to the blotter. This is required on a restart
+    as the order stream does not include
+    EXECUTION_COMPLETE orders
+    """
+
+    def __init__(self, flumine):
+        self.flumine = flumine
+
+    def add_market(self, market) -> None:
+        resp = self.flumine.client.betting_client.betting.list_current_orders(
+            customer_strategy_refs=[config.hostname],
+            order_projection="EXECUTION_COMPLETE",
+        )
+        for current_order in resp.orders:
+            logger.info(
+                "OrdersMiddleware: Processing order {0}".format(current_order.bet_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,
+                },
+            )
+            order = self._create_order_from_current(current_order, market)
+            if order:
+                order.update_current_order(current_order)
+                order.execution_complete()
+
+    def _create_order_from_current(self, current_order, market):
+        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 strategy
+        strategy = self.flumine.strategies.hashes.get(strategy_name_hash)
+        if strategy is None:
+            logger.warning(
+                "OrdersMiddleware: Strategy 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": str(strategy),
+                },
+            )
+            return
+        # add trade/order
+        trade = Trade(
+            market.market_id,
+            current_order.selection_id,
+            current_order.handicap,
+            strategy,
+        )
+        order = trade.create_order_from_current(current_order, order_id)
+        market.blotter[order.id] = order
+        runner_context = strategy.get_runner_context(*order.lookup)
+        runner_context.place(trade.id)
+        logger.info(
+            "OrdersMiddleware: New order trade created",
+            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": str(strategy),
+            },
+        )
+        return order
diff --git a/flumine/__version__.py b/flumine/__version__.py
index e1bb5bb..54f0901 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.10"
+__version__ = "1.18.11"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/markets/middleware.py b/flumine/markets/middleware.py
index 4a14ad1..61eff19 100644
--- a/flumine/markets/middleware.py
+++ b/flumine/markets/middleware.py
@@ -13,7 +13,7 @@ PLACE_MINIMUM_ADJUSTMENT_FACTOR = 0  # todo implement correctly (https://en-betf
 
 class Middleware:
     def __call__(self, market) -> None:
-        raise NotImplementedError
+        pass
 
     def add_market(self, market) -> None:
         pass
diff --git a/flumine/order/process.py b/flumine/order/process.py
index 560face..a72fcce 100644
--- a/flumine/order/process.py
+++ b/flumine/order/process.py
@@ -133,4 +133,6 @@ def create_order_from_current(markets: Markets, strategies: Strategies, current_
     )
     order = trade.create_order_from_current(current_order, order_id)
     market.blotter[order.id] = order
+    runner_context = strategy.get_runner_context(*order.lookup)
+    runner_context.place(trade.id)
     return order
diff --git a/flumine/streams/orderstream.py b/flumine/streams/orderstream.py
index d816c17..cc800cd 100644
--- a/flumine/streams/orderstream.py
+++ b/flumine/streams/orderstream.py
@@ -1,3 +1,4 @@
+import time
 import queue
 import logging
 from betfairlightweight import BetfairError, filters
@@ -10,6 +11,8 @@ from .. import config
 logger = logging.getLogger(__name__)
 
 RETRY_WAIT = BaseStream.RETRY_WAIT
+START_DELAY = 2
+SNAP_DELTA = 5
 
 
 class OrderStream(BaseStream):
@@ -29,6 +32,7 @@ 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
@@ -56,19 +60,27 @@ class OrderStream(BaseStream):
         logger.info("Stopped OrderStream {0}".format(self.stream_id))
 
     def handle_output(self) -> None:
-        """Handles output from stream."""
+        """Handles output from stream, snaps
+        if live orders or time delta greater
+        than SNAP_DELTA.
+        """
+        last_snap = 1
         while self.is_alive():
             try:
                 order_books = self._output_queue.get(
                     block=True, timeout=self.streaming_timeout
                 )
-            except queue.Empty:  # todo snap every 5s or so anyway
-                if self.flumine.markets.live_orders:
+            except queue.Empty:
+                if (
+                    self.flumine.markets.live_orders
+                    or (time.time() - last_snap) > SNAP_DELTA
+                ):
                     order_books = self._listener.snap(
                         market_ids=self.flumine.markets.open_market_ids
                     )
                 else:
                     continue
+            last_snap = time.time()
             if order_books:
                 self.flumine.handler_queue.put(CurrentOrdersEvent(order_books))
 
diff --git a/flumine/streams/streams.py b/flumine/streams/streams.py
index 9cd92e0..3ba5bdb 100644
--- a/flumine/streams/streams.py
+++ b/flumine/streams/streams.py
@@ -95,7 +95,7 @@ class Streams:
                 streaming_timeout=strategy.streaming_timeout,
                 conflate_ms=strategy.conflate_ms,
             )
-            self._streams.append(stream)
+            self._streams.insert(0, stream)
             return stream
 
     def add_historical_stream(


Test patch:
diff --git a/tests/test_middleware.py b/tests/test_middleware.py
index 73806e3..5feebd9 100644
--- a/tests/test_middleware.py
+++ b/tests/test_middleware.py
@@ -17,8 +17,7 @@ class MiddlewareTest(unittest.TestCase):
         self.middleware = Middleware()
 
     def test_call(self):
-        with self.assertRaises(NotImplementedError):
-            self.middleware(None)
+        self.middleware(None)
 
     def test_add_market(self):
         mock_market = mock.Mock()
diff --git a/tests/test_streams.py b/tests/test_streams.py
index b84e3af..9192796 100644
--- a/tests/test_streams.py
+++ b/tests/test_streams.py
@@ -4,6 +4,7 @@ from unittest import mock
 from flumine.streams import streams, datastream, historicalstream
 from flumine.streams.basestream import BaseStream
 from flumine.streams.simulatedorderstream import CurrentOrders
+from flumine.streams import orderstream
 from flumine.exceptions import ListenerError
 
 
@@ -701,6 +702,8 @@ class TestOrderStream(unittest.TestCase):
         self.assertEqual(self.stream.streaming_timeout, 0.01)
         self.assertEqual(self.stream.conflate_ms, 100)
         self.assertIsNone(self.stream._stream)
+        self.assertEqual(orderstream.START_DELAY, 2)
+        self.assertEqual(orderstream.SNAP_DELTA, 5)
 
     # def test_run(self):
     #     pass