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