Deepvest/ProfilingAI

1# src/monitoring/notification_system.py2from ..core.market_states import MarketConditions3from typing import Callable, List, Dict4import pandas as pd5from datetime import datetime6 7class NotificationSystem:8    def __init__(self):9        self.subscribers = set()10        self.notification_rules = {}11        self.alert_history = []12        13    def add_notification_rule(self, event_type: str, condition: Callable, 14                            message_template: str):15        self.notification_rules[event_type] = {16            'condition': condition,17            'template': message_template18        }19        20    def check_portfolio_alerts(self, portfolio: pd.DataFrame, 21                             market_conditions: MarketConditions):22        alerts = []23        for event_type, rule in self.notification_rules.items():24            if rule['condition'](portfolio, market_conditions):25                alert = self._create_alert(event_type, rule['template'])26                alerts.append(alert)27                self.alert_history.append(alert)28        return alerts29        30    def _create_alert(self, event_type: str, template: str) -> Dict:31        return {32            'type': event_type,33            'message': template,34            'timestamp': datetime.now()35        }