alert_system.py

"""
Alert System Module

Real-time alerts for trades, price targets, and risk events.
"""

from typing import Dict, List, Any, Callable, Optional
from datetime import datetime
from dataclasses import dataclass
import threading

from logger import trading_logger, system_logger


@dataclass
class Alert:
    """Represents a trading alert."""
    alert_id: str
    alert_type: str  # 'price', 'trade', 'risk', 'system'
    symbol: Optional[str]
    message: str
    priority: str  # 'low', 'medium', 'high', 'critical'
    timestamp: datetime
    acknowledged: bool = False


class AlertSystem:
    """
    Real-time alert and notification system.
    """
    
    def __init__(self):
        """Initialize alert system."""
        self.alerts: List[Alert] = []
        self.alert_callbacks: List[Callable] = []
        self.alert_counter = 0
        self.lock = threading.Lock()
    
    def register_callback(self, callback: Callable):
        """
        Register a callback function to be called when alerts are triggered.
        
        Args:
            callback: Function to call with alert as argument
        """
        self.alert_callbacks.append(callback)
    
    def trigger_alert(
        self,
        alert_type: str,
        message: str,
        priority: str = 'medium',
        symbol: Optional[str] = None
    ) -> Alert:
        """
        Trigger a new alert.
        
        Args:
            alert_type: Type of alert
            message: Alert message
            priority: Alert priority level
            symbol: Optional stock symbol
            
        Returns:
            Created Alert object
        """
        with self.lock:
            self.alert_counter += 1
            alert_id = f"ALERT_{self.alert_counter:05d}"
        
        alert = Alert(
            alert_id=alert_id,
            alert_type=alert_type,
            symbol=symbol,
            message=message,
            priority=priority,
            timestamp=datetime.now()
        )
        
        self.alerts.append(alert)
        
        # Log based on priority
        log_message = f"[{alert_type.upper()}] {message}"
        if symbol:
            log_message = f"[{symbol}] {log_message}"
        
        if priority == 'critical':
            trading_logger.error(log_message)
        elif priority == 'high':
            trading_logger.warning(log_message)
        else:
            trading_logger.info(log_message)
        
        # Call registered callbacks
        for callback in self.alert_callbacks:
            try:
                callback(alert)
            except Exception as e:
                system_logger.error(f"Alert callback error: {e}")
        
        return alert
    
    def trade_executed_alert(self, symbol: str, side: str, quantity: int, price: float):
        """Alert for trade execution."""
        message = f"{side.upper()} {quantity} shares @ ${price:.2f}"
        self.trigger_alert('trade', message, 'high', symbol)
    
    def stop_hit_alert(self, symbol: str, price: float, pnl: float):
        """Alert for stop loss hit."""
        pnl_sign = '+' if pnl >= 0 else ''
        message = f"Stop hit @ ${price:.2f} (P&L: {pnl_sign}${pnl:.2f})"
        self.trigger_alert('trade', message, 'high', symbol)
    
    def target_hit_alert(self, symbol: str, price: float, pnl: float):
        """Alert for profit target hit."""
        message = f"Target hit @ ${price:.2f} (Profit: +${pnl:.2f})"
        self.trigger_alert('trade', message, 'high', symbol)
    
    def risk_limit_alert(self, message: str):
        """Alert for risk limit breach."""
        self.trigger_alert('risk', message, 'critical')
    
    def circuit_breaker_alert(self, reason: str):
        """Alert for circuit breaker activation."""
        self.trigger_alert('risk', f"CIRCUIT BREAKER: {reason}", 'critical')
    
    def daily_limit_alert(self, limit_type: str):
        """Alert for daily limit reached."""
        message = f"Daily {limit_type} limit reached"
        self.trigger_alert('risk', message, 'high')
    
    def price_alert(self, symbol: str, price: float, condition: str):
        """Alert for price level reached."""
        message = f"Price {condition}: ${price:.2f}"
        self.trigger_alert('price', message, 'medium', symbol)
    
    def get_unacknowledged_alerts(self) -> List[Alert]:
        """Get all unacknowledged alerts."""
        return [a for a in self.alerts if not a.acknowledged]
    
    def get_recent_alerts(self, count: int = 10) -> List[Alert]:
        """
        Get recent alerts.
        
        Args:
            count: Number of recent alerts to return
            
        Returns:
            List of recent alerts
        """
        return sorted(self.alerts, key=lambda a: a.timestamp, reverse=True)[:count]
    
    def acknowledge_alert(self, alert_id: str):
        """Mark an alert as acknowledged."""
        for alert in self.alerts:
            if alert.alert_id == alert_id:
                alert.acknowledged = True
                break
    
    def clear_acknowledged(self):
        """Remove acknowledged alerts."""
        self.alerts = [a for a in self.alerts if not a.acknowledged]
    
    def get_summary(self) -> Dict[str, Any]:
        """Get alert system summary."""
        unack = self.get_unacknowledged_alerts()
        critical = [a for a in unack if a.priority == 'critical']
        high = [a for a in unack if a.priority == 'high']
        
        return {
            'total_alerts': len(self.alerts),
            'unacknowledged': len(unack),
            'critical_unack': len(critical),
            'high_unack': len(high),
            'recent_alerts': [
                {
                    'type': a.alert_type,
                    'message': a.message,
                    'priority': a.priority,
                    'time': a.timestamp.strftime('%H:%M:%S')
                }
                for a in self.get_recent_alerts(5)
            ]
        }