Coverage for adrs/oms/rate_limit/rate_limiter.py: 86%
460 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-29 05:07 +0800
1import asyncio
3from abc import abstractmethod, ABC
4from collections import deque
5from contextlib import asynccontextmanager
6from decimal import Decimal
7import random
8import time
10import logging
11from typing import Any, Dict, AsyncGenerator
13from adrs.oms.config import ConfigManager
14from adrs.oms.rate_limit.error_policy import (
15 BINANCE_ORDER_RATE_LIMIT_CODES,
16 binance_banned_until_ms,
17 is_binance_rate_limit_error,
18)
19from adrs.oms.rate_limit.exchange_limit_profiles import (
20 BinanceLimitProfile,
21 BINANCE_DEPTH_WEIGHTS,
22 BINANCE_FUTURES_COSTS,
23 BybitLimitState,
24 DYNAMIC_WEIGHT,
25 Endpoints,
26 OMS_DEPTH_LIMIT,
27 get_depth_weight,
28 BybitRateLimitPool,
29 BybitLimitProfile,
30 BYBIT_FUTURES_COSTS,
31)
33from cybotrade.binance import BinanceLinearClient, BinanceError
34from cybotrade.bybit import BybitLinearClient, BybitError
36logger = logging.getLogger(__name__)
38RESERVE_TIMEOUT_SEC = 1.5
39_MIN_RESERVE_SLEEP = 0.005
41# Wall-clock ceiling on a single guarded/reserved exchange call.
42EXCHANGE_CALL_TIMEOUT_SEC = 10.0
44# Only these header prefixes carry rate-limit signal; everything else (auth,
45# account/IP identifiers, cookies) is dropped before logging.
46_RATE_LIMIT_HEADER_PREFIXES = (
47 "retry-after",
48 "x-mbx-used-weight",
49 "x-mbx-order-count",
50 "x-bapi-limit",
51)
54def _redact_headers(headers: dict[str, Any]) -> dict[str, Any]:
55 """Keep only rate-limit headers; never log raw exchange headers."""
56 return {
57 k: v
58 for k, v in headers.items()
59 if k.lower().startswith(_RATE_LIMIT_HEADER_PREFIXES)
60 }
63def _get_header(headers: dict[str, Any], name: str) -> Any | None:
64 """Case-insensitive header lookup; exchange clients don't guarantee casing."""
65 if name in headers:
66 return headers[name]
67 lname = name.lower()
68 for k, v in headers.items():
69 if k.lower() == lname:
70 return v
71 return None
74def _int_header(headers: dict[str, Any], name: str) -> int | None:
75 """Case-insensitive header lookup coerced to int; None if absent or junk."""
76 raw = _get_header(headers, name)
77 if raw is None:
78 return None
79 try:
80 return int(raw)
81 except (TypeError, ValueError):
82 logger.warning(f"[RATE_LIMIT] Ignoring unparseable {name} header: {raw!r}")
83 return None
86# A parsed cooldown deadline further out than this is treated as garbage rather
87# than trusted, so one malformed header cannot park the OMS indefinitely.
88#
89# Sized off what Binance actually hands out, not off what feels reasonable: a
90# repeat offender's IP ban escalates to three days, and a day-long horizon threw
91# every one of those away as implausible. With no deadline left, _arm_cooldown
92# fell back to the 65s blind cooldown, after which the OMS polled straight
93# through the remaining ban and renewed it — the exact production failure this
94# code exists to prevent, reserved for its most expensive case. Four days leaves
95# headroom over the longest real ban while still catching absurd values.
96_MAX_TRUSTED_COOLDOWN_MS = 4 * 24 * 60 * 60 * 1000
97# Cooldown when the exchange says "too many requests" but supplies neither a
98# Retry-After header nor a ban deadline. One weight window plus a margin: long
99# enough to stop the bleeding, short enough not to strand the OMS.
100_BLIND_COOLDOWN_MS = 65_000
101_COOLDOWN_SAFETY_MS = 1_000
103# Extra, randomised delay added when a cooldown is armed.
104#
105# Every process sharing a rate-limit bucket parses the SAME deadline out of the
106# same ban message, so without jitter they all resume on the same millisecond and
107# hit the bucket in lockstep -- the classic thundering herd, and the one shape of
108# retry most likely to earn the next ban. On Binance testnet the bucket is a
109# shared CloudFront PoP (see the -1003 investigation), so the co-tenants waking
110# alongside us are not even ours to coordinate with.
111#
112# Only ever ADDS: the standing contract is that a cooldown is never shortened.
113#
114# Applied as a FRACTION of the wait, capped here, because the waits differ by
115# three orders of magnitude: a Bybit UID pool refills on a one-second window and
116# must not be held for five extra seconds, while a Binance -1003 ban runs for
117# minutes and can absorb seconds of spread. A flat jitter would either be useless
118# on the long waits or crippling on the short ones.
119_COOLDOWN_JITTER_MAX_MS = 5_000
120_COOLDOWN_JITTER_FRACTION = 0.10
122# After a cooldown releases, the weight budget climbs back to its normal ceiling
123# over this window instead of snapping to full.
124#
125# Resuming at full rate the instant a ban expires is what turns one ban into the
126# next: the first minute after release is exactly when the bucket is most likely
127# to still be sore, and a burst then is indistinguishable (to the exchange) from
128# the behaviour that caused the ban.
129_COOLDOWN_RAMP_MS = 30_000
131# Fraction of the normal ceiling allowed at the moment of release. Climbs
132# linearly to 1.0 across _COOLDOWN_RAMP_MS.
133_COOLDOWN_RAMP_START_FRACTION = 0.25
135# Only order counts are reconciled from headers. x-mbx-used-weight-1m has no
136# prefix constant on purpose: it is deliberately never read, because it does not
137# report the egress IP's spend. See the REQUEST_WEIGHT branch in
138# BinanceRateLimiter.__init__ for the measurements.
139_BINANCE_ORDER_COUNT_HEADER_PREFIX = "x-mbx-order-count"
141# Binance suffixes each usage header with the interval of the limiter it reports:
142# X-MBX-USED-WEIGHT-(intervalNum)(intervalLetter). The letters are the first
143# character of the exchangeInfo `interval` enum, so 1/MINUTE -> "1M" and
144# 10/SECOND -> "10S".
145_BINANCE_INTERVAL_LETTERS = {
146 "SECOND": "S",
147 "MINUTE": "M",
148 "HOUR": "H",
149 "DAY": "D",
150}
153def _binance_usage_header(prefix: str, interval: str, interval_num: int) -> str | None:
154 """
155 Usage header reporting the limiter at `interval_num` x `interval`, lowercased
156 to match what reqwest hands back. None when the interval is one we have no
157 letter for, so the caller can say so rather than build a header name that
158 will never match.
159 """
160 letter = _BINANCE_INTERVAL_LETTERS.get(str(interval).upper())
161 if letter is None:
162 return None
163 return f"{prefix}-{interval_num}{letter}".lower()
166class LocalRateLimitError(Exception):
167 """
168 Raised when the local rate limiter blocks a request
169 before it is sent to the exchange.
170 """
172 def __init__(self, message="Local rate limit exceeded"):
173 self.message = message
174 super().__init__(self.message)
177# This limiter is per process, while an exchange's heaviest budgets are metered
178# per source IP and every tenant on a shard shares one NAT address. Two things
179# stand in for the cross-process lock this used to ask for: IP-scoped budgets are
180# divided by Config.tenants_per_egress_ip, and on Binance the IP-wide spend is
181# read off x-mbx-used-weight-1m, which already reflects what co-tenants have
182# spent. They are deliberately tracked as two separate counters against two
183# separate ceilings — this process's own tally against its divided share, and
184# the IP-wide figure against the undivided IP budget. Folding one into the other
185# compares an IP-wide count against a per-tenant budget and caps the whole shard
186# at one tenant's share. A real shared token bucket would still beat both, since
187# the static split cannot lend unused budget between quiet and busy tenants.
188class RateLimiter(ABC):
189 # epoch ms until which all calls are locally blocked after a rate-limit
190 # error; see _arm_cooldown
191 retry_after: int = 0
193 # Ceiling on the body of guard()/reserve(), i.e. one exchange call
194 call_timeout_sec: float = EXCHANGE_CALL_TIMEOUT_SEC
196 def __init__(
197 self,
198 config: ConfigManager,
199 ):
200 self.config = config
201 self.soft_limit_percentage = config.config.soft_limit_percent
202 # Processes sharing this egress IP. IP-scoped budgets are split this
203 # many ways; account-scoped ones are not. See Config for why.
204 self.tenants_per_egress_ip = config.config.tenants_per_egress_ip
205 # Per-pool FIFO queue for reserve(); asyncio.Lock wakes waiters in
206 # acquisition order, so only one counts down at a time (no stampede)
207 self._reserve_locks: dict[Any, asyncio.Lock] = {}
208 # Per-pool count of callers currently in reserve(); guard() yields
209 # while this is > 0 so reserved calls take priority
210 self._waiters: dict[Any, int] = {}
212 @abstractmethod
213 async def init(self): ...
215 @abstractmethod
216 def get_synced_time_ms(self) -> int:
217 """
218 Current exchange-synced time in epoch ms
219 """
220 ...
222 @asynccontextmanager
223 @abstractmethod
224 async def guard(self, endpoint: Endpoints) -> AsyncGenerator[None, None]:
225 """
226 Context manager to wrap exchange calls with rate limit checks.
228 Usage:
229 async with self.rate_limiter.guard(Endpoints.PLACE_ORDER):
230 await exchange.place_order(...)
232 Raises:
233 LocalRateLimitError: If local limits are exhausted (Pre-check).
234 """
235 yield
237 def _reserve_lock(self, key: Any) -> asyncio.Lock:
238 lock = self._reserve_locks.get(key)
239 if lock is None:
240 lock = asyncio.Lock()
241 self._reserve_locks[key] = lock
242 return lock
244 def _waiters_repr(self) -> str:
245 """Per-pool reserve() queue depth, for the stats log; empty if idle."""
246 active = {k: n for k, n in self._waiters.items() if n > 0}
247 if not active:
248 return ""
249 parts = ", ".join(f"{getattr(k, 'name', k)}={n}" for k, n in active.items())
250 return f" Reserving[{parts}]"
252 @asynccontextmanager
253 async def reserve(self, endpoint: Endpoints) -> AsyncGenerator[None, None]:
254 """
255 Like guard(), but waits for a slot instead of failing fast when the
256 pool is full. While a caller is queued here, guard() callers on the
257 same pool yield, so a reserved (delta-critical) call gets priority
258 when capacity frees.
260 Waiters on a pool are serialised by a lock, so they acquire slots in
261 FIFO order and only one is ever counting down — no wake-up stampede.
263 Usage:
264 async with self.rate_limiter.reserve(Endpoints.GET_OPEN_ORDERS):
265 await exchange.get_open_orders(...)
267 Raises:
268 LocalRateLimitError: on wait timeout, or immediately while
269 retry_after is active — so callers keep their guard fallback.
270 TimeoutError: if the guarded call outlives call_timeout_sec.
271 """
272 key = self._pool_key(endpoint)
273 deadline = time.monotonic() + RESERVE_TIMEOUT_SEC
274 self._waiters[key] = self._waiters.get(key, 0) + 1
275 try:
276 async with self._reserve_lock(key):
277 while True:
278 if self.retry_after >= self.get_synced_time_ms():
279 raise LocalRateLimitError(
280 f"reserve aborted, rate limiter cooling down {self}"
281 )
282 if self._has_capacity(endpoint):
283 self.record_usage(endpoint)
284 break
285 delay = self._next_free_delay(endpoint)
286 if time.monotonic() + delay > deadline:
287 raise LocalRateLimitError(
288 f"reserve timed out waiting for a slot {self}"
289 )
290 await asyncio.sleep(max(delay, _MIN_RESERVE_SLEEP))
291 finally:
292 self._waiters[key] -= 1
293 try:
294 async with asyncio.timeout(self.call_timeout_sec):
295 yield
296 except Exception as e:
297 self._handle_call_error(e, endpoint)
298 raise e
299 else:
300 self._on_call_success(endpoint)
301 @abstractmethod
302 def _pool_key(self, endpoint: Endpoints) -> Any:
303 """Key identifying the contended pool, for reserve queueing/priority."""
304 ...
306 @abstractmethod
307 def _has_capacity(self, endpoint: Endpoints, **kwargs) -> bool:
308 """
309 Raw capacity check for one call, ignoring retry_after and the reserve
310 yield rule (reserve() and check_limits() layer those on top).
311 """
312 ...
314 @abstractmethod
315 def _next_free_delay(self, endpoint: Endpoints) -> float:
316 """Seconds until the pool frees at least one slot for this endpoint."""
317 ...
319 @abstractmethod
320 def _handle_call_error(
321 self, e: Exception, endpoint: Endpoints | None = None
322 ) -> None:
323 """Fold an exchange rate-limit error into local retry_after state."""
324 ...
326 @abstractmethod
327 async def on_resync_time(self):
328 """
329 To resync with server time
330 """
331 ...
333 @abstractmethod
334 def reset_limits(self):
335 """
336 To reset limits based on exchange rules
337 """
338 ...
340 @abstractmethod
341 def check_limits(self, endpoint: Endpoints, **kwargs) -> bool:
342 """
343 To determine whether to allow the endpoint request
344 """
345 ...
347 @abstractmethod
348 def record_usage(self, endpoint: Endpoints, **kwargs):
349 """
350 To record usage once endpoint request was successful
351 """
352 ...
354 @abstractmethod
355 def local_cache_error(self, headers: dict[str, Any], **kwargs: Any) -> None:
356 """
357 To record usage once endpoint request was failure, local store has been desyncronized
359 Subclasses may take extra keyword-only detail from the error (Binance
360 needs the body code and the ban deadline, which are not in the headers).
361 """
362 ...
364 def _on_call_success(self, endpoint: Endpoints) -> None:
365 """
366 Hook for exchanges that can reconcile local state from a successful
367 response (e.g. Bybit's rate-limit headers). No-op by default.
368 """
370 def _arm_cooldown(
371 self,
372 *deadlines_ms: int | None,
373 reason: str,
374 blind_cooldown_ms: int = _BLIND_COOLDOWN_MS,
375 ) -> None:
376 """
377 Block every call until the latest of `deadlines_ms`.
379 Rules that apply whichever exchange armed it:
381 - An existing cooldown is never shortened. A later response can arrive
382 carrying a nearer deadline (a 429's Retry-After after a 418's ban
383 deadline, say) and must not release the block early.
384 - A deadline beyond _MAX_TRUSTED_COOLDOWN_MS (four days, comfortably
385 past Binance's longest real ban) is discarded as garbage rather than
386 trusted, so one malformed header cannot park the OMS indefinitely.
387 - If no usable deadline survives, back off for `blind_cooldown_ms`
388 anyway. The exchange has said it is over budget, and continuing to
389 send is what escalates a throttle into a ban. Callers whose pool
390 refills quickly (a Bybit UID pool, on a one-second window) pass a
391 correspondingly short fallback.
392 """
393 now = self.get_synced_time_ms()
394 horizon = now + _MAX_TRUSTED_COOLDOWN_MS
395 usable = [d for d in deadlines_ms if d is not None and now < d <= horizon]
396 for rejected in (d for d in deadlines_ms if d is not None and d > horizon):
397 logger.warning(
398 f"[RATE_LIMIT] Ignoring implausible cooldown deadline {rejected} "
399 f"({(rejected - now) / 3_600_000:.1f}h out)"
400 )
401 exchange_supplied = bool(usable)
402 if not usable:
403 usable = [now + blind_cooldown_ms]
404 logger.warning(
405 f"[RATE_LIMIT] {reason} carried no usable deadline, "
406 f"backing off {blind_cooldown_ms / 1000:.1f}s"
407 )
408 # Jitter only ever delays, so the never-shorten rule still holds. Applied
409 # to the candidate rather than to self.retry_after so re-arming does not
410 # stack jitter on jitter.
411 deadline = max(usable)
412 # Jitter only a deadline the EXCHANGE supplied. That is the one every
413 # co-tenant sharing the bucket also parsed, so it is the only one they
414 # would wake on together. A blind fallback is chosen locally, from the
415 # moment this process happened to see the error, so it is already
416 # unsynchronised and jittering it would just delay recovery for nothing.
417 spread = (
418 min(
419 _COOLDOWN_JITTER_MAX_MS,
420 int((deadline - now) * _COOLDOWN_JITTER_FRACTION),
421 )
422 if exchange_supplied
423 else 0
424 )
425 jittered = deadline + (random.randint(0, spread) if spread > 0 else 0)
426 self.retry_after = max(self.retry_after, jittered)
427 logger.warning(
428 f"[RATE_LIMIT] {reason}: cooling down for "
429 f"{(self.retry_after - now) / 1000:.1f}s {self}"
430 )
432 @abstractmethod
433 def __repr__(self) -> str: ...
436class BinanceRateLimiter(RateLimiter):
437 def __init__(self, config: ConfigManager):
438 if not isinstance(config.exchange, BinanceLinearClient):
439 raise Exception("Exchange mismatch with rate limiter")
440 super().__init__(config)
441 exchange_info = config.exchange.exchange_info
442 if exchange_info is None:
443 raise Exception("Init the config before passing into Rate Limiter")
445 rateLimits: list[dict[str, Any]] | None = exchange_info.get("rateLimits")
446 if rateLimits is None:
447 raise Exception("Check Binance Api, rateLimits shouldn't be None here")
449 limit_profile = BinanceLimitProfile(
450 request_weight_limit_per_minute=0,
451 order_limit_per_minute=0,
452 order_limit_per_10_sec=0,
453 )
455 # Header name -> the limit_profile/current_limit_state field it reports
456 # usage for. Built from the rules below rather than hard-coded, because
457 # Binance names each header after the limiter it belongs to: "Every
458 # request will contain X-MBX-USED-WEIGHT-(intervalNum)(intervalLetter)
459 # ... for all request rate limiters defined". Today USDⓈ-M advertises
460 # REQUEST_WEIGHT 1/MINUTE and ORDERS at both 1/MINUTE and 10/SECOND, so
461 # this resolves to x-mbx-used-weight-1m, x-mbx-order-count-1m and
462 # x-mbx-order-count-10s -- but deriving it means the limit and the header
463 # reporting its usage can never disagree about the interval.
464 usage_headers: dict[str, str] = {}
466 for rule in rateLimits:
467 try:
468 limit_type = rule["rateLimitType"]
469 interval = rule["interval"]
470 interval_num = rule["intervalNum"]
471 limit = rule["limit"]
472 except Exception as e:
473 logger.error(
474 f"[BinanceRateLimiter] Failed parsing for {rule} due to {e}"
475 )
476 continue
478 if (
479 limit_type == "REQUEST_WEIGHT"
480 and interval == "MINUTE"
481 and interval_num == 1
482 ):
483 # REQUEST_WEIGHT is metered per source IP, so it is the one
484 # budget that has to be shared with every co-tenant behind the
485 # same NAT gateway. Floor, and never below 1, so a
486 # generously-sized shard still leaves each tenant able to send.
487 limit_profile.request_weight_limit_per_minute = max(
488 1,
489 int(
490 limit * self.soft_limit_percentage / self.tenants_per_egress_ip
491 ),
492 )
493 # Deliberately no header reconciliation for weight. Measured on a
494 # live shard: x-mbx-used-weight-1m does not report this egress
495 # IP's spend. Public endpoints answer with our own small figure
496 # (/fapi/v1/time and /fapi/v1/exchangeInfo returned 8 and 9),
497 # while signed account endpoints answer from some larger shared
498 # pool in the same breath (/fapi/v3/positionRisk 1389,
499 # /fapi/v2/balance 2293), and a /fapi/v1/time immediately after
500 # dropped back to 10. The signed figures also swing about
501 # tenfold between samples while our own traffic is unchanged.
502 # Presence of an API key makes no difference -- one endpoint
503 # probed with and without it returned 4/5, 6/7, 2/3, 4/5 -- so
504 # the split is by endpoint family, not by authentication.
505 #
506 # Since the OMS reads signed endpoints constantly, adopting that
507 # number meant a max() over unrelated readings, and it was
508 # gating request admission: live readings sawtoothed to 3337 and
509 # a direct probe hit 4433 against a 4800 refusal threshold. One
510 # more step and the OMS would have stopped trading on a quantity
511 # that measures nothing about us. Binance's own 429/418 plus
512 # Retry-After is the authoritative "you exceeded" signal, and it
513 # is already handled in note_error(); the local tally below is
514 # what this process controls. Nothing here needs the header.
515 continue
516 elif limit_type == "ORDERS" and interval == "MINUTE" and interval_num == 1:
517 # Order counts are metered per account, so each tenant's own API
518 # key gets the whole budget; dividing would throttle for no
519 # reason.
520 limit_profile.order_limit_per_minute = int(
521 limit * self.soft_limit_percentage
522 )
523 field = "order_limit_per_minute"
524 prefix = _BINANCE_ORDER_COUNT_HEADER_PREFIX
525 elif limit_type == "ORDERS" and interval == "SECOND" and interval_num == 10:
526 limit_profile.order_limit_per_10_sec = int(
527 limit * self.soft_limit_percentage
528 )
529 field = "order_limit_per_10_sec"
530 prefix = _BINANCE_ORDER_COUNT_HEADER_PREFIX
531 else:
532 logger.warning(f"Unknown rate limit {rule}")
533 continue
535 header = _binance_usage_header(prefix, interval, interval_num)
536 if header is None:
537 logger.error(
538 f"[BinanceRateLimiter] No interval letter known for {interval!r}; "
539 f"{field} will be tracked from local estimates only"
540 )
541 else:
542 usage_headers[header] = field
544 # A REQUEST_WEIGHT rule we do not understand leaves the budget at 0, and
545 # _has_capacity then refuses every single call — an OMS that runs, logs a
546 # warning, and silently never trades. Fail at startup instead, where the
547 # crash names the cause.
548 if limit_profile.request_weight_limit_per_minute <= 0:
549 raise Exception(
550 "Binance exchangeInfo declared no REQUEST_WEIGHT limit this "
551 f"limiter understands, so every request would be refused. "
552 f"rateLimits={rateLimits}"
553 )
555 self.limit_profile = limit_profile
556 self._usage_headers = usage_headers
557 logger.info(
558 f"[BinanceRateLimiter] Weight(1m) budget "
559 f"{limit_profile.request_weight_limit_per_minute} "
560 f"(this tenant's share, split "
561 f"{self.tenants_per_egress_ip} way(s); tracked locally, because "
562 f"x-mbx-used-weight-1m does not report this IP's spend), "
563 f"Orders(1m) {limit_profile.order_limit_per_minute}, "
564 f"Orders(10s) {limit_profile.order_limit_per_10_sec} (account-scoped). "
565 f"Reconciling usage from {sorted(usage_headers)}"
566 )
567 self.current_limit_state = BinanceLimitProfile(
568 request_weight_limit_per_minute=0,
569 order_limit_per_minute=0,
570 order_limit_per_10_sec=0,
571 )
572 self.exchange = config.exchange
573 self.exchange_time_offset = 0
575 self.last_reset_10s_timestamp = 0
576 self.last_reset_1m_timestamp = 0
578 # to block all calls until
579 self.retry_after = 0
581 async def init(self):
582 await self.on_resync_time()
583 self.last_reset_10s_timestamp = self.get_synced_time_ms() // 10000 # 10 seconds
584 self.last_reset_1m_timestamp = self.get_synced_time_ms() // 60000 # 1 minute
586 @asynccontextmanager
587 async def guard(self, endpoint: Endpoints) -> AsyncGenerator[None, None]:
588 if not self.check_limits(endpoint=endpoint):
589 raise LocalRateLimitError(f"Failed due to rate limits {self}")
591 self.record_usage(endpoint=endpoint)
592 try:
593 # See EXCHANGE_CALL_TIMEOUT_SEC: nothing below this layer bounds the
594 # call, and a hung one silently retires the cron job that made it.
595 async with asyncio.timeout(self.call_timeout_sec):
596 yield
597 except Exception as e:
598 self._handle_call_error(e, endpoint)
599 raise e
600 else:
601 self._on_call_success(endpoint)
603 def _handle_call_error(
604 self, e: Exception, endpoint: Endpoints | None = None
605 ) -> None:
606 if not isinstance(e, BinanceError) or not is_binance_rate_limit_error(e):
607 return
608 self.local_cache_error(
609 e.response_headers if e.response_headers else {},
610 code=e.code,
611 banned_until_ms=binance_banned_until_ms(str(e)),
612 )
614 def _on_call_success(self, endpoint: Endpoints) -> None:
615 headers = getattr(self.exchange, "last_response_headers", None)
616 if headers:
617 self._reconcile_from_headers(headers)
619 def _reconcile_from_headers(self, headers: dict[str, Any]) -> None:
620 """
621 Adopt Binance's own usage counters from the response just returned.
623 Order counts only. The weight header is never registered in
624 _usage_headers — see the REQUEST_WEIGHT branch in __init__ for the
625 measurements — so nothing here reads it and weight stays on the local
626 tally that record_usage() maintains.
628 Order counts are metered per account, so the header and the local tally
629 really are two readings of one number and the header simply wins.
631 Takes the max rather than overwriting: record_usage() charges a call
632 before it is sent, while the header only reflects requests Binance has
633 already counted, so overwriting would refund every in-flight
634 reservation.
636 Source: https://developers.binance.com/docs/derivatives/usds-margined-futures/general-info
637 """
638 state = self.current_limit_state
639 # reset_limits() first, or a counter this rolls forward can be zeroed
640 # immediately afterwards by a window boundary that has already passed.
641 window_before = self.last_reset_1m_timestamp
642 self.reset_limits()
643 if self.last_reset_1m_timestamp != window_before:
644 # Our window rolled while this response was in flight, so the header
645 # describes Binance's *previous* minute. Adopting a nearly-full
646 # previous-minute count into a freshly zeroed window would block
647 # weighted calls for up to a full minute for no reason. Skip it; the
648 # next response re-establishes the figure milliseconds later.
649 return
651 # Header names come from the rateLimits rules parsed in __init__, so each
652 # one lines up with the budget it reports usage against. An absent header
653 # just means no reading this time (order counts only ride on order
654 # responses), and the local estimate stands.
655 for header, field in self._usage_headers.items():
656 reported = _int_header(headers, header)
657 if reported is None:
658 continue
659 setattr(state, field, max(getattr(state, field), reported))
661 def _pool_key(self, endpoint: Endpoints) -> Any:
662 # Binance draws almost everything from the shared weight budget, so a
663 # single queue suffices; reserve on Binance usually times out anyway
664 return "binance_weight"
666 async def on_resync_time(self):
667 endpoint = Endpoints.GET_SERVER_TIME
668 try:
669 async with self.guard(endpoint=endpoint):
670 exchange_time = await self.exchange.get_server_time()
671 current_time = int(time.time() * 1000)
672 self.exchange_time_offset = current_time - exchange_time
673 except Exception as e:
674 logger.warning(f"[ON_RESYNC_TIME] {e}")
676 def get_synced_time_ms(self) -> int:
677 current_time = int(time.time() * 1000)
678 return current_time - self.exchange_time_offset
680 def _next_free_delay(self, endpoint: Endpoints) -> float:
681 # Weight and 1m orders reset on the minute boundary, 10s orders on the
682 # 10s boundary. Return the sooner one and let the reserve loop recheck;
683 # on Binance this usually exceeds the reserve timeout and degrades to
684 # guard behaviour, which is acceptable.
685 now = self.get_synced_time_ms()
686 next_1m = ((now // 60000) + 1) * 60000
687 next_10s = ((now // 10000) + 1) * 10000
688 return (min(next_1m, next_10s) - now) / 1000.0
690 def reset_limits(self):
691 """
692 Will reset limits based on time intervals
694 Source: https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/rate-limits
695 """
696 synced_time = self.get_synced_time_ms()
697 synced_time_10s = synced_time // 10000 # 10 seconds
698 synced_time_1m = synced_time // 60000 # 1 minute
699 # if different means next interval has passed
700 if synced_time_10s != self.last_reset_10s_timestamp:
701 self.last_reset_10s_timestamp = synced_time_10s
702 self.current_limit_state.order_limit_per_10_sec = 0
703 if synced_time_1m != self.last_reset_1m_timestamp:
704 self.last_reset_1m_timestamp = synced_time_1m
705 self.current_limit_state.order_limit_per_minute = 0
706 self.current_limit_state.request_weight_limit_per_minute = 0
708 def find_cost_info(self, endpoint: Endpoints, **kwargs) -> tuple[int, int]:
709 """
710 (weight, orders) charged for one call to `endpoint`.
712 Endpoints marked DYNAMIC_WEIGHT are resolved here from the request
713 parameters, so the capacity check and the usage record can never
714 disagree about what a call costs — before this, only the capacity check
715 resolved them, and record_usage() would have added the raw -1 marker to
716 the counter, crediting weight back on every call.
717 """
718 cost_info = BINANCE_FUTURES_COSTS.get(endpoint)
719 if cost_info is None:
720 logger.error(
721 f"[CHECK_LIMITS] Severe error no cost info is defined for this endpoint: {endpoint.name}"
722 )
723 raise KeyError(f"Endpoint doesn't exist, {endpoint.name}")
725 weight_cost = cost_info["weight"]
726 if weight_cost == DYNAMIC_WEIGHT:
727 weight_cost = self._resolve_dynamic_weight(endpoint, **kwargs)
728 return (weight_cost, cost_info["orders"])
730 def _resolve_dynamic_weight(self, endpoint: Endpoints, **kwargs) -> int:
731 """
732 Weight for an endpoint whose cost depends on its parameters.
734 guard()/reserve() do not thread per-call parameters through, so in
735 practice depth reads resolve to OMS_DEPTH_LIMIT — the depth
736 OrderUtils.get_order_book actually requests, which is the OMS's only
737 depth call site. Charging anything else here undercounts (a cheaper
738 tier) or wastes budget (a heavier one); an explicit limit/depth kwarg,
739 if one is ever threaded, still wins.
740 """
741 if endpoint == Endpoints.GET_ORDERBOOK_SNAPSHOT:
742 limit = kwargs.get("limit", kwargs.get("depth"))
743 return get_depth_weight(limit if limit is not None else OMS_DEPTH_LIMIT)
744 logger.error(
745 f"[CHECK_LIMITS] {endpoint.name} is marked dynamic but has no weight "
746 f"rule; charging the heaviest known cost"
747 )
748 return max(get_depth_weight(BINANCE_DEPTH_WEIGHTS[-1][0]), 1)
750 def check_limits(self, endpoint: Endpoints, **kwargs) -> bool:
751 """
752 Whether a guard() call may proceed: blocked while retry_after is
753 active, and yields to anything queued in reserve() for this pool.
754 POST_ORDER/ Order Creation is exempted from this rule
755 """
756 # Absulute condition if retry after is active will not do anything
757 if self.retry_after >= self.get_synced_time_ms():
758 return False
759 # Yield to callers waiting in reserve() so reserved calls take priority
760 if self._waiters.get(self._pool_key(endpoint), 0) > 0:
761 return False
762 return self._has_capacity(endpoint, **kwargs)
764 def _effective_weight_ceiling(self) -> int:
765 """
766 The weight budget this instant, reduced while ramping back up after a
767 cooldown.
769 A cooldown ends at a timestamp the exchange chose, which every co-tenant
770 sharing the bucket also parsed, so the moment of release is the single
771 most contended instant in the cycle. Going straight back to the full
772 budget there is what makes one ban beget the next. Instead the ceiling
773 starts at _COOLDOWN_RAMP_START_FRACTION and climbs linearly to the real
774 budget over _COOLDOWN_RAMP_MS.
776 A limiter that has never cooled down (retry_after == 0) is never ramped.
777 """
778 ceiling = self.limit_profile.request_weight_limit_per_minute
779 if self.retry_after <= 0:
780 return ceiling
781 elapsed = self.get_synced_time_ms() - self.retry_after
782 if elapsed < 0 or elapsed > _COOLDOWN_RAMP_MS:
783 return ceiling
784 fraction = _COOLDOWN_RAMP_START_FRACTION + (
785 (1.0 - _COOLDOWN_RAMP_START_FRACTION) * (elapsed / _COOLDOWN_RAMP_MS)
786 )
787 # Never below 1, or a small budget would refuse everything and the ramp
788 # would be a stall rather than a ramp.
789 return max(1, int(ceiling * fraction))
791 def _has_capacity(self, endpoint: Endpoints, **kwargs) -> bool:
792 self.reset_limits()
793 try:
794 (weight_cost, order_cost) = self.find_cost_info(endpoint=endpoint, **kwargs)
795 # Check in decending order by timescale
796 # Checking REQUEST_WEIGHT
797 # One ceiling: this process's own spend against this tenant's share.
798 # There is deliberately no second, IP-wide check — the header it
799 # would have to come from does not report this IP's spend (see the
800 # REQUEST_WEIGHT branch in __init__), and refusing on it stalled
801 # trading on a quantity unrelated to our traffic. Co-tenant pressure
802 # is handled by dividing the budget, and real exhaustion by
803 # Binance's 429/418 and the cooldown in note_error().
804 current_weight = self.current_limit_state.request_weight_limit_per_minute
805 max_weight = self._effective_weight_ceiling()
806 if weight_cost != 0 and max_weight <= (current_weight + weight_cost):
807 logger.warning(
808 f"[CHECK_LIMITS] REQUEST_WEIGHT 1m reached this tenant's share\n{max_weight} <= {current_weight} + {weight_cost}"
809 )
810 return False
811 # Checking ORDERS 1m
812 current_orders_1m = self.current_limit_state.order_limit_per_minute
813 max_orders_1m = self.limit_profile.order_limit_per_minute
814 if order_cost != 0 and max_orders_1m <= (current_orders_1m + order_cost):
815 logger.warning(
816 f"[CHECK_LIMITS] ORDERS 1m reached its limit\n{max_orders_1m} <= {current_orders_1m} + {order_cost}"
817 )
818 return False
819 # Checking ORDERS 10s
820 current_orders_10s = self.current_limit_state.order_limit_per_10_sec
821 max_orders_10s = self.limit_profile.order_limit_per_10_sec
822 if order_cost != 0 and max_orders_10s <= (current_orders_10s + order_cost):
823 logger.warning(
824 f"[CHECK_LIMITS] ORDERS 10s reached its limit\n{max_orders_10s} <= {current_orders_10s} + {order_cost}"
825 )
826 return False
827 # Passed all checks
828 return True
829 except Exception as e:
830 logger.error(f"Failed to check limits due to, {e}")
831 return False
833 def record_usage(self, endpoint: Endpoints, **kwargs):
834 """
835 Update limit values after successful endpoint request
836 """
837 try:
838 (weight_cost, order_cost) = self.find_cost_info(endpoint=endpoint, **kwargs)
839 self.current_limit_state.order_limit_per_10_sec += order_cost
840 self.current_limit_state.order_limit_per_minute += order_cost
841 self.current_limit_state.request_weight_limit_per_minute += weight_cost
842 except Exception as e:
843 logger.error(f"Failed to record usage due to {e}")
845 def local_cache_error(
846 self,
847 headers: dict[str, Any],
848 *,
849 code: int | None = None,
850 banned_until_ms: int | None = None,
851 **_: Any,
852 ) -> None:
853 """
854 Use when Binance reports a rate-limit breach: HTTP 429 (budget
855 exhausted) or 418 (IP banned), both carrying body code -1003, or -1015
856 for an order-rate breach.
858 Whichever it is, the local counters have provably disagreed with the
859 exchange, so the budget the breach came from is marked spent.
860 reset_limits() clears it again at the next window boundary.
862 An order-rate breach is scoped to the account's order budget, so it only
863 blocks placement and leaves reads working. Anything else is weight- or
864 IP-scoped, and continuing to send during it is exactly what makes
865 Binance extend a ban, so every endpoint is blocked until it lapses.
866 """
868 logger.warning(f"[LOCAL CACHE ERROR] HEADERS {_redact_headers(headers)}")
870 if code in BINANCE_ORDER_RATE_LIMIT_CODES:
871 self.current_limit_state.order_limit_per_10_sec = (
872 self.limit_profile.order_limit_per_10_sec
873 )
874 self.current_limit_state.order_limit_per_minute = (
875 self.limit_profile.order_limit_per_minute
876 )
877 logger.warning(
878 f"[LOCAL CACHE ERROR] order rate exhausted (code {code}), "
879 f"orders blocked until the next window {self}"
880 )
881 return
883 self.current_limit_state.request_weight_limit_per_minute = (
884 self.limit_profile.request_weight_limit_per_minute
885 )
887 # Retry-After is seconds; the ban deadline parsed out of the -1003
888 # message is already epoch ms. A 418 often carries only the latter.
889 retry_after_sec = _int_header(headers, "Retry-After")
890 self._arm_cooldown(
891 banned_until_ms + _COOLDOWN_SAFETY_MS
892 if banned_until_ms is not None
893 else None,
894 self.get_synced_time_ms() + retry_after_sec * 1000 + _COOLDOWN_SAFETY_MS
895 if retry_after_sec is not None
896 else None,
897 reason=f"Binance rate limit (code {code})",
898 )
900 def __repr__(self) -> str:
901 retry_message = ""
902 if self.retry_after > self.get_synced_time_ms():
903 retry_message = f" [RETRYING_AFTER: {self.retry_after}]"
904 return (
905 f"<RateLimitState "
906 f"Weight(1m): {self.current_limit_state.request_weight_limit_per_minute}"
907 f"/{self.limit_profile.request_weight_limit_per_minute}, "
908 f"Orders(1m): {self.current_limit_state.order_limit_per_minute}, "
909 f"Orders(10s): {self.current_limit_state.order_limit_per_10_sec}"
910 f"{f'Retry-After: {self.retry_after}' if self.retry_after > self.get_synced_time_ms() else ''}"
911 f"{self._waiters_repr()}"
912 f"{retry_message}>"
913 )
916class BybitRateLimiter(RateLimiter):
917 def __init__(self, config: ConfigManager):
918 if not isinstance(config.exchange, BybitLinearClient):
919 raise Exception("Exchange mismatch with rate limiter")
921 super().__init__(config)
922 limit_profile = BybitLimitProfile.with_buffer(
923 buffer_pct=Decimal("1.0") - self.soft_limit_percentage,
924 tenants_per_egress_ip=self.tenants_per_egress_ip,
925 )
927 self.limit_profile = limit_profile
928 self.current_limit_state: Dict[BybitRateLimitPool, BybitLimitState] = {
929 limit_pool: BybitLimitState(timestamps=deque())
930 for limit_pool in BybitRateLimitPool
931 }
932 self.exchange = config.exchange
933 self.exchange_time_offset = 0
935 # to block all calls until
936 self.retry_after = 0
938 async def init(self):
939 await self.on_resync_time()
941 @asynccontextmanager
942 async def guard(self, endpoint: Endpoints) -> AsyncGenerator[None, None]:
943 if not self.check_limits(endpoint=endpoint):
944 raise LocalRateLimitError(f"Failed due to rate limits {self}")
946 self.record_usage(endpoint=endpoint)
947 try:
948 async with asyncio.timeout(self.call_timeout_sec):
949 yield
950 except BaseException as e:
951 self._handle_call_error(e, endpoint)
952 raise e
953 else:
954 self._on_call_success(endpoint)
956 def _handle_call_error(
957 self, e: BaseException, endpoint: Endpoints | None = None
958 ) -> None:
959 if isinstance(e, BybitError) and (e.http_status == 403 or e.retCode == 10006):
960 self.local_cache_error(e.response_headers if e.response_headers else {})
961 return
962 # Not a rate-limit signal, so the optimistic pre-call decrement in
963 # record_usage() described a request that never counted against the pool.
964 # Give it back.
965 #
966 # Without this the decrement is permanent: _on_call_success is the only
967 # place headers are adopted, so a failure leaves `limit` and `reset_ts`
968 # unset, and _uid_pool_snapshot's `if state.reset_ts and ...` can then
969 # never fire. Once remaining reached 0 the pool refused for the life of
970 # the process, even after the call started working again. Reported from
971 # production as "UID_WALLET: 0/None, never recovers" -- the wallet guard
972 # wrapped a ClickHouse write as well as the exchange call, so an aegis
973 # outage walked it to zero one aegis tick at a time.
974 #
975 # BaseException, not Exception, because a cancelled task must refund too.
976 self._refund_optimistic_decrement(endpoint)
978 def _refund_optimistic_decrement(self, endpoint: Endpoints | None) -> None:
979 if endpoint is None:
980 return
981 cost_info = BYBIT_FUTURES_COSTS.get(endpoint)
982 if cost_info is None:
983 return
984 state = self.current_limit_state[cost_info]
985 if state.remaining is None:
986 return
987 ceiling = (
988 state.limit
989 if state.limit is not None
990 else self.limit_profile.limits[cost_info]
991 )
992 state.remaining = min(ceiling, state.remaining + 1)
994 def _on_call_success(self, endpoint: Endpoints) -> None:
995 cost_info = BYBIT_FUTURES_COSTS.get(endpoint)
996 if cost_info is None:
997 return
998 headers = getattr(self.exchange, "last_response_headers", None)
999 if headers:
1000 self._reconcile_uid_pool(cost_info, headers)
1002 def _reconcile_uid_pool(
1003 self, pool: BybitRateLimitPool, headers: dict[str, Any]
1004 ) -> None:
1005 """
1006 Overwrite this UID pool's tracked capacity with the exchange's own
1007 rate-limit headers from the response that was just returned for it.
1009 Source: https://bybit-exchange.github.io/docs/v5/rate-limit
1010 """
1011 limit = _get_header(headers, "X-Bapi-Limit")
1012 status = _get_header(headers, "X-Bapi-Limit-Status")
1013 if limit is None or status is None:
1014 return
1015 reset_ts = _get_header(headers, "X-Bapi-Limit-Reset-Timestamp")
1016 state = self.current_limit_state[pool]
1017 state.limit = int(limit)
1018 state.remaining = int(status)
1019 if reset_ts is not None:
1020 state.reset_ts = int(reset_ts)
1022 def _pool_key(self, endpoint: Endpoints) -> Any:
1023 # Endpoints with a dedicated UID pool queue on it; the rest share the
1024 # IP-global pool that every call also counts against
1025 return BYBIT_FUTURES_COSTS.get(endpoint, BybitRateLimitPool.IP_GLOBAL)
1027 def _next_free_delay(self, endpoint: Endpoints) -> float:
1028 now = self.get_synced_time_ms()
1029 interval_ms = self.limit_profile.interval * 1000
1030 delay_ms = 0
1032 # IP_GLOBAL: rolling 1s window, no exchange header to draw on. Frees
1033 # when its oldest timestamp ages out.
1034 ip_limit = BybitRateLimitPool.IP_GLOBAL
1035 ip_state = self.current_limit_state[ip_limit]
1036 ip_pool_limit = self.limit_profile.limits[ip_limit]
1037 if ip_state.timestamps and len(ip_state.timestamps) + 1 > ip_pool_limit:
1038 delay_ms = max(delay_ms, ip_state.timestamps[0] + interval_ms - now)
1040 # UID_* pool: exchange reports its own reset time once exhausted.
1041 cost_info = BYBIT_FUTURES_COSTS.get(endpoint)
1042 if cost_info:
1043 effective_remaining, _ = self._uid_pool_snapshot(cost_info, now)
1044 if effective_remaining < 1:
1045 delay_ms = max(
1046 delay_ms, self.current_limit_state[cost_info].reset_ts - now
1047 )
1049 return max(delay_ms, 0) / 1000.0
1051 async def on_resync_time(self):
1052 endpoint = Endpoints.GET_SERVER_TIME
1053 try:
1054 async with self.guard(endpoint=endpoint):
1055 exchange_time = await self.exchange.get_server_time()
1056 current_time = int(time.time() * 1000)
1057 self.exchange_time_offset = current_time - exchange_time
1058 except Exception as e:
1059 logger.warning(f"[ON_RESYNC_TIME] {e}")
1061 def get_synced_time_ms(self) -> int:
1062 current_time = int(time.time() * 1000)
1063 return current_time - self.exchange_time_offset
1065 def reset_limits(self):
1066 """
1067 Rolls the IP_GLOBAL window. UID_* pools have no local window to roll
1068 any more; their capacity comes from the exchange's own headers (see
1069 _uid_pool_snapshot), refreshed lazily once their reset_ts passes.
1071 Source: https://bybit-exchange.github.io/docs/v5/rate-limit
1072 """
1073 synced_time = self.get_synced_time_ms()
1074 state = self.current_limit_state[BybitRateLimitPool.IP_GLOBAL]
1075 while state.timestamps and synced_time - state.timestamps[0] >= 1000:
1076 state.timestamps.popleft()
1078 def _uid_pool_snapshot(self, pool: BybitRateLimitPool, now: int) -> tuple[int, int]:
1079 """
1080 (effective_remaining, limit) for a UID_* pool as of `now`. Before the
1081 first response for this pool has been seen, capacity is assumed full
1082 at the hard-coded buffered limit. reset_ts is 0 until a real header
1083 sets it, so `now >= reset_ts` must not fire on that default — it
1084 would wipe out every optimistic decrement made between responses.
1085 """
1086 state = self.current_limit_state[pool]
1087 limit = (
1088 state.limit if state.limit is not None else self.limit_profile.limits[pool]
1089 )
1090 if state.remaining is None:
1091 return limit, limit
1092 if state.reset_ts and now >= state.reset_ts:
1093 return limit, limit
1094 return state.remaining, limit
1096 def check_limits(self, endpoint: Endpoints, **kwargs) -> bool:
1097 """
1098 Whether a guard() call may proceed: blocked while retry_after is
1099 active, and yields to anything queued in reserve() for this pool.
1100 POST_ORDER/ Order Creation is exempted from this rule
1101 """
1102 # Absulute condition if retry after is active will not do anything
1103 if self.retry_after >= self.get_synced_time_ms():
1104 return False
1105 # Yield to callers waiting in reserve() so reserved calls take priority
1106 if self._waiters.get(self._pool_key(endpoint), 0) > 0:
1107 return False
1108 return self._has_capacity(endpoint, **kwargs)
1110 def _has_capacity(self, endpoint: Endpoints, **kwargs) -> bool:
1111 self.reset_limits()
1112 try:
1113 cost_info = BYBIT_FUTURES_COSTS.get(endpoint)
1114 now = self.get_synced_time_ms()
1116 # IP rate limit takes priority; Bybit exposes no header for it,
1117 # so it stays on the local rolling window.
1118 ip_limit = BybitRateLimitPool.IP_GLOBAL
1119 if (
1120 len(self.current_limit_state[ip_limit].timestamps) + 1
1121 > self.limit_profile.limits[ip_limit]
1122 ):
1123 logger.warning(
1124 f"[CHECK_LIMITS] {ip_limit.name} reached its limit\n{self.limit_profile.limits[ip_limit]} <= {len(self.current_limit_state[ip_limit].timestamps) + 1}"
1125 )
1126 return False
1128 if cost_info:
1129 effective_remaining, limit = self._uid_pool_snapshot(cost_info, now)
1130 if effective_remaining < 1:
1131 logger.warning(
1132 f"[CHECK_LIMITS] {cost_info.name} reached its limit\n0 remaining of {limit} (X-Bapi-Limit-Status)"
1133 )
1134 return False
1135 # Passed all checks
1136 return True
1137 except Exception as e:
1138 logger.error(f"Failed to check limits due to, {e}")
1139 return False
1141 def record_usage(self, endpoint: Endpoints, **kwargs):
1142 """
1143 Update limit values after successful endpoint request
1144 """
1145 current_time = self.get_synced_time_ms()
1146 ip_limit = BybitRateLimitPool.IP_GLOBAL
1147 self.current_limit_state[ip_limit].timestamps.append(current_time)
1149 cost_info = BYBIT_FUTURES_COSTS.get(endpoint)
1150 if cost_info:
1151 # Optimistic pre-call decrement so concurrent callers on the same
1152 # pool don't all admit off one stale header snapshot;
1153 # _on_call_success() overwrites this with the exchange's own
1154 # count once the response for this call lands.
1155 state = self.current_limit_state[cost_info]
1156 effective_remaining, _ = self._uid_pool_snapshot(cost_info, current_time)
1157 state.remaining = max(0, effective_remaining - 1)
1159 def local_cache_error(self, headers: dict[str, Any], **_: Any) -> None:
1160 """
1161 Use when bybit api return 403 (IP rate limit exhausted) and retCode 10006 endpoint exhausted
1163 A UID pool breach blocks until that pool's own reset timestamp, which is
1164 usually well under a second away. Only an IP breach — which Bybit
1165 reports with no reset header at all — falls back to the 10 minute block,
1166 matching the ban it hands out.
1168 The header lookup has to be case-insensitive: reqwest, which cybotrade's
1169 HTTP client is built on, lowercases header names, so matching
1170 "X-Bapi-Limit-Reset-Timestamp" exactly never fired and every UID breach
1171 took the IP branch and blocked all trading for 10 minutes.
1172 """
1174 logger.warning(f"[LOCAL CACHE ERROR] HEADERS {_redact_headers(headers)}")
1175 uid_pool_reset_ms = _int_header(headers, "X-Bapi-Limit-Reset-Timestamp")
1176 if uid_pool_reset_ms is not None:
1177 # UID ENDPOINT EXHAUSTED
1178 self._arm_cooldown(
1179 uid_pool_reset_ms + 50, # safety buffer
1180 reason="Bybit UID pool exhausted",
1181 # The pool rolls on a one-second window, so a reset timestamp
1182 # that has already passed needs no more than that much backoff.
1183 blind_cooldown_ms=self.limit_profile.interval * 1000,
1184 )
1185 return
1186 # IP RATE LIMIT EXHAUSTED
1187 self._arm_cooldown(
1188 self.get_synced_time_ms() + (10 * 60 * 1000) + _COOLDOWN_SAFETY_MS,
1189 reason="Bybit IP rate limit exhausted",
1190 )
1192 def __repr__(self) -> str:
1193 retry_message = ""
1194 if self.retry_after > self.get_synced_time_ms():
1195 retry_message = f" [RETRYING_AFTER: {self.retry_after}]"
1197 def _pool_repr(pool: BybitRateLimitPool, state: BybitLimitState) -> str:
1198 if pool is BybitRateLimitPool.IP_GLOBAL:
1199 return f"{pool.name}: Usage: {len(state.timestamps)}"
1200 return f"{pool.name}: Remaining: {state.remaining}/{state.limit}"
1202 state_dump = ", ".join(
1203 _pool_repr(k, v) for k, v in self.current_limit_state.items()
1204 )
1205 return (
1206 f"<RateLimitState "
1207 f"{state_dump}"
1208 f"{f'Retry-After: {self.retry_after}' if self.retry_after > self.get_synced_time_ms() else ''}"
1209 f"{self._waiters_repr()}"
1210 f"{retry_message}>"
1211 )