(fix) sync throttler pair-templated rate limits on dynamic pair registration

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}"

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,20 +220,25 @@ 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
     ):
         """
         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__}")
 
     # ========================================

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")

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"]