import logging.config from app.bybit import get_bybit_client from app.bybit.logger_bybit.logger_bybit import LOGGING_CONFIG logging.config.dictConfig(LOGGING_CONFIG) logger = logging.getLogger("get_positions") async def get_active_positions(tg_id: int) -> list | None: """ Get active positions for a user """ try: client = await get_bybit_client(tg_id) response = client.get_positions(category="linear", settleCoin="USDT") if response["retCode"] == 0: positions = response.get("result", {}).get("list", []) active_symbols = [ pos.get("symbol") for pos in positions if float(pos.get("size", 0)) > 0 ] if active_symbols: logger.info("Active positions for user: %s", tg_id) return positions else: logger.warning("No active positions found for user: %s", tg_id) return ["No active positions found"] else: logger.error( "Error getting active positions for user %s: %s", tg_id, response["retMsg"], ) return None except Exception as e: logger.error("Error getting active positions for user %s: %s", tg_id, e) return None async def get_active_positions_by_symbol(tg_id: int, symbol: str) -> dict | None: """ Get active positions for a user by symbol """ try: client = await get_bybit_client(tg_id) response = client.get_positions(category="linear", symbol=symbol) if response["retCode"] == 0: positions = response.get("result", {}).get("list", []) if positions: logger.info("Active positions for user: %s", tg_id) return positions else: logger.warning("No active positions found for user: %s", tg_id) return None else: logger.error( "Error getting active positions for user %s: %s", tg_id, response["retMsg"], ) return None except Exception as e: logger.error("Error getting active positions for user %s: %s", tg_id, e) return None async def get_active_orders(tg_id: int) -> list | None: """ Get active orders """ try: client = await get_bybit_client(tg_id) response = client.get_open_orders( category="linear", settleCoin="USDT", limit=50, ) if response["retCode"] == 0: orders = response.get("result", {}).get("list", []) active_orders = [ pos.get("symbol") for pos in orders if float(pos.get("qty", 0)) > 0 ] if active_orders: logger.info("Active orders for user: %s", tg_id) return orders else: logger.warning("No active orders found for user: %s", tg_id) return ["No active orders found"] else: logger.error( "Error getting active orders for user %s: %s", tg_id, response["retMsg"] ) return None except Exception as e: logger.error("Error getting active orders for user %s: %s", tg_id, e) return None async def get_active_orders_by_symbol(tg_id: int, symbol: str) -> dict | None: """ Get active orders by symbol """ try: client = await get_bybit_client(tg_id) response = client.get_open_orders( category="linear", symbol=symbol, limit=50, ) if response["retCode"] == 0: orders = response.get("result", {}).get("list", []) if orders: logger.info("Active orders for user: %s", tg_id) return orders else: logger.warning("No active orders found for user: %s", tg_id) return None else: logger.error( "Error getting active orders for user %s: %s", tg_id, response["retMsg"] ) return None except Exception as e: logger.error("Error getting active orders for user %s: %s", tg_id, e) return None