feat: Implement Redis backend for parameter storage and retrieval in AsyncCache; update version to 0.2.19

Author: River.Shi

nexustrader/base/connector.py

@@ -87,7 +87,9 @@ def __init__(
 
         # Aggregator management - key: symbol, value: list of aggregators
         self._aggregators: Dict[str, List[KlineAggregator]] = {}
-        self._msgbus.subscribe(topic="trade", handler=self._handle_trade_for_aggregators)
+        self._msgbus.subscribe(
+            topic="trade", handler=self._handle_trade_for_aggregators
+        )
 
     @property
     def account_type(self):
@@ -173,7 +175,9 @@ async def subscribe_kline_aggregator(
         await self.subscribe_trade(symbol)
 
         # Create and register time kline aggregator
-        aggregator = self._create_time_kline_aggregator(symbol, interval, build_with_no_updates)
+        aggregator = self._create_time_kline_aggregator(
+            symbol, interval, build_with_no_updates
+        )
         self._add_aggregator(symbol, aggregator)
 
         self._log.info(

nexustrader/base/ems.py

@@ -58,8 +58,6 @@ def __init__(
         ] = {}
         self._private_connectors: Dict[AccountType, PrivateConnector] | None = None
         self._is_mock = is_mock
-    
-
 
     def _build(self, private_connectors: Dict[AccountType, PrivateConnector]):
         self._private_connectors = private_connectors
@@ -134,7 +132,7 @@ def _price_to_precision(
                 precision_decimal, rounding=ROUND_FLOOR
             ) * exp
         return format_price
-    
+
     @abstractmethod
     def _instrument_id_to_account_type(
         self, instrument_id: InstrumentId

nexustrader/base/ws_client.py

@@ -78,7 +78,7 @@ def on_ws_disconnected(self, transport: WSTransport) -> None:
             transport (picows.WSTransport): WebSocket transport instance
         """
         self._log.debug("Disconnected from Websocket.")
-    
+
     def _decode_frame(self, frame: WSFrame) -> str:
         """Decode the payload of a WebSocket frame safely.
 
@@ -119,7 +119,7 @@ def on_ws_frame(self, transport: WSTransport, frame: WSFrame) -> None:
                     return
         except Exception as e:
             import traceback
-        
+
             self._log.error(
                 f"Error processing message: {str(e)}\nTraceback: {traceback.format_exc()}\nws_frame: {self._decode_frame(frame)}"
             )
@@ -205,6 +205,9 @@ async def _connection_handler(self):
             await asyncio.sleep(self._reconnect_interval)
 
     def _send(self, payload: dict):
+        if not self.connected:
+            self._log.warning(f"Websocket not connected. drop msg: {str(payload)}")
+            return
         self._transport.send(WSMsgType.TEXT, msgspec.json.encode(payload))
 
     def disconnect(self):

nexustrader/constants.py

@@ -8,6 +8,8 @@
 from throttled.asyncio import Throttled
 from nautilus_trader.core.nautilus_pyo3 import LogColor  # noqa
 
+BACKEND_LITERAL = Literal["memory", "redis"]
+
 
 def is_sphinx_build():
     return "sphinx" in sys.modules
@@ -39,15 +41,8 @@ def get_postgresql_config():
     }
 
 
-def get_redis_config(in_docker: bool = False):
+def get_redis_config():
     try:
-        if in_docker:
-            return {
-                "host": "redis",
-                "db": settings.REDIS_DB,
-                "password": settings.REDIS_PASSWORD,
-            }
-
         return {
             "host": settings.REDIS_HOST,
             "port": settings.REDIS_PORT,
@@ -356,6 +351,11 @@ class StorageType(Enum):
     POSTGRESQL = "postgresql"
 
 
+class ParamBackend(Enum):
+    MEMORY = "memory"
+    REDIS = "redis"
+
+
 class RateLimiter:
     @abstractmethod
     def __call__(self, endpoint: str) -> Throttled: ...

nexustrader/core/cache.py

@@ -23,9 +23,9 @@
     BookL2,
 )
 from nexustrader.constants import STATUS_TRANSITIONS, AccountType, KlineInterval
-from nexustrader.core.entity import TaskManager
+from nexustrader.core.entity import TaskManager, get_redis_client_if_available
 from nexustrader.core.nautilius_core import LiveClock, MessageBus, Logger
-from nexustrader.constants import StorageType
+from nexustrader.constants import StorageType, ParamBackend
 from nexustrader.backends import SQLiteBackend, PostgreSQLBackend
 
 
@@ -112,7 +112,10 @@ def __init__(
         self._position_lock = threading.RLock()  # Lock for position updates
         self._order_lock = threading.RLock()  # Lock for order updates
         self._balance_lock = threading.RLock()  # Lock for balance updates
-        self._param_lock = threading.RLock()  # Lock for parameter updates
+
+        # Redis client for parameter storage (lazy initialization)
+        self._redis_client = None
+        self._redis_client_initialized = False
 
     ################# # base functions ####################
 
@@ -463,24 +466,151 @@ def get_open_orders(
 
     ################ # parameter cache  ###################
 
-    def get_param(self, key: str, default: Any = None) -> Any:
-        """Get a parameter from the cache"""
-        with self._param_lock:
+    def _ensure_redis_client(self) -> None:
+        """
+        Lazy initialization of Redis client.
+        Raises RuntimeError if Redis is not available.
+        """
+        if not self._redis_client_initialized:
+            self._redis_client = get_redis_client_if_available()
+            self._redis_client_initialized = True
+            if self._redis_client is None:
+                raise RuntimeError(
+                    "Redis client is not available. Please ensure Redis is running and configured properly."
+                )
+
+    def _get_redis_param_key(self, key: str) -> str:
+        """Generate Redis key for parameter storage"""
+        return f"nexus:param:{self.strategy_id}:{self.user_id}:{key}"
+
+    def get_param(
+        self,
+        key: str,
+        default: Any = None,
+        param_backend: ParamBackend = ParamBackend.MEMORY,
+    ) -> Any:
+        """
+        Get a parameter from the cache
+
+        Args:
+            key: Parameter key
+            default: Default value if key not found
+            param_backend: Storage backend (MEMORY or REDIS)
+
+        Returns:
+            Parameter value or default
+
+        Raises:
+            RuntimeError: If param_backend is REDIS but Redis is not available
+        """
+        if param_backend == ParamBackend.REDIS:
+            self._ensure_redis_client()
+            redis_key = self._get_redis_param_key(key)
+            value = self._redis_client.get(redis_key)
+            if value is None:
+                return default
+            # Deserialize the value
+            return msgspec.json.decode(value)
+        else:
             return self._mem_params.get(key, default)
 
-    def set_param(self, key: str, value: Any) -> None:
-        """Set a parameter in the cache"""
-        with self._param_lock:
+    def set_param(
+        self, key: str, value: Any, param_backend: ParamBackend = ParamBackend.MEMORY
+    ) -> None:
+        """
+        Set a parameter in the cache
+
+        Args:
+            key: Parameter key
+            value: Parameter value
+            param_backend: Storage backend (MEMORY or REDIS)
+
+        Raises:
+            RuntimeError: If param_backend is REDIS but Redis is not available
+        """
+        if param_backend == ParamBackend.REDIS:
+            self._ensure_redis_client()
+            redis_key = self._get_redis_param_key(key)
+            # Serialize the value
+            serialized_value = msgspec.json.encode(value)
+            self._redis_client.set(redis_key, serialized_value)
+        else:
             self._mem_params[key] = value
 
-    def get_all_params(self) -> Dict[str, Any]:
-        """Get all parameters from the cache"""
-        with self._param_lock:
+    def get_all_params(
+        self, param_backend: ParamBackend = ParamBackend.MEMORY
+    ) -> Dict[str, Any]:
+        """
+        Get all parameters from the cache
+
+        Args:
+            param_backend: Storage backend (MEMORY or REDIS)
+
+        Returns:
+            Dictionary of all parameters
+
+        Raises:
+            RuntimeError: If param_backend is REDIS but Redis is not available
+        """
+        if param_backend == ParamBackend.REDIS:
+            self._ensure_redis_client()
+            pattern = f"nexus:param:{self.strategy_id}:{self.user_id}:*"
+            params = {}
+            # Use SCAN to iterate over keys matching the pattern
+            cursor = 0
+            while True:
+                cursor, keys = self._redis_client.scan(cursor, match=pattern, count=100)
+                for redis_key in keys:
+                    # Extract the parameter key from the Redis key
+                    key = (
+                        redis_key.decode()
+                        if isinstance(redis_key, bytes)
+                        else redis_key
+                    )
+                    param_key = key.split(":", 4)[-1]  # Get the part after the last ':'
+                    value = self._redis_client.get(redis_key)
+                    if value is not None:
+                        params[param_key] = msgspec.json.decode(value)
+                if cursor == 0:
+                    break
+            return params
+        else:
             return self._mem_params.copy()
 
-    def clear_param(self, key: Optional[str] = None) -> None:
-        """Clear parameter(s) from the cache"""
-        with self._param_lock:
+    def clear_param(
+        self,
+        key: Optional[str] = None,
+        param_backend: ParamBackend = ParamBackend.MEMORY,
+    ) -> None:
+        """
+        Clear parameter(s) from the cache
+
+        Args:
+            key: Parameter key to clear (None to clear all)
+            param_backend: Storage backend (MEMORY or REDIS)
+
+        Raises:
+            RuntimeError: If param_backend is REDIS but Redis is not available
+        """
+        if param_backend == ParamBackend.REDIS:
+            self._ensure_redis_client()
+            if key is None:
+                # Clear all parameters matching the pattern
+                pattern = f"nexus:param:{self.strategy_id}:{self.user_id}:*"
+                cursor = 0
+                while True:
+                    cursor, keys = self._redis_client.scan(
+                        cursor, match=pattern, count=100
+                    )
+                    if keys:
+                        self._redis_client.delete(*keys)
+                    if cursor == 0:
+                        break
+            else:
+                # Clear specific parameter
+                redis_key = self._get_redis_param_key(key)
+                self._redis_client.delete(redis_key)
+        else:
             if key is None:
                 # Clear all parameters
                 self._mem_params.clear()

nexustrader/core/entity.py

@@ -69,16 +69,8 @@ def is_redis_available() -> bool:
 
         # Check if Redis server is accessible
         try:
-            # Detect if running in Docker
-            in_docker = False
-            try:
-                socket.gethostbyname("redis")
-                in_docker = True
-            except socket.gaierror:
-                pass
-
             # Get Redis config and test connection
-            redis_config = get_redis_config(in_docker)
+            redis_config = get_redis_config()
             client = redis.Redis(**redis_config)
             client.ping()  # This will raise an exception if Redis is not accessible
             client.close()
@@ -98,18 +90,8 @@ def get_redis_client_if_available():
         return None
 
     try:
-        import socket
         from nexustrader.constants import get_redis_config
-
-        # Detect if running in Docker
-        in_docker = False
-        try:
-            socket.gethostbyname("redis")
-            in_docker = True
-        except socket.gaierror:
-            pass
-
-        redis_config = get_redis_config(in_docker)
+        redis_config = get_redis_config()
         return redis.Redis(**redis_config)
     except Exception:
         return None

nexustrader/strategy.py

@@ -49,12 +49,14 @@
     OrderSide,
     OrderType,
     TimeInForce,
+    ParamBackend,
     # PositionSide,
     AccountType,
     SubmitType,
     ExchangeType,
     KlineInterval,
     TriggerType,
+    BACKEND_LITERAL,
 )
 
 
@@ -1145,7 +1147,11 @@ def _on_mark_price(self, mark_price: MarkPrice):
         self._subscriptions_ready[DataType.MARK_PRICE].input(mark_price)
 
     def param(
-        self, name: str, value: Optional[Any] = None, default: Optional[Any] = None
+        self,
+        name: str,
+        value: Optional[Any] = None,
+        default: Optional[Any] = None,
+        backend: BACKEND_LITERAL = "memory",
     ) -> Any:
         """
         Get or set a parameter in the cache.
@@ -1164,15 +1170,18 @@ def param(
             # Get a parameter
             rolling_n = self.param('rolling_n')
         """
+        param_backend = ParamBackend(backend)
         if value is not None:
             # Set parameter
-            self.cache.set_param(name, value)
+            self.cache.set_param(name, value, param_backend)
             return None
         else:
             # Get parameter
-            return self.cache.get_param(name, default)
+            return self.cache.get_param(name, default, param_backend)
 
-    def clear_param(self, name: Optional[str] = None) -> None:
+    def clear_param(
+        self, name: Optional[str] = None, backend: BACKEND_LITERAL = "memory"
+    ) -> None:
         """
         Clear parameter(s) from the cache.
 
@@ -1186,7 +1195,7 @@ def clear_param(self, name: Optional[str] = None) -> None:
             # Clear all parameters
             self.clear_param()
         """
-        self.cache.clear_param(name)
+        self.cache.clear_param(name, ParamBackend(backend))
 
     def set_timer(
         self,

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "nexustrader"
-version = "0.2.18"
+version = "0.2.19"
 description = "fastest python trading bot"
 authors = [
     {name = "River-Shi",email = "[email protected]"}