Repository: liampauling/flumine
Pull request #493 (2021-09-10 09:49:33)

Task/async cleanup


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index 06b165b..cb5e79c 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,6 +3,21 @@
 Release History
 ---------------
 
+1.19.14 (2021-09-10)
++++++++++++++++++++
+
+**Improvements**
+
+- `process_current_orders` optimisation
+- `market_version` and `elapsed_seconds_created` added to order
+- `OrderStream` logic improvement
+- `market.event` refactored to filter on start time as well as eventId (FORECAST limitation)
+- Example improvement (@petercoles)
+
+**Bug Fixes**
+
+- Correctly complete order in blotter when live
+
 1.19.13 (2021-09-08)
 +++++++++++++++++++
 
diff --git a/examples/marketrecorder.py b/examples/marketrecorder.py
index 5a2d2ea..81747a2 100644
--- a/examples/marketrecorder.py
+++ b/examples/marketrecorder.py
@@ -34,6 +34,7 @@ strategy = MarketRecorder(
         "local_dir": "/tmp",
         "force_update": False,
         "remove_file": True,
+        "remove_gz_file": False,
     },
 )
 
diff --git a/flumine/__version__.py b/flumine/__version__.py
index 566dfb9..2a81cfd 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.13"
+__version__ = "1.19.14"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/baseflumine.py b/flumine/baseflumine.py
index f89b147..ab744ee 100644
--- a/flumine/baseflumine.py
+++ b/flumine/baseflumine.py
@@ -219,13 +219,18 @@ class BaseFlumine:
 
     def _process_current_orders(self, event: events.CurrentOrdersEvent) -> None:
         # update state
-        process_current_orders(
-            self.markets, self.strategies, event, self.log_control, self._add_market
-        )
+        if event.event:
+            process_current_orders(
+                self.markets, self.strategies, event, self.log_control, self._add_market
+            )
         for market in self.markets:
-            if market.closed is False:
+            if market.closed is False and market.blotter.active:
+                blotter = market.blotter
+                for order in blotter.live_orders:
+                    if order.complete:
+                        blotter.complete_order(order)
                 for strategy in self.strategies:
-                    strategy_orders = market.blotter.strategy_orders(strategy)
+                    strategy_orders = blotter.strategy_orders(strategy)
                     utils.call_process_orders_error_handling(
                         strategy, market, strategy_orders
                     )
diff --git a/flumine/execution/transaction.py b/flumine/execution/transaction.py
index 93ced92..0c2398d 100644
--- a/flumine/execution/transaction.py
+++ b/flumine/execution/transaction.py
@@ -54,7 +54,11 @@ class Transaction:
         ):
             return False
         # place
-        order.place(self.market.market_book.publish_time)
+        order.place(
+            self.market.market_book.publish_time,
+            market_version,
+            self._async_place_orders,
+        )
         if self.market.blotter.has_trade(order.trade.id):
             new_trade = False
         else:
diff --git a/flumine/markets/blotter.py b/flumine/markets/blotter.py
index 88850a2..1f49818 100644
--- a/flumine/markets/blotter.py
+++ b/flumine/markets/blotter.py
@@ -33,6 +33,7 @@ class Blotter:
 
     def __init__(self, market_id: str):
         self.market_id = market_id
+        self.active = False
         self._orders = {}  # {Order.id: Order}
         # cached lists/dicts for faster lookup
         self._trades = defaultdict(list)  # {Trade.id: [Order,]}
@@ -184,6 +185,7 @@ class Blotter:
     __contains__ = has_order
 
     def __setitem__(self, customer_order_ref: str, order) -> None:
+        self.active = True
         self._orders[customer_order_ref] = order
         self._live_orders.append(order)
         self._trades[order.trade.id].append(order)
diff --git a/flumine/markets/market.py b/flumine/markets/market.py
index f211e8e..708fb9f 100644
--- a/flumine/markets/market.py
+++ b/flumine/markets/market.py
@@ -103,7 +103,10 @@ class Market:
     def event(self) -> dict:
         event = defaultdict(list)
         for market in self.flumine.markets:
-            if market.event_id == self.event_id:
+            if (
+                self.event_id == market.event_id
+                and self.market_start_datetime == market.market_start_datetime
+            ):
                 event[market.market_type].append(market)
         return event
 
diff --git a/flumine/order/order.py b/flumine/order/order.py
index 6f6b24d..1e9b8ba 100644
--- a/flumine/order/order.py
+++ b/flumine/order/order.py
@@ -75,6 +75,8 @@ class BaseOrder:
         self.simulated = Simulated(self)  # used in simulated execution
         self._simulated = bool(self.simulated)  # cache in current class (2x quicker)
         self.publish_time = None  # marketBook.publish_time
+        self.market_version = None  # marketBook.version
+        self.async_ = None
 
         self.date_time_created = datetime.datetime.utcnow()
         self.date_time_execution_complete = None
@@ -119,7 +121,7 @@ class BaseOrder:
         self.update_data.clear()
 
     # updates
-    def place(self, publish_time: int) -> None:
+    def place(self, publish_time: int, market_version: int, async_: bool) -> None:
         raise NotImplementedError
 
     def cancel(self, size_reduction: float = None) -> None:
@@ -226,6 +228,10 @@ class BaseOrder:
         else:
             return
 
+    @property
+    def elapsed_seconds_created(self) -> float:
+        return (datetime.datetime.utcnow() - self.date_time_created).total_seconds()
+
     @property
     def elapsed_seconds_executable(self) -> Optional[float]:
         if self.date_time_execution_complete and self.responses.date_time_placed:
@@ -260,6 +266,8 @@ class BaseOrder:
             "bet_id": self.bet_id,
             "date_time_created": str(self.date_time_created),
             "publish_time": str(self.publish_time) if self.publish_time else None,
+            "market_version": self.market_version,
+            "async": self.async_,
             "trade": self.trade.info,
             "order_type": self.order_type.info,
             "info": {
@@ -300,8 +308,10 @@ class BetfairOrder(BaseOrder):
     EXCHANGE = ExchangeType.BETFAIR
 
     # updates
-    def place(self, publish_time: int) -> None:
+    def place(self, publish_time: int, market_version: int, async_: bool) -> None:
         self.publish_time = publish_time
+        self.market_version = market_version
+        self.async_ = async_
         self.placing()
 
     def cancel(self, size_reduction: float = None) -> None:
diff --git a/flumine/order/process.py b/flumine/order/process.py
index ce6bf5a..4fab944 100644
--- a/flumine/order/process.py
+++ b/flumine/order/process.py
@@ -74,7 +74,7 @@ def process_current_order(order: BaseOrder, current_order, log_control) -> None:
     # update
     order.update_current_order(current_order)
     # pickup async orders
-    if config.async_place_orders and order.bet_id is None and current_order.bet_id:
+    if order.async_ and order.bet_id is None and current_order.bet_id:
         order.bet_id = current_order.bet_id
         log_control(OrderEvent(order))
     # update status
diff --git a/flumine/order/responses.py b/flumine/order/responses.py
index 1717c0a..e5d87e3 100644
--- a/flumine/order/responses.py
+++ b/flumine/order/responses.py
@@ -1,4 +1,5 @@
 import datetime
+from typing import Optional
 
 
 class Responses:
@@ -11,17 +12,24 @@ class Responses:
         self.cancel_responses = []
         self.replace_responses = []
         self.update_responses = []
-        self.date_time_placed = None
+        self._date_time_placed = None
 
-    def placed(self, response):
+    def placed(self, response) -> None:
         self.place_response = response
-        self.date_time_placed = datetime.datetime.utcnow()
+        self._date_time_placed = datetime.datetime.utcnow()
 
-    def cancelled(self, response):
+    def cancelled(self, response) -> None:
         self.cancel_responses.append(response)
 
-    def replaced(self, response):
+    def replaced(self, response) -> None:
         self.replace_responses.append(response)
 
-    def updated(self, response):
+    def updated(self, response) -> None:
         self.update_responses.append(response)
+
+    @property
+    def date_time_placed(self) -> Optional[datetime.datetime]:
+        if self._date_time_placed:
+            return self._date_time_placed
+        elif self.current_order:
+            return self.current_order.placed_date
diff --git a/flumine/streams/orderstream.py b/flumine/streams/orderstream.py
index 4cc3537..74a184f 100644
--- a/flumine/streams/orderstream.py
+++ b/flumine/streams/orderstream.py
@@ -80,7 +80,7 @@ class OrderStream(BaseStream):
                 else:
                     continue
             last_snap = time.time()
-            if order_books:
+            if order_books or self.flumine.markets.live_orders:
                 self.flumine.handler_queue.put(CurrentOrdersEvent(order_books))
 
         logger.info("Stopped output_thread (OrderStream {0})".format(self.stream_id))


Test patch:
diff --git a/tests/test_blotter.py b/tests/test_blotter.py
index b6d9b86..1a4dc15 100644
--- a/tests/test_blotter.py
+++ b/tests/test_blotter.py
@@ -12,6 +12,7 @@ class BlotterTest(unittest.TestCase):
 
     def test_init(self):
         self.assertEqual(self.blotter.market_id, "1.23")
+        self.assertFalse(self.blotter.active)
         self.assertEqual(self.blotter._orders, {})
         self.assertEqual(self.blotter._live_orders, [])
         self.assertEqual(self.blotter._trades, {})
@@ -454,6 +455,7 @@ class BlotterTest(unittest.TestCase):
     def test__setitem(self):
         mock_order = mock.Mock(lookup=(1, 2, 3))
         self.blotter["123"] = mock_order
+        self.assertTrue(self.blotter.active)
         self.assertEqual(self.blotter._orders, {"123": mock_order})
         self.assertEqual(self.blotter._live_orders, [mock_order])
         self.assertEqual(self.blotter._trades, {mock_order.trade.id: [mock_order]})
diff --git a/tests/test_markets.py b/tests/test_markets.py
index 6190b19..bb3ce55 100644
--- a/tests/test_markets.py
+++ b/tests/test_markets.py
@@ -178,18 +178,19 @@ class MarketTest(unittest.TestCase):
         mock_transaction.replace_order.assert_called_with(mock_order, 2, False, True)
 
     def test_event(self):
-        mock_market_catalogue = mock.Mock()
+        mock_market_catalogue = mock.Mock(market_start_time=12)
         mock_market_catalogue.event.id = 12
         self.market.market_catalogue = mock_market_catalogue
 
         self.market.flumine.markets = []
         self.assertEqual(self.market.event, {})
 
-        m_one = mock.Mock(market_type=1, event_id=12)
-        m_two = mock.Mock(market_type=2, event_id=12)
-        m_three = mock.Mock(market_type=3, event_id=123)
-        m_four = mock.Mock(market_type=1, event_id=12)
-        self.market.flumine.markets = [m_one, m_two, m_three, m_four]
+        m_one = mock.Mock(market_type=1, event_id=12, market_start_datetime=12)
+        m_two = mock.Mock(market_type=2, event_id=12, market_start_datetime=12)
+        m_three = mock.Mock(market_type=3, event_id=123, market_start_datetime=12)
+        m_four = mock.Mock(market_type=1, event_id=12, market_start_datetime=12)
+        m_five = mock.Mock(market_type=2, event_id=12, market_start_datetime=13)
+        self.market.flumine.markets = [m_one, m_two, m_three, m_four, m_five]
         self.assertEqual(self.market.event, {1: [m_one, m_four], 2: [m_two]})
 
     def test_event_type_id_mc(self):
diff --git a/tests/test_order.py b/tests/test_order.py
index 48eddb0..de1fd4b 100644
--- a/tests/test_order.py
+++ b/tests/test_order.py
@@ -47,6 +47,8 @@ class BaseOrderTest(unittest.TestCase):
         self.assertIsNone(self.order.EXCHANGE)
         self.assertEqual(self.order.update_data, {})
         self.assertIsNone(self.order.publish_time)
+        self.assertIsNone(self.order.market_version)
+        self.assertIsNone(self.order.async_)
         self.assertIsNotNone(self.order.date_time_created)
         self.assertIsNone(self.order.date_time_execution_complete)
         self.assertFalse(self.order.simulated)
@@ -105,7 +107,7 @@ class BaseOrderTest(unittest.TestCase):
 
     def test_place(self):
         with self.assertRaises(NotImplementedError):
-            self.order.place(123)
+            self.order.place(123, 456, False)
 
     def test_cancel(self):
         with self.assertRaises(NotImplementedError):
@@ -207,6 +209,9 @@ class BaseOrderTest(unittest.TestCase):
         self.order.responses = mock_responses
         self.assertGreaterEqual(self.order.elapsed_seconds, 0)
 
+    def elapsed_seconds_created(self):
+        self.assertGreaterEqual(self.order.elapsed_seconds_created, 0)
+
     def test_elapsed_seconds_executable(self):
         self.assertIsNone(self.order.elapsed_seconds_executable)
         mock_responses = mock.Mock()
@@ -267,9 +272,11 @@ class BetfairOrderTest(unittest.TestCase):
 
     @mock.patch("flumine.order.order.BetfairOrder.placing")
     def test_place(self, mock_placing):
-        self.order.place(123)
+        self.order.place(123, 456, False)
         mock_placing.assert_called_with()
         self.assertEqual(self.order.publish_time, 123)
+        self.assertEqual(self.order.market_version, 456)
+        self.assertFalse(self.order.async_)
 
     @mock.patch(
         "flumine.order.order.BetfairOrder.size_remaining",
@@ -504,6 +511,8 @@ class BetfairOrderTest(unittest.TestCase):
                 "market_id": self.mock_trade.market_id,
                 "selection_id": self.mock_trade.selection_id,
                 "publish_time": None,
+                "market_version": None,
+                "async": None,
                 "status": None,
                 "status_log": "Pending, Execution complete",
                 "trade": self.mock_trade.info,
diff --git a/tests/test_responses.py b/tests/test_responses.py
index df5355e..b6c97ff 100644
--- a/tests/test_responses.py
+++ b/tests/test_responses.py
@@ -1,4 +1,5 @@
 import unittest
+from unittest import mock
 
 from flumine.order.responses import Responses
 
@@ -13,12 +14,12 @@ class ResponsesTest(unittest.TestCase):
         self.assertEqual(self.responses.cancel_responses, [])
         self.assertEqual(self.responses.replace_responses, [])
         self.assertEqual(self.responses.cancel_responses, [])
-        self.assertIsNone(self.responses.date_time_placed)
+        self.assertIsNone(self.responses._date_time_placed)
         self.assertIsNone(self.responses.current_order)
 
     def test_placed(self):
         self.responses.placed(12)
-        self.assertIsNotNone(self.responses.date_time_placed)
+        self.assertIsNotNone(self.responses._date_time_placed)
         self.assertEqual(self.responses.place_response, 12)
 
     def test_cancelled(self):
@@ -33,3 +34,11 @@ class ResponsesTest(unittest.TestCase):
     def test_updated(self):
         self.responses.updated(12)
         self.assertEqual(self.responses.update_responses, [12])
+
+    def test_date_time_placed(self):
+        self.assertIsNone(self.responses.date_time_placed)
+        self.responses._date_time_placed = 1
+        self.responses.current_order = mock.Mock(placed_date=2)
+        self.assertEqual(self.responses.date_time_placed, 1)
+        self.responses._date_time_placed = None
+        self.assertEqual(self.responses.date_time_placed, 2)
diff --git a/tests/test_transaction.py b/tests/test_transaction.py
index 1ba3e5c..7ec7c98 100644
--- a/tests/test_transaction.py
+++ b/tests/test_transaction.py
@@ -37,7 +37,7 @@ class TransactionTest(unittest.TestCase):
         mock_order.trade.market_notes = None
         self.assertTrue(self.transaction.place_order(mock_order))
         mock_order.place.assert_called_with(
-            self.transaction.market.market_book.publish_time
+            self.transaction.market.market_book.publish_time, None, False
         )
         self.transaction.market.flumine.log_control.assert_called_with(
             mock_events.TradeEvent()
@@ -70,7 +70,7 @@ class TransactionTest(unittest.TestCase):
         mock_order = mock.Mock(id="123")
         self.assertTrue(self.transaction.place_order(mock_order, execute=False))
         mock_order.place.assert_called_with(
-            self.transaction.market.market_book.publish_time
+            self.transaction.market.market_book.publish_time, None, False
         )
         self.transaction.market.flumine.log_control.assert_called_with(
             mock_events.TradeEvent()