Coverage for adrs/oms/ops/order_pool.py: 61%
127 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-29 05:07 +0800
1import logging
2import asyncio
3import random
4import time
6from dataclasses import dataclass
7from decimal import Decimal
8from datetime import datetime, timezone, timedelta
9from pydantic import AwareDatetime, BaseModel, Field
10from contextlib import asynccontextmanager
12from adrs.oms.config import ConfigManager
13from adrs.oms.rate_limit.rate_limiter import RateLimiter
14from adrs.oms.rate_limit.exchange_limit_profiles import (
15 Endpoints,
16 OpenOrdersFetchProfile,
17 OPEN_ORDERS_FETCH_PROFILES,
18)
20from cybotrade.io import ExchangeClient
21from cybotrade import Symbol
22from cybotrade.models import (
23 OrderSide,
24 OrderUpdate,
25)
27logger = logging.getLogger(__name__)
29# How long a registered-but-unconfirmed placement counts as in-flight. Past
30# this, an unknown WS update for the id is treated as a real desync again.
31PENDING_INSERT_TTL_SEC = 60
34class BacklogDetails(BaseModel):
35 symbol: str
36 total_retries: int
37 next_retry_at: AwareDatetime | None = None # None = due now
38 client_order_id: str
41class OrderBacklogs(BacklogDetails):
42 side: OrderSide
43 ori_entry_time: AwareDatetime
44 qty: Decimal
45 offset: Decimal
46 replace_best_bid_ask_time: AwareDatetime
47 max_replace_limit_order_time: AwareDatetime
48 package_id: str
49 initial_price: Decimal | None = None
50 initial_time: datetime | None = None
53class CancelBacklogs(BacklogDetails):
54 pass
57class ExpiredBacklogs(BacklogDetails):
58 # For OrderBacklogs
59 side: OrderSide
60 qty: Decimal
61 replace_best_bid_ask_time: AwareDatetime
62 max_replace_limit_order_time: AwareDatetime
63 # For CancelBacklogs
64 is_bbo: bool
65 # For Aegis
66 package_id: str
67 initial_price: Decimal
68 initial_time: datetime
71class OrderDetails(BaseModel):
72 # Order Management
73 symbol: str
74 side: OrderSide
75 price: Decimal
76 remain_size: Decimal
77 client_order_id: str
78 replace_best_bid_ask_time: AwareDatetime
79 max_replace_limit_order_time: AwareDatetime
80 # Which signal group does it belong to
81 package_id: str
82 initial_price: Decimal
83 initial_time: datetime
84 # When this entry landed in the local pool; unlike initial_time it is not
85 # carried across reprices, so resync can tell fresh entries from ghosts
86 inserted_at: AwareDatetime = Field(
87 default_factory=lambda: datetime.now(timezone.utc)
88 )
91@dataclass
92class OpenOrdersSnapshot:
93 taken_at: datetime # when the REST pass began
94 orders: dict[Symbol, list[OrderUpdate]]
97class OrderPoolHandler:
98 def __init__(
99 self,
100 exchange: ExchangeClient,
101 config: ConfigManager,
102 rate_limiter: RateLimiter,
103 ):
104 self.order_pool: dict[str, OrderDetails] = {}
105 self._pool_lock = asyncio.Lock()
106 # Placements whose REST response hasn't landed yet: a WS update can
107 # beat the pool insert, and must not be read as a desync
108 self.pending_inserts: dict[str, float] = {} # client_order_id -> monotonic ts
109 self.order_backlog: list[BacklogDetails] = []
110 self._backlog_lock = asyncio.Lock()
111 self.order_value_update: dict[str, Decimal] = {}
112 self.order_records: dict[
113 str, list[tuple[Symbol, str, Decimal, datetime]]
114 ] = {} # package id, list of client order id
115 self.exchange = exchange
116 self.config = config
117 self.rate_limiter = rate_limiter
119 @asynccontextmanager
120 async def get_order_backlog(self):
121 await self._backlog_lock.acquire()
122 try:
123 yield self.order_backlog
124 finally:
125 self._backlog_lock.release()
127 @staticmethod
128 def dedup_append(order_backlog: list[BacklogDetails], item: BacklogDetails) -> bool:
129 """Append only if no entry with the same client_order_id is queued."""
130 if any(b.client_order_id == item.client_order_id for b in order_backlog):
131 return False
132 order_backlog.append(item)
133 return True
135 def register_pending_insert(self, client_order_id: str):
136 self.pending_inserts[client_order_id] = time.monotonic()
138 def confirm_pending_insert(self, client_order_id: str):
139 self.pending_inserts.pop(client_order_id, None)
141 def is_pending_insert(self, client_order_id: str) -> bool:
142 """
143 Entries self-expire after PENDING_INSERT_TTL_SEC, so failure paths
144 (ambiguous REST errors where the order may still exist) never need
145 explicit cleanup and can't leak.
146 """
147 now = time.monotonic()
148 expired = [
149 cid
150 for cid, ts in self.pending_inserts.items()
151 if now - ts > PENDING_INSERT_TTL_SEC
152 ]
153 for cid in expired:
154 del self.pending_inserts[cid]
155 return client_order_id in self.pending_inserts
157 @asynccontextmanager
158 async def get_order_pool(self):
159 await self._pool_lock.acquire()
160 try:
161 yield self.order_pool
162 finally:
163 self._pool_lock.release()
165 async def snapshot(
166 self,
167 ) -> tuple[dict[str, "OrderDetails"], list["BacklogDetails"]]:
168 """Return frozen copies of the pool and backlog for assertions in tests."""
169 async with self.get_order_pool() as pool:
170 pool_copy = dict(pool)
171 async with self.get_order_backlog() as backlog:
172 backlog_copy = list(backlog)
173 return pool_copy, backlog_copy
175 async def fetch_open_orders_snapshot(self) -> OpenOrdersSnapshot:
176 """
177 One REST pass over open orders, shared within a single event
178 (placement tick / desync resync / init).
180 A symbol absent from the result means its fetch failed; consumers
181 keep their previous state for it rather than treating it as empty.
182 """
183 taken_at = datetime.now(timezone.utc)
184 symbols = [
185 Symbol(s) for s in self.config.config.base_asset_to_symbol_table.values()
186 ]
188 profile = OPEN_ORDERS_FETCH_PROFILES.get(
189 self.config.config.credentials.exchange
190 )
191 if profile is not None and len(symbols) >= profile.min_symbols_for_bulk:
192 orders = await self._fetch_bulk_open_orders(symbols, profile)
193 if orders is not None:
194 return OpenOrdersSnapshot(taken_at=taken_at, orders=orders)
196 orders = {}
197 for symbol in symbols:
198 try:
199 async with self.rate_limiter.reserve(
200 endpoint=Endpoints.GET_OPEN_ORDERS
201 ):
202 orders[symbol] = await self.exchange.get_open_orders(symbol=symbol)
203 except Exception as e:
204 logger.warning(f"Failed to fetch open orders for {symbol} due to {e}")
205 return OpenOrdersSnapshot(taken_at=taken_at, orders=orders)
207 async def _fetch_bulk_open_orders(
208 self, symbols: list[Symbol], profile: OpenOrdersFetchProfile
209 ) -> dict[Symbol, list[OrderUpdate]] | None:
210 """
211 All symbols in one request. None means bulk is unusable this round
212 (error, or the response may be truncated at the exchange's page
213 limit) and the caller must fall back to per-symbol fetches.
214 """
215 try:
216 async with self.rate_limiter.reserve(
217 endpoint=Endpoints.GET_OPEN_ORDERS_ALL
218 ):
219 all_orders = await self.exchange.get_open_orders(
220 symbol=None, **profile.bulk_kwargs
221 )
222 except Exception as e:
223 logger.warning(
224 f"Bulk open-orders fetch failed due to {e}, falling back to per-symbol"
225 )
226 return None
227 if (
228 profile.truncation_row_limit is not None
229 and len(all_orders) >= profile.truncation_row_limit
230 ):
231 logger.warning(
232 "Bulk open-orders response filled the page, possible truncation, falling back to per-symbol"
233 )
234 return None
235 # Bulk saw everything, so absent symbol = no open orders (empty list,
236 # unlike a failed per-symbol fetch which omits the key). Orders on
237 # symbols outside the config table are dropped to match per-symbol
238 # behavior — the OMS must not adopt orders it doesn't trade.
239 orders: dict[Symbol, list[OrderUpdate]] = {symbol: [] for symbol in symbols}
240 for order in all_orders:
241 if order.symbol in orders:
242 orders[order.symbol].append(order)
243 return orders
245 async def resync_order_pool(self, snapshot: OpenOrdersSnapshot):
246 """
247 Expensive function only resync when needed
248 """
249 current_time = datetime.now(timezone.utc)
250 replace_order_interval_in_sec = current_time + timedelta(
251 seconds=round(
252 random.uniform(
253 self.config.config.min_limit_replace_interval,
254 self.config.config.max_limit_replace_interval,
255 ),
256 2,
257 )
258 )
259 new_order_pool: dict[str, OrderDetails] = {}
260 client_to_package = {
261 record[1]: pkg_id
262 for pkg_id, records in self.order_records.items()
263 for record in records
264 }
265 try:
266 # Snapshot the old pool once so timer carryover is read from stable
267 # state, not the dict we are rebuilding mid-loop.
268 old_order_pool = self.order_pool
269 for order_list in snapshot.orders.values():
270 for order in order_list:
271 prev = old_order_pool.get(order.client_order_id)
272 new_order_pool[order.client_order_id] = OrderDetails(
273 client_order_id=order.client_order_id,
274 replace_best_bid_ask_time=prev.replace_best_bid_ask_time
275 if prev is not None
276 else datetime.now(timezone.utc)
277 + timedelta(
278 seconds=self.config.config.replace_best_bid_ask_time
279 ),
280 max_replace_limit_order_time=(
281 prev.max_replace_limit_order_time
282 if prev is not None
283 else replace_order_interval_in_sec
284 ),
285 symbol=str(order.symbol),
286 remain_size=order.remain_size,
287 side=order.side,
288 price=order.price,
289 package_id=package_id
290 if (package_id := client_to_package.get(order.client_order_id))
291 else "",
292 initial_price=Decimal("0"),
293 initial_time=datetime.now(tz=timezone.utc),
294 )
295 # Merge, then swap under the pool lock. An entry missing from the
296 # snapshot is only a ghost if REST could have seen it: entries for
297 # failed-fetch symbols and entries inserted after the REST pass
298 # began are carried over, not dropped.
299 fetched_symbols = {str(s) for s in snapshot.orders.keys()}
300 async with self.get_order_pool():
301 for cid, details in self.order_pool.items():
302 if cid in new_order_pool:
303 continue
304 if (
305 details.symbol not in fetched_symbols
306 or details.inserted_at > snapshot.taken_at
307 ):
308 new_order_pool[cid] = details
309 self.order_pool = new_order_pool
310 except Exception as e:
311 logger.warning(f"Failed to resync_order_pool due to {e}")