Coverage for adrs/oms/config.py: 60%
149 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-29 05:07 +0800
1from abc import abstractmethod
2import json
3import time
5from typing import TYPE_CHECKING
7from cybotrade import Symbol
8from pydantic import BaseModel, Field, ValidationError
9from decimal import Decimal
11from cybotrade.models import Exchange, SymbolInfo
12from cybotrade.io import ExchangeClient, ExchangeEvent
13from cybotrade.bybit import BybitLinearClient, BybitPrivateWS, BybitPublicWS
14from cybotrade.binance import BinanceLinearClient, BinancePrivateWS, BinancePublicWS
15from cybotrade.kucoin import KucoinLinearClient, KucoinPrivateWS
16from cybotrade.edgex import EdgeXClient, EdgeXPrivateWS
17from cybotrade.hyperliquid import (
18 HyperliquidClient,
19 HyperliquidPrivateWS,
20 HyperliquidPublicWS,
21)
23from adrs.oms.logging import PrefixedLogger
24from adrs.oms.rate_limit.exchange_limit_profiles import Endpoints
25from adrs.oms.rate_limit.error_policy import (
26 ExchangeErrorPolicy,
27 BybitErrorPolicy,
28 BinanceErrorPolicy,
29 DefaultErrorPolicy,
30 HyperliquidErrorPolicy,
31)
33if TYPE_CHECKING:
34 from adrs.oms.rate_limit.rate_limiter import RateLimiter
36# Symbol filters (tick size, lot step, notional bounds) change rarely; a
37# stale read self-heals via the order-reject path, so refreshing faster
38# than this only burns request weight
39SYMBOL_INFO_TTL_SEC = 15 * 60
42class Credentials(BaseModel):
43 exchange: Exchange
44 api_key: str
45 api_secret: str
46 api_passphrase: str | None = None
47 testnet: bool = False
48 demo: bool = False
50 def to_exchange_client(self) -> ExchangeClient:
51 match self.exchange:
52 case Exchange.BYBIT_LINEAR:
53 return BybitLinearClient(
54 api_key=self.api_key,
55 api_secret=self.api_secret,
56 testnet=self.testnet,
57 demo=self.demo,
58 )
59 case Exchange.BINANCE_LINEAR:
60 return BinanceLinearClient(
61 api_key=self.api_key,
62 api_secret=self.api_secret,
63 testnet=self.testnet,
64 )
65 case Exchange.KUCOIN_LINEAR:
66 if self.api_passphrase is None:
67 raise ValueError("'api_passphrase' is required for Kucoin")
69 return KucoinLinearClient(
70 api_key=self.api_key,
71 api_secret=self.api_secret,
72 api_passphrase=self.api_passphrase,
73 sandbox=self.testnet,
74 )
75 case Exchange.EDGEX:
76 return EdgeXClient(
77 account_id=self.api_key,
78 private_key=self.api_secret,
79 )
80 case Exchange.HYPERLIQUID:
81 # api_key is the MASTER account address, not a key. cybotrade
82 # can resolve it from userRole, but that is an awaited call and
83 # to_exchange_event() below is synchronous, so adrs requires it
84 # up front -- as Kucoin requires api_passphrase. Putting the
85 # agent wallet's own address here makes every state read return
86 # empty while orders still place, so it is validated, not
87 # defaulted.
88 if not self.api_key:
89 raise ValueError(
90 "'api_key' must be the Hyperliquid master account address"
91 )
92 return HyperliquidClient(
93 private_key=self.api_secret,
94 account_address=self.api_key,
95 vault_address=self.api_passphrase,
96 testnet=self.testnet,
97 )
98 case _:
99 raise Exception(f"Unsupported exchange {self.exchange}")
101 def to_exchange_event(self) -> ExchangeEvent:
102 match self.exchange:
103 case Exchange.BYBIT_LINEAR:
104 return BybitPrivateWS(
105 api_key=self.api_key,
106 api_secret=self.api_secret,
107 testnet=self.testnet,
108 demo=self.demo,
109 # Bybit only pushes topics that were subscribed, unlike
110 # Binance's listenKey stream which carries every event type.
111 topics=["order", "position"],
112 )
113 case Exchange.BINANCE_LINEAR:
114 return BinancePrivateWS(
115 api_key=self.api_key,
116 api_secret=self.api_secret,
117 testnet=self.testnet,
118 )
119 case Exchange.KUCOIN_LINEAR:
120 if self.api_passphrase is None:
121 raise ValueError("'api_passphrase' is required for Kucoin")
123 return KucoinPrivateWS(
124 api_key=self.api_key,
125 api_secret=self.api_secret,
126 api_passphrase=self.api_passphrase,
127 topics=["/contractMarket/tradeOrders"],
128 )
129 case Exchange.EDGEX:
130 return EdgeXPrivateWS(
131 account_id=self.api_key,
132 private_key=self.api_secret,
133 )
134 case Exchange.HYPERLIQUID:
135 # Takes an address, not credentials: Hyperliquid keys user
136 # subscriptions on the address and that data is public.
137 if not self.api_key:
138 raise ValueError(
139 "'api_key' must be the Hyperliquid master account address"
140 )
141 return HyperliquidPrivateWS(address=self.api_key, testnet=self.testnet)
142 case _:
143 raise Exception(f"Unsupported exchange {self.exchange}")
145 def to_public_exchange_event(self, symbols: list[str]) -> ExchangeEvent | None:
146 """
147 The public top-of-book adapter for this exchange, or None where there is
148 no adapter.
150 None is a supported configuration, not an error: Kucoin and EdgeX have no
151 public feed and read prices from REST exactly as every exchange did
152 before the feed existed. Raising here would break them.
154 testnet must match the credentials — the feed and the REST fallback have
155 to price against the same book.
156 """
157 match self.exchange:
158 case Exchange.BYBIT_LINEAR:
159 return BybitPublicWS(
160 symbols=symbols, testnet=self.testnet, demo=self.demo
161 )
162 case Exchange.BINANCE_LINEAR:
163 return BinancePublicWS(symbols=symbols, testnet=self.testnet)
164 case Exchange.HYPERLIQUID:
165 # HyperliquidPublicWS resolves each symbol to a coin via
166 # base_from_symbol(), which needs a real Symbol -- a plain
167 # string resolves to the whole pair as the coin name with no
168 # error raised, silently subscribing to nothing on the book.
169 return HyperliquidPublicWS(
170 symbols=[Symbol(s) for s in symbols], testnet=self.testnet
171 )
172 case _:
173 return None
175 def to_exchange_topic(self) -> str:
176 match self.exchange:
177 case Exchange.BYBIT_LINEAR:
178 return "bybit-linear"
179 case Exchange.BINANCE_LINEAR:
180 return "binance-linear"
181 # TODO Get actual values
182 case Exchange.KUCOIN_LINEAR:
183 return "kucoin-linear"
184 case Exchange.EDGEX:
185 return "edgex"
186 case Exchange.HYPERLIQUID:
187 return "hyperliquid"
188 case _:
189 raise Exception(f"Unsupported exchange {self.exchange}")
191 def to_error_policy(self) -> ExchangeErrorPolicy:
192 match self.exchange:
193 case Exchange.BYBIT_LINEAR:
194 return BybitErrorPolicy()
195 case Exchange.BINANCE_LINEAR:
196 return BinanceErrorPolicy()
197 case Exchange.HYPERLIQUID:
198 return HyperliquidErrorPolicy()
199 case _:
200 # No exchange-specific policy: keep legacy retry-everything behaviour
201 return DefaultErrorPolicy()
204class Config(BaseModel):
205 oms_id: str
206 portfolio_id: str
208 initial_balance: Decimal
209 leverage: Decimal
210 base_asset_to_symbol_table: dict[str, str]
212 credentials: Credentials
214 order_placement_interval: int
215 expiry_check: int
216 replace_best_bid_ask_time: int
217 max_limit_replace_interval: int
218 min_limit_replace_interval: int
219 # Attempts a backlogged order/cancel gets before it is abandoned. The retry
220 # cron runs every 2s and backs off by min(2 * attempt, 30)s, so 10 attempts
221 # is roughly 108 seconds of trying. Defaulted to the value production
222 # already runs, so a config that omits it behaves the same as the fleet
223 # rather than failing to load.
224 max_retries_allowed: int = 10
225 soft_limit_percent: Decimal
227 # How many OMS processes share this process's egress IP address.
228 #
229 # Exchanges meter their heaviest limits per source IP: Binance's
230 # REQUEST_WEIGHT (2400/min) and Bybit's IP_GLOBAL pool. A rate limiter is
231 # per process, so every tenant behind one NAT gateway used to claim the
232 # whole IP budget for itself — three tenants on a shard admitted 2.4x what
233 # the IP allows, and a full 14-tenant shard about 11x. They then banned each
234 # other, which is what made this look like a multi-tenancy problem rather
235 # than an accounting one.
236 #
237 # Set this to the shard's tenant cap (prime's COMPUTE_MAX_TENANTS_PER_SHARD)
238 # so IP-scoped budgets are divided that many ways. Account-scoped budgets —
239 # Binance's order counts, Bybit's UID pools — are per API key and are never
240 # divided. Dedicated-tier tenants own their IP, so 1 is correct there.
241 #
242 # This is a static worst-case split. On Binance the limiter also reconciles
243 # against x-mbx-used-weight-1m, which is itself IP-scoped and so already
244 # accounts for co-tenant traffic; this bound is what stops a process
245 # overshooting before its first response comes back.
246 tenants_per_egress_ip: int = Field(default=1, ge=1)
249class ConfigManager:
250 # Subclasses override to validate against an extended config schema,
251 # e.g. class BqFileConfigManager(FileConfigManager): config_cls = BqConfig
252 config_cls: type[Config] = Config
254 config: Config
255 exchange: ExchangeClient
256 symbol_infos: dict[Symbol, SymbolInfo]
258 def __init__(self, prefix: str = "[ConfigManager]"):
259 self.logger: PrefixedLogger = PrefixedLogger(prefix=prefix, name=__name__)
260 self.symbol_infos = {}
261 self._symbol_info_refreshed_at: float = 0.0 # monotonic, 0 = never
263 @abstractmethod
264 async def setup(self, *args, **kwars):
265 pass
267 @abstractmethod
268 async def load(self) -> Config:
269 pass
271 async def refresh(self):
272 old_credentials = self.config.credentials if hasattr(self, "config") else None
273 self.config = await self.load()
274 if old_credentials != self.config.credentials:
275 self.exchange = self.config.credentials.to_exchange_client()
277 async def update_symbol_info(
278 self, rate_limiter: "RateLimiter | None" = None, force: bool = False
279 ):
280 """
281 Refreshes at most once per SYMBOL_INFO_TTL_SEC unless forced
282 (config/symbol-table changes must not wait out the TTL).
283 """
284 if (
285 not force
286 and self.symbol_infos
287 and time.monotonic() - self._symbol_info_refreshed_at < SYMBOL_INFO_TTL_SEC
288 ):
289 return
290 for symbol in self.config.base_asset_to_symbol_table.values():
291 if rate_limiter is not None:
292 async with rate_limiter.guard(endpoint=Endpoints.GET_SYMBOL_INFO):
293 self.symbol_infos[
294 Symbol(symbol)
295 ] = await self.exchange.get_symbol_info(symbol=Symbol(symbol))
296 else: # bootstrap (setup) — no limiter yet, single startup burst
297 self.symbol_infos[Symbol(symbol)] = await self.exchange.get_symbol_info(
298 symbol=Symbol(symbol)
299 )
300 # Only stamped after a full pass: a mid-loop failure propagates and
301 # leaves the TTL expired, so the next tick retries
302 self._symbol_info_refreshed_at = time.monotonic()
305class FileConfigManager(ConfigManager):
306 def __init__(
307 self,
308 path: str,
309 ):
310 super().__init__(prefix="[FileConfigManager]")
311 self.path = path
313 async def setup(self):
314 await self.refresh()
315 await self.update_symbol_info()
317 async def load(self):
318 try:
319 with open(self.path, "r") as f:
320 data = json.load(f)
321 return self.config_cls.model_validate(obj=data)
322 except FileNotFoundError:
323 self.logger.error("Config file does not exist")
324 raise
325 except json.JSONDecodeError:
326 self.logger.error("Config file contains invalid json data")
327 raise
328 except ValidationError as e:
329 self.logger.error(f"'{self.path}' does not match the Config schema: {e}")
330 raise
331 except Exception as e:
332 self.logger.error(f"An unexpected error occurred: {e}")
333 raise
336class NATSConfigManager(ConfigManager):
337 def __init__(
338 self,
339 url: str,
340 bucket: str,
341 key: str,
342 user: str | None = None,
343 password: str | None = None,
344 ):
345 super().__init__(prefix="[NATSConfigManager]")
346 self.url = url
347 self.bucket = bucket
348 self.key = key
350 async def setup(self):
351 await self.refresh()
353 async def load(self):
354 raise NotImplementedError(
355 "NATSConfigManager.load is not implemented. "
356 "Subclass NATSConfigManager and override load() to fetch config from NATS."
357 )