#!/usr/bin/env python3
"""
Trading Statistics Tracker (Refactored Version)
Main class that coordinates between specialized manager components.
"""
import logging
from datetime import datetime, timezone
from typing import Dict, List, Any, Optional, Tuple
import math
import numpy as np
import uuid
from .database_manager import DatabaseManager
from .order_manager import OrderManager
from .trade_lifecycle_manager import TradeLifecycleManager
from .aggregation_manager import AggregationManager
from .performance_calculator import PerformanceCalculator
from src.utils.token_display_formatter import get_formatter
logger = logging.getLogger(__name__)
def _normalize_token_case(token: str) -> str:
"""Normalize token case for consistency."""
if any(c.isupper() for c in token):
return token # Keep original case for mixed-case tokens
else:
return token.upper() # Convert to uppercase for all-lowercase
class TradingStats:
"""Refactored trading statistics tracker using modular components."""
def __init__(self, db_path: str = "data/trading_stats.sqlite"):
"""Initialize with all manager components."""
# Initialize core database manager
self.db_manager = DatabaseManager(db_path)
# Initialize specialized managers
self.order_manager = OrderManager(self.db_manager)
self.trade_manager = TradeLifecycleManager(self.db_manager)
self.aggregation_manager = AggregationManager(self.db_manager)
self.performance_calculator = PerformanceCalculator(self.db_manager)
logger.info("š TradingStats initialized with modular components")
def close(self):
"""Close database connection."""
self.db_manager.close()
# =============================================================================
# COMPATIBILITY METHODS - Direct exposure of internal methods
# =============================================================================
def _get_metadata(self, key: str) -> Optional[str]:
"""Get metadata from database."""
return self.db_manager._get_metadata(key)
def _set_metadata(self, key: str, value: str):
"""Set metadata in database."""
return self.db_manager._set_metadata(key, value)
# =============================================================================
# DATABASE MANAGEMENT DELEGATION
# =============================================================================
def set_initial_balance(self, balance: float):
"""Set initial balance."""
return self.db_manager.set_initial_balance(balance)
def get_initial_balance(self) -> float:
"""Get initial balance."""
return self.db_manager.get_initial_balance()
def record_balance_snapshot(self, balance: float, unrealized_pnl: float = 0.0,
timestamp: Optional[str] = None, notes: Optional[str] = None):
"""Record balance snapshot."""
return self.db_manager.record_balance_snapshot(balance, unrealized_pnl, timestamp, notes)
def purge_old_balance_history(self, days_to_keep: int = 30) -> int:
"""Purge old balance history."""
return self.db_manager.purge_old_balance_history(days_to_keep)
def get_balance_history_record_count(self) -> int:
"""Get balance history record count."""
return self.db_manager.get_balance_history_record_count()
def purge_old_daily_aggregated_stats(self, days_to_keep: int = 365) -> int:
"""Purge old daily aggregated stats."""
return self.db_manager.purge_old_daily_aggregated_stats(days_to_keep)
# =============================================================================
# ORDER MANAGEMENT DELEGATION
# =============================================================================
def record_order_placed(self, symbol: str, side: str, order_type: str,
amount_requested: float, price: Optional[float] = None,
bot_order_ref_id: Optional[str] = None,
exchange_order_id: Optional[str] = None,
timestamp: Optional[str] = None) -> bool:
"""Record order placement."""
result = self.order_manager.record_order_placed(
symbol, side, order_type, amount_requested, price,
bot_order_ref_id, exchange_order_id
)
return result is not None
def update_order_exchange_id(self, bot_order_ref_id: str, exchange_order_id: str) -> bool:
"""Update order with exchange ID."""
return self.order_manager.update_order_exchange_id(bot_order_ref_id, exchange_order_id)
def record_order_filled(self, exchange_order_id: str, actual_amount: float,
actual_price: float, fees: float = 0.0,
timestamp: Optional[str] = None,
exchange_fill_id: Optional[str] = None) -> bool:
"""Record order fill."""
return self.order_manager.record_order_filled(
exchange_order_id, actual_amount, actual_price, fees, timestamp, exchange_fill_id
)
def record_order_cancelled(self, exchange_order_id: str, reason: str = "user_cancelled",
timestamp: Optional[str] = None) -> bool:
"""Record order cancellation."""
return self.order_manager.record_order_cancelled(exchange_order_id, reason, timestamp)
def update_order_status(self, exchange_order_id: str, new_status: str,
notes: Optional[str] = None, timestamp: Optional[str] = None) -> bool:
"""Update order status."""
return self.order_manager.update_order_status(exchange_order_id, new_status, notes, timestamp)
def get_order_by_exchange_id(self, exchange_order_id: str) -> Optional[Dict[str, Any]]:
"""Get order by exchange ID."""
return self.order_manager.get_order_by_exchange_id(exchange_order_id)
def get_order_by_bot_ref_id(self, bot_order_ref_id: str) -> Optional[Dict[str, Any]]:
"""Get order by bot reference ID."""
return self.order_manager.get_order_by_bot_ref_id(bot_order_ref_id)
def get_orders_by_symbol(self, symbol: str, limit: int = 50) -> List[Dict[str, Any]]:
"""Get orders by symbol."""
return self.order_manager.get_orders_by_symbol(symbol, limit)
def get_orders_by_status(self, status: str, limit: Optional[int] = 50,
order_type_filter: Optional[str] = None,
parent_bot_order_ref_id: Optional[str] = None) -> List[Dict[str, Any]]:
"""Get orders by status with optional filters."""
# OrderManager expects (status, order_type_filter, parent_bot_order_ref_id) without limit
return self.order_manager.get_orders_by_status(status, order_type_filter, parent_bot_order_ref_id)
def get_recent_orders(self, limit: int = 20) -> List[Dict[str, Any]]:
"""Get recent orders."""
return self.order_manager.get_recent_orders(limit)
def cleanup_old_cancelled_orders(self, days_old: int = 7) -> int:
"""Clean up old cancelled orders."""
return self.order_manager.cleanup_old_cancelled_orders(days_old)
# =============================================================================
# TRADE LIFECYCLE DELEGATION
# =============================================================================
def create_trade_lifecycle(self, symbol: str, side: str, entry_order_id: Optional[str] = None,
entry_bot_order_ref_id: Optional[str] = None,
stop_loss_price: Optional[float] = None,
take_profit_price: Optional[float] = None,
trade_type: str = 'manual') -> Optional[str]:
"""Create trade lifecycle."""
return self.trade_manager.create_trade_lifecycle(
symbol, side, entry_order_id, entry_bot_order_ref_id,
stop_loss_price, take_profit_price, trade_type
)
def update_trade_position_opened(self, lifecycle_id: str, entry_price: float,
entry_amount: float, exchange_fill_id: str) -> bool:
"""Update trade position opened."""
return self.trade_manager.update_trade_position_opened(
lifecycle_id, entry_price, entry_amount, exchange_fill_id
)
def update_trade_position_closed(self, lifecycle_id: str, exit_price: float,
realized_pnl: float, exchange_fill_id: str) -> bool:
"""Update trade position closed."""
return self.trade_manager.update_trade_position_closed(
lifecycle_id, exit_price, realized_pnl, exchange_fill_id
)
def update_trade_cancelled(self, lifecycle_id: str, reason: str = "order_cancelled") -> bool:
"""Update trade cancelled."""
return self.trade_manager.update_trade_cancelled(lifecycle_id, reason)
def link_stop_loss_to_trade(self, lifecycle_id: str, stop_loss_order_id: str,
stop_loss_price: float) -> bool:
"""Link stop loss to trade."""
return self.trade_manager.link_stop_loss_to_trade(
lifecycle_id, stop_loss_order_id, stop_loss_price
)
def link_take_profit_to_trade(self, lifecycle_id: str, take_profit_order_id: str,
take_profit_price: float) -> bool:
"""Link take profit to trade."""
return self.trade_manager.link_take_profit_to_trade(
lifecycle_id, take_profit_order_id, take_profit_price
)
def get_trade_by_lifecycle_id(self, lifecycle_id: str) -> Optional[Dict[str, Any]]:
"""Get trade by lifecycle ID."""
return self.trade_manager.get_trade_by_lifecycle_id(lifecycle_id)
def get_trade_by_symbol_and_status(self, symbol: str, status: str = 'position_opened') -> Optional[Dict[str, Any]]:
"""Get trade by symbol and status."""
return self.trade_manager.get_trade_by_symbol_and_status(symbol, status)
def get_open_positions(self, symbol: Optional[str] = None) -> List[Dict[str, Any]]:
"""Get open positions."""
return self.trade_manager.get_open_positions(symbol)
def get_trades_by_status(self, status: str, limit: int = 50) -> List[Dict[str, Any]]:
"""Get trades by status."""
return self.trade_manager.get_trades_by_status(status, limit)
def get_lifecycle_by_entry_order_id(self, entry_exchange_order_id: str, status: Optional[str] = None) -> Optional[Dict[str, Any]]:
"""Get lifecycle by entry order ID."""
return self.trade_manager.get_lifecycle_by_entry_order_id(entry_exchange_order_id, status)
def get_lifecycle_by_sl_order_id(self, sl_exchange_order_id: str, status: str = 'position_opened') -> Optional[Dict[str, Any]]:
"""Get lifecycle by stop loss order ID."""
return self.trade_manager.get_lifecycle_by_sl_order_id(sl_exchange_order_id, status)
def get_lifecycle_by_tp_order_id(self, tp_exchange_order_id: str, status: str = 'position_opened') -> Optional[Dict[str, Any]]:
"""Get lifecycle by take profit order ID."""
return self.trade_manager.get_lifecycle_by_tp_order_id(tp_exchange_order_id, status)
def get_pending_stop_loss_activations(self) -> List[Dict[str, Any]]:
"""Get pending stop loss activations."""
return self.trade_manager.get_pending_stop_loss_activations()
def cleanup_old_cancelled_trades(self, days_old: int = 7) -> int:
"""Clean up old cancelled trades."""
return self.trade_manager.cleanup_old_cancelled_trades(days_old)
def confirm_position_with_exchange(self, symbol: str, exchange_position_size: float,
exchange_open_orders: List[Dict]) -> bool:
"""Confirm position with exchange."""
return self.trade_manager.confirm_position_with_exchange(
symbol, exchange_position_size, exchange_open_orders
)
def update_trade_market_data(self, trade_lifecycle_id: str, **kwargs) -> bool:
"""Update trade market data."""
return self.trade_manager.update_trade_market_data(trade_lifecycle_id, **kwargs)
def get_recent_trades(self, limit: int = 10) -> List[Dict[str, Any]]:
"""Get recent trades."""
return self.trade_manager.get_recent_trades(limit)
def get_all_trades(self) -> List[Dict[str, Any]]:
"""Get all trades."""
return self.trade_manager.get_all_trades()
# =============================================================================
# AGGREGATION MANAGEMENT DELEGATION
# =============================================================================
def migrate_trade_to_aggregated_stats(self, trade_lifecycle_id: str):
"""Migrate trade to aggregated stats."""
return self.aggregation_manager.migrate_trade_to_aggregated_stats(trade_lifecycle_id)
def record_deposit(self, amount: float, timestamp: Optional[str] = None,
deposit_id: Optional[str] = None, description: Optional[str] = None):
"""Record deposit."""
return self.aggregation_manager.record_deposit(amount, timestamp, deposit_id, description)
def record_withdrawal(self, amount: float, timestamp: Optional[str] = None,
withdrawal_id: Optional[str] = None, description: Optional[str] = None):
"""Record withdrawal."""
return self.aggregation_manager.record_withdrawal(amount, timestamp, withdrawal_id, description)
def get_balance_adjustments_summary(self) -> Dict[str, Any]:
"""Get balance adjustments summary."""
return self.aggregation_manager.get_balance_adjustments_summary()
def get_daily_stats(self, limit: int = 10) -> List[Dict[str, Any]]:
"""Get daily stats."""
return self.aggregation_manager.get_daily_stats(limit)
def get_weekly_stats(self, limit: int = 10) -> List[Dict[str, Any]]:
"""Get weekly stats."""
return self.aggregation_manager.get_weekly_stats(limit)
def get_monthly_stats(self, limit: int = 10) -> List[Dict[str, Any]]:
"""Get monthly stats."""
return self.aggregation_manager.get_monthly_stats(limit)
# =============================================================================
# PERFORMANCE CALCULATION DELEGATION
# =============================================================================
def get_performance_stats(self) -> Dict[str, Any]:
"""Get performance stats."""
return self.performance_calculator.get_performance_stats()
def get_token_performance(self, limit: int = 20) -> List[Dict[str, Any]]:
"""Get token performance."""
return self.performance_calculator.get_token_performance(limit)
def get_balance_history(self, days: int = 30) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]:
"""Get balance history."""
return self.performance_calculator.get_balance_history(days)
def get_live_max_drawdown(self) -> Tuple[float, float]:
"""Get live max drawdown."""
return self.performance_calculator.get_live_max_drawdown()
def update_live_max_drawdown(self, current_balance: float) -> bool:
"""Update live max drawdown."""
return self.performance_calculator.update_live_max_drawdown(current_balance)
def calculate_sharpe_ratio(self, days: int = 30) -> Optional[float]:
"""Calculate Sharpe ratio."""
return self.performance_calculator.calculate_sharpe_ratio(days)
def calculate_max_consecutive_losses(self) -> int:
"""Calculate max consecutive losses."""
return self.performance_calculator.calculate_max_consecutive_losses()
def get_risk_metrics(self) -> Dict[str, Any]:
"""Get risk metrics."""
return self.performance_calculator.get_risk_metrics()
def get_period_performance(self, start_date: str, end_date: str) -> Dict[str, Any]:
"""Get period performance."""
return self.performance_calculator.get_period_performance(start_date, end_date)
def get_recent_performance_trend(self, days: int = 7) -> Dict[str, Any]:
"""Get recent performance trend."""
return self.performance_calculator.get_recent_performance_trend(days)
# =============================================================================
# COMPATIBILITY METHODS - Legacy API Support
# =============================================================================
def get_basic_stats(self, current_balance: Optional[float] = None) -> Dict[str, Any]:
"""Get basic trading statistics from DB, primarily using aggregated tables."""
# Get counts of open positions (trades that are not yet migrated)
open_positions_count = self._get_open_positions_count_from_db()
# Get overall aggregated stats from token_stats table
query_token_stats_summary = """
SELECT
SUM(total_realized_pnl) as total_pnl_from_cycles,
SUM(total_completed_cycles) as total_completed_cycles_sum,
MIN(first_cycle_closed_at) as overall_first_cycle_closed,
MAX(last_cycle_closed_at) as overall_last_cycle_closed
FROM token_stats
"""
token_stats_summary = self.db_manager._fetchone_query(query_token_stats_summary)
total_pnl_from_cycles = token_stats_summary['total_pnl_from_cycles'] if token_stats_summary and token_stats_summary['total_pnl_from_cycles'] is not None else 0.0
total_completed_cycles_sum = token_stats_summary['total_completed_cycles_sum'] if token_stats_summary and token_stats_summary['total_completed_cycles_sum'] is not None else 0
# Total trades considered as sum of completed cycles and currently open positions
total_trades_redefined = total_completed_cycles_sum + open_positions_count
initial_balance_str = self._get_metadata('initial_balance')
initial_balance = float(initial_balance_str) if initial_balance_str else 0.0
start_date_iso = self._get_metadata('start_date')
start_date_obj = datetime.fromisoformat(start_date_iso) if start_date_iso else datetime.now(timezone.utc)
days_active = (datetime.now(timezone.utc) - start_date_obj).days + 1
# 'last_trade' timestamp could be the last update to token_stats or an open trade
last_activity_ts = token_stats_summary['overall_last_cycle_closed'] if token_stats_summary else None
last_open_trade_ts_row = self.db_manager._fetchone_query("SELECT MAX(updated_at) as last_update FROM trades WHERE status = 'position_opened'")
if last_open_trade_ts_row and last_open_trade_ts_row['last_update']:
if not last_activity_ts or datetime.fromisoformat(last_open_trade_ts_row['last_update']) > datetime.fromisoformat(last_activity_ts):
last_activity_ts = last_open_trade_ts_row['last_update']
return {
'total_trades': total_trades_redefined,
'completed_trades': total_completed_cycles_sum,
'initial_balance': initial_balance,
'total_pnl': total_pnl_from_cycles,
'days_active': days_active,
'start_date': start_date_obj.strftime('%Y-%m-%d'),
'last_trade': last_activity_ts,
'open_positions_count': open_positions_count
}
def _get_open_positions_count_from_db(self) -> int:
"""Get count of open positions from trades table."""
row = self.db_manager._fetchone_query("SELECT COUNT(DISTINCT symbol) as count FROM trades WHERE status = 'position_opened'")
return row['count'] if row else 0
def get_token_detailed_stats(self, token: str) -> Dict[str, Any]:
"""Get detailed statistics for a specific token."""
try:
# Normalize token case
upper_token = _normalize_token_case(token)
# Get aggregated stats from token_stats table
token_agg_stats = self.db_manager._fetchone_query(
"SELECT * FROM token_stats WHERE token = ?", (upper_token,)
)
# Get open trades for this token
open_trades_for_token = self.db_manager._fetch_query(
"SELECT * FROM trades WHERE status = 'position_opened' AND symbol LIKE ? ORDER BY position_opened_at DESC",
(f"{upper_token}/%",)
)
# Initialize performance stats
perf_stats = {
'completed_trades': 0,
'total_pnl': 0.0,
'pnl_percentage': 0.0,
'win_rate': 0.0,
'profit_factor': 0.0,
'avg_win': 0.0,
'avg_loss': 0.0,
'largest_win': 0.0,
'largest_loss': 0.0,
'expectancy': 0.0,
'total_wins': 0,
'total_losses': 0,
'completed_entry_volume': 0.0,
'completed_exit_volume': 0.0,
'total_cancelled': 0,
'total_duration_seconds': 0,
'avg_trade_duration': "N/A"
}
if token_agg_stats:
total_cycles = token_agg_stats.get('total_completed_cycles', 0)
winning_cycles = token_agg_stats.get('winning_cycles', 0)
losing_cycles = token_agg_stats.get('losing_cycles', 0)
sum_winning_pnl = token_agg_stats.get('sum_of_winning_pnl', 0.0)
sum_losing_pnl = token_agg_stats.get('sum_of_losing_pnl', 0.0)
perf_stats.update({
'completed_trades': total_cycles,
'total_pnl': token_agg_stats.get('total_realized_pnl', 0.0),
'win_rate': (winning_cycles / total_cycles * 100) if total_cycles > 0 else 0.0,
'profit_factor': (sum_winning_pnl / sum_losing_pnl) if sum_losing_pnl > 0 else float('inf') if sum_winning_pnl > 0 else 0.0,
'avg_win': (sum_winning_pnl / winning_cycles) if winning_cycles > 0 else 0.0,
'avg_loss': (sum_losing_pnl / losing_cycles) if losing_cycles > 0 else 0.0,
'largest_win': token_agg_stats.get('largest_winning_cycle_pnl', 0.0),
'largest_loss': token_agg_stats.get('largest_losing_cycle_pnl', 0.0),
'total_wins': winning_cycles,
'total_losses': losing_cycles,
'completed_entry_volume': token_agg_stats.get('total_entry_volume', 0.0),
'completed_exit_volume': token_agg_stats.get('total_exit_volume', 0.0),
'total_cancelled': token_agg_stats.get('total_cancelled_cycles', 0),
'total_duration_seconds': token_agg_stats.get('total_duration_seconds', 0)
})
# Calculate expectancy
win_rate_decimal = perf_stats['win_rate'] / 100
perf_stats['expectancy'] = (perf_stats['avg_win'] * win_rate_decimal) - (perf_stats['avg_loss'] * (1 - win_rate_decimal))
# Format average trade duration
if total_cycles > 0:
avg_duration_seconds = token_agg_stats.get('total_duration_seconds', 0) / total_cycles
perf_stats['avg_trade_duration'] = self._format_duration(avg_duration_seconds)
# Calculate open positions summary
open_positions_summary = []
total_open_value = 0.0
total_open_unrealized_pnl = 0.0
for op_trade in open_trades_for_token:
open_positions_summary.append({
'lifecycle_id': op_trade.get('trade_lifecycle_id'),
'side': op_trade.get('position_side'),
'amount': op_trade.get('current_position_size'),
'entry_price': op_trade.get('entry_price'),
'mark_price': op_trade.get('mark_price'),
'unrealized_pnl': op_trade.get('unrealized_pnl'),
'opened_at': op_trade.get('position_opened_at')
})
total_open_value += op_trade.get('value', 0.0)
total_open_unrealized_pnl += op_trade.get('unrealized_pnl', 0.0)
# Get open orders count for this token
open_orders_count_row = self.db_manager._fetchone_query(
"SELECT COUNT(*) as count FROM orders WHERE symbol LIKE ? AND status IN ('open', 'submitted', 'pending_trigger')",
(f"{upper_token}/%",)
)
current_open_orders_for_token = open_orders_count_row['count'] if open_orders_count_row else 0
effective_total_trades = perf_stats['completed_trades'] + len(open_trades_for_token)
return {
'token': upper_token,
'performance': perf_stats,
'open_positions': {
'count': len(open_trades_for_token),
'total_value': total_open_value,
'total_unrealized_pnl': total_open_unrealized_pnl,
'positions': open_positions_summary
},
'summary': {
'total_trades': effective_total_trades,
'open_orders': current_open_orders_for_token,
}
}
except Exception as e:
logger.error(f"ā Error getting detailed stats for {token}: {e}")
return {}
def _format_duration(self, seconds: float) -> str:
"""Format duration in seconds to a human-readable string."""
if seconds <= 0:
return "0s"
days = int(seconds // 86400)
hours = int((seconds % 86400) // 3600)
minutes = int((seconds % 3600) // 60)
secs = int(seconds % 60)
parts = []
if days > 0:
parts.append(f"{days}d")
if hours > 0:
parts.append(f"{hours}h")
if minutes > 0:
parts.append(f"{minutes}m")
if secs > 0 or not parts:
parts.append(f"{secs}s")
return " ".join(parts)
def format_stats_message(self, current_balance: Optional[float] = None) -> str:
"""Format stats for Telegram display using data from DB."""
try:
basic = self.get_basic_stats(current_balance)
perf = self.get_performance_stats()
risk = self.get_risk_metrics()
formatter = get_formatter()
effective_current_balance = current_balance if current_balance is not None else (basic['initial_balance'] + basic['total_pnl'])
initial_bal = basic['initial_balance']
total_pnl_val = effective_current_balance - initial_bal if initial_bal > 0 and current_balance is not None else basic['total_pnl']
total_return_pct = (total_pnl_val / initial_bal * 100) if initial_bal > 0 else 0.0
pnl_emoji = "š¢" if total_pnl_val >= 0 else "š“"
open_positions_count = basic['open_positions_count']
stats_text_parts = []
stats_text_parts.append(f"š Trading Statistics\n")
# Account Overview
stats_text_parts.append(f"\nš° Account Overview:")
stats_text_parts.append(f"ā¢ Current Balance: {formatter.format_price_with_symbol(effective_current_balance)}")
stats_text_parts.append(f"ā¢ Initial Balance: {formatter.format_price_with_symbol(initial_bal)}")
stats_text_parts.append(f"ā¢ Open Positions: {open_positions_count}")
stats_text_parts.append(f"ā¢ {pnl_emoji} Total P&L: {formatter.format_price_with_symbol(total_pnl_val)} ({total_return_pct:+.2f}%)")
stats_text_parts.append(f"ā¢ Days Active: {basic['days_active']}\n")
# Trading Performance
stats_text_parts.append(f"š Trading Performance:")
stats_text_parts.append(f"ā¢ Total Cycles: {perf['total_completed_cycles']}")
stats_text_parts.append(f"ā¢ Win Rate: {perf['win_rate']:.1f}% ({perf['total_winning_cycles']}/{perf['total_completed_cycles']})")
stats_text_parts.append(f"ā¢ Profit Factor: {perf['profit_factor']:.2f}")
stats_text_parts.append(f"ā¢ Expectancy: {formatter.format_price_with_symbol(perf['expectancy'])}")
return "\n".join(stats_text_parts)
except Exception as e:
logger.error(f"ā Error formatting stats message: {e}")
return f"ā Error generating statistics: {str(e)}"
# =============================================================================
# CONVENIENCE METHODS & HIGH-LEVEL OPERATIONS
# =============================================================================
def process_trade_complete_cycle(self, symbol: str, side: str, entry_price: float,
exit_price: float, amount: float,
timestamp: Optional[str] = None) -> str:
"""Process a complete trade cycle in one operation."""
# Create lifecycle
lifecycle_id = self.create_trade_lifecycle(symbol, side, trade_type='complete_cycle')
if not lifecycle_id:
raise Exception("Failed to create trade lifecycle")
# Update to position opened
success = self.update_trade_position_opened(lifecycle_id, entry_price, amount, "manual_entry")
if not success:
raise Exception("Failed to update position opened")
# Calculate PnL
if side.lower() == 'buy':
realized_pnl = (exit_price - entry_price) * amount
else: # sell
realized_pnl = (entry_price - exit_price) * amount
# Update to position closed
success = self.update_trade_position_closed(lifecycle_id, exit_price, realized_pnl, "manual_exit")
if not success:
raise Exception("Failed to update position closed")
# Migrate to aggregated stats
self.migrate_trade_to_aggregated_stats(lifecycle_id)
logger.info(f"ā
Processed complete trade cycle: {symbol} {side.upper()} P&L: ${realized_pnl:.2f}")
return lifecycle_id
def get_summary_report(self) -> Dict[str, Any]:
"""Get comprehensive summary report."""
try:
perf_stats = self.get_performance_stats()
token_performance = self.get_token_performance(limit=10)
daily_stats = self.get_daily_stats(limit=7)
risk_metrics = self.get_risk_metrics()
balance_adjustments = self.get_balance_adjustments_summary()
# Get current positions
open_positions = self.get_open_positions()
return {
'performance_stats': perf_stats,
'top_tokens': token_performance,
'recent_daily_stats': daily_stats,
'risk_metrics': risk_metrics,
'balance_adjustments': balance_adjustments,
'open_positions_count': len(open_positions),
'open_positions': open_positions,
'generated_at': datetime.now(timezone.utc).isoformat()
}
except Exception as e:
logger.error(f"ā Error generating summary report: {e}")
return {'error': str(e)}
def health_check(self) -> Dict[str, Any]:
"""Perform health check on all components."""
try:
health = {
'database': 'ok',
'order_manager': 'ok',
'trade_manager': 'ok',
'aggregation_manager': 'ok',
'performance_calculator': 'ok',
'overall': 'ok'
}
# Test database connection
self.db_manager._fetch_query("SELECT 1")
# Test each component with basic operations
self.get_recent_orders(limit=1)
self.get_recent_trades(limit=1)
self.get_daily_stats(limit=1)
self.get_performance_stats()
return health
except Exception as e:
logger.error(f"ā Health check failed: {e}")
return {'overall': 'error', 'error': str(e)}