import logging
from typing import Dict, Any, List
from datetime import datetime
from telegram import Update
from telegram.ext import ContextTypes
from .base import InfoCommandsBase
from src.utils.token_display_formatter import get_formatter, normalize_token_case
logger = logging.getLogger(__name__)
class StatsCommands(InfoCommandsBase):
"""Handles all statistics-related commands."""
def __init__(self, trading_engine, notification_manager):
super().__init__(trading_engine, notification_manager)
self.formatter = get_formatter()
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
# If a token is provided, show detailed stats for that token
if context.args and len(context.args) > 0:
token_name = normalize_token_case(context.args[0])
stats_message = await self.trading_engine.stats.format_token_stats_message(token_name)
await self._reply(update, stats_message)
return
# For the general /stats command, create a detailed report
balance_info = self.trading_engine.get_balance()
current_balance = float(balance_info.get('total', {}).get('USDC', 0.0))
report = self.trading_engine.stats.get_summary_report()
if not report or not report.get('performance_stats'):
await self._reply(update, "❌ Trading stats not available yet.")
return
stats_message = await self._format_comprehensive_stats(report, current_balance)
await self._reply(update, stats_message, disable_web_page_preview=True)
except Exception as e:
logger.error(f"Error in stats command: {e}", exc_info=True)
await self._reply(update, f"❌ Error getting statistics: {str(e)}")
def _format_duration(self, seconds: float) -> str:
"""Helper to format seconds into a human-readable string."""
if seconds < 60:
return f"{seconds:.1f}s"
minutes = seconds / 60
if minutes < 60:
return f"{minutes:.1f}m"
hours = minutes / 60
if hours < 24:
return f"{hours:.1f}h"
days = hours / 24
return f"{days:.1f}d"
async def _format_comprehensive_stats(self, report: Dict[str, Any], current_balance: float) -> str:
"""Formats the main, detailed statistics message."""
perf = report.get('performance_stats', {})
risk = report.get('risk_metrics', {})
top_tokens = report.get('top_tokens', [])
# --- Header ---
initial_balance = perf.get('initial_balance', 0.0)
total_pnl = perf.get('total_pnl', 0.0)
total_return_pct = (total_pnl / initial_balance * 100) if initial_balance > 0 else 0.0
pnl_emoji = "🟢" if total_pnl >= 0 else "🔴"
header = [
"📊 Overall Trading Performance\n",
f"• Current Balance: {await self.formatter.format_price_with_symbol(current_balance)}",
f"• Initial Balance: {await self.formatter.format_price_with_symbol(initial_balance)}",
f"• {pnl_emoji} Total Net P&L: {await self.formatter.format_price_with_symbol(total_pnl)} ({total_return_pct:+.2f}%)"
]
# --- Core Metrics ---
max_drawdown_pct = risk.get('max_drawdown_percentage', 0.0)
drawdown_start_date_iso = risk.get('drawdown_start_date')
drawdown_date_str = ""
if drawdown_start_date_iso:
try:
drawdown_date = datetime.fromisoformat(drawdown_start_date_iso).strftime('%Y-%m-%d')
drawdown_date_str = f" (since {drawdown_date})"
except (ValueError, TypeError):
logger.warning(f"Could not parse drawdown start date: {drawdown_start_date_iso}")
sharpe_ratio = risk.get('sharpe_ratio')
sharpe_str = f"{sharpe_ratio:.2f}" if sharpe_ratio is not None else "N/A"
core_metrics = [
"\nCore Metrics:",
f"• Win Rate: {perf.get('win_rate', 0.0):.2f}% ({perf.get('total_wins', 0)}W / {perf.get('total_losses', 0)}L)",
f"• Profit Factor: {perf.get('profit_factor', 'N/A')}",
f"• Max Drawdown: {max_drawdown_pct:.2f}%{drawdown_date_str}",
f"• Sharpe Ratio: {sharpe_str}",
f"• Total Trades: {perf.get('total_trades', 0)}",
f"• Total Volume: {await self.formatter.format_price_with_symbol(perf.get('total_entry_volume', 0.0))}"
]
# --- P&L Analysis ---
pnl_analysis = [
"\nP&L Analysis:",
f"• Avg Profit per Trade: {await self.formatter.format_price_with_symbol(perf.get('avg_trade_pnl', 0.0))}",
f"• Avg Winning Trade: {await self.formatter.format_price_with_symbol(perf.get('avg_win_pnl', 0.0))}",
f"• Avg Losing Trade: {await self.formatter.format_price_with_symbol(perf.get('avg_loss_pnl', 0.0))}",
f"• Largest Win: {await self.formatter.format_price_with_symbol(perf.get('largest_win_pnl', 0.0))}",
f"• Largest Loss: {await self.formatter.format_price_with_symbol(perf.get('largest_loss_pnl', 0.0))}"
]
# --- ROE Analysis (with fixes) ---
best_roe = perf.get('best_roe_trade', {})
worst_roe = perf.get('worst_roe_trade', {})
roe_analysis = ["\nReturn on Equity (ROE) Analysis:"]
if best_roe:
roe_analysis.append(f"• Best ROE Trade: {best_roe.get('percentage', 0.0):+.2f}% ({best_roe.get('token')})")
if worst_roe:
roe_analysis.append(f"• Worst ROE Trade: {worst_roe.get('percentage', 0.0):+.2f}% ({worst_roe.get('token')})")
# --- Trade Duration ---
duration_analysis = [
"\nTrade Duration:",
f"• Avg Duration: {self._format_duration(perf.get('avg_trade_duration_seconds', 0.0))}",
f"• Avg Win Duration: {self._format_duration(perf.get('avg_win_duration_seconds', 0.0))}",
f"• Avg Loss Duration: {self._format_duration(perf.get('avg_loss_duration_seconds', 0.0))}"
]
# --- Top Movers ---
top_movers = ["\nTop Performing Tokens (by P&L):"]
if top_tokens:
# Sort by total_pnl descending
sorted_tokens = sorted(top_tokens, key=lambda x: x.get('total_pnl', 0.0), reverse=True)
for token_stat in sorted_tokens[:5]:
token_pnl = token_stat.get('total_pnl', 0.0)
token_pnl_emoji = "🟢" if token_pnl >= 0 else "🔴"
top_movers.append(
f"• {token_stat.get('token')}: {token_pnl_emoji} "
f"{await self.formatter.format_price_with_symbol(token_pnl)} "
f"({token_stat.get('win_rate', 0):.1f}% WR)"
)
else:
top_movers.append("• No token-specific data yet.")
# --- Footer ---
footer = [
"\nStatistics are based on completed trades. Use /positions for live data."
]
return "\n".join(header + core_metrics + pnl_analysis + roe_analysis + duration_analysis + top_movers + footer)