From f418e7779b351d65b9dee04a2b566d1ebf1fc414 Mon Sep 17 00:00:00 2001 From: Michael Feng Date: Thu, 6 Aug 2026 23:26:15 -0700 Subject: [PATCH 1/2] (fix) sync throttler pair-templated rate limits on dynamic pair registration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Connectors are created with trading_pairs=[], so AsyncThrottler builds its rate limits from an empty pair list at init. Pairs registered dynamically never reach the throttler, and pair-scoped limit_ids (e.g. bybit's v5/order/create-{PAIR}, v5/position/set-leverage-{PAIR}) resolve to None: AttributeError 'NoneType' object has no attribute 'weight'. Adds UnifiedConnectorService.sync_pair_derived_state(), which appends the pair and mutates the existing throttler in place via add_rate_limits() (the instance WebAssistantsFactory captured at init — reassigning _throttler provably has no effect). Wired into every dynamic registration site: TradingService.add_market (hoisted before its early return so a transiently failed sync retries), ensure_data_connector_started, _add_trading_pair_to_tracker (covers the data-connector bootstrap path), and set_leverage (pair-scoped endpoint that bypassed registration). Fixes #207 Co-Authored-By: Claude Fable 5 --- services/perpetual_trading_service.py | 6 ++ services/trading_service.py | 30 ++++++---- services/unified_connector_service.py | 50 ++++++++++++++++- test/test_sync_pair_derived_state.py | 79 +++++++++++++++++++++++++++ 4 files changed, 151 insertions(+), 14 deletions(-) create mode 100644 test/test_sync_pair_derived_state.py 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..f788fc77 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.") + + # Sync pair-derived state (throttler limits, per-pair trading rules) BEFORE + # the early return below: a no-op when already synced, and it retries a + # previously failed sync even when the market is already tracked with a + # healthy order book — otherwise a transient failure (e.g. node outage + # during the rules fetch) would stick until restart (#207/#208). + 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, per-pair trading rules) is refreshed too — see issues #207 / #208. + 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..faf0f22a 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,47 @@ 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. + """ + # 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. + throttler = getattr(connector, "_throttler", None) + if ( + throttler is not None + and hasattr(throttler, "add_rate_limits") + and hasattr(connector, "rate_limits_rules") + ): + try: + throttler.add_rate_limits(connector.rate_limits_rules) + 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 +468,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..a8e2d29b --- /dev/null +++ b/test/test_sync_pair_derived_state.py @@ -0,0 +1,79 @@ +"""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"] From 7a51a8ca38abbc9cdb77e98abe5b8b1eb9df2799 Mon Sep 17 00:00:00 2001 From: Michael Feng Date: Thu, 6 Aug 2026 23:47:20 -0700 Subject: [PATCH 2/2] (fix) adversarial-review fixes: validate pairs before registration, guard property eval - Validate the pair against the connector's symbol map BEFORE appending to _trading_pairs, raising ValueError for unknown pairs. There is no rollback path, and a poisoned entry breaks every per-pair consumer: status polling raises KeyError on the symbol map inside gathers without return_exceptions (killing balance/position/order updates until restart), and per-pair trading-rules rebuilds fail for ALL pairs. Previously a typo'd pair via add_market or set_leverage was registered permanently. Already-registered pairs skip re-validation (the symbol map may be transiently unavailable). - Evaluate rate_limits_rules inside the try: it is a property that can itself raise, and hasattr() evaluated it outside the guard (any non-AttributeError escaping aborted registration; an internal AttributeError silently skipped the sync with zero logging). - Scope comments to what this PR implements (the rules-fetch retry rationale belongs to the stacked #212). Findings from adversarial review; tests added for rejection-without- registration and skip-revalidation-when-registered. Co-Authored-By: Claude Fable 5 --- services/trading_service.py | 12 ++++---- services/unified_connector_service.py | 41 +++++++++++++++++++++++---- test/test_sync_pair_derived_state.py | 40 ++++++++++++++++++++++++++ 3 files changed, 81 insertions(+), 12 deletions(-) diff --git a/services/trading_service.py b/services/trading_service.py index f788fc77..7ce176f5 100644 --- a/services/trading_service.py +++ b/services/trading_service.py @@ -132,11 +132,11 @@ async def add_market( if not connector: raise ValueError(f"Connector {connector_name} not available. Check credentials.") - # Sync pair-derived state (throttler limits, per-pair trading rules) BEFORE - # the early return below: a no-op when already synced, and it retries a - # previously failed sync even when the market is already tracked with a - # healthy order book — otherwise a transient failure (e.g. node outage - # during the rules fetch) would stick until restart (#207/#208). + # 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: @@ -230,7 +230,7 @@ async def _register_trading_pair_with_connector( Delegates to UnifiedConnectorService.sync_pair_derived_state so that state built from the pair list at connector init (throttler pair-templated rate - limits, per-pair trading rules) is refreshed too — see issues #207 / #208. + limits) is refreshed too — see issue #207. Args: connector: The connector instance (ExchangePyBase) diff --git a/services/unified_connector_service.py b/services/unified_connector_service.py index faf0f22a..3b12d2c2 100644 --- a/services/unified_connector_service.py +++ b/services/unified_connector_service.py @@ -427,7 +427,34 @@ async def sync_pair_derived_state(connector: ConnectorBase, trading_pair: str) - 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: @@ -437,15 +464,17 @@ async def sync_pair_derived_state(connector: ConnectorBase, trading_pair: str) - # 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. + # 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") - and hasattr(connector, "rate_limits_rules") - ): + 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 " diff --git a/test/test_sync_pair_derived_state.py b/test/test_sync_pair_derived_state.py index a8e2d29b..16c5f24a 100644 --- a/test/test_sync_pair_derived_state.py +++ b/test/test_sync_pair_derived_state.py @@ -77,3 +77,43 @@ class Minimal: 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"]