feat: Add SubscriptionManagementSystem with dynamic subscribe/unsubscribe support across all exchanges

Introduce a new SubscriptionManagementSystem (SMS) to centralize subscription management with support for both subscribing and unsubscribing from market data streams. This enhancement enables dynamic strategy execution where data subscriptions can be added or removed at runtime.

Key changes:
- Add new SubscriptionManagementSystem class with queue-based subscription handling
- Implement unsubscribe methods for all data types (trade, bookl1, bookl2, kline, funding rate, index price, mark price)
- Add support for kline and volume kline aggregator unsubscription with proper cleanup
- Extend PublicConnector abstract base class with unsubscribe method signatures
- Update all exchange connectors (Binance, Bybit, OKX, Bitget, Hyperliquid) with unsubscribe implementations
- Integrate SMS into Engine for centralized subscription lifecycle management
- Add DataReady tracking for subscription readiness with timeout support
- Update version to 0.2.20

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: River.Shi

nexustrader/base/__init__.py

@@ -3,6 +3,7 @@
 from nexustrader.base.api_client import ApiClient
 from nexustrader.base.oms import OrderManagementSystem
 from nexustrader.base.ems import ExecutionManagementSystem
+from nexustrader.base.sms import SubscriptionManagementSystem
 from nexustrader.base.connector import (
     PublicConnector,
     PrivateConnector,
@@ -18,6 +19,7 @@
     "OrderManagementSystem",
     "ExecutionManagementSystem",
     "PublicConnector",
+    "SubscriptionManagementSystem",
     "PrivateConnector",
     "MockLinearConnector",
     "RetryManager",

nexustrader/base/connector.py

@@ -139,11 +139,21 @@ async def subscribe_trade(self, symbol: str | List[str]):
         """Subscribe to the trade data"""
         pass
 
+    @abstractmethod
+    async def unsubscribe_trade(self, symbol: str | List[str]):
+        """Unsubscribe from the trade data"""
+        pass
+
     @abstractmethod
     async def subscribe_bookl1(self, symbol: str | List[str]):
         """Subscribe to the bookl1 data"""
         pass
 
+    @abstractmethod
+    async def unsubscribe_bookl1(self, symbol: str | List[str]):
+        """Unsubscribe from the bookl1 data"""
+        pass
+
     @abstractmethod
     async def subscribe_kline(
         self,
@@ -159,6 +169,15 @@ async def subscribe_kline(
         """
         pass
 
+    @abstractmethod
+    async def unsubscribe_kline(
+        self,
+        symbol: str | List[str],
+        interval: KlineInterval,
+    ):
+        """Unsubscribe from the kline data"""
+        pass
+
     async def subscribe_kline_aggregator(
         self,
         symbol: str,
@@ -184,6 +203,33 @@ async def subscribe_kline_aggregator(
             f"Time kline aggregator created for {symbol} with interval {interval}"
         )
 
+    async def unsubscribe_kline_aggregator(
+        self,
+        symbol: str,
+        interval: KlineInterval,
+    ):
+        """Unsubscribe from time-based kline data using TimeKlineAggregator
+
+        Args:
+            symbol: Symbol to unsubscribe from
+            interval: Kline interval
+        """
+        aggregators = self._aggregators.get(symbol, [])
+        # Create a copy to avoid "list changed size during iteration" error
+        for aggregator in aggregators.copy():
+            if (
+                isinstance(aggregator, TimeKlineAggregator)
+                and aggregator.interval == interval
+            ):
+                aggregator.stop()
+                aggregators.remove(aggregator)
+                self._log.info(
+                    f"Time kline aggregator stopped for {symbol} with interval {interval}"
+                )
+        if not aggregators:
+            self._aggregators.pop(symbol, None)
+            await self.unsubscribe_trade(symbol)
+
     async def subscribe_volume_kline_aggregator(
         self, symbol: str, volume_threshold: float, volume_type: str
     ):
@@ -206,26 +252,70 @@ async def subscribe_volume_kline_aggregator(
             f"Volume kline aggregator created for {symbol} with threshold {volume_threshold} and type {volume_type}"
         )
 
+    async def unsubscribe_volume_kline_aggregator(
+        self, symbol: str, volume_threshold: float, volume_type: str
+    ):
+        """Unsubscribe from volume-based kline data using VolumeKlineAggregator
+
+        Args:
+            symbol: Symbol to unsubscribe from
+            volume_threshold: Volume threshold for the kline aggregator
+        """
+        aggregators = self._aggregators.get(symbol, [])
+        # Create a copy to avoid "list changed size during iteration" error
+        for aggregator in aggregators.copy():
+            if (
+                isinstance(aggregator, VolumeKlineAggregator)
+                and aggregator.volume_threshold == volume_threshold
+            ):
+                aggregators.remove(aggregator)
+                self._log.info(
+                    f"Volume kline aggregator stopped for {symbol} with threshold {volume_threshold} and type {volume_type}"
+                )
+        if not aggregators:
+            self._aggregators.pop(symbol, None)
+            await self.unsubscribe_trade(symbol)
+
     @abstractmethod
     async def subscribe_bookl2(self, symbol: str | List[str], level: BookLevel):
         """Subscribe to the bookl2 data"""
         pass
 
+    @abstractmethod
+    async def unsubscribe_bookl2(self, symbol: str | List[str], level: BookLevel):
+        """Unsubscribe from the bookl2 data"""
+        pass
+
     @abstractmethod
     async def subscribe_funding_rate(self, symbol: str | List[str]):
         """Subscribe to the funding rate data"""
         pass
 
+    @abstractmethod
+    async def unsubscribe_funding_rate(self, symbol: str | List[str]):
+        """Unsubscribe from the funding rate data"""
+        pass
+
     @abstractmethod
     async def subscribe_index_price(self, symbol: str | List[str]):
         """Subscribe to the index price data"""
         pass
 
+    @abstractmethod
+    async def unsubscribe_index_price(self, symbol: str | List[str]):
+        """Unsubscribe from the index price data"""
+        pass
+
     @abstractmethod
     async def subscribe_mark_price(self, symbol: str | List[str]):
         """Subscribe to the mark price data"""
         pass
 
+    @abstractmethod
+    async def unsubscribe_mark_price(self, symbol: str | List[str]):
+        """Unsubscribe from the mark price data"""
+        pass
+
     def _create_time_kline_aggregator(
         self,
         symbol: str,