import logging from typing import Dict, Any, List, Optional from datetime import datetime, timedelta, timezone from telegram import Update from telegram.ext import ContextTypes from .base import InfoCommandsBase from src.utils.token_display_formatter import get_formatter from src.config.config import Config logger = logging.getLogger(__name__) class StatsCommands(InfoCommandsBase): """Handles all statistics-related commands.""" async def _format_token_specific_stats_message(self, token_stats_data: Dict[str, Any], token_name: str) -> str: """Format detailed statistics for a specific token, matching the main /stats style.""" formatter = get_formatter() if not token_stats_data or token_stats_data.get('summary_total_trades', 0) == 0: return ( f"šŸ“Š {token_name} Statistics\n\n" f"šŸ“­ No trading data found for {token_name}.\n\n" f"šŸ’” To trade this token, try commands like:\n" f" /long {token_name} 100\n" f" /short {token_name} 100" ) perf_summary = token_stats_data.get('performance_summary', {}) open_positions = token_stats_data.get('open_positions', []) session = token_stats_data.get('session_info', {}) # --- Account Overview --- account_lines = [ f"šŸ’° {token_name.upper()} Account Overview:", f"• Current Balance: {formatter.format_price_with_symbol(perf_summary.get('current_balance', 0.0))}", f"• Initial Balance: {formatter.format_price_with_symbol(perf_summary.get('initial_balance', 0.0))}", f"• Open Positions: {len(open_positions)}", ] total_pnl = perf_summary.get('total_pnl', 0.0) entry_vol = perf_summary.get('completed_entry_volume', 0.0) total_pnl_pct = (total_pnl / entry_vol * 100) if entry_vol > 0 else 0.0 pnl_emoji = "🟢" if total_pnl >= 0 else "šŸ”“" account_lines.append(f"• {pnl_emoji} Total P&L: {formatter.format_price_with_symbol(total_pnl)} ({total_pnl_pct:+.2f}%)") account_lines.append(f"• Days Active: {perf_summary.get('days_active', 0)}") # --- Performance Metrics --- perf_lines = [ "šŸ† Performance Metrics:", f"• Total Completed Trades: {perf_summary.get('completed_trades', 0)}", f"• Win Rate: {perf_summary.get('win_rate', 0.0):.1f}% ({perf_summary.get('total_wins', 0)}/{perf_summary.get('completed_trades', 0)})", f"• Trading Volume (Entry Vol.): {formatter.format_price_with_symbol(perf_summary.get('completed_entry_volume', 0.0))}", f"• Profit Factor: {perf_summary.get('profit_factor', 0.0):.2f}", f"• Expectancy: {formatter.format_price_with_symbol(perf_summary.get('expectancy', 0.0))}", f"• Largest Winning Trade: {formatter.format_price_with_symbol(perf_summary.get('largest_win', 0.0))} ({perf_summary.get('largest_win_pct', 0.0):+.2f}%)", f"• Largest Losing Trade: {formatter.format_price_with_symbol(perf_summary.get('largest_loss', 0.0))}", f"• Best ROE Trade: {formatter.format_price_with_symbol(perf_summary.get('best_roe_trade', 0.0))} ({perf_summary.get('best_roe_trade_pct', 0.0):+.2f}%)", f"• Worst ROE Trade: {formatter.format_price_with_symbol(perf_summary.get('worst_roe_trade', 0.0))} ({perf_summary.get('worst_roe_trade_pct', 0.0):+.2f}%)", f"• Average Trade Duration: {perf_summary.get('avg_trade_duration', 'N/A')}", f"• Max Drawdown: {perf_summary.get('max_drawdown', 0.0):.2f}% (Live)", ] # --- Session Info --- session_lines = [ "ā° Session Info:", f"• Bot Started: {session.get('bot_started', 'N/A')}", f"• Stats Last Updated: {session.get('last_updated', 'N/A')}", ] # Combine all sections stats_text = ( f"šŸ“Š {token_name.upper()} Trading Statistics\n\n" + "\n".join(account_lines) + "\n\n" + "\n".join(perf_lines) + "\n\n" + "\n".join(session_lines) ) return stats_text.strip() async def stats_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle the /stats command. Shows overall stats or stats for a specific token.""" 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 if context.args and len(context.args) > 0: # Token-specific stats token_name = context.args[0].upper() token_stats = stats.get_token_stats(token_name) if not token_stats: await self._reply(update, f"āŒ No trading data found for {token_name}.") return stats_message = await self._format_token_specific_stats_message(token_stats, token_name) await self._reply(update, stats_message) return # Get current balance current_balance = self.trading_engine.get_balance() # Fix: If current_balance is a dict, extract the numeric value if isinstance(current_balance, dict): # Try common keys for USDC or total if 'USDC' in current_balance and isinstance(current_balance['USDC'], dict): current_balance = current_balance['USDC'].get('total', 0) elif 'total' in current_balance: current_balance = current_balance['total'] else: # Fallback: try to find a float value in the dict for v in current_balance.values(): if isinstance(v, (float, int)): current_balance = v break else: current_balance = 0 # Get performance stats perf = stats.get_performance_stats() if not perf: await self._reply(update, "āŒ Could not get performance stats.") return # Format the message formatter = get_formatter() stats_text_parts = ["šŸ“Š Trading Statistics\n"] # Account Overview stats_text_parts.append("šŸ’° Account Overview:") stats_text_parts.append(f"• Current Balance: {formatter.format_price_with_symbol(current_balance)}") stats_text_parts.append(f"• Open Positions: {perf.get('open_positions', 0)}") stats_text_parts.append(f"• Total P&L: {formatter.format_price_with_symbol(perf.get('total_pnl', 0))}") # Performance Metrics stats_text_parts.append("\nšŸ† Performance Metrics:") stats_text_parts.append(f"• Total Trades: {perf.get('total_trades', 0)}") stats_text_parts.append(f"• Win Rate: {perf.get('win_rate', 0.0):.1f}% ({perf.get('total_wins', 0)}/{perf.get('total_trades', 0)})") stats_text_parts.append(f"• Trading Volume: {formatter.format_price_with_symbol(perf.get('total_entry_volume', 0.0))}") stats_text_parts.append(f"• Profit Factor: {perf.get('profit_factor', 0.0):.2f}") stats_text_parts.append(f"• Expectancy: {formatter.format_price_with_symbol(perf.get('expectancy', 0.0))}") # Largest Trades stats_text_parts.append("\nšŸ“ˆ Largest Trades:") stats_text_parts.append(f"• Largest Win: {formatter.format_price_with_symbol(perf.get('largest_win', 0.0))} ({perf.get('largest_win_pct', 0.0):+.2f}%) ({perf.get('largest_win_token', 'N/A')})") stats_text_parts.append(f"• Largest Loss: {formatter.format_price_with_symbol(perf.get('largest_loss', 0.0))} ({perf.get('largest_loss_pct', 0.0):+.2f}%) ({perf.get('largest_loss_token', 'N/A')})") # Best/Worst Tokens stats_text_parts.append("\nšŸ† Token Performance:") stats_text_parts.append(f"• Best Token: {perf.get('best_token', 'N/A')} {formatter.format_price_with_symbol(perf.get('best_token_pnl', 0.0))} ({perf.get('best_token_pct', 0.0):+.2f}%)") stats_text_parts.append(f"• Worst Token: {perf.get('worst_token', 'N/A')} {formatter.format_price_with_symbol(perf.get('worst_token_pnl', 0.0))} ({perf.get('worst_token_pct', 0.0):+.2f}%)") # Risk Metrics stats_text_parts.append("\nāš ļø Risk Metrics:") stats_text_parts.append(f"• Max Drawdown: {perf.get('max_drawdown_pct', 0.0):.2f}%") # Session Info stats_text_parts.append("\nā° Session Info:") stats_text_parts.append(f"• Stats Last Updated: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')}") await self._reply(update, "\n".join(stats_text_parts)) except Exception as e: logger.error(f"Error in stats command: {e}") await self._reply(update, f"āŒ Error getting statistics: {str(e)}")