Coverage for adrs/oms/rate_limit/hyperliquid_limiter.py: 94%
138 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-29 04:56 +0800
1"""
2Rate limiting for Hyperliquid.
4Hyperliquid rate-limits on two independent axes:
6* an aggregated per-IP weight budget of 1200 per minute, which this limiter
7 models exactly from a static cost table, and
8* a cumulative per-address allowance of one request per USDC traded, with an
9 initial buffer of 10,000 requests, which it does not.
11The second is not a rate. It is a lifetime allowance that grows with traded
12volume, so mirroring it would mean polling volume (paying weight for the
13privilege) and acting on an estimate that is wrong either way: throttling when
14there was headroom, or failing to protect when there was not. It is enormous
15once trading -- $1M of volume earns 1M requests -- and its failure mode is a
16clean, detectable throttle to one request per 10s. So it is handled reactively
17via HyperliquidErrorPolicy, and counted here only to warn as the initial buffer
18depletes. Counted for observability; never enforced as a limit we cannot see.
19"""
21import asyncio
22import logging
23import time
24from collections import deque
25from contextlib import asynccontextmanager
26from typing import Any, AsyncGenerator, TYPE_CHECKING
28from cybotrade.exceptions import DeserializationError
29from cybotrade.hyperliquid import HyperliquidClient, HyperliquidError
31from adrs.oms.rate_limit.error_policy import is_hyperliquid_rate_limit_error
32from adrs.oms.rate_limit.exchange_limit_profiles import (
33 Endpoints,
34 HYPERLIQUID_ADDRESS_ACTION_BUFFER,
35 HYPERLIQUID_COSTS,
36 HYPERLIQUID_IP_WEIGHT_PER_MINUTE,
37 HYPERLIQUID_RATE_LIMIT_COOLDOWN_MS,
38 HYPERLIQUID_WEIGHT_WINDOW_SEC,
39 HyperliquidLimitProfile,
40 HyperliquidRateLimitPool,
41)
42from adrs.oms.rate_limit.rate_limiter import LocalRateLimitError, RateLimiter
44if TYPE_CHECKING:
45 from adrs.oms.config import ConfigManager
47logger = logging.getLogger(__name__)
49# Fraction of the initial address buffer at which to start warning.
50_ACTION_WARN_RATIO = 0.8
52# Endpoints whose real cost is amortised over a window rather than charged per
53# guard() call, because cybotrade serves the repeat calls from a client-side
54# cache and they never reach Hyperliquid. See _effective_weight().
55_AMORTISED_ENDPOINTS = frozenset({Endpoints.GET_SYMBOL_INFO})
58class HyperliquidRateLimiter(RateLimiter):
59 def __init__(self, config: "ConfigManager"):
60 if not isinstance(config.exchange, HyperliquidClient):
61 raise Exception("Exchange mismatch with rate limiter")
62 super().__init__(config)
64 # Weight is per IP, so a shard behind one NAT address must divide it;
65 # the address allowance is account-scoped and is not divided. This is
66 # the distinction the RateLimiter base class documents.
67 ceiling = int(
68 HYPERLIQUID_IP_WEIGHT_PER_MINUTE
69 * self.soft_limit_percentage
70 / self.tenants_per_egress_ip
71 )
72 self.limit_profile = HyperliquidLimitProfile(
73 request_weight_limit_per_minute=ceiling,
74 address_action_buffer=HYPERLIQUID_ADDRESS_ACTION_BUFFER,
75 )
76 # (epoch_ms, weight) charged, trimmed to the trailing window
77 self.weight_window: deque[tuple[int, int]] = deque()
78 self.address_actions = 0
79 self._warned_actions = False
80 # epoch_ms of the last full charge for each amortised endpoint; see
81 # _effective_weight(). None until the first charge in a fresh window.
82 self._amortised_charged_at: dict[Endpoints, int] = {}
84 logger.info(
85 f"[HYPERLIQUID_LIMITS] Weight(1m) {ceiling} (IP-scoped, split "
86 f"{self.tenants_per_egress_ip} ways). Address actions counted "
87 f"against a {HYPERLIQUID_ADDRESS_ACTION_BUFFER} buffer, not enforced."
88 )
90 # ---- clock ---------------------------------------------------------
92 async def init(self):
93 """
94 Nothing to initialise.
96 Hyperliquid publishes no server-time endpoint and nonces are
97 client-generated, so there is no clock to sync against.
98 """
99 return None
101 def get_synced_time_ms(self) -> int:
102 """Local clock; see init() for why there is nothing to sync to."""
103 return int(time.time() * 1000)
105 async def on_resync_time(self):
106 """No-op, for the same reason as init()."""
107 return None
109 # ---- window --------------------------------------------------------
111 def current_weight(self) -> int:
112 """Weight charged inside the trailing window, after trimming."""
113 self._trim_window()
114 return sum(weight for _, weight in self.weight_window)
116 def _cost(self, endpoint: Endpoints) -> dict[str, int]:
117 # KeyError rather than a zero default: an unpriced endpoint would be
118 # charged nothing, and undercounting is what gets an IP banned.
119 return HYPERLIQUID_COSTS[endpoint]
121 def _effective_weight(self, endpoint: Endpoints) -> int:
122 """
123 Weight to charge for this call: the table price, except for an endpoint
124 whose repeat calls cybotrade serves from its own cache.
126 GET_SYMBOL_INFO is the one such endpoint. config.py's
127 update_symbol_info() takes one guard per symbol, but cybotrade's
128 HyperliquidClient caches `metaAndAssetCtxs` (METADATA_TTL, 5 minutes)
129 and only the first call in a refresh reaches the exchange -- so a
130 20-symbol sweep issues exactly one real weight-20 request while the
131 per-symbol charge bills 400 of the minute's 1200. On the documented
132 14-tenant shard the ceiling is 68/min, which the phantom charge can
133 never fit: the sweep never completes, _symbol_info_refreshed_at is
134 never stamped, and every tick retries it -- a livelock that starves
135 PLACE_ORDER and CANCEL_ORDER. The Binance table documents the same
136 phantom-charge failure.
138 So the full weight is charged for the first call in a rolling window
139 and 0 for the rest of that window, capping the charge at once per
140 HYPERLIQUID_WEIGHT_WINDOW_SEC.
142 This is deliberately conservative rather than exact. It is coupled to
143 cybotrade's HyperliquidClient.METADATA_TTL: at 5 minutes, at most one
144 real call occurs per 5 minutes while this charges up to one per minute
145 -- a 5x overcharge, in the safe direction. If that TTL were ever
146 shortened below HYPERLIQUID_WEIGHT_WINDOW_SEC (60s), two real calls
147 could land inside one window and this would undercount, which is the
148 direction that gets the shared egress IP banned. Re-derive the cap
149 against METADATA_TTL before assuming it still holds.
150 """
151 weight = self._cost(endpoint)["weight"]
152 if endpoint not in _AMORTISED_ENDPOINTS:
153 return weight
154 charged_at = self._amortised_charged_at.get(endpoint)
155 if charged_at is None:
156 return weight
157 age_ms = self.get_synced_time_ms() - charged_at
158 if 0 <= age_ms < HYPERLIQUID_WEIGHT_WINDOW_SEC * 1000:
159 return 0
160 return weight
162 def _trim_window(self):
163 """Drop entries that have aged out of the trailing window."""
164 cutoff = self.get_synced_time_ms() - HYPERLIQUID_WEIGHT_WINDOW_SEC * 1000
165 while self.weight_window and self.weight_window[0][0] <= cutoff:
166 self.weight_window.popleft()
168 def reset_limits(self):
169 """
170 Fully clear local state, as opposed to the passive time-based trimming
171 _trim_window() does on every check. Public because the ABC requires it;
172 nothing in adrs calls it today.
174 WARNING: clearing the window discards weight the exchange is still
175 counting. Hyperliquid's budget is a trailing 60s window on its side and
176 there is no exchange-side signal to resync from (see init()), so a call
177 here does not make that spend go away -- it only makes this limiter
178 blind to it, and the next 60s of calls are admitted on top of weight
179 already spent. That is an over-admission into a budget shared by every
180 tenant on the egress IP. In particular it must not be called to recover
181 from a connection event: a reconnect tells us nothing about what the
182 exchange has already counted. Contrast _trim_window(), which only drops
183 entries that have genuinely aged out and is safe on every check.
184 """
185 self.weight_window.clear()
186 self.address_actions = 0
187 self._warned_actions = False
188 # Cleared too, so the next amortised call pays in full rather than
189 # riding a window this reset just erased.
190 self._amortised_charged_at.clear()
192 def _pool_key(self, endpoint: Endpoints) -> Any:
193 # One aggregated budget, so every call contends for the same pool.
194 return HyperliquidRateLimitPool.IP_WEIGHT
196 def _has_capacity(self, endpoint: Endpoints, **kwargs) -> bool:
197 self._trim_window()
198 # _effective_weight(), not the raw table price: record_usage() charges
199 # the amortised figure, so projecting the raw one here would deny a
200 # call that is about to cost nothing.
201 projected = sum(w for _, w in self.weight_window) + self._effective_weight(
202 endpoint
203 )
204 ceiling = self.limit_profile.request_weight_limit_per_minute
205 if projected > ceiling:
206 logger.warning(
207 f"[CHECK_LIMITS] Hyperliquid weight window is full: "
208 f"{projected} > {ceiling}"
209 )
210 return False
211 return True
213 def _next_free_delay(self, endpoint: Endpoints) -> float:
214 """Seconds until the oldest entry ages out and frees weight."""
215 self._trim_window()
216 if self._has_capacity(endpoint):
217 return 0.0
218 if not self.weight_window:
219 return 0.0
220 oldest_ts = self.weight_window[0][0]
221 expires_at = oldest_ts + HYPERLIQUID_WEIGHT_WINDOW_SEC * 1000
222 return max(0.0, (expires_at - self.get_synced_time_ms()) / 1000.0)
224 def check_limits(self, endpoint: Endpoints, **kwargs) -> bool:
225 """
226 Whether a guard() call may proceed: blocked while retry_after is
227 active, and yields to anything queued in reserve() for this pool.
228 """
229 # Absolute condition: while retry_after is active nothing proceeds. >=
230 # rather than >, matching reserve() and both sibling limiters -- on the
231 # exact millisecond of the deadline the hold still applies.
232 if self.retry_after >= self.get_synced_time_ms():
233 logger.warning(f"[CHECK_LIMITS] cooling down until {self.retry_after}")
234 return False
235 # Yield to callers waiting in reserve() so reserved calls take priority.
236 # Without this the four reserve() sites lose their whole mechanism:
237 # guard() callers keep taking the capacity a queued reserver is waiting
238 # for, so delta-critical position and open-order reads are starved.
239 if self._waiters.get(self._pool_key(endpoint), 0) > 0:
240 return False
241 return self._has_capacity(endpoint, **kwargs)
243 def record_usage(self, endpoint: Endpoints, **kwargs):
244 cost = self._cost(endpoint)
245 now = self.get_synced_time_ms()
246 weight = self._effective_weight(endpoint)
247 if weight and endpoint in _AMORTISED_ENDPOINTS:
248 # Opens the amortisation window for this endpoint; the rest of the
249 # sweep rides it at 0. See _effective_weight().
250 self._amortised_charged_at[endpoint] = now
251 self.weight_window.append((now, weight))
252 if cost["actions"]:
253 self.address_actions += cost["actions"]
254 self._maybe_warn_actions()
256 def _maybe_warn_actions(self) -> None:
257 buffer = self.limit_profile.address_action_buffer
258 if self._warned_actions or self.address_actions < buffer * _ACTION_WARN_RATIO:
259 return
260 self._warned_actions = True
261 logger.warning(
262 f"[HYPERLIQUID_LIMITS] {self.address_actions} exchange actions sent "
263 f"against an initial address buffer of {buffer}. The real allowance "
264 f"grows with traded volume and is not observable here; if the "
265 f"exchange starts throttling, that is why."
266 )
268 # ---- guard ---------------------------------------------------------
270 @asynccontextmanager
271 async def guard(self, endpoint: Endpoints) -> AsyncGenerator[None, None]:
272 if not self.check_limits(endpoint=endpoint):
273 raise LocalRateLimitError(f"Failed due to rate limits {self}")
275 self.record_usage(endpoint=endpoint)
276 try:
277 async with asyncio.timeout(self.call_timeout_sec):
278 yield
279 except BaseException as e:
280 self._handle_call_error(e, endpoint)
281 raise e
282 else:
283 self._on_call_success(endpoint)
285 # ---- error handling -----------------------------------------------
287 def _handle_call_error(
288 self, e: BaseException, endpoint: Endpoints | None = None
289 ) -> None:
290 """
291 Arm the local cooldown when Hyperliquid is throttling us.
293 Delegates the whole decision to local_cache_error(), which is the
294 ABC-designated place for folding a failure into local state. Keeping
295 one classifier means the two entry points cannot drift apart, and it
296 stops local_cache_error() being dead code reachable from nowhere.
298 Unlike the Bybit limiter there is no optimistic pre-call decrement to
299 refund: the weight window records what was actually sent, and a request
300 that failed still cost its weight at the exchange.
301 """
302 # Hyperliquid sends no rate-limit headers at all, so there is nothing
303 # to pass; the exception itself carries every signal there is.
304 if isinstance(e, Exception):
305 self.local_cache_error({}, error=e)
307 def local_cache_error(self, headers: dict[str, Any], **kwargs: Any) -> None:
308 """
309 Fold a rate-limit failure into local state.
311 Hyperliquid returns no rate-limit headers at all -- an action failure
312 arrives in the body of an HTTP 200 -- so `headers` is always empty here
313 and the signal comes from the exception. Callers pass it as `error=...`,
314 or as `message=...` when only the text is to hand.
316 Classification is delegated to is_hyperliquid_rate_limit_error() rather
317 than re-matched here, so the needles live in exactly one place; that
318 also picks up the HTTP 429 case, which has no Hyperliquid body to match
319 against.
320 """
321 error = kwargs.get("error")
322 if error is None:
323 message = str(kwargs.get("message") or "")
324 if not message:
325 return
326 error = HyperliquidError(message)
327 if not isinstance(error, Exception):
328 return
329 if isinstance(error, DeserializationError):
330 # Deliberately broad. cybotrade's HyperliquidClient._post_info /
331 # _post_exchange never look at the HTTP status: they hand the body
332 # straight to json.loads, so a 429's non-JSON body (an edge/proxy
333 # page, not a Hyperliquid payload) surfaces here as
334 # DeserializationError and as nothing else. At this layer a
335 # malformed response is therefore indistinguishable from a 429, so
336 # it is treated as a suspected throttle. Holding briefly on a read
337 # failure costs some latency; missing a real 429 means polling
338 # through the throttle and renewing an IP ban that takes down every
339 # tenant sharing the egress address.
340 logger.warning(
341 f"[HYPERLIQUID_LIMITS] undecodable response ({error}); treating "
342 f"it as a suspected throttle, since a 429 body reaches us as "
343 f"exactly this and nothing else."
344 )
345 self._arm_throttle_cooldown()
346 return
347 if is_hyperliquid_rate_limit_error(error):
348 self._arm_throttle_cooldown()
350 def _arm_throttle_cooldown(self) -> None:
351 deadline = self.get_synced_time_ms() + HYPERLIQUID_RATE_LIMIT_COOLDOWN_MS
352 self.retry_after = max(self.retry_after, deadline)
353 logger.warning(
354 f"[HYPERLIQUID_LIMITS] rate limited by the exchange; holding calls "
355 f"until {self.retry_after}. Either axis can produce this: the "
356 f"address allowance is cumulative and clears as traded volume "
357 f"accrues, while an IP weight overrun clears as the window drains."
358 )
360 # ---- repr ----------------------------------------------------------
362 def __repr__(self) -> str:
363 retry = ""
364 if self.retry_after > self.get_synced_time_ms():
365 retry = f" [RETRYING_AFTER: {self.retry_after}]"
366 return (
367 f"HyperliquidRateLimiter(weight: {self.current_weight()}"
368 f"/{self.limit_profile.request_weight_limit_per_minute}, "
369 f"actions: {self.address_actions}"
370 f"/{self.limit_profile.address_action_buffer}){retry}"
371 )