diff --git a/services/perpetual_trading_service.py b/services/perpetual_trading_service.py index b0b24485..da757f8e 100644 --- a/services/perpetual_trading_service.py +++ b/services/perpetual_trading_service.py @@ -66,6 +66,12 @@ async def set_leverage(self, account_name: str, connector_name: str, if not hasattr(connector, '_execute_set_leverage'): raise HTTPException(status_code=400, detail=f"Connector '{connector_name}' does not support leverage setting") + # Set-leverage endpoints can be pair-scoped (e.g. bybit's + # v5/position/set-leverage-{PAIR}); register the pair so the throttler + # learns its rate limit before the request — see issue #207. + from services.unified_connector_service import UnifiedConnectorService + await UnifiedConnectorService.sync_pair_derived_state(connector, trading_pair) + try: await connector._execute_set_leverage(trading_pair, leverage) message = f"Leverage for {trading_pair} set to {leverage} on {connector_name}" diff --git a/services/trading_service.py b/services/trading_service.py index 11041d61..7ce176f5 100644 --- a/services/trading_service.py +++ b/services/trading_service.py @@ -128,6 +128,17 @@ async def add_market( """ await self.ensure_connector(connector_name) + connector = self.connectors.get(connector_name) + if not connector: + raise ValueError(f"Connector {connector_name} not available. Check credentials.") + + # Register the pair and sync pair-derived state (throttler limits) BEFORE + # the early return below: a no-op when already synced, and running it here + # means an already-tracked market still re-syncs any state a previous + # attempt failed to apply (#207). Raises ValueError for pairs the + # connector does not recognize — nothing is registered in that case. + await self._register_trading_pair_with_connector(connector, trading_pair) + if connector_name not in self._markets: self._markets[connector_name] = set() @@ -151,11 +162,6 @@ async def add_market( self._markets[connector_name].add(trading_pair) - # Get connector from our account's connectors - connector = self.connectors.get(connector_name) - if not connector: - raise ValueError(f"Connector {connector_name} not available. Check credentials.") - # Initialize order book via MarketDataService (uses best available connector) logger.info(f"Initializing order book for {connector_name}/{trading_pair}") success = await self._market_data_service.initialize_order_book( @@ -170,9 +176,6 @@ async def add_market( logger.info(f"Order book initialized successfully for {connector_name}/{trading_pair}") - # Register trading pair with connector - self._register_trading_pair_with_connector(connector, trading_pair) - # Update balances to include tokens from new trading pair if hasattr(connector, '_update_balances'): try: @@ -217,7 +220,7 @@ async def remove_market( logger.info(f"Removed market {connector_name}/{trading_pair}") - def _register_trading_pair_with_connector( + async def _register_trading_pair_with_connector( self, connector: ConnectorBase, trading_pair: str @@ -225,12 +228,17 @@ def _register_trading_pair_with_connector( """ Register a trading pair with the connector's internal structures. + Delegates to UnifiedConnectorService.sync_pair_derived_state so that state + built from the pair list at connector init (throttler pair-templated rate + limits) is refreshed too — see issue #207. + Args: connector: The connector instance (ExchangePyBase) trading_pair: Trading pair to register """ - if trading_pair not in connector._trading_pairs: - connector._trading_pairs.append(trading_pair) + already_registered = trading_pair in (getattr(connector, "_trading_pairs", None) or []) + await self._connector_service.sync_pair_derived_state(connector, trading_pair) + if not already_registered: logger.debug(f"Registered {trading_pair} with connector {type(connector).__name__}") # ======================================== diff --git a/services/unified_connector_service.py b/services/unified_connector_service.py index 48f0cedf..3b12d2c2 100644 --- a/services/unified_connector_service.py +++ b/services/unified_connector_service.py @@ -244,9 +244,8 @@ async def ensure_data_connector_started( connector = self.get_data_connector(connector_name) try: - # Add trading pair before starting network - if trading_pair not in connector._trading_pairs: - connector._trading_pairs.append(trading_pair) + # Add trading pair and sync pair-derived state before starting network + await self.sync_pair_derived_state(connector, trading_pair) # Start network await connector.start_network() @@ -412,6 +411,76 @@ def _is_tracker_running(self, tracker) -> bool: return True return False + @staticmethod + async def sync_pair_derived_state(connector: ConnectorBase, trading_pair: str) -> None: + """Sync connector state that was derived from the trading-pair list at init. + + Connectors are created with ``trading_pairs=[]`` and pairs are registered + dynamically, but state built FROM that list during ``__init__`` is never + refreshed afterwards. ``AsyncThrottler`` pair-templated rate limits (issue + #207): connectors like bybit_perpetual build ``rate_limits_rules`` from + ``trading_pairs``, so a pair added later has no rate limit and pair-scoped + requests crash with ``AttributeError: 'NoneType' object has no attribute + 'weight'``. The throttler must be mutated in place — + ``WebAssistantsFactory`` captured the instance at connector init, so + reassigning ``connector._throttler`` has no effect. + + Idempotent — safe to call on every dynamic pair registration, including + pairs that arrive via the data-connector bootstrap path. + + Raises ValueError when the connector's symbol map does not know the pair: + an unresolvable pair must never enter ``_trading_pairs``. There is no + rollback path, and a poisoned entry breaks every consumer that iterates + the list — per-pair status polling raises ``KeyError`` on the symbol map + inside gathers without ``return_exceptions`` (killing balance/position + updates until restart), and per-pair trading-rules rebuilds fail for ALL + pairs. Registered pairs are also enrolled in per-pair status polling — + deliberate for pairs about to be traded, but callers should not register + speculatively. + """ + # 0. Validate BEFORE registering. exchange_symbol_associated_to_pair + # raises for pairs the exchange does not know; connectors without a + # symbol map (Gateway, minimal test doubles) skip validation. + already_registered = trading_pair in (getattr(connector, "_trading_pairs", None) or []) + if not already_registered: + resolver = getattr(connector, "exchange_symbol_associated_to_pair", None) + if resolver is not None: + try: + await resolver(trading_pair) + except asyncio.CancelledError: + raise + except Exception as e: + raise ValueError( + f"Cannot register '{trading_pair}' on {type(connector).__name__}: " + f"the connector does not recognize this trading pair ({e})" + ) from e + + # 1. The pair list itself — everything below derives from it. + pairs = getattr(connector, "_trading_pairs", None) + if pairs is None: + connector._trading_pairs = [trading_pair] + elif trading_pair not in pairs: + pairs.append(trading_pair) + + # 2. Throttler rate limits (#207). add_rate_limits() skips known limit_ids. + # Synced on data connectors too — their REST fetches go through the same + # throttler and can hit pair-templated limit_ids. rate_limits_rules is + # evaluated inside the try: it is a property that can itself raise, and a + # failed sync must degrade to a warning, never abort registration. + throttler = getattr(connector, "_throttler", None) + if throttler is not None and hasattr(throttler, "add_rate_limits"): + try: + throttler.add_rate_limits(connector.rate_limits_rules) + except AttributeError: + logger.debug( + f"{type(connector).__name__} has no rate_limits_rules; throttler sync skipped" + ) + except Exception as e: + logger.warning( + f"Could not sync throttler rate limits for {trading_pair} on " + f"{type(connector).__name__}: {e}" + ) + async def _add_trading_pair_to_tracker( self, connector: ExchangePyBase, @@ -428,6 +497,10 @@ async def _add_trading_pair_to_tracker( 2. Otherwise, register the pair and start the tracker """ try: + # Sync pair-derived state (throttler limits, trading rules) regardless of + # which path below registers the order book — see sync_pair_derived_state. + await self.sync_pair_derived_state(connector, trading_pair) + # Safety check - gateway/AMM connectors don't have order book trackers if not hasattr(connector, 'order_book_tracker') or connector.order_book_tracker is None: logger.debug(f"Connector {type(connector).__name__} doesn't have order book tracker") diff --git a/test/test_sync_pair_derived_state.py b/test/test_sync_pair_derived_state.py new file mode 100644 index 00000000..16c5f24a --- /dev/null +++ b/test/test_sync_pair_derived_state.py @@ -0,0 +1,119 @@ +"""Tests for UnifiedConnectorService.sync_pair_derived_state (issue #207). + +Connectors are created with trading_pairs=[] and pairs registered dynamically, but +throttler pair-templated rate limits are built from that list at init. The helper +must re-sync them, idempotently, on every registration. +""" +import asyncio + +from hummingbot.core.api_throttler.async_throttler import AsyncThrottler +from hummingbot.core.api_throttler.data_types import RateLimit + +from services.unified_connector_service import UnifiedConnectorService + + +class FakePairLimitsConnector: + """Mimics a connector with pair-templated rate limits (bybit-style).""" + + def __init__(self, trading_pairs=None): + self._trading_pairs = trading_pairs + self._throttler = AsyncThrottler(rate_limits=self._build_limits(trading_pairs or [])) + + @staticmethod + def _build_limits(trading_pairs): + limits = [RateLimit(limit_id="global", limit=10, time_interval=1)] + for pair in trading_pairs: + limits.append(RateLimit(limit_id=f"order/create-{pair}", limit=5, time_interval=1)) + return limits + + @property + def rate_limits_rules(self): + return self._build_limits(self._trading_pairs or []) + + +def _run(coro): + return asyncio.run(coro) + + +def test_pair_appended(): + connector = FakePairLimitsConnector(trading_pairs=[]) + _run(UnifiedConnectorService.sync_pair_derived_state(connector, "BTC-USDT")) + assert connector._trading_pairs == ["BTC-USDT"] + + +def test_throttler_learns_pair_scoped_limit_in_place(): + connector = FakePairLimitsConnector(trading_pairs=[]) + # WebAssistantsFactory captures the throttler instance at init — the fix must + # mutate that same object, not replace it. + original_throttler = connector._throttler + + _run(UnifiedConnectorService.sync_pair_derived_state(connector, "BTC-USDT")) + + assert connector._throttler is original_throttler + limit_ids = {limit.limit_id for limit in original_throttler._rate_limits} + assert "order/create-BTC-USDT" in limit_ids + + +def test_idempotent_no_duplicate_limits(): + connector = FakePairLimitsConnector(trading_pairs=[]) + _run(UnifiedConnectorService.sync_pair_derived_state(connector, "BTC-USDT")) + _run(UnifiedConnectorService.sync_pair_derived_state(connector, "BTC-USDT")) + + assert connector._trading_pairs == ["BTC-USDT"] + limit_ids = [limit.limit_id for limit in connector._throttler._rate_limits] + assert limit_ids.count("order/create-BTC-USDT") == 1 + + +def test_none_trading_pairs_initialized(): + connector = FakePairLimitsConnector(trading_pairs=None) + _run(UnifiedConnectorService.sync_pair_derived_state(connector, "RLUSD-XRP")) + assert connector._trading_pairs == ["RLUSD-XRP"] + + +def test_minimal_connector_does_not_raise(): + class Minimal: + pass + + minimal = Minimal() + _run(UnifiedConnectorService.sync_pair_derived_state(minimal, "BTC-USDT")) + assert minimal._trading_pairs == ["BTC-USDT"] + + +def test_unknown_pair_rejected_and_not_registered(): + """A pair the connector's symbol map cannot resolve must never enter + _trading_pairs: there is no rollback, and a poisoned entry breaks per-pair + status polling and rules rebuilds for the connector's lifetime.""" + + class ValidatingConnector(FakePairLimitsConnector): + KNOWN = {"BTC-USDT"} + + async def exchange_symbol_associated_to_pair(self, trading_pair): + if trading_pair not in self.KNOWN: + raise KeyError(trading_pair) + return trading_pair.replace("-", "") + + connector = ValidatingConnector(trading_pairs=[]) + + try: + _run(UnifiedConnectorService.sync_pair_derived_state(connector, "BTC-USD")) + raise AssertionError("expected ValueError for unknown pair") + except ValueError: + pass + assert connector._trading_pairs == [] # nothing registered, nothing to roll back + + # A valid pair still registers normally afterwards + _run(UnifiedConnectorService.sync_pair_derived_state(connector, "BTC-USDT")) + assert connector._trading_pairs == ["BTC-USDT"] + + +def test_already_registered_pair_skips_validation(): + """Re-syncing an already-registered pair must not re-validate — the symbol + map may be temporarily unavailable, and the pair was validated on entry.""" + + class BrokenResolverConnector(FakePairLimitsConnector): + async def exchange_symbol_associated_to_pair(self, trading_pair): + raise ConnectionError("symbol map fetch failed") + + connector = BrokenResolverConnector(trading_pairs=["BTC-USDT"]) + _run(UnifiedConnectorService.sync_pair_derived_state(connector, "BTC-USDT")) + assert connector._trading_pairs == ["BTC-USDT"]