Increased priority for order update events so they get dispatched before other events like bar events.

Author: gbeced

CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Bug fixes
 
 * Backtesting stop orders were being canceled if the stop price was not hit.
+* Backtesting order updates were sometimes being processed after bar events. Order update events should get processed first.
 
 ### Misc
 

basana/backtesting/order_mgr.py

@@ -25,7 +25,7 @@
 from basana.backtesting import account_balances, config, errors, fees, helpers, lending, loan_mgr, liquidity, prices
 from basana.backtesting.orders import Order, OrderInfo
 from basana.backtesting.value_map import ValueMap, ValueMapDict
-from basana.core import dispatcher, helpers as core_helpers, logs
+from basana.core import dispatcher, event, helpers as core_helpers, logs
 from basana.core.bar import Bar, BarEvent
 from basana.core.enums import OrderOperation
 from basana.core.pair import Pair
@@ -36,6 +36,7 @@
 
 LiquidityStrategyFactory = Callable[[], liquidity.LiquidityStrategy]
 OrderEventHandler = Callable[["OrderEvent"], Awaitable[Any]]
+ORDER_UPDATE_PRIORITY = event.DEFAULT_EVENT_SOURCE_PRIORITY + 1
 
 
 @dataclasses.dataclass
@@ -81,7 +82,7 @@ def __init__(self, exchange_ctx: ExchangeContext, immediate_order_processing: bo
         )
         self._orders = helpers.ExchangeObjectContainer[Order]()
         self._holds_by_order: Dict[str, ValueMap] = {}
-        self._order_updates = core_helpers.LazyProxy(bs.FifoQueueEventSource)
+        self._order_updates = core_helpers.LazyProxy(lambda: bs.FifoQueueEventSource(priority=ORDER_UPDATE_PRIORITY))
 
     def on_bar_event(self, bar_event: BarEvent):
         liquidity_strategy = self._liquidity_strategies[bar_event.bar.pair]

basana/core/event.py

@@ -22,8 +22,9 @@
 from . import dt
 
 
-MIN_EVENTSOURCE_PRIORITY = -1000
-MAX_EVENTSOURCE_PRIORITY = 1000
+MIN_EVENT_SOURCE_PRIORITY = -1000
+MAX_EVENT_SOURCE_PRIORITY = 1000
+DEFAULT_EVENT_SOURCE_PRIORITY = 0
 
 
 class Producer:
@@ -84,17 +85,19 @@ class EventSource(metaclass=abc.ABCMeta):
     :param producer: An optional producer associated with this event source.
     """
 
-    def __init__(self, producer: Optional[Producer] = None):
+    def __init__(self, producer: Optional[Producer] = None, priority: int = DEFAULT_EVENT_SOURCE_PRIORITY):
+        assert priority >= MIN_EVENT_SOURCE_PRIORITY and priority <= MAX_EVENT_SOURCE_PRIORITY, "Invalid priority"
+
         self.producer = producer
-        self._priority = 0
+        self._priority = priority
 
     @property
     def priority(self):
         return self._priority
 
     @priority.setter
     def priority(self, value: int):
-        assert value >= MIN_EVENTSOURCE_PRIORITY and value <= MAX_EVENTSOURCE_PRIORITY, "Invalid priority"
+        assert value >= MIN_EVENT_SOURCE_PRIORITY and value <= MAX_EVENT_SOURCE_PRIORITY, "Invalid priority"
         self._priority = value
 
     @abc.abstractmethod
@@ -113,8 +116,11 @@ class FifoQueueEventSource(EventSource):
     :param producer: An optional producer associated with this event source.
     :param events: An optional list of initial events.
     """
-    def __init__(self, producer: Optional[Producer] = None, events: List[Event] = []):
-        super().__init__(producer)
+    def __init__(
+            self, producer: Optional[Producer] = None, events: List[Event] = [],
+            priority: int = DEFAULT_EVENT_SOURCE_PRIORITY
+    ):
+        super().__init__(producer, priority)
         self._queue = deque(events)
 
     def push(self, event: Event):

tests/test_backtesting_exchange_orders.py

@@ -47,9 +47,11 @@ def test_create_get_and_cancel_order(backtesting_dispatcher):
     bar_events = []
 
     async def on_order_updated(event: exchange.OrderEvent):
-        order_events.append(event.order)
+        order_events.append(event)
 
     async def on_bar(bar_event):
+        bar_events.append(bar_event)
+
         order_request = requests.MarketOrder(OrderOperation.BUY, pair, Decimal("1"))
         created_order = await e.create_order(order_request)
         assert created_order is not None
@@ -87,11 +89,54 @@ async def on_bar(bar_event):
         assert sum(e._balances.holds.values()) == 0
         assert len(e._order_mgr._holds_by_order) == 0
 
+    async def impl():
+        e.subscribe_to_bar_events(pair, on_bar)
+        e.subscribe_to_order_events(on_order_updated)
+
+        e.add_bar_source(
+            event.FifoQueueEventSource(events=[
+                bar.BarEvent(
+                    dt.local_datetime(2001, 1, 3),
+                    bar.Bar(
+                        dt.local_datetime(2001, 1, 2), pair,
+                        Decimal(5), Decimal(10), Decimal(1), Decimal(5), Decimal(100)
+                    )
+                ),
+            ])
+        )
+
+        await backtesting_dispatcher.run()
+
+        assert len(bar_events) == 1
+        assert len(order_events) == 2
+        assert order_events[0].order.is_open is True
+        assert order_events[1].order.is_open is False
+
+    asyncio.run(impl())
+
+
+def test_order_updates_have_priority(backtesting_dispatcher):
+    e = exchange.Exchange(
+        backtesting_dispatcher,
+        {"USD": Decimal("50000")}
+    )
+    pair = Pair("BTC", "USD")
+    order_events = []
+    bar_events = []
+    all_events = []
+
+    async def on_order_updated(event: exchange.OrderEvent):
+        order_events.append(event)
+        all_events.append(event)
+
+    async def on_bar(bar_event):
         bar_events.append(bar_event)
+        all_events.append(bar_event)
 
     async def impl():
         e.subscribe_to_bar_events(pair, on_bar)
         e.subscribe_to_order_events(on_order_updated)
+
         e.add_bar_source(
             event.FifoQueueEventSource(events=[
                 bar.BarEvent(
@@ -104,12 +149,13 @@ async def impl():
             ])
         )
 
+        await e.create_limit_order(OrderOperation.BUY, pair, Decimal(1), Decimal(10))
         await backtesting_dispatcher.run()
 
         assert len(bar_events) == 1
-        assert len(order_events) == 2
-        assert order_events[0].is_open is True
-        assert order_events[1].is_open is False
+        assert len(order_events) == 1
+        assert order_events[0].order.amount_filled == Decimal(1)
+        assert all_events == order_events + bar_events
 
     asyncio.run(impl())