Fixed the work of the websocket

app/bybit/open_positions.py

@@ -233,9 +233,6 @@ async def trading_cycle(
         )
         current_step += 1
 
-        if max_bets_in_series < current_step:
-            return "Max bets in series"
-
         await set_margin_mode(tg_id=tg_id, margin_mode=margin_type)
         await set_leverage(
             tg_id=tg_id,

app/bybit/telegram_message_handler.py

@@ -1,10 +1,9 @@
 import logging.config
 import math
-# import json
+import json
 import app.telegram.keyboards.inline as kbi
 import database.request as rq
 from app.bybit.get_functions.get_instruments_info import get_instruments_info
-from app.bybit.get_functions.get_positions import get_active_positions_by_symbol
 from app.bybit.logger_bybit.logger_bybit import LOGGING_CONFIG
 from app.bybit.open_positions import trading_cycle, trading_cycle_profit
 from app.bybit.set_functions.set_tp_sl import set_tp_sl_for_position
@@ -123,7 +122,6 @@ class TelegramMessageHandler:
                     current_step = user_deals_data.current_step
                     order_quantity = user_deals_data.order_quantity
                     pnl_series = user_deals_data.pnl_series
-                    margin_type = user_deals_data.margin_type
                     take_profit_percent = user_deals_data.take_profit_percent
                     stop_loss_percent = user_deals_data.stop_loss_percent
                     leverage = safe_float(user_deals_data.leverage)
@@ -167,16 +165,6 @@ class TelegramMessageHandler:
                             if commission_place == "Commission_for_tp":
                                 total_commission = safe_float(total_fee) / qty_formatted
 
-                        if margin_type == "ISOLATED_MARGIN":
-                            if side == "Buy":
-                                take_profit_price = safe_float(exec_price) * (
-                                        1 + take_profit_percent / 100) + total_commission
-                                stop_loss_price = None
-                            else:
-                                take_profit_price = safe_float(exec_price) * (
-                                        1 - take_profit_percent / 100) - total_commission
-                                stop_loss_price = None
-                        else:
                         if side == "Buy":
                             take_profit_price = safe_float(exec_price) * (
                                     1 + take_profit_percent / 100) + total_commission
@@ -193,27 +181,11 @@ class TelegramMessageHandler:
                                                             position_idx=0)
                         if ress or ress == "not modified":
                             take_profit_truncated = await truncate_float(take_profit_price, 6)
+                            stop_loss_truncated = await truncate_float(stop_loss_price, 6)
                             text += (f"Движение: {side_rus}\n"
                                      f"Тейк-профит: {take_profit_truncated}\n"
+                                     f"Стоп-лосс: {stop_loss_truncated}\n"
                                      )
-                            if stop_loss_price is not None:
-                                stop_loss_truncated = await truncate_float(stop_loss_price, 6)
-                            else:
-                                stop_loss_truncated = None
-
-                            if stop_loss_truncated is not None:
-                                text += f"Стоп-лосс: {stop_loss_truncated}\n"
-                            else:
-                                deals = await get_active_positions_by_symbol(
-                                    tg_id=tg_id, symbol=symbol
-                                )
-                                position = next((d for d in deals if d.get("symbol") == symbol), None)
-
-                                if position:
-                                    liq_price = position.get("liqPrice", 0)
-                                    text += f"Цена ликвидации: {liq_price}\n"
-
-
                         else:
                             text += (f"Движение: {side_rus}\n"
                                      "Не удалось установить ТП и СЛ\n")
@@ -258,7 +230,6 @@ class TelegramMessageHandler:
                             pass
                         else:
                             errors = {
-                                "Max bets in series": "❗️ Максимальное количество сделок в серии достигнуто",
                                 "Risk is too high for this trade": "❗️ Риск сделки слишком высок для продолжения",
                                 "ab not enough for new order": "❗️ Недостаточно средств для продолжения торговли",
                                 "InvalidRequestError": "❗️ Недостаточно средств для размещения нового ордера с заданным количеством и плечом.",
@@ -285,6 +256,63 @@ class TelegramMessageHandler:
                             )
 
                     elif stop_order_type == "StopLoss" or exec_type == "BustTrade":
+                        current_step = user_deals_data.current_step
+                        max_bets_in_series = user_deals_data.max_bets_in_series
+                        current_step += 1
+
+                        if max_bets_in_series < current_step:
+                            text_series = ("\n❗️ Максимальное количество сделок в серии достигнуто.\n"
+                                           "📈 Начинаю новую серию с базовой ставки\n")
+                            await self.telegram_bot.send_message(
+                                chat_id=tg_id, text=text_series
+                            )
+                            if side == "Buy":
+                                r_side = "Sell"
+                            else:
+                                r_side = "Buy"
+
+                            await rq.set_last_side_by_symbol(
+                                tg_id=tg_id, symbol=symbol, last_side=r_side)
+                            await rq.set_total_fee_user_auto_trading(
+                                tg_id=tg_id, symbol=symbol, total_fee=0
+                            )
+                            await rq.set_fee_user_auto_trading(
+                                tg_id=tg_id, symbol=symbol, fee=0
+                            )
+                            await rq.set_pnl_series_by_symbol(tg_id=tg_id, symbol=symbol, pnl_series=0)
+
+                            res = await trading_cycle_profit(
+                                tg_id=tg_id, symbol=symbol, side=r_side
+                            )
+                            if res == "OK":
+                                pass
+                            else:
+                                errors = {
+                                    "Risk is too high for this trade": "❗️ Риск сделки слишком высок для продолжения",
+                                    "ab not enough for new order": "❗️ Недостаточно средств для продолжения торговли",
+                                    "InvalidRequestError": "❗️ Недостаточно средств для размещения нового ордера с заданным количеством и плечом.",
+                                    "The number of contracts exceeds maximum limit allowed": "❗️ Превышен максимальный лимит ставки",
+                                    "Order placement failed as your position may exceed the max": "❗️ Превышен максимальный лимит ставки",
+                                }
+                                error_text = errors.get(
+                                    res, "❗️ Не удалось открыть новую сделку"
+                                )
+                                await rq.set_auto_trading(
+                                    tg_id=tg_id, symbol=symbol, auto_trading=False
+                                )
+
+                                await rq.set_total_fee_user_auto_trading(
+                                    tg_id=tg_id, symbol=symbol, total_fee=0
+                                )
+                                await rq.set_fee_user_auto_trading(
+                                    tg_id=tg_id, symbol=symbol, fee=0
+                                )
+                                await self.telegram_bot.send_message(
+                                    chat_id=tg_id,
+                                    text=error_text,
+                                    reply_markup=kbi.profile_bybit,
+                                )
+                        else:
                             open_order_text = "\n❗️ Открываю новую сделку с увеличенной ставкой.\n"
                             await self.telegram_bot.send_message(
                                 chat_id=tg_id, text=open_order_text
@@ -303,7 +331,6 @@ class TelegramMessageHandler:
                                 pass
                             else:
                                 errors = {
-                                "Max bets in series": "❗️ Максимальное количество сделок в серии достигнуто",
                                     "Risk is too high for this trade": "❗️ Риск сделки слишком высок для продолжения",
                                     "ab not enough for new order": "❗️ Недостаточно средств для продолжения торговли",
                                     "InvalidRequestError": "❗️ Недостаточно средств для размещения нового ордера с заданным количеством и плечом.",
@@ -346,5 +373,7 @@ class TelegramMessageHandler:
                         )
                         logger.info("Stop trading for symbol: %s, create_type: %s, stop_order_type: %s: %s",
                                     symbol, create_type, stop_order_type, tg_id)
+            else:
+                logger.info("Execution update: %s", json.dumps(message))
         except Exception as e:
             logger.error("Error in telegram_message_handler: %s", e, exc_info=True)

app/bybit/web_socket.py

@@ -1,4 +1,5 @@
 import asyncio
+from collections import deque
 import logging.config
 
 from pybit.unified_trading import WebSocket
@@ -11,25 +12,49 @@ logging.config.dictConfig(LOGGING_CONFIG)
 logger = logging.getLogger("web_socket")
 
 
+class CustomWebSocket(WebSocket):
+    """Custom WebSocket wrapper with enhanced error handling."""
+
+    def _on_error(self, error):
+        logger.error(f"WebSocket error: {error}")
+        return super()._on_error(error)
+
+    def _on_close(self):
+        logger.warning("WebSocket connection closed")
+        super()._on_close()
+
+
 class WebSocketBot:
     """
-    Class to handle WebSocket connections and messages.
+    Manages multiple Bybit private WebSocket connections for Telegram users.
+    Uses queue-based message processing to handle thread-safe async calls.
     """
 
     def __init__(self, telegram_bot):
-        """Initialize the TradingBot class."""
+        """
+        Initialize WebSocketBot.
+
+        Args:
+            telegram_bot: Telegram bot instance for message handling
+        """
         self.telegram_bot = telegram_bot
-        self.ws_private = None
-        self.user_messages = {}
         self.user_sockets = {}
+        self.user_messages = {}
         self.user_keys = {}
         self.loop = None
         self.message_handler = TelegramMessageHandler(telegram_bot)
 
+        self.order_queues = {}  # {tg_id: deque}
+        self.execution_queues = {}  # {tg_id: deque}
+        self.processing_tasks = {}  # {tg_id: task}
+
     async def run_user_check_loop(self):
-        """Run a loop to check for users and connect them to the WebSocket."""
+        """Main loop that continuously checks users and maintains connections."""
         self.loop = asyncio.get_running_loop()
+        logger.info("Starting WebSocket user check loop")
+
         while True:
+            try:
                 users = await WebSocketBot.get_users_from_db()
                 for user in users:
                     tg_id = user.tg_id
@@ -39,16 +64,13 @@ class WebSocketBot:
                         continue
 
                     keys_stored = self.user_keys.get(tg_id)
-                if tg_id in self.user_sockets and keys_stored == (api_key, api_secret):
+                    socket_exists = tg_id in self.user_sockets
+
+                    if socket_exists and keys_stored == (api_key, api_secret):
                         continue
 
-                if tg_id in self.user_sockets:
+                    if socket_exists:
-                    self.user_sockets.clear()
+                        await self.close_user_socket(tg_id)
-                    self.user_messages.clear()
-                    self.user_keys.clear()
-                    logger.info(
-                        "Closed old websocket for user %s due to key change", tg_id
-                    )
 
                     success = await self.try_connect_user(api_key, api_secret, tg_id)
                     if success:
@@ -56,59 +78,112 @@ class WebSocketBot:
                         self.user_messages.setdefault(
                             tg_id, {"position": None, "order": None, "execution": None}
                         )
-                    logger.info("User %s connected to WebSocket", tg_id)
+                        logger.info("User %s successfully connected", tg_id)
-                else:
+
-                    await asyncio.sleep(5)
+            except Exception as e:
-                    await self.try_connect_user(api_key, api_secret, tg_id)
+                logger.error("Error in user check loop: %s", e)
 
             await asyncio.sleep(10)
 
-    async def clear_user_sockets(self):
-        """Clear the user_sockets and user_messages dictionaries."""
-        self.user_sockets.clear()
-        self.user_messages.clear()
-        self.user_keys.clear()
-        logger.info("Cleared user_sockets")
-
     async def try_connect_user(self, api_key, api_secret, tg_id):
-        """Try to connect a user to the WebSocket."""
+        """
+        Create and setup WebSocket streams with thread-safe queues.
+        """
         try:
-            self.ws_private = WebSocket(
+            ws = CustomWebSocket(
                 demo=True,
                 testnet=False,
                 channel_type="private",
                 api_key=api_key,
-                api_secret=api_secret,
+                api_secret=api_secret
             )
 
-            self.user_sockets[tg_id] = self.ws_private
+            self.user_sockets[tg_id] = ws
-            # Connect to the WebSocket private channel
+
-            # Handle order updates
+            self.order_queues[tg_id] = deque()
-            self.ws_private.order_stream(
+            self.execution_queues[tg_id] = deque()
-                lambda msg: self.loop.call_soon_threadsafe(
+
-                    asyncio.create_task, self.handle_order_update(msg, tg_id)
+            self.processing_tasks[tg_id] = asyncio.create_task(
-                )
+                self._process_order_queue(tg_id)
-            )
-            # Handle execution updates
-            self.ws_private.execution_stream(
-                lambda msg: self.loop.call_soon_threadsafe(
-                    asyncio.create_task, self.handle_execution_update(msg, tg_id)
             )
+            self.processing_tasks[tg_id + 1] = asyncio.create_task(
+                self._process_execution_queue(tg_id)
             )
+
+            def order_callback(msg):
+                self.order_queues[tg_id].append(msg)
+
+            def execution_callback(msg):
+                self.execution_queues[tg_id].append(msg)
+
+            ws.order_stream(order_callback)
+            ws.execution_stream(execution_callback)
+
+            logger.info("WebSocket streams configured for user %s", tg_id)
             return True
+
         except Exception as e:
             logger.error("Error connecting user %s: %s", tg_id, e)
+            self.user_sockets.pop(tg_id, None)
             return False
 
+    async def _process_order_queue(self, tg_id):
+        """Continuously process order queue for user."""
+        while tg_id in self.user_sockets:
+            try:
+                if self.order_queues[tg_id]:
+                    msg = self.order_queues[tg_id].popleft()
+                    await self.handle_order_update(msg, tg_id)
+            except Exception as e:
+                logger.error("Error processing order queue %s: %s", tg_id, e)
+            await asyncio.sleep(0.01)
+
+    async def _process_execution_queue(self, tg_id):
+        """Continuously process execution queue for user."""
+        while tg_id in self.user_sockets:
+            try:
+                if self.execution_queues[tg_id]:
+                    msg = self.execution_queues[tg_id].popleft()
+                    await self.handle_execution_update(msg, tg_id)
+            except Exception as e:
+                logger.error("Error processing execution queue %s: %s", tg_id, e)
+            await asyncio.sleep(0.01)
+
+    async def close_user_socket(self, tg_id):
+        """Gracefully close user connection."""
+        if tg_id in self.user_sockets:
+            self.user_sockets.pop(tg_id, None)
+
+            for key in (tg_id, tg_id + 1):
+                task = self.processing_tasks.pop(key, None)
+                if task and not task.done():
+                    task.cancel()
+
+            self.order_queues.pop(tg_id, None)
+            self.execution_queues.pop(tg_id, None)
+            self.user_messages.pop(tg_id, None)
+            self.user_keys.pop(tg_id, None)
+            logger.info("Cleaned up user %s", tg_id)
+
     async def handle_order_update(self, message, tg_id):
-        """Handle order updates."""
+        """Process order updates."""
+        try:
             await self.message_handler.format_order_update(message, tg_id)
+        except Exception as e:
+            logger.error("Error handling order update for %s: %s", tg_id, e)
 
     async def handle_execution_update(self, message, tg_id):
-        """Handle execution updates without duplicate processing."""
+        """Process execution updates."""
+        try:
             await self.message_handler.format_execution_update(message, tg_id)
+        except Exception as e:
+            logger.error("Error handling execution update for %s: %s", tg_id, e)
 
     @staticmethod
     async def get_users_from_db():
-        """Get all users from the database."""
+        """Fetch all users from database."""
+        try:
             return await rq.get_users()
+        except Exception as e:
+            logger.error("Error getting users from DB: %s", e)
+            return []

run.py

@@ -31,7 +31,6 @@ async def main():
         dp = Dispatcher(storage=storage)
         dp.include_router(router)
         web_socket = WebSocketBot(telegram_bot=bot)
-        await web_socket.clear_user_sockets()
         ws_task = asyncio.create_task(web_socket.run_user_check_loop())
         tg_task = asyncio.create_task(dp.start_polling(bot))