Feat/add ws to execution

config.py

@@ -1,11 +1,12 @@
 from typing import List
+
 from pydantic import Field
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
 class BrokerSettings(BaseSettings):
     """MQTT Broker configuration for bot communication."""
-    
+
     host: str = Field(default="localhost", description="MQTT broker host")
     port: int = Field(default=1883, description="MQTT broker port")
     username: str = Field(default="admin", description="MQTT broker username")
@@ -16,7 +17,7 @@ class BrokerSettings(BaseSettings):
 
 class DatabaseSettings(BaseSettings):
     """Database configuration."""
-    
+
     url: str = Field(
         default="postgresql+asyncpg://hbot:hummingbot-api@localhost:5432/hummingbot_api",
         description="Database connection URL"
@@ -27,7 +28,7 @@ class DatabaseSettings(BaseSettings):
 
 class MarketDataSettings(BaseSettings):
     """Market data feed manager configuration."""
-    
+
     cleanup_interval: int = Field(
         default=300,
         description="How often to run feed cleanup in seconds"
@@ -40,13 +41,25 @@ class MarketDataSettings(BaseSettings):
         default=30,
         description="How long to wait for a candle feed to become ready in seconds"
     )
+    ws_heartbeat_interval: int = Field(
+        default=30,
+        description="WebSocket heartbeat interval in seconds"
+    )
+    ws_min_update_interval: float = Field(
+        default=0.25,
+        description="Minimum allowed WebSocket subscription update interval in seconds"
+    )
+    ws_max_update_interval: float = Field(
+        default=60.0,
+        description="Maximum allowed WebSocket subscription update interval in seconds"
+    )
 
     model_config = SettingsConfigDict(env_prefix="MARKET_DATA_", extra="ignore")
 
 
 class SecuritySettings(BaseSettings):
     """Security and authentication configuration."""
-    
+
     username: str = Field(default="admin", description="API basic auth username")
     password: str = Field(default="admin", description="API basic auth password")
     debug_mode: bool = Field(default=False, description="Enable debug mode (disables auth)")
@@ -81,18 +94,18 @@ class GatewaySettings(BaseSettings):
 
 class AppSettings(BaseSettings):
     """Main application settings."""
-    
+
     # Static paths
     controllers_path: str = "bots/conf/controllers"
     controllers_module: str = "bots.controllers"
     password_verification_path: str = "credentials/master_account/.password_verification"
-    
+
     # Environment-configurable settings
     logfire_environment: str = Field(
         default="dev",
         description="Logfire environment name"
     )
-    
+
     # Account state update interval
     account_update_interval: int = Field(
         default=5,
@@ -117,7 +130,7 @@ class Settings(BaseSettings):
     aws: AWSSettings = Field(default_factory=AWSSettings)
     gateway: GatewaySettings = Field(default_factory=GatewaySettings)
     app: AppSettings = Field(default_factory=AppSettings)
-    
+
     # Direct banned_tokens field to handle env parsing
     banned_tokens: List[str] = Field(
         default=["NAV", "ARS", "ETHW", "ETHF"],

deps.py

@@ -1,14 +1,17 @@
 from fastapi import Request
-from services.bots_orchestrator import BotsOrchestrator
+
+from database import AsyncDatabaseManager
 from services.accounts_service import AccountsService
+from services.bots_orchestrator import BotsOrchestrator
 from services.docker_service import DockerService
+from services.executor_service import ExecutorService
+from services.executor_ws_manager import ExecutorWebSocketManager
 from services.gateway_service import GatewayService
-from services.unified_connector_service import UnifiedConnectorService
 from services.market_data_service import MarketDataService
 from services.trading_service import TradingService
-from services.executor_service import ExecutorService
+from services.unified_connector_service import UnifiedConnectorService
+from services.websocket_manager import WebSocketManager
 from utils.bot_archiver import BotArchiver
-from database import AsyncDatabaseManager
 
 
 def get_bots_orchestrator(request: Request) -> BotsOrchestrator:
@@ -59,3 +62,13 @@ def get_bot_archiver(request: Request) -> BotArchiver:
 def get_database_manager(request: Request) -> AsyncDatabaseManager:
     """Get AsyncDatabaseManager from app state."""
     return request.app.state.db_manager
+
+
+def get_executor_ws_manager(request: Request) -> ExecutorWebSocketManager:
+    """Get ExecutorWebSocketManager from app state."""
+    return request.app.state.executor_ws_manager
+
+
+def get_websocket_manager(request: Request) -> WebSocketManager:
+    """Get WebSocketManager from app state."""
+    return request.app.state.websocket_manager

main.py

@@ -58,15 +58,18 @@ def patched_save_to_yml(yml_path, cm):
     rate_oracle,
     scripts,
     trading,
+    websocket,
 )
 from services.accounts_service import AccountsService  # noqa: E402
 from services.bots_orchestrator import BotsOrchestrator  # noqa: E402
 from services.docker_service import DockerService  # noqa: E402
 from services.executor_service import ExecutorService  # noqa: E402
+from services.executor_ws_manager import ExecutorWebSocketManager  # noqa: E402
 from services.gateway_service import GatewayService  # noqa: E402
 from services.market_data_service import MarketDataService  # noqa: E402
 from services.trading_service import TradingService  # noqa: E402
 from services.unified_connector_service import UnifiedConnectorService  # noqa: E402
+from services.websocket_manager import WebSocketManager  # noqa: E402
 from utils.bot_archiver import BotArchiver  # noqa: E402
 from utils.security import BackendAPISecurity  # noqa: E402
 
@@ -211,14 +214,6 @@ async def lifespan(app: FastAPI):
         max_retries=10
     )
     logging.info("ExecutorService initialized")
-    # Ensure lp_executor is in the registry (workspace hummingbot may load after class definition)
-    try:
-        from hummingbot.strategy_v2.executors.lp_executor.data_types import LPExecutorConfig
-        from hummingbot.strategy_v2.executors.lp_executor.lp_executor import LPExecutor
-        ExecutorService.EXECUTOR_REGISTRY["lp_executor"] = (LPExecutor, LPExecutorConfig)
-        logging.debug("lp_executor registered in ExecutorService")
-    except Exception as e:
-        logging.warning(f"Failed to register lp_executor: {e}")
 
     # =========================================================================
     # 5. Other Services
@@ -266,11 +261,18 @@ async def lifespan(app: FastAPI):
     app.state.trading_service = trading_service
     app.state.accounts_service = accounts_service
     app.state.executor_service = executor_service
+    websocket_manager = WebSocketManager(market_data_service)
+    app.state.websocket_manager = websocket_manager
+
     app.state.bots_orchestrator = bots_orchestrator
     app.state.docker_service = docker_service
     app.state.gateway_service = gateway_service
     app.state.bot_archiver = bot_archiver
 
+    # WebSocket manager for executor streaming
+    executor_ws_manager = ExecutorWebSocketManager(executor_service, market_data_service)
+    app.state.executor_ws_manager = executor_ws_manager
+
     logging.info("All services started successfully")
 
     yield
@@ -281,6 +283,8 @@ async def lifespan(app: FastAPI):
 
     logging.info("Shutting down services...")
 
+    websocket_manager.shutdown()
+    await executor_ws_manager.shutdown()
     bots_orchestrator.stop()
     await accounts_service.stop()
     await executor_service.stop()
@@ -381,6 +385,9 @@ def auth_user(
 app.include_router(executors.router, dependencies=[Depends(auth_user)])
 app.include_router(gateway_proxy.router, dependencies=[Depends(auth_user)])
 
+# WebSocket router (handles its own auth)
+app.include_router(websocket.router)
+
 
 @app.get("/")
 async def root():