Coverage for adrs/oms/ops/order_placement_manager.py: 65%
353 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-29 05:07 +0800
1import asyncio
2import logging
4from dataclasses import dataclass
5from decimal import Decimal
6from datetime import datetime, timezone, timedelta
7from typing import Literal, Dict, List
10from adrs.oms.config import ConfigManager
11from adrs.oms.ops.order_executer import (
12 CancelFatal,
13 CancelMultiResult,
14 OrderExecutor,
15 MAX_CONCURRENT_ORDER_OPS,
16)
17from adrs.oms.ops.order_utils import OrderUtils
18from adrs.oms.position import PositionManager
19from adrs.oms.price_feed import PriceFeed
20from adrs.oms.ops.order_pool import (
21 BacklogDetails,
22 CancelBacklogs,
23 ExpiredBacklogs,
24 OrderBacklogs,
25 OrderDetails,
26 OrderPoolHandler,
27)
29from cybotrade import Symbol
30from cybotrade.io import Event, EventType
31from cybotrade.models import OrderSide, OrderUpdate, OrderStatus
33from adrs.oms.rate_limit.rate_limiter import RateLimiter
34from adrs.oms.rate_limit.exchange_limit_profiles import Endpoints
35from adrs.oms.rate_limit.error_policy import ExchangeErrorPolicy
37logger = logging.getLogger(__name__)
39# on_exchange_event below matches on EventType.PositionUpdate, which does not
40# exist in cybotrade 2.1.0. `match` value patterns are attribute lookups
41# evaluated in arm order, so on 2.1.0 the PositionUpdate arm itself raises
42# AttributeError *while matching* -- not while handling -- and that exception
43# propagates out of on_exchange_event into cybotrade's _consume_stream, which
44# catches it, logs one generic warning, and moves on. Every ACCOUNT_UPDATE is
45# then silently dropped: no exception surfaces anywhere near the OMS, no
46# metric moves, and with the incremental fill writer removed there is nothing
47# left to notice on its own. Fail loudly here instead, at import time, so a
48# version mismatch is a startup crash with a clear cause rather than a single
49# swallowed log line followed by silent over-placement.
50if not hasattr(EventType, "PositionUpdate"):
51 raise RuntimeError(
52 "cybotrade>=2.2.0 is required: EventType.PositionUpdate is missing. "
53 "This OMS drives PositionManager.exchange from the ACCOUNT_UPDATE "
54 "stream (see PositionManager.apply_stream_positions) and no longer "
55 "reconciles positions incrementally from fills, so a build without "
56 "this event silently stops updating positions from the stream."
57 )
59# Caps the per-tick burst below the ~8/s order/cancel pool limit
60MAX_CONCURRENT_BACKLOG_RETRIES = 3
62# Consecutive on_order_placement ticks a symbol can fail to cancel its
63# opposite-side orders before the circuit breaker logs loudly. Every such
64# tick already skips that symbol's replacement (see on_order_placement); this
65# only escalates a persistent problem (e.g. sustained rate limiting) from a
66# per-tick warning to a standing alert, instead of it silently piling into
67# the retry backlog tick after tick.
68CANCEL_FAILURE_CIRCUIT_BREAKER_THRESHOLD = 3
71class OrderPlacementManager:
72 def __init__(
73 self,
74 position: PositionManager,
75 config: ConfigManager,
76 rate_limiter: RateLimiter,
77 error_policy: ExchangeErrorPolicy,
78 executor_cls: type[OrderExecutor] = OrderExecutor,
79 price_feed: PriceFeed | None = None,
80 ) -> None:
81 self.position = position
82 self.config_manager = config
83 self.price_feed = price_feed
84 self.order_pools = OrderPoolHandler(
85 exchange=config.exchange,
86 config=self.config_manager,
87 rate_limiter=rate_limiter,
88 )
89 # Forwarded only when a feed exists: executor_cls is a documented subclass
90 # extension point (oms.py), and an override that predates this parameter must
91 # keep working unchanged when no feed is configured.
92 executor_kwargs = {} if price_feed is None else {"price_feed": price_feed}
93 self.executor = executor_cls(
94 config_manager=self.config_manager,
95 order_pools=self.order_pools,
96 rate_limiter=rate_limiter,
97 error_policy=error_policy,
98 **executor_kwargs,
99 )
100 self.rate_limiter = rate_limiter
101 # Consecutive on_order_placement ticks in a row where this symbol's
102 # cancel_multi_limit_order had at least one failure. Reset on any
103 # tick where every chosen cancel succeeds.
104 self._consecutive_cancel_failures: Dict[Symbol, int] = {}
106 def _note_cancel_outcome(self, symbol: Symbol, cancel_result: CancelMultiResult):
107 if cancel_result.failed_count == 0:
108 self._consecutive_cancel_failures.pop(symbol, None)
109 return
110 streak = self._consecutive_cancel_failures.get(symbol, 0) + 1
111 self._consecutive_cancel_failures[symbol] = streak
112 if streak >= CANCEL_FAILURE_CIRCUIT_BREAKER_THRESHOLD:
113 logger.error(
114 f"[CIRCUIT_BREAKER] {symbol} has failed to cancel opposite-side "
115 f"orders for {streak} consecutive ticks ({cancel_result.failed_count} "
116 f"failure(s) this tick) - placement has been skipped every one of "
117 f"those ticks, but this many in a row means the problem is not "
118 f"self-resolving. Check the rate limiter / exchange connectivity."
119 )
121 async def on_exchange_event(self, event: Event):
122 """
123 Handles various events sent by exchange
124 """
125 match event.event_type:
126 case EventType.Authenticated | EventType.Subscribed:
127 logger.info(f"[ON_EVENT] '{event.event_type}': {event.data}")
128 # cybotrade emits Authenticated on every connect, including
129 # reconnects -- read it as "feed restarted". Everything that
130 # happened during the gap (ORDER_TRADE_UPDATE as well as
131 # ACCOUNT_UPDATE) is lost with nothing else to notice, so the
132 # REST anchor must not still look fresh: force the next
133 # delta_calculation to read.
134 self.position.invalidate_exchange_anchor()
135 case EventType.OrderUpdate:
136 logger.debug(f"[ON_EVENT] '{event.event_type}': {event.data}")
137 await self.on_order_update(event.data)
138 case EventType.Error:
139 logger.error(f"[ON_EVENT] '{event.event_type}': {event.data}")
140 case EventType.PositionUpdate:
141 logger.debug(f"[ON_EVENT] '{event.event_type}': {event.data}")
142 self.position.apply_stream_positions(event.data)
143 case _:
144 logger.warning(f"[ON_EVENT] '{event.event_type}': {event.data}")
146 async def validate_oms_state(self, update: OrderUpdate) -> bool:
147 """
148 Return True when the update can be applied to local state.
150 An unknown id only warrants a full open-orders resync when the order
151 is still live on the exchange; in-flight placements and terminal
152 events are handled with at most a position refresh.
153 """
154 async with self.order_pools.get_order_pool() as order_pool:
155 if update.client_order_id in order_pool:
156 return True
158 is_terminal = update.status in (
159 OrderStatus.FILLED,
160 OrderStatus.PARTIALLY_FILLED_CANCELLED,
161 )
163 if self.order_pools.is_pending_insert(update.client_order_id):
164 if is_terminal:
165 await self.position.update_exchange()
166 logger.info(
167 f"[VALIDATE] {update.client_order_id} in flight, WS beat the REST response"
168 )
169 return False
171 if is_terminal:
172 await self.position.update_exchange()
173 logger.info(
174 f"[VALIDATE] Terminal update for unknown order {update.client_order_id}, refreshed positions"
175 )
176 return False
178 logger.warning(
179 f"[VALIDATE] Live order {update.client_order_id} missing from local pool, resyncing"
180 )
181 snapshot = await self.order_pools.fetch_open_orders_snapshot()
182 await self.position.update_exchange()
183 self.position.update_pending(snapshot)
184 await self.order_pools.resync_order_pool(snapshot)
185 return False
187 def update_positions(self, update: OrderUpdate):
188 # filled_size is cumulative. Apply only the new slice and store the
189 # cumulative total (not the slice) as the next baseline, otherwise the
190 # baseline drifts and fills over-count from the third update onward.
191 #
192 # Only `pending` is maintained here now. The exchange position is owned
193 # absolutely by ACCOUNT_UPDATE (see PositionManager.apply_stream_positions):
194 # this used to also do `exchange += asset_filled`, which drifted for the
195 # same reason the baseline does and relied on REST polling to correct it.
196 # `pending` keeps its increment because the open-orders snapshot corrects
197 # it every placement tick.
198 last_filled = self.order_pools.order_value_update.get(
199 update.client_order_id, Decimal("0")
200 )
201 increment = update.filled_size - last_filled
202 self.order_pools.order_value_update[update.client_order_id] = update.filled_size
203 asset_filled = increment if update.side == OrderSide.BUY else -increment
204 self.position.pending[update.symbol].quantity -= asset_filled
206 async def on_order_update(self, update: OrderUpdate):
207 """
208 Handles Event OrderUpdate where position update and calculation would be done
209 """
210 symbol = str(update.symbol)
212 if update.status == OrderStatus.CREATED:
213 logger.info(
214 f"[ORDER_UPDATE|CREATED] Limit {symbol} order: {update.client_order_id} is {update.status}"
215 )
216 # Update local store in case rest api hits rate limits
217 self.position.pending[update.symbol].quantity += (
218 update.size if update.side == OrderSide.BUY else -update.size
219 )
220 return
222 if update.status == OrderStatus.CANCELLED:
223 # Pop covers externally cancelled orders (manual UI, exchange purge)
224 # which would otherwise stay in the pool and fail cancel forever;
225 # already removed by our own cancel path is fine, pop is idempotent
226 async with self.order_pools.get_order_pool() as order_pool:
227 order_pool.pop(update.client_order_id, None)
228 # Remove unfulffilled amount from pending for to remake order in next delta calculation
229 self.position.pending[update.symbol].quantity -= (
230 update.remain_size
231 if update.side == OrderSide.BUY
232 else -update.remain_size
233 )
234 logger.info(
235 f"[ORDER_UPDATE|CANCELLED] Removed {update.client_order_id} from order_pool due to order {update.status}"
236 )
237 return
239 # Majority of cases if when price moved during order placement invalidating post-only orders
240 # Hence retry with bbo
241 if update.status == OrderStatus.REJECTED:
242 logger.info(
243 f"[ORDER_UPDATE|REJECTED] {update.client_order_id} has been rejected due to order {update.status}"
244 )
245 # TEST WILL NEED TO WATCH OUT THE FLOW WHEN THIS ACTUALLY TRIGGERS
246 async with self.order_pools.get_order_pool() as order_pool:
247 # pop: WS REJECTED can beat the REST response that inserts the
248 # pool entry, del would raise and kill the event handler
249 rejected_order = order_pool.pop(update.client_order_id, None)
250 if rejected_order is not None:
251 # Mirror CANCELLED accounting: CREATED added this qty to pending
252 # and the reprice below adds its own on its CREATED, without
253 # this the rejected order's qty leaks in pending until resync
254 self.position.pending[update.symbol].quantity -= (
255 update.remain_size
256 if update.side == OrderSide.BUY
257 else -update.remain_size
258 )
259 result = await self.executor.reprice_at_bbo(
260 symbol=update.symbol,
261 qty=update.remain_size,
262 side=update.side,
263 replace_best_bid_ask_time=datetime.now(timezone.utc)
264 + timedelta(
265 seconds=self.config_manager.config.replace_best_bid_ask_time
266 ),
267 package_id=self.executor.package_id,
268 )
269 if result:
270 async with self.order_pools.get_order_backlog() as order_backlog:
271 self.order_pools.dedup_append(order_backlog, result)
272 return
274 # Create and cancel don't need as order might not be in order pool yet
275 if not await self.validate_oms_state(update=update):
276 # validate logged the specific cause and refreshed what it needed
277 return
279 is_terminal = (
280 update.status == OrderStatus.FILLED
281 or update.status == OrderStatus.PARTIALLY_FILLED_CANCELLED
282 )
284 # Update Order Details. validate_oms_state released the pool lock, so the
285 # entry may have been popped concurrently (e.g. a FILLED beat us) — bail
286 # if it's gone. All ops below are synchronous, so the lock stays a leaf.
287 async with self.order_pools.get_order_pool() as order_pool:
288 entry = order_pool.get(update.client_order_id)
289 if entry is None:
290 return
291 entry.remain_size = update.remain_size
292 if is_terminal:
293 order_pool.pop(update.client_order_id, None)
294 pool_size = len(order_pool)
296 if update.status == OrderStatus.PARTIALLY_FILLED:
297 logger.info(
298 f"[ORDER_UPDATE|PARTIALLY_FILLED] {update.client_order_id} has filled {update.filled_size}, remain {update.remain_size}"
299 )
300 self.update_positions(update=update)
301 return
303 if is_terminal:
304 self.update_positions(update=update)
305 if update.status == OrderStatus.PARTIALLY_FILLED_CANCELLED:
306 # Unfilled remainder was added to pending at CREATED but the
307 # cancel leaves it working forever; mirror the CANCELLED branch
308 self.position.pending[update.symbol].quantity -= (
309 update.remain_size
310 if update.side == OrderSide.BUY
311 else -update.remain_size
312 )
313 self.order_pools.order_value_update.pop(update.client_order_id, None)
314 logger.info(
315 f"[ORDER_UPDATE|FILLED] Removed {update.client_order_id} from order_pool due to order {update.status}, current order_pool size: {pool_size}"
316 )
317 return
319 def on_position_check(self):
320 desired = self.position.desired
321 pending = self.position.pending
322 exchange = self.position.exchange
323 logger.info("[OMS STATUS CHECK]")
324 for position in (
325 (desired, "DESIRED"),
326 (pending, "PENDING"),
327 (exchange, "EXCHANGE"),
328 ):
329 logger.info(
330 f"[{position[1]} POSITION] {' '.join([(parts[0] if (parts := symbol.split()) else 'Unknown') + ': ' + str(position.quantity) for symbol, position in position[0].items()])}"
331 )
332 logger.info(f"[RATE LIMITER STATS] {self.rate_limiter}")
334 async def on_order_placement(self):
335 """
336 Periodic rebalance execution logic.
338 Rebalance logic:
339 - delta > 0 => buy, delta < 0 => sell
340 - pending tracks already-submitted orders (same sign = same direction)
342 Rules:
343 - If delta and pending have opposite signs → cancel pending
344 - Else (same direction) → place extra orders
346 TLDR: Same direction → add orders, Opposite → cancel
347 """
348 self.executor.update_package_id()
349 self.on_position_check()
350 snapshot = await self.order_pools.fetch_open_orders_snapshot()
351 deltas = await self.position.delta_calculation(snapshot)
352 queued: dict[Symbol, Decimal] = {}
353 async with self.order_pools.get_order_backlog() as order_backlog:
354 for backlog in order_backlog:
355 if not isinstance(backlog, OrderBacklogs):
356 continue
357 symbol = Symbol(backlog.symbol)
358 signed = backlog.qty if backlog.side == OrderSide.BUY else -backlog.qty
359 queued[symbol] = queued.get(symbol, Decimal("0")) + signed
361 for symbol, queued_qty in queued.items():
362 if symbol not in deltas:
363 continue
364 adjusted = deltas[symbol] - queued_qty
365 if deltas[symbol] >= 0:
366 deltas[symbol] = max(Decimal("0"), adjusted)
367 else:
368 deltas[symbol] = min(Decimal("0"), adjusted)
370 if all(delta == Decimal("0") for delta in deltas.values()):
371 logger.info("Delta between desired and pending/exchange is currently zero")
372 return
374 logger.info(f"[ON_ORDER_PLACEMENT] Deltas {deltas}")
376 try:
377 await self.config_manager.update_symbol_info(self.rate_limiter)
378 except Exception as e:
379 logger.warning(f"update symbol info failed due to {e}")
381 for symbol, delta in deltas.items():
382 if delta == 0:
383 continue
385 # To determine whether the delta is big enough to proceed
386 residual_threshold = self.config_manager.symbol_infos[symbol].min_limit_qty
387 if abs(delta) < residual_threshold:
388 logger.info(
389 f"{symbol}'s delta {delta} is lower than provided threshold {residual_threshold}, it is ignored"
390 )
391 continue
393 # To double check just in case exchange didn't properly update min limit
394 notional_threshold = self.config_manager.symbol_infos[symbol].min_notional
395 current_price = await self.executor.get_current_price(symbol=symbol)
396 if current_price is None:
397 logger.warning(
398 f"{symbol}'s current price couldn't be fetched, skipping this tick"
399 )
400 continue
401 notional_value = abs(delta) * current_price
402 if notional_value < notional_threshold:
403 logger.info(
404 f"{symbol}'s delta {delta} is lower than provided notional threshold {notional_threshold}, it is ignored"
405 )
406 continue
408 pending = self.position.pending[symbol].quantity
409 direction = pending * delta if pending != 0 else delta
411 # if there is orders on the opposite side cancel first
412 if direction < 0 and pending != 0:
413 open_orders = snapshot.orders.get(symbol)
414 if open_orders is None:
415 # Fetch failed for this symbol; delta retries next tick
416 logger.warning(
417 f"{symbol} missing from open-orders snapshot, skipping reversal this tick"
418 )
419 continue
420 cancel_result = await self.executor.cancel_multi_limit_order(
421 symbol=symbol, target=delta, open_orders=open_orders
422 )
423 self._note_cancel_outcome(symbol, cancel_result)
424 if cancel_result.failed_count:
425 # Some opposite-side orders are still live/unconfirmed;
426 # placing a same-tick replacement now would size it as if
427 # they were gone. Skip - the failed cancels are already
428 # queued for retry, and the next tick's fresh delta
429 # (informed by whatever actually happens to them) decides.
430 logger.warning(
431 f"[ON_ORDER_PLACEMENT] {cancel_result.failed_count} cancel(s) "
432 f"failed for {symbol}; skipping this tick's replacement order"
433 )
434 continue
435 delta = cancel_result.remainder
437 # place order based on delta or remander
438 await self.executor.place_multiple_limit_order(
439 symbol=symbol,
440 quantity=delta,
441 )
442 self.on_position_check()
444 async def on_retry_backlog(self):
445 """
446 To reattempt all failed orders/cancels
447 """
448 if self.rate_limiter.retry_after >= self.rate_limiter.get_synced_time_ms():
449 logger.info("[ON_RETRY_BACKLOG] Skipping tick, rate limiter cooling down")
450 return
452 now = datetime.now(timezone.utc)
454 # Snapshot due items under the lock; run the I/O with it released
455 async with self.order_pools.get_order_backlog() as order_backlog:
456 if len(order_backlog) == 0:
457 return
458 logger.info(f"[ON_RETRY_BACKLOG] Current backlog size {len(order_backlog)}")
459 due: list[BacklogDetails] = []
460 for backlog in list(order_backlog):
461 if backlog.next_retry_at is not None and backlog.next_retry_at > now:
462 continue
463 if (
464 backlog.total_retries
465 >= self.config_manager.config.max_retries_allowed
466 ):
467 order_backlog.remove(backlog)
468 # Giving up on a CANCEL is not the same as giving up on a
469 # placement. If the order is still in the pool it may still be
470 # resting on the exchange, against the strategy's intent, and
471 # nothing will try to remove it again — so it is an error, not
472 # the routine warning an abandoned placement gets. Rate-limit
473 # refusals burn retries without a request ever reaching the
474 # exchange, so this path is reached fastest exactly when the
475 # cancel pool is saturated. _retry_one already treats a
476 # departed order as done, so `still_ours` is what separates
477 # "nothing left to cancel" from "we stopped trying".
478 still_ours = (
479 isinstance(backlog, CancelBacklogs | ExpiredBacklogs)
480 and backlog.client_order_id in self.order_pools.order_pool
481 )
482 if still_ours:
483 logger.error(
484 f"[ON_RETRY_BACKLOG] Gave up cancelling "
485 f"{backlog.client_order_id} on {backlog.symbol} after "
486 f"{backlog.total_retries} attempts; it is still in the "
487 f"order pool and may be resting on the exchange. "
488 f"Backlog size {len(self.order_pools.order_backlog)}"
489 )
490 else:
491 logger.warning(
492 f"[ON_RETRY_BACKLOG] Removed backlog {backlog} exceeded max retries, Current backlog size {len(self.order_pools.order_backlog)}"
493 )
494 continue
495 due.append(backlog)
497 if not due:
498 return
500 sem = asyncio.Semaphore(MAX_CONCURRENT_BACKLOG_RETRIES)
502 async def _bounded(backlog: BacklogDetails):
503 async with sem:
504 return await self._retry_one(backlog, now)
506 outcomes = await asyncio.gather(
507 *[_bounded(backlog) for backlog in due],
508 return_exceptions=True,
509 )
511 # Re-acquire only to apply list mutations
512 async with self.order_pools.get_order_backlog() as order_backlog:
513 for backlog, outcome in zip(due, outcomes):
514 if isinstance(outcome, BaseException):
515 logger.error(
516 f"[ON_RETRY_BACKLOG] Retry crashed for {backlog} due to {outcome}"
517 )
518 backlog.total_retries += 1
519 backlog.next_retry_at = now + timedelta(
520 seconds=min(2 * backlog.total_retries, 30)
521 )
522 continue
524 should_remove, replacement = outcome
525 if should_remove and backlog in order_backlog:
526 order_backlog.remove(backlog)
527 if replacement is not None:
528 self.order_pools.dedup_append(order_backlog, replacement)
530 async def _retry_one(self, backlog: BacklogDetails, now: datetime):
531 """Retry one item. Returns (should_remove, replacement_order)."""
532 if isinstance(backlog, OrderBacklogs):
533 symbol = Symbol(backlog.symbol)
534 result = await self.executor.place_single_limit_order(
535 symbol=symbol,
536 offset=backlog.offset,
537 qty=backlog.qty,
538 side=backlog.side,
539 symbol_info=self.config_manager.symbol_infos[symbol],
540 package_id=backlog.package_id,
541 initial_price=backlog.initial_price,
542 initial_time=backlog.initial_time,
543 client_order_id=backlog.client_order_id,
544 )
545 elif isinstance(backlog, CancelBacklogs | ExpiredBacklogs):
546 async with self.order_pools.get_order_pool() as order_pool:
547 still_ours = backlog.client_order_id in order_pool
548 if not still_ours:
549 logger.info(
550 f"[ON_RETRY_BACKLOG] {backlog.client_order_id} already left the pool, dropping backlog without replacement"
551 )
552 return (True, None)
553 result = await self.executor.cancel_single_order(
554 symbol=Symbol(backlog.symbol),
555 client_order_id=backlog.client_order_id,
556 )
557 else:
558 raise Exception("Unknown Backlog Type")
560 if isinstance(result, CancelFatal):
561 return (True, None)
563 # Failed: strike + backoff, keep
564 if result is not None:
565 backlog.total_retries += 1
566 delay = min(2 * backlog.total_retries, 30)
567 backlog.next_retry_at = now + timedelta(seconds=delay)
568 return (False, None)
570 # Success: remove; expired cancel also re-places
571 if isinstance(backlog, ExpiredBacklogs):
572 return (True, await self._reprice_expired(backlog))
573 return (True, None)
575 async def _reprice_expired(self, backlog: ExpiredBacklogs):
576 """Reconfirm remaining qty after cancel, then re-place. Returns replacement or None."""
577 qty = backlog.qty
578 details = None
579 try:
580 async with self.rate_limiter.guard(endpoint=Endpoints.GET_ORDER_DETAILS):
581 details = (
582 await self.config_manager.exchange.get_order_details_from_history(
583 symbol=Symbol(backlog.symbol),
584 client_order_id=backlog.client_order_id,
585 )
586 )
587 except Exception as e:
588 logger.warning(
589 f"[ON_RETRY_BACKLOG] Could not reconfirm remaining qty for {backlog.client_order_id} due to {e}, leaving requantification to the delta path"
590 )
591 return None
592 if details is not None:
593 if details.status in (
594 OrderStatus.CREATED,
595 OrderStatus.PARTIALLY_FILLED,
596 ):
597 logger.warning(
598 f"[ON_RETRY_BACKLOG] {backlog.client_order_id} still shows open in history, skipping replacement"
599 )
600 return None
601 qty = details.remain_size
602 if qty <= Decimal("0"):
603 logger.info(
604 f"[ON_RETRY_BACKLOG] {backlog.client_order_id} fully filled during backlog wait, nothing to replace"
605 )
606 return None
607 # Now try placing order will turn expire into order backlog if fails
608 if backlog.is_bbo:
609 return await self.executor.reprice_at_bbo(
610 symbol=Symbol(backlog.symbol),
611 qty=qty,
612 side=backlog.side,
613 replace_best_bid_ask_time=backlog.replace_best_bid_ask_time,
614 package_id=backlog.package_id,
615 initial_price=backlog.initial_price,
616 initial_time=backlog.initial_time,
617 )
618 return await self.executor.reprice_at_mid(
619 symbol=Symbol(backlog.symbol),
620 qty=qty,
621 side=backlog.side,
622 replace_best_bid_ask_time=backlog.replace_best_bid_ask_time,
623 package_id=backlog.package_id,
624 initial_price=backlog.initial_price,
625 initial_time=backlog.initial_time,
626 )
628 async def on_order_expiry_check(self):
629 """
630 Remove orders that have exceeded the allowed lifetime.
632 Replaces them at Update Price (MID) or BBO depending on the trigger.
633 """
635 @dataclass
636 class PendingExpiry:
637 order: OrderDetails
638 strategy: Literal["BBO", "MID"]
640 # Decide which orders expired (no delta_lock — read-only price discovery)
641 async with self.order_pools.get_order_pool() as order_pool:
642 orders_to_check = list(order_pool.values())
644 if not orders_to_check:
645 return
647 async with self.order_pools.get_order_backlog() as order_backlog:
648 queued_ids = {b.client_order_id for b in order_backlog}
649 if queued_ids:
650 orders_to_check = [
651 o for o in orders_to_check if o.client_order_id not in queued_ids
652 ]
653 if not orders_to_check:
654 return
656 pending_actions: dict[str, PendingExpiry] = {}
657 current_time = datetime.now(timezone.utc)
659 get_depth_endpoint = Endpoints.GET_ORDERBOOK_SNAPSHOT
660 current_prices: Dict[Symbol, List[Decimal]] = {}
661 for order in orders_to_check:
662 symbol = Symbol(order.symbol)
664 reason = None
665 if order.replace_best_bid_ask_time <= current_time:
666 reason = "BBO"
667 elif order.max_replace_limit_order_time <= current_time:
668 reason = "MID"
670 if reason:
671 if symbol not in current_prices:
672 try:
673 order_book = await OrderUtils.get_order_book(
674 exchange=self.executor.exchange,
675 pair=symbol,
676 need_log=False,
677 rate_limiter=self.rate_limiter,
678 endpoint=get_depth_endpoint,
679 price_feed=self.price_feed,
680 )
681 current_prices[symbol] = order_book
682 except Exception as e:
683 logger.warning(
684 f"Current price for {order.symbol} couldn't be fetch due to {e}"
685 )
686 order_book = current_prices.get(symbol)
687 current_price = (
688 (order_book[0] if order.side == OrderSide.BUY else order_book[1])
689 if order_book
690 else None
691 )
692 if not current_price or current_price != order.price:
693 # Entry stays in the pool until cancel succeeds:
694 # cancel_single_order pops on success, and treats a failed
695 # cancel for an already-vanished entry as done — popping here
696 # would make every transient cancel failure look like success
697 # and reprice a still-live order
698 pending_actions[order.client_order_id] = PendingExpiry(
699 order, reason
700 )
702 if not pending_actions:
703 return
705 logger.info(
706 f"[ON_ORDER_EXPIRY_CHECK] Processing {len(pending_actions)} expired orders"
707 )
709 # Cancel + replace under delta_lock (the in-flight window); collect
710 # backlog locally so the backlog lock isn't held across the gathers
711 backlog_to_add: list[BacklogDetails] = []
712 sem = asyncio.Semaphore(MAX_CONCURRENT_ORDER_OPS)
714 async def _bounded(coro):
715 async with sem:
716 return await coro
718 async with self.position.delta_lock:
719 cancel_tasks = [
720 _bounded(
721 self.executor.cancel_single_order(
722 symbol=Symbol(action.order.symbol),
723 client_order_id=action.order.client_order_id,
724 )
725 )
726 for action in pending_actions.values()
727 ]
729 cancel_results = await asyncio.gather(*cancel_tasks, return_exceptions=True)
731 replacement_tasks = []
733 for action, result in zip(pending_actions.values(), cancel_results):
734 if isinstance(result, BaseException):
735 logger.error(
736 f"[ON_ORDER_EXPIRY_CHECK] Critical Cancel Error: {result}"
737 )
738 continue
740 if isinstance(result, CancelFatal):
741 logger.error(
742 f"[ON_ORDER_EXPIRY_CHECK] FATAL cancel for "
743 f"{action.order.client_order_id}, skipping replacement"
744 )
745 continue
747 if result is not None:
748 expire_backlog = ExpiredBacklogs(
749 side=action.order.side,
750 qty=action.order.remain_size,
751 is_bbo=(action.strategy == "BBO"),
752 max_replace_limit_order_time=action.order.max_replace_limit_order_time,
753 replace_best_bid_ask_time=action.order.replace_best_bid_ask_time,
754 package_id=action.order.package_id,
755 initial_price=action.order.initial_price,
756 initial_time=action.order.initial_time,
757 **result.model_dump(),
758 )
759 backlog_to_add.append(expire_backlog)
760 continue
762 if action.strategy == "BBO":
763 logger.info(
764 f"[ON_ORDER_EXPIRY_CHECK] Repricing {action.order.symbol} order at BBO with {action.order.remain_size}"
765 )
766 new_task = self.executor.reprice_at_bbo(
767 symbol=Symbol(action.order.symbol),
768 qty=action.order.remain_size,
769 side=action.order.side,
770 replace_best_bid_ask_time=action.order.replace_best_bid_ask_time,
771 package_id=action.order.package_id,
772 initial_price=action.order.initial_price,
773 initial_time=action.order.initial_time,
774 )
775 else:
776 logger.info(
777 f"[ON_ORDER_EXPIRY_CHECK] Repricing {action.order.symbol} order at MID with {action.order.remain_size}"
778 )
779 new_task = self.executor.reprice_at_mid(
780 symbol=Symbol(action.order.symbol),
781 qty=action.order.remain_size,
782 side=action.order.side,
783 replace_best_bid_ask_time=action.order.replace_best_bid_ask_time,
784 package_id=action.order.package_id,
785 initial_price=action.order.initial_price,
786 initial_time=action.order.initial_time,
787 )
788 replacement_tasks.append(_bounded(new_task))
790 if replacement_tasks:
791 logger.info(
792 f"[ON_ORDER_EXPIRY_CHECK] Sending {len(replacement_tasks)} replacement orders. "
793 f"({len(pending_actions) - len(replacement_tasks)} failed/skipped)"
794 )
796 replace_results = await asyncio.gather(
797 *replacement_tasks, return_exceptions=True
798 )
800 for res in replace_results:
801 if res is None:
802 continue
803 if isinstance(res, BaseException):
804 logger.error(
805 f"[ON_ORDER_EXPIRY_CHECK] Critical Replace Error: {res}"
806 )
807 else:
808 backlog_to_add.append(res)
810 if backlog_to_add:
811 async with self.order_pools.get_order_backlog() as order_backlog:
812 for item in backlog_to_add:
813 self.order_pools.dedup_append(order_backlog, item)