import logging.config from aiogram import F, Router from aiogram.fsm.context import FSMContext from aiogram.types import CallbackQuery import app.telegram.keyboards.inline as kbi from app.bybit.get_functions.get_positions import ( get_active_orders, get_active_orders_by_symbol, get_active_positions, get_active_positions_by_symbol, ) from logger_helper.logger_helper import LOGGING_CONFIG logging.config.dictConfig(LOGGING_CONFIG) logger = logging.getLogger("get_positions_handlers") router_get_positions_handlers = Router(name="get_positions_handlers") @router_get_positions_handlers.callback_query(F.data == "my_deals") async def get_positions_handlers( callback_query: CallbackQuery, state: FSMContext ) -> None: """ Gets the user's active positions. :param callback_query: CallbackQuery object. :param state: FSMContext :return: None """ try: await state.clear() await callback_query.message.edit_text( text="Выберите какие сделки вы хотите посмотреть:", reply_markup=kbi.change_position, ) except Exception as e: logger.error("Error in get_positions_handler: %s", e) await callback_query.answer(text="Произошла ошибка при получении сделок.") @router_get_positions_handlers.callback_query(F.data == "change_position") async def get_positions_handler( callback_query: CallbackQuery, state: FSMContext ) -> None: """ Gets the user's active positions. :param callback_query: CallbackQuery object. :param state: FSMContext :return: None """ try: res = await get_active_positions(tg_id=callback_query.from_user.id) if res is None: await callback_query.answer( text="Произошла ошибка при получении активных позиций." ) return if res == ["No active positions found"]: await callback_query.answer(text="Нет активных позиций.") return active_positions = [pos for pos in res if float(pos.get("size", 0)) > 0] if not active_positions: await callback_query.answer(text="Нет активных позиций.") return active_symbols_sides = [ (pos.get("symbol"), pos.get("side")) for pos in active_positions ] await callback_query.message.edit_text( text="Ваши активные позиции:", reply_markup=kbi.create_active_positions_keyboard( symbols=active_symbols_sides ), ) except Exception as e: logger.error("Error in get_positions_handler: %s", e) await callback_query.answer( text="Произошла ошибка при получении активных позиций." ) finally: await state.clear() @router_get_positions_handlers.callback_query( lambda c: c.data.startswith("get_position_") ) async def get_position_handler(callback_query: CallbackQuery, state: FSMContext): try: data = callback_query.data parts = data.split("_") symbol = parts[2] get_side = parts[3] res = await get_active_positions_by_symbol( tg_id=callback_query.from_user.id, symbol=symbol ) if res is None: await callback_query.answer( text="Произошла ошибка при получении активных позиций." ) return position = next((pos for pos in res if pos.get("side") == get_side), None) if position: side = position.get("side") symbol = position.get("symbol") or "Нет данных" avg_price = position.get("avgPrice") or "Нет данных" size = position.get("size") or "Нет данных" take_profit = position.get("takeProfit") or "Нет данных" stop_loss = position.get("stopLoss") or "Нет данных" position_idx = position.get("positionIdx") or "Нет данных" liq_price = position.get("liqPrice") or "Нет данных" else: side = "Нет данных" symbol = "Нет данных" avg_price = "Нет данных" size = "Нет данных" take_profit = "Нет данных" stop_loss = "Нет данных" position_idx = "Нет данных" liq_price = "Нет данных" side_rus = ( "Покупка" if side == "Buy" else "Продажа" if side == "Sell" else "Нет данных" ) position_idx_rus = ( "Односторонний" if position_idx == 0 else ( "Покупка в режиме хеджирования" if position_idx == 1 else ( "Продажа в режиме хеджирования" if position_idx == 2 else "Нет данных" ) ) ) text_lines = [ f"Торговая пара: {symbol}", f"Режим позиции: {position_idx_rus}", f"Цена входа: {avg_price}", f"Количество: {size}", f"Движение: {side_rus}", ] if take_profit and take_profit != "Нет данных": text_lines.append(f"Тейк-профит: {take_profit}") if stop_loss and stop_loss != "Нет данных": text_lines.append(f"Стоп-лосс: {stop_loss}") if liq_price and liq_price != "Нет данных": text_lines.append(f"Цена ликвидации: {liq_price}") text = "\n".join(text_lines) await callback_query.message.edit_text( text=text, reply_markup=kbi.make_close_position_keyboard( symbol_pos=symbol, side=side, position_idx=position_idx, qty=size ), ) except Exception as e: logger.error("Error in get_position_handler: %s", e) await callback_query.answer( text="Произошла ошибка при получении активных позиций." ) finally: await state.clear() @router_get_positions_handlers.callback_query(F.data == "open_orders") async def get_open_orders_handler( callback_query: CallbackQuery, state: FSMContext ) -> None: """ Gets the user's open orders. :param callback_query: CallbackQuery object. :param state: FSMContext :return: None """ try: res = await get_active_orders(tg_id=callback_query.from_user.id) if res is None: await callback_query.answer( text="Произошла ошибка при получении активных ордеров." ) return if res == ["No active orders found"]: await callback_query.answer(text="Нет активных ордеров.") return active_positions = [pos for pos in res if pos.get("orderStatus", 0) == "New"] if not active_positions: await callback_query.answer(text="Нет активных ордеров.") return active_orders_sides = [ (pos.get("symbol"), pos.get("side")) for pos in active_positions ] await callback_query.message.edit_text( text="Ваши активные ордера:", reply_markup=kbi.create_active_orders_keyboard(orders=active_orders_sides), ) except Exception as e: logger.error("Error in get_open_orders_handler: %s", e) await callback_query.answer( text="Произошла ошибка при получении активных ордеров." ) finally: await state.clear() @router_get_positions_handlers.callback_query(lambda c: c.data.startswith("get_order_")) async def get_order_handler(callback_query: CallbackQuery, state: FSMContext): try: data = callback_query.data parts = data.split("_") symbol = parts[2] get_side = parts[3] res = await get_active_orders_by_symbol( tg_id=callback_query.from_user.id, symbol=symbol ) if res is None: await callback_query.answer( text="Произошла ошибка при получении активных ордеров." ) return orders = next((pos for pos in res if pos.get("side") == get_side), None) if orders: side = orders.get("side") symbol = orders.get("symbol") price = orders.get("price") qty = orders.get("qty") order_type = orders.get("orderType") trigger_price = orders.get("triggerPrice") take_profit = orders.get("takeProfit") stop_loss = orders.get("stopLoss") order_id = orders.get("orderId") else: side = "Нет данных" symbol = "Нет данных" price = "Нет данных" qty = "Нет данных" order_type = "Нет данных" trigger_price = "Нет данных" take_profit = "Нет данных" stop_loss = "Нет данных" order_id = "Нет данных" side_rus = ( "Покупка" if side == "Buy" else "Продажа" if side == "Sell" else "Нет данных" ) order_type_rus = ( "Рыночный" if order_type == "Market" else "Лимитный" if order_type == "Limit" else "Нет данных" ) text_lines = [ f"Торговая пара: {symbol}", f"Количество: {qty}", f"Движение: {side_rus}", f"Тип ордера: {order_type_rus}", ] if price: text_lines.append(f"Цена: {price}") if trigger_price and trigger_price != "Нет данных": text_lines.append(f"Триггер цена: {trigger_price}") if take_profit and take_profit != "Нет данных": text_lines.append(f"Тейк-профит: {take_profit}") if stop_loss and stop_loss != "Нет данных": text_lines.append(f"Стоп-лосс: {stop_loss}") text = "\n".join(text_lines) await callback_query.message.edit_text( text=text, reply_markup=kbi.make_close_orders_keyboard( symbol_order=symbol, order_id=order_id ), ) except Exception as e: logger.error("Error in get_order_handler: %s", e) await callback_query.answer( text="Произошла ошибка при получении активных ордеров." ) finally: await state.clear()