Coverage for adrs/oms/ops/order_executer.py: 94%
219 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-29 05:07 +0800
1import time
2import logging
3import random
4import asyncio
6from dataclasses import dataclass
7from datetime import datetime, timezone, timedelta
8from decimal import Decimal
10from pydantic.types import AwareDatetime
11from cybotrade import Symbol
12from cybotrade.io import ExchangeClient
13from cybotrade.models import OrderSide, OrderUpdate, SymbolInfo
15from adrs.oms.config import ConfigManager, Config
16from adrs.oms.calculation import Calculate
17from adrs.oms.ops.order_pool import (
18 OrderBacklogs,
19 OrderPoolHandler,
20 CancelBacklogs,
21 OrderDetails,
22)
23from adrs.oms.ops.order_utils import OrderUtils
24from adrs.oms.price_feed import PriceFeed
25from adrs.oms.rate_limit.rate_limiter import RateLimiter
26from adrs.oms.rate_limit.exchange_limit_profiles import Endpoints
27from adrs.oms.rate_limit.error_policy import ErrorAction, ExchangeErrorPolicy
29logger = logging.getLogger(__name__)
31# Limit orders post away from the touch (buys below market, sells above), so the
32# value at fill differs from value at the current market price. Derive the
33# notional-based qty bounds from a price nudged this fraction toward a worse fill
34# (buys: cheaper → need more qty to clear min_notional). Keeps an order from
35# clearing min_notional at market yet getting rejected for order value once it
36# rests at the limit price.
37NOTIONAL_PRICE_SLIPPAGE = Decimal("0.005")
39# Caps concurrent place/cancel requests per burst. Keeps the in-flight count
40# at roughly the pool drain rate (~8/s buffered) x per-op latency, and bounds
41# how many requests are already on the wire if the exchange rate-limits us
42# while the local accounting thinks we are fine.
43MAX_CONCURRENT_ORDER_OPS = 3
46def make_package_id(portfolio_id: str) -> str:
47 return f"{portfolio_id}_{time.time_ns() // 1_000_000}"
50@dataclass
51class PlacementContext:
52 """
53 Inputs available to the strategy hooks for one placement decision.
55 New fields may be added over time; additions do not break existing
56 subclass hook signatures.
57 """
59 symbol: Symbol
60 side: OrderSide
61 market_price: Decimal
62 symbol_info: SymbolInfo
63 # Exchange floor/ceiling per order, derived from lot size and notional limits
64 min_qty: Decimal
65 max_qty: Decimal
68@dataclass
69class CancelFatal:
70 """
71 Cancel hit a FATAL error: retrying is futile, but the order may still be
72 live on the exchange. Distinct from None (confirmed gone) so callers never
73 treat the quantity as freed or place a replacement for it - the entry
74 stays in the pool for the delta/expiry logic to keep seeing.
75 """
77 client_order_id: str
80@dataclass
81class CancelMultiResult:
82 # Quantity still to place as a new order, computed from cancels that
83 # actually succeeded - never from the pre-cancel selection.
84 remainder: Decimal
85 # How many of the chosen cancels did NOT succeed (still live/unknown on
86 # the exchange). Callers must not place a same-tick replacement sized on
87 # `remainder` while this is nonzero, or they risk placing on top of an
88 # order that is still resting.
89 failed_count: int
92class OrderExecutor:
93 def __init__(
94 self,
95 config_manager: ConfigManager,
96 order_pools: OrderPoolHandler,
97 rate_limiter: RateLimiter,
98 error_policy: ExchangeErrorPolicy,
99 price_feed: PriceFeed | None = None,
100 ):
101 self.exchange: ExchangeClient = config_manager.exchange
102 self.config: Config = config_manager.config
103 self.order_pools = order_pools
104 self.symbol_infos = config_manager.symbol_infos
105 self.rate_limiter = rate_limiter
106 self.error_policy = error_policy
107 self.price_feed = price_feed
108 self.package_id = make_package_id(self.config.portfolio_id)
110 def update_package_id(self):
111 self.package_id = make_package_id(self.config.portfolio_id)
113 def _make_context(
114 self, symbol: Symbol, side: OrderSide, market_price: Decimal
115 ) -> PlacementContext:
116 symbol_info = self.symbol_infos[symbol]
117 # Notional bounds against the worst-case resting price, not the current
118 # market: a buy that fills cheaper needs MORE qty to clear min_notional,
119 # a sell that fills dearer needs LESS qty to stay under max_notional.
120 # Using market_price directly lets an order pass here yet get rejected
121 # for order value once it rests at the limit price.
122 min_notional_price = market_price * (Decimal("1") - NOTIONAL_PRICE_SLIPPAGE)
123 max_notional_price = market_price * (Decimal("1") + NOTIONAL_PRICE_SLIPPAGE)
124 return PlacementContext(
125 symbol=symbol,
126 side=side,
127 market_price=market_price,
128 symbol_info=symbol_info,
129 min_qty=max(
130 symbol_info.min_limit_qty,
131 symbol_info.min_notional / min_notional_price,
132 ),
133 max_qty=min(
134 symbol_info.max_limit_qty,
135 symbol_info.max_notional / max_notional_price,
136 ),
137 )
139 async def split_order_quantity(
140 self, qty: Decimal, ctx: PlacementContext
141 ) -> list[Decimal]:
142 """
143 Strategy hook: break a target quantity into individual order sizes.
145 The number of entries decides how many orders are placed; entries of
146 zero or below are skipped. ctx.min_qty/max_qty are the exchange's
147 per-order bounds; implementations may tighten them but not exceed.
149 Naive default: a single order for the full quantity, chunked only
150 when it exceeds the exchange's max order size.
152 Every returned size is rounded (truncated) to the instrument's
153 `quantity_precision` — exchanges reject a qty that isn't a multiple of
154 the lot step (e.g. bybit "10001 Qty invalid"). Sizes that truncate below
155 `min_qty` are dropped (too small to place).
156 """
157 precision = ctx.symbol_info.quantity_precision
159 def round_qty(x: Decimal) -> Decimal:
160 return Calculate.round_with_precision(x, precision)
162 if qty <= ctx.max_qty:
163 rounded = round_qty(qty)
164 return [rounded] if rounded >= ctx.min_qty else []
165 chunks, rest = divmod(qty, ctx.max_qty)
166 sizes = [round_qty(ctx.max_qty)] * int(chunks)
167 rest = round_qty(rest)
168 if rest >= ctx.min_qty:
169 sizes.append(rest)
170 return sizes
172 async def compute_limit_offsets(
173 self, level: int, ctx: PlacementContext
174 ) -> list[Decimal]:
175 """
176 Strategy hook: price offset from best bid/ask for each order,
177 index-aligned with the sizes from split_order_quantity.
179 Offset 0 posts at BBO; positive offsets post deeper into the book.
181 Naive default: post everything at BBO.
182 """
183 return [Decimal("0")] * level
185 async def cancel_single_order(
186 self,
187 symbol: Symbol,
188 client_order_id: str,
189 ) -> CancelBacklogs | CancelFatal | None:
190 """
191 To cancel a specified order
192 """
193 endpoint = Endpoints.CANCEL_ORDER
194 try:
195 async with self.rate_limiter.guard(endpoint=endpoint):
196 await self.exchange.cancel_order(
197 symbol=symbol,
198 client_order_id=client_order_id,
199 )
200 async with self.order_pools.get_order_pool() as order_pool:
201 order_pool.pop(client_order_id, None)
202 return None
203 except Exception as e:
204 action = self.error_policy.classify(e)
205 async with self.order_pools.get_order_pool() as order_pool:
206 already_left = client_order_id not in order_pool
207 if action is ErrorAction.TERMINAL_SUCCESS:
208 order_pool.pop(client_order_id, None)
210 if action is ErrorAction.TERMINAL_SUCCESS or already_left:
211 # Order already gone on the exchange (110001) or our WS already
212 # popped it; nothing left to cancel, retrying only burns rate limit
213 logger.info(
214 f"Cancel for {client_order_id} treated as done "
215 f"(action={action.name}, already_left={already_left})"
216 )
217 return None
218 if action is ErrorAction.FATAL:
219 # Order may still be live, so keep it in the pool for the expiry/
220 # delta logic, but stop retrying a cancel that will never succeed
221 logger.error(
222 f"Cancel for {client_order_id} hit FATAL error, not retrying: {e}"
223 )
224 return CancelFatal(client_order_id=client_order_id)
226 logger.warning(f"Failed to cancel order {client_order_id} because {e}")
227 return CancelBacklogs(
228 symbol=str(symbol),
229 total_retries=1,
230 client_order_id=client_order_id,
231 )
233 async def cancel_multi_limit_order(
234 self, symbol: Symbol, target: Decimal, open_orders: list[OrderUpdate]
235 ) -> CancelMultiResult:
236 """
237 Cancel pending orders to approach the target.
239 open_orders comes from the caller's tick snapshot so the tick costs a
240 single open-orders fetch instead of one per decision.
242 The remainder is computed from cancels that actually succeeded, not
243 from the pre-cancel selection: a cancel blocked by the rate limiter
244 (or rejected by the exchange) leaves its order live, so counting it
245 as freed would size a same-tick replacement on top of an order that
246 never went away - doubling the position if both then fill.
247 """
248 # remove more than needed then return value to compensate it
249 # REASON better to use the rate limits than to pay the fees twice
250 if target == Decimal("0"):
251 raise ValueError("Target shouldn't be zero")
253 side = OrderSide.BUY if target > Decimal("0") else OrderSide.SELL
254 candidates = sorted(
255 # BUY target: smallest opposite-side orders first to minimise over-cancel
256 # SELL target: largest opposite-side orders first to reach target quickly
257 (order for order in open_orders if order.side != side),
258 key=lambda order: order.remain_size,
259 reverse=(side != OrderSide.BUY),
260 )
262 selected_sum = Decimal("0")
263 chosen_orders: list[OrderUpdate] = []
265 for order in candidates:
266 if side == OrderSide.BUY and selected_sum + order.remain_size < target:
267 selected_sum += order.remain_size
268 chosen_orders.append(order)
269 elif side == OrderSide.SELL and selected_sum - order.remain_size > target:
270 selected_sum -= order.remain_size
271 chosen_orders.append(order)
272 # To add the final order to exceed target if available
273 else:
274 selected_sum += (
275 order.remain_size if side == OrderSide.BUY else -order.remain_size
276 )
277 chosen_orders.append(order)
278 break
280 sem = asyncio.Semaphore(MAX_CONCURRENT_ORDER_OPS)
282 async def _bounded_cancel(order: OrderUpdate):
283 async with sem:
284 return await self.cancel_single_order(
285 symbol=order.symbol,
286 client_order_id=order.client_order_id,
287 )
289 cancelled_sum = Decimal("0")
290 failed_count = 0
291 to_backlog: list[CancelBacklogs] = []
293 cancel_results = await asyncio.gather(
294 *[_bounded_cancel(order) for order in chosen_orders],
295 return_exceptions=True,
296 )
298 for order, result in zip(chosen_orders, cancel_results):
299 if isinstance(result, BaseException):
300 logger.error(f"Failed to cancel order due to, {result}")
301 failed_count += 1
302 continue
303 if isinstance(result, CancelBacklogs):
304 to_backlog.append(result)
305 failed_count += 1
306 continue
307 if isinstance(result, CancelFatal):
308 failed_count += 1
309 continue
310 cancelled_sum += (
311 order.remain_size if side == OrderSide.BUY else -order.remain_size
312 )
314 if to_backlog:
315 async with self.order_pools.get_order_backlog() as order_backlog:
316 for item in to_backlog:
317 self.order_pools.dedup_append(order_backlog, item)
319 logger.info(
320 f"[CANCEL_MULTI_LIMIT_ORDER] Cancelled total {cancelled_sum} {symbol} worth of open orders"
321 + (f", {failed_count} cancel(s) failed" if failed_count else "")
322 )
323 return CancelMultiResult(
324 remainder=target - cancelled_sum, failed_count=failed_count
325 )
327 async def place_single_limit_order(
328 self,
329 symbol: Symbol,
330 offset: Decimal,
331 qty: Decimal,
332 side: OrderSide,
333 symbol_info: SymbolInfo,
334 replace_best_bid_ask_time: AwareDatetime | None = None,
335 package_id: str | None = None,
336 initial_price: Decimal | None = None,
337 initial_time: datetime | None = None,
338 client_order_id: str | None = None,
339 order_book: list[Decimal] | None = None,
340 ) -> OrderBacklogs | None:
341 """
342 Place a single limit order on the exchange
344 OrderDetails replace_best_bid_ask always updates to the latest version
345 unless replace_best_bid_ask_time is specified
347 Package ID is for to be repriced orders, ignore if it is a new order
349 client_order_id is for backlog retries only: reusing the original id lets
350 the exchange dedupe when the first attempt was sent but its response was lost
352 return
353 None if successful
354 OrderBacklogs if fail
355 """
356 post_order_endpoint = Endpoints.PLACE_ORDER
357 get_depth_endpoint = Endpoints.GET_ORDERBOOK_SNAPSHOT
359 current_time = datetime.now(timezone.utc)
360 replace_order_interval_in_sec = current_time + timedelta(
361 seconds=round(
362 random.uniform(
363 self.config.min_limit_replace_interval,
364 self.config.max_limit_replace_interval,
365 ),
366 2,
367 )
368 )
369 replace_best_bid_ask_time = (
370 replace_best_bid_ask_time
371 if replace_best_bid_ask_time is not None
372 else current_time + timedelta(seconds=self.config.replace_best_bid_ask_time)
373 )
374 client_order_id = (
375 client_order_id
376 if client_order_id is not None
377 else OrderUtils.make_client_order_id()
378 )
379 current_package_id = package_id if package_id else self.package_id
380 try:
381 if order_book is None:
382 order_book = await OrderUtils.get_order_book(
383 exchange=self.exchange,
384 pair=symbol,
385 need_log=True,
386 rate_limiter=self.rate_limiter,
387 endpoint=get_depth_endpoint,
388 price_feed=self.price_feed,
389 )
391 price = order_book[0] if side == OrderSide.BUY else order_book[1]
392 adjusted_price = Calculate.align_price(
393 limit_price=price,
394 side=side,
395 offset=offset,
396 tick_size=symbol_info.tick_size,
397 precision=symbol_info.price_precision,
398 )
400 # Register before the REST send: a WS fill can beat the REST
401 # response, and validate_oms_state must not read that as a desync
402 self.order_pools.register_pending_insert(client_order_id)
403 async with self.rate_limiter.guard(endpoint=post_order_endpoint):
404 await self.exchange.place_order(
405 limit=adjusted_price,
406 side=side,
407 quantity=qty,
408 symbol=symbol,
409 client_order_id=client_order_id,
410 post_only=True,
411 )
413 except Exception as e:
414 if self.error_policy.classify(e) is ErrorAction.FATAL:
415 # Nothing was placed; dropping is correct, retrying is futile
416 logger.error(
417 f"[PLACE_SINGLE_LIMIT_ORDER] FATAL error, not retrying: {e}"
418 )
419 return None
420 logger.warning(
421 f"[PLACE_SINGLE_LIMIT_ORDER] Order placement failed, reason {e}"
422 )
423 return OrderBacklogs(
424 offset=offset,
425 ori_entry_time=datetime.now(timezone.utc),
426 qty=qty,
427 max_replace_limit_order_time=replace_order_interval_in_sec,
428 side=side,
429 replace_best_bid_ask_time=replace_best_bid_ask_time,
430 symbol=str(symbol),
431 total_retries=1,
432 client_order_id=client_order_id,
433 package_id=current_package_id,
434 initial_price=initial_price,
435 initial_time=initial_time,
436 )
438 logger.info(f"[PLACE_LIMIT] Placed {side} {qty} {symbol} @ {adjusted_price}")
439 initial_price = initial_price if initial_price else adjusted_price
440 initial_time = initial_time if initial_time else current_time
441 async with self.order_pools.get_order_pool() as order_pool:
442 order_pool[client_order_id] = OrderDetails(
443 client_order_id=client_order_id,
444 symbol=str(symbol),
445 max_replace_limit_order_time=replace_order_interval_in_sec,
446 replace_best_bid_ask_time=replace_best_bid_ask_time,
447 remain_size=qty,
448 side=side,
449 price=adjusted_price,
450 package_id=current_package_id,
451 initial_price=initial_price,
452 initial_time=initial_time,
453 )
454 self.order_pools.confirm_pending_insert(client_order_id)
456 if current_package_id not in self.order_pools.order_records:
457 self.order_pools.order_records[current_package_id] = []
458 self.order_pools.order_records[current_package_id].append(
459 (symbol, client_order_id, initial_price, initial_time)
460 )
462 async def get_current_price(self, symbol: Symbol) -> Decimal | None:
463 """
464 Canonical current-price fetch: reserves a rate-limit slot and returns
465 None on any failure. All price reads go through here so behaviour is
466 consistent (waits under contention, never raises to the caller).
468 Prefers the websocket feed, which costs no request weight, and falls
469 back to the same depth read as every other price path — deliberately not
470 cybotrade's get_current_price, which sends no depth limit and so would
471 fetch 500 levels (weight 10) while the limiter charges OMS_DEPTH_LIMIT's
472 weight (2). One call shape means one limit and one weight. The mid
473 formula matches cybotrade's get_current_price so both sources agree.
474 """
475 try:
476 best_bid, best_ask = await OrderUtils.get_order_book(
477 exchange=self.exchange,
478 pair=symbol,
479 need_log=False,
480 rate_limiter=self.rate_limiter,
481 endpoint=Endpoints.GET_ORDERBOOK_SNAPSHOT,
482 price_feed=self.price_feed,
483 )
484 return (best_bid + best_ask) / Decimal("2.0")
485 except Exception as e:
486 logger.warning(f"Failed to fetch current price due to {e}")
487 return None
489 async def reprice_at_mid(
490 self,
491 symbol: Symbol,
492 qty: Decimal,
493 side: OrderSide,
494 replace_best_bid_ask_time: AwareDatetime,
495 package_id: str,
496 initial_price: Decimal,
497 initial_time: datetime | None = None,
498 ) -> OrderBacklogs | None:
499 """
500 Replaces orders at a refreshed price so they are more likely to be filled.
502 The offset comes from the compute_limit_offsets hook (level=1); the
503 naive base hook makes this equivalent to repricing at BBO.
505 The order being repriced was already cancelled, so a failure to compute
506 the offset must not drop the quantity — fall back to BBO (offset 0)
507 and let place_single_limit_order handle placement failures via backlog
508 """
509 symbol_info = self.symbol_infos[symbol]
511 offset = Decimal("0")
512 market_price = await self.get_current_price(symbol=symbol)
513 if market_price is not None:
514 try:
515 ctx = self._make_context(
516 symbol=symbol, side=side, market_price=market_price
517 )
518 offset = (await self.compute_limit_offsets(1, ctx))[0]
519 except Exception as e:
520 logger.warning(
521 f"[REPRICE_AT_MID] Offset calculation for {symbol} failed due to {e}, falling back to BBO"
522 )
524 return await self.place_single_limit_order(
525 symbol=symbol,
526 offset=offset,
527 qty=qty,
528 side=side,
529 symbol_info=symbol_info,
530 replace_best_bid_ask_time=replace_best_bid_ask_time,
531 package_id=package_id,
532 initial_price=initial_price,
533 initial_time=initial_time,
534 )
536 async def reprice_at_bbo(
537 self,
538 symbol: Symbol,
539 qty: Decimal,
540 side: OrderSide,
541 replace_best_bid_ask_time: AwareDatetime,
542 package_id: str,
543 initial_price: Decimal | None = None,
544 initial_time: datetime | None = None,
545 ) -> OrderBacklogs | None:
546 """
547 Replaces expired orders immediately at the current Best Bid or Offer (BBO).
549 This function is a fallback mechanism for orders that have passed their
550 expiry time without being executed or cancelled by other logic.
552 Rationale (Why at BBO?):
553 The fact that the order expired implies the system critically needs this position filled.
554 If a significant market movement occurred (causing a material change in delta),
555 the order should have been cancelled by delta logic before it ever reached its expiry time.
556 """
557 return await self.place_single_limit_order(
558 symbol=symbol,
559 offset=Decimal("0"),
560 qty=qty,
561 side=side,
562 symbol_info=self.symbol_infos[symbol],
563 replace_best_bid_ask_time=replace_best_bid_ask_time,
564 package_id=package_id,
565 initial_price=initial_price,
566 initial_time=initial_time,
567 )
569 async def place_multiple_limit_order(
570 self,
571 symbol: Symbol,
572 quantity: Decimal,
573 ):
574 """
575 Place a set of limit orders which total up to quantity.
577 Sizing and pricing come from the split_order_quantity and
578 compute_limit_offsets hooks; orchestration and failure handling stay
579 here. A hook failure degrades to the naive defaults rather than
580 dropping the rebalance.
581 """
583 order_side = OrderSide.BUY if quantity > Decimal("0") else OrderSide.SELL
584 qty = abs(quantity)
586 try:
587 shared_order_book = await OrderUtils.get_order_book(
588 exchange=self.exchange,
589 pair=symbol,
590 need_log=True,
591 rate_limiter=self.rate_limiter,
592 endpoint=Endpoints.GET_ORDERBOOK_SNAPSHOT,
593 price_feed=self.price_feed,
594 )
595 except Exception as e:
596 logger.warning(
597 f"[PLACE_MULTI_LIMIT_ORDER] No orderbook for {symbol} ({e}), skipping this cycle; delta will retry next placement"
598 )
599 return
601 market_price = (shared_order_book[0] + shared_order_book[1]) / Decimal("2")
603 ctx = self._make_context(
604 symbol=symbol, side=order_side, market_price=market_price
605 )
607 try:
608 random_order_size = await self.split_order_quantity(qty, ctx)
609 except Exception as e:
610 logger.warning(
611 f"[PLACE_MULTI_LIMIT_ORDER] split_order_quantity failed due to {e}, using naive sizing"
612 )
613 # Pinned to the base implementation so a broken override cannot recurse
614 random_order_size = await OrderExecutor.split_order_quantity(self, qty, ctx)
616 try:
617 limit_offsets = await self.compute_limit_offsets(
618 len(random_order_size), ctx
619 )
620 except Exception as e:
621 logger.warning(
622 f"[PLACE_MULTI_LIMIT_ORDER] compute_limit_offsets failed due to {e}, falling back to BBO"
623 )
624 limit_offsets = [Decimal("0")] * len(random_order_size)
626 if len(limit_offsets) != len(random_order_size):
627 logger.warning(
628 "[PLACE_MULTI_LIMIT_ORDER] Offset count does not match order count, falling back to BBO"
629 )
630 limit_offsets = [Decimal("0")] * len(random_order_size)
632 logger.info(
633 f"[PLACE_MULTI_LIMIT_ORDER] Placing {len(random_order_size)} limit orders, with price offset of {limit_offsets}"
634 )
636 sem = asyncio.Semaphore(MAX_CONCURRENT_ORDER_OPS)
638 async def _bounded_place(i: int):
639 async with sem:
640 return await self.place_single_limit_order(
641 symbol=symbol,
642 offset=limit_offsets[i],
643 qty=random_order_size[i],
644 side=order_side,
645 symbol_info=ctx.symbol_info,
646 order_book=shared_order_book,
647 )
649 order_tasks = [
650 _bounded_place(i)
651 for i in range(len(random_order_size))
652 if random_order_size[i] > Decimal("0")
653 ]
655 order_results = await asyncio.gather(*order_tasks, return_exceptions=True)
657 to_backlog: list[OrderBacklogs] = []
658 for result in order_results:
659 if isinstance(result, BaseException):
660 logger.error(f"Failed to place order due to, {result}")
661 continue
662 if isinstance(result, OrderBacklogs):
663 to_backlog.append(result)
665 if to_backlog:
666 async with self.order_pools.get_order_backlog() as order_backlog:
667 order_backlog.extend(to_backlog)