performance_analytics.py

"""
Performance Analytics Module

Comprehensive performance tracking and analysis for trading strategies.
"""

from typing import Dict, List, Any, Optional
from datetime import datetime, timedelta
from dataclasses import dataclass
import pandas as pd
import numpy as np

from logger import system_logger


@dataclass
class TradeStats:
    """Statistics for a single trade."""
    symbol: str
    entry_time: datetime
    exit_time: datetime
    entry_price: float
    exit_price: float
    quantity: int
    pnl: float
    pnl_pct: float
    hold_duration: timedelta
    trade_type: str  # 'win' or 'loss'


class PerformanceAnalytics:
    """
    Advanced performance analytics and statistics.
    """
    
    def __init__(self, database=None):
        """
        Initialize performance analytics.
        
        Args:
            database: Database instance for loading trade history
        """
        self.database = database
        self.trades: List[TradeStats] = []
    
    def add_trade(
        self,
        symbol: str,
        entry_time: datetime,
        exit_time: datetime,
        entry_price: float,
        exit_price: float,
        quantity: int
    ):
        """
        Add a completed trade for analysis.
        
        Args:
            symbol: Stock symbol
            entry_time: Entry timestamp
            exit_time: Exit timestamp
            entry_price: Entry price
            exit_price: Exit price
            quantity: Position size
        """
        pnl = (exit_price - entry_price) * quantity
        pnl_pct = ((exit_price / entry_price) - 1) * 100
        hold_duration = exit_time - entry_time
        trade_type = 'win' if pnl > 0 else 'loss'
        
        trade = TradeStats(
            symbol=symbol,
            entry_time=entry_time,
            exit_time=exit_time,
            entry_price=entry_price,
            exit_price=exit_price,
            quantity=quantity,
            pnl=pnl,
            pnl_pct=pnl_pct,
            hold_duration=hold_duration,
            trade_type=trade_type
        )
        
        self.trades.append(trade)
    
    def get_overall_stats(self) -> Dict[str, Any]:
        """Get overall performance statistics."""
        if not self.trades:
            return self._empty_stats()
        
        wins = [t for t in self.trades if t.trade_type == 'win']
        losses = [t for t in self.trades if t.trade_type == 'loss']
        
        total_trades = len(self.trades)
        win_count = len(wins)
        loss_count = len(losses)
        
        # Win rate
        win_rate = (win_count / total_trades * 100) if total_trades > 0 else 0
        
        # P&L stats
        total_pnl = sum(t.pnl for t in self.trades)
        gross_profit = sum(t.pnl for t in wins)
        gross_loss = abs(sum(t.pnl for t in losses))
        
        # Profit factor (gross profit / gross loss)
        profit_factor = (gross_profit / gross_loss) if gross_loss > 0 else 0
        
        # Average stats
        avg_win = (gross_profit / win_count) if win_count > 0 else 0
        avg_loss = (gross_loss / loss_count) if loss_count > 0 else 0
        avg_pnl = total_pnl / total_trades if total_trades > 0 else 0
        
        # Best/worst trades
        best_trade = max(self.trades, key=lambda t: t.pnl)
        worst_trade = min(self.trades, key=lambda t: t.pnl)
        
        # Hold time stats
        avg_hold_time = sum([t.hold_duration.total_seconds() for t in self.trades]) / total_trades
        avg_hold_hours = avg_hold_time / 3600
        
        # Consecutive wins/losses
        max_consecutive_wins = self._max_consecutive('win')
        max_consecutive_losses = self._max_consecutive('loss')
        
        return {
            'total_trades': total_trades,
            'win_count': win_count,
            'loss_count': loss_count,
            'win_rate': win_rate,
            'total_pnl': total_pnl,
            'gross_profit': gross_profit,
            'gross_loss': gross_loss,
            'profit_factor': profit_factor,
            'avg_win': avg_win,
            'avg_loss': avg_loss,
            'avg_pnl': avg_pnl,
            'avg_rr_ratio': (avg_win / avg_loss) if avg_loss > 0 else 0,
            'best_trade': best_trade.pnl,
            'best_trade_symbol': best_trade.symbol,
            'worst_trade': worst_trade.pnl,
            'worst_trade_symbol': worst_trade.symbol,
            'avg_hold_hours': avg_hold_hours,
            'max_consecutive_wins': max_consecutive_wins,
            'max_consecutive_losses': max_consecutive_losses
        }
    
    def get_stats_by_symbol(self) -> Dict[str, Dict[str, Any]]:
        """Get performance statistics broken down by symbol."""
        symbol_trades = {}
        
        for trade in self.trades:
            if trade.symbol not in symbol_trades:
                symbol_trades[trade.symbol] = []
            symbol_trades[trade.symbol].append(trade)
        
        stats_by_symbol = {}
        for symbol, trades in symbol_trades.items():
            wins = [t for t in trades if t.trade_type == 'win']
            total_pnl = sum(t.pnl for t in trades)
            win_rate = (len(wins) / len(trades) * 100) if trades else 0
            
            stats_by_symbol[symbol] = {
                'trades': len(trades),
                'win_rate': win_rate,
                'total_pnl': total_pnl,
                'avg_pnl': total_pnl / len(trades) if trades else 0
            }
        
        return stats_by_symbol
    
    def get_monthly_performance(self) -> List[Dict[str, Any]]:
        """Get performance broken down by month."""
        if not self.trades:
            return []
        
        # Group trades by month
        monthly = {}
        for trade in self.trades:
            month_key = trade.exit_time.strftime('%Y-%m')
            if month_key not in monthly:
                monthly[month_key] = []
            monthly[month_key].append(trade)
        
        # Calculate stats for each month
        results = []
        for month, trades in sorted(monthly.items()):
            total_pnl = sum(t.pnl for t in trades)
            wins = [t for t in trades if t.trade_type == 'win']
            win_rate = (len(wins) / len(trades) * 100) if trades else 0
            
            results.append({
                'month': month,
                'trades': len(trades),
                'pnl': total_pnl,
                'win_rate': win_rate
            })
        
        return results
    
    def calculate_sharpe_ratio(self, risk_free_rate: float = 0.0) -> float:
        """
        Calculate Sharpe ratio.
        
        Args:
            risk_free_rate: Risk-free rate (annualized)
            
        Returns:
            Sharpe ratio
        """
        if len(self.trades) < 2:
            return 0.0
        
        returns = [t.pnl_pct for t in self.trades]
        avg_return = np.mean(returns)
        std_return = np.std(returns)
        
        if std_return == 0:
            return 0.0
        
        # Annualize (assuming ~250 trading days)
        sharpe = (avg_return - risk_free_rate) / std_return
        return sharpe * np.sqrt(250)
    
    def _max_consecutive(self, trade_type: str) -> int:
        """Calculate maximum consecutive wins or losses."""
        max_consecutive = 0
        current_consecutive = 0
        
        for trade in self.trades:
            if trade.trade_type == trade_type:
                current_consecutive += 1
                max_consecutive = max(max_consecutive, current_consecutive)
            else:
                current_consecutive = 0
        
        return max_consecutive
    
    def _empty_stats(self) -> Dict[str, Any]:
        """Return empty stats structure."""
        return {
            'total_trades': 0,
            'win_count': 0,
            'loss_count': 0,
            'win_rate': 0.0,
            'total_pnl': 0.0,
            'gross_profit': 0.0,
            'gross_loss': 0.0,
            'profit_factor': 0.0,
            'avg_win': 0.0,
            'avg_loss': 0.0,
            'avg_pnl': 0.0,
            'avg_rr_ratio': 0.0,
            'best_trade': 0.0,
            'best_trade_symbol': 'N/A',
            'worst_trade': 0.0,
            'worst_trade_symbol': 'N/A',
            'avg_hold_hours': 0.0,
            'max_consecutive_wins': 0,
            'max_consecutive_losses': 0
        }