import logging from typing import Dict, Any, List, Optional from telegram import Update from telegram.ext import ContextTypes from .base import InfoCommandsBase logger = logging.getLogger(__name__) class OrdersCommands(InfoCommandsBase): """Handles all order-related commands.""" async def orders_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle the /orders command.""" try: if not self._is_authorized(update): await self._reply(update, "❌ Unauthorized access.") return # Get open orders from the exchange open_orders = self.trading_engine.get_orders() or [] # Get pending stop loss orders from the database stats = self.trading_engine.get_stats() pending_sl_lifecycles = [] if stats: pending_sl_lifecycles = stats.get_pending_stop_losses_from_lifecycles() # Combine both lists all_orders = open_orders + pending_sl_lifecycles if not all_orders: await self._reply(update, "📭 No open or pending orders") return # Format orders text orders_text = "📋 Open & Pending Orders\n\n" for order in all_orders: try: is_pending_sl = 'trade_lifecycle_id' in order and order.get('stop_loss_price') is not None symbol = order.get('symbol', 'unknown') base_asset = symbol.split('/')[0] if '/' in symbol else symbol.split(':')[0] if is_pending_sl: # This is a pending SL from a trade lifecycle entry_side = order.get('side', 'unknown').upper() order_type = "STOP (PENDING)" side = "SELL" if entry_side == "BUY" else "BUY" price = float(order.get('stop_loss_price', 0)) amount = float(order.get('amount', 0)) # Amount from the entry order status = f"Awaiting {order.get('status', '').upper()} Entry" # e.g. Awaiting PENDING Entry order_id = order.get('trade_lifecycle_id', 'unknown') else: # This is a regular exchange order order_type = order.get('type', 'unknown').upper() side = order.get('side', 'unknown').upper() price = float(order.get('price', 0)) amount = float(order.get('amount', 0)) filled = float(order.get('filled', 0)) amount = amount - filled # Show remaining amount status = order.get('status', 'unknown').upper() order_id = order.get('id', 'unknown') # Skip fully filled orders if amount <= 0 and not is_pending_sl: continue # Format order details formatter = self._get_formatter() price_str = await formatter.format_price_with_symbol(price, base_asset) amount_str = await formatter.format_amount(amount, base_asset) if amount > 0 else "N/A" # Order header side_emoji = "🟢" if side == "BUY" else "🔴" orders_text += f"{side_emoji} {base_asset} {side} {order_type}\n" orders_text += f" Status: {status.replace('_', ' ')}\n" orders_text += f" 📏 Amount: {amount_str}\n" orders_text += f" 💰 {'Trigger Price' if 'STOP' in order_type else 'Price'}: {price_str}\n" # Add order type specific info for exchange orders if not is_pending_sl and (order_type == "STOP_LOSS" or order_type == "TAKE_PROFIT"): trigger_price = order.get('info', {}).get('triggerPrice') if trigger_price: trigger_price_str = await formatter.format_price_with_symbol(float(trigger_price), base_asset) orders_text += f" 🎯 Trigger: {trigger_price_str}\n" # Add order ID id_label = "Lifecycle ID" if is_pending_sl else "Order ID" orders_text += f" 🆔 {id_label}: {str(order_id)[:12]}\n\n" except Exception as e: logger.error(f"Error processing order {order.get('symbol', 'unknown')}: {e}", exc_info=True) continue # Add footer orders_text += "💡 Pending SL orders are activated when the entry order fills." await self._reply(update, orders_text.strip()) except Exception as e: logger.error(f"Error in orders command: {e}", exc_info=True) await self._reply(update, "❌ Error retrieving order information.")