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.")