import logging
from typing import Dict, Any, List, Optional
from datetime import datetime, timezone
from telegram import Update
from telegram.ext import ContextTypes
from .base import InfoCommandsBase
logger = logging.getLogger(__name__)
class PositionsCommands(InfoCommandsBase):
"""Handles all position-related commands."""
async def positions_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Handle the /positions command."""
try:
if not self._is_authorized(update):
await self._reply(update, "ā Unauthorized access.")
return
stats = self.trading_engine.get_stats()
if not stats:
await self._reply(update, "ā Trading stats not available.")
return
# Get open positions from DB
open_positions = stats.get_open_positions()
if not open_positions:
await self._reply(update, "š No open positions\n\nš” Use /long or /short to open a position")
return
# Get current exchange orders for stop loss detection
exchange_orders = self.trading_engine.get_orders() or []
# Initialize totals
total_position_value = 0.0
total_unrealized = 0.0
total_margin_used = 0.0
# Build position details
positions_text = "š Open Positions\n\n"
for position_trade in open_positions:
try:
symbol = position_trade['symbol']
base_asset = symbol.split('/')[0] if '/' in symbol else symbol.split(':')[0]
position_side = position_trade.get('position_side', 'unknown')
abs_current_amount = abs(float(position_trade.get('current_position_size', 0)))
entry_price = float(position_trade.get('entry_price', 0))
mark_price = position_trade.get('mark_price', entry_price)
trade_type = position_trade.get('trade_type', 'unknown')
# Calculate duration
duration_str = "N/A"
position_opened_at_str = position_trade.get('position_opened_at')
if position_opened_at_str:
try:
position_opened_at = datetime.fromisoformat(position_opened_at_str.replace('Z', '+00:00'))
duration = datetime.now(timezone.utc) - position_opened_at
days = duration.days
hours = duration.seconds // 3600
minutes = (duration.seconds % 3600) // 60
parts = []
if days > 0:
parts.append(f"{days}d")
if hours > 0:
parts.append(f"{hours}h")
if minutes > 0 or (days == 0 and hours == 0):
parts.append(f"{minutes}m")
duration_str = " ".join(parts) if parts else "0m"
except ValueError:
logger.warning(f"Could not parse position_opened_at: {position_opened_at_str} for {symbol}")
duration_str = "Error"
# Calculate unrealized PnL
unrealized_pnl = position_trade.get('unrealized_pnl')
if unrealized_pnl is None:
if position_side == 'long':
unrealized_pnl = (mark_price - entry_price) * abs_current_amount
else: # Short position
unrealized_pnl = (entry_price - mark_price) * abs_current_amount
unrealized_pnl = unrealized_pnl or 0.0
# ROE Percentage from database
roe_percentage = position_trade.get('roe_percentage', 0.0)
# Add to totals
individual_position_value = position_trade.get('position_value')
if individual_position_value is None:
individual_position_value = abs_current_amount * mark_price
total_position_value += individual_position_value
total_unrealized += unrealized_pnl
margin_used = position_trade.get('margin_used')
if margin_used is not None:
total_margin_used += margin_used
# Format position details
formatter = self._get_formatter()
entry_price_str = formatter.format_price_with_symbol(entry_price, base_asset)
mark_price_str = formatter.format_price_with_symbol(mark_price, base_asset)
size_str = formatter.format_amount(abs_current_amount, base_asset)
# Position header
pos_emoji = "š¢" if position_side == 'long' else "š“"
direction_text = position_side.upper()
# Add leverage if available
leverage = position_trade.get('leverage')
if leverage is not None:
try:
leverage_val = float(leverage)
leverage_str = f"x{leverage_val:.1f}".rstrip('0').rstrip('.') if '.' in f"{leverage_val:.1f}" else f"x{int(leverage_val)}"
direction_text = f"{direction_text} {leverage_str}"
except ValueError:
logger.warning(f"Could not parse leverage value: {leverage} for {symbol}")
# Add type indicator
type_indicator = ""
if position_trade.get('trade_lifecycle_id'):
type_indicator = " š¤"
elif trade_type == 'external':
type_indicator = " š"
# Build position text
positions_text += f"{pos_emoji} {base_asset} ({direction_text}){type_indicator}\n"
positions_text += f" š Size: {size_str} {base_asset}\n"
positions_text += f" š° Entry: {entry_price_str}\n"
positions_text += f" ā³ Duration: {duration_str}\n"
positions_text += f" š¦ Value: ${individual_position_value:,.2f}\n"
if mark_price != 0 and abs(mark_price - entry_price) > 1e-9:
positions_text += f" š Mark: {mark_price_str}\n"
pnl_emoji = "š¢" if unrealized_pnl >= 0 else "š“"
positions_text += f" {pnl_emoji} P&L: ${unrealized_pnl:,.2f}\n"
roe_emoji = "š¢" if roe_percentage >= 0 else "š“"
positions_text += f" {roe_emoji} ROE: {roe_percentage:+.2f}%\n"
# Add risk management info
if margin_used is not None:
positions_text += f" š³ Margin Used: ${margin_used:,.2f}\n"
if position_trade.get('liquidation_price') is not None and position_trade.get('liquidation_price') > 0:
liq_price_str = formatter.format_price_with_symbol(position_trade.get('liquidation_price'), base_asset)
positions_text += f" ā ļø Liquidation: {liq_price_str}\n"
if position_trade.get('stop_loss_price'):
sl_price = position_trade['stop_loss_price']
positions_text += f" š Stop Loss: {formatter.format_price_with_symbol(sl_price, base_asset)}\n"
if position_trade.get('take_profit_price'):
tp_price = position_trade['take_profit_price']
positions_text += f" šÆ Take Profit: {formatter.format_price_with_symbol(tp_price, base_asset)}\n"
# Add external stop losses
external_sls = self._get_external_stop_losses(symbol, position_side, entry_price, abs_current_amount, exchange_orders)
for ext_sl_price in external_sls:
positions_text += f" š”ļø External SL: {formatter.format_price_with_symbol(ext_sl_price, base_asset)}\n"
positions_text += f" š Lifecycle ID: {position_trade['trade_lifecycle_id'][:8]}\n\n"
except Exception as e:
logger.error(f"Error processing position {position_trade.get('symbol', 'unknown')}: {e}")
continue
# Add portfolio summary
portfolio_emoji = "š¢" if total_unrealized >= 0 else "š“"
positions_text += f"š¼ Total Portfolio:\n"
positions_text += f" š¦ Total Positions Value: ${total_position_value:,.2f}\n"
if total_margin_used > 0:
positions_text += f" š³ Total Margin Used: ${total_margin_used:,.2f}\n"
leverage_ratio = total_position_value / total_margin_used if total_margin_used > 0 else 1.0
positions_text += f" āļø Portfolio Leverage: {leverage_ratio:.2f}x\n"
positions_text += f" {portfolio_emoji} Total Unrealized P&L: ${total_unrealized:,.2f}\n"
if total_margin_used > 0:
margin_pnl_percentage = (total_unrealized / total_margin_used) * 100
positions_text += f" š Portfolio Return: {margin_pnl_percentage:+.2f}% (on margin)\n"
positions_text += "\n"
positions_text += f"š¤ Legend: š¤ Bot-created ā¢ š External/synced ā¢ š”ļø External SL\n"
positions_text += f"š” Use /sl [token] [price] or /tp [token] [price] to set risk management"
await self._reply(update, positions_text.strip())
except Exception as e:
logger.error(f"Error in positions command: {e}")
await self._reply(update, "ā Error retrieving position information.")
def _get_external_stop_losses(self, symbol: str, position_side: str, entry_price: float,
current_amount: float, exchange_orders: List[Dict[str, Any]]) -> List[float]:
"""Get external stop losses for a position."""
external_sls = []
for order in exchange_orders:
try:
order_symbol = order.get('symbol')
order_side = order.get('side', '').lower()
order_type = order.get('type', '').lower()
order_price = float(order.get('price', 0))
trigger_price = order.get('info', {}).get('triggerPrice')
is_reduce_only = order.get('reduceOnly', False) or order.get('info', {}).get('reduceOnly', False)
order_amount = float(order.get('amount', 0))
if (order_symbol == symbol and is_reduce_only and
abs(order_amount - current_amount) < 0.01 * current_amount):
sl_trigger_price = 0
if trigger_price:
try:
sl_trigger_price = float(trigger_price)
except (ValueError, TypeError):
pass
if not sl_trigger_price and order_price > 0:
sl_trigger_price = order_price
is_valid_sl = False
if position_side == 'long' and order_side == 'sell':
if sl_trigger_price > 0 and sl_trigger_price < entry_price:
is_valid_sl = True
elif position_side == 'short' and order_side == 'buy':
if sl_trigger_price > 0 and sl_trigger_price > entry_price:
is_valid_sl = True
if is_valid_sl:
external_sls.append(sl_trigger_price)
except Exception as e:
logger.warning(f"Error processing order for external SL: {e}")
continue
return external_sls