Spaces:
Running
Running
| # MCP Alert Engine Module | |
| """ | |
| Smart Alert and Notification Engine for MCP Integration. | |
| Features: | |
| - Threshold-based alerts | |
| - Anomaly detection alerts | |
| - Trend-based alerts | |
| - Scheduled monitoring | |
| - Alert prioritization | |
| """ | |
| from dataclasses import dataclass, field | |
| from typing import List, Dict, Optional, Any, Callable | |
| from enum import Enum | |
| from datetime import datetime, timedelta | |
| import json | |
| class AlertPriority(Enum): | |
| """Priority levels for alerts""" | |
| LOW = "low" | |
| MEDIUM = "medium" | |
| HIGH = "high" | |
| CRITICAL = "critical" | |
| class AlertType(Enum): | |
| """Types of alerts""" | |
| THRESHOLD = "threshold" | |
| ANOMALY = "anomaly" | |
| TREND = "trend" | |
| COMPARISON = "comparison" | |
| MISSING_DATA = "missing_data" | |
| CUSTOM = "custom" | |
| class Alert: | |
| """A single alert""" | |
| id: str | |
| type: AlertType | |
| priority: AlertPriority | |
| title: str | |
| message: str | |
| metric: str | |
| current_value: Any | |
| threshold_value: Optional[Any] = None | |
| timestamp: datetime = field(default_factory=datetime.now) | |
| metadata: Dict = field(default_factory=dict) | |
| suggested_action: Optional[str] = None | |
| class AlertRule: | |
| """Definition of an alert rule""" | |
| name: str | |
| metric: str | |
| condition: str # 'gt', 'lt', 'eq', 'gte', 'lte', 'change_pct', 'anomaly' | |
| threshold: Any | |
| priority: AlertPriority = AlertPriority.MEDIUM | |
| message_template: str = "" | |
| enabled: bool = True | |
| class AlertEngine: | |
| """ | |
| Enterprise Alert Engine MCP. | |
| Provides intelligent alerting: | |
| - Threshold monitoring | |
| - Anomaly detection | |
| - Trend analysis | |
| - Smart prioritization | |
| """ | |
| def __init__(self): | |
| self.rules: List[AlertRule] = [] | |
| self.alert_history: List[Alert] = [] | |
| self._alert_counter = 0 | |
| def add_rule(self, rule: AlertRule) -> None: | |
| """Add an alert rule""" | |
| self.rules.append(rule) | |
| def add_rules(self, rules: List[Dict]) -> None: | |
| """Add multiple rules from dict definitions""" | |
| for rule_dict in rules: | |
| rule = AlertRule( | |
| name=rule_dict.get('name', 'Unnamed Rule'), | |
| metric=rule_dict.get('metric', ''), | |
| condition=rule_dict.get('condition', 'gt'), | |
| threshold=rule_dict.get('threshold', 0), | |
| priority=AlertPriority(rule_dict.get('priority', 'medium')), | |
| message_template=rule_dict.get('message', ''), | |
| enabled=rule_dict.get('enabled', True) | |
| ) | |
| self.rules.append(rule) | |
| def evaluate( | |
| self, | |
| data: Any, | |
| context: Optional[Dict] = None | |
| ) -> Dict[str, Any]: | |
| """ | |
| Evaluate data against all rules and generate alerts. | |
| Args: | |
| data: DataFrame or dict with metrics | |
| context: Additional context (previous values, etc.) | |
| Returns: | |
| Generated alerts | |
| """ | |
| try: | |
| import pandas as pd | |
| import numpy as np | |
| if isinstance(data, pd.DataFrame): | |
| df = data.copy() | |
| else: | |
| df = pd.DataFrame([data]) if isinstance(data, dict) else pd.DataFrame(data) | |
| alerts = [] | |
| context = context or {} | |
| # Evaluate each rule | |
| for rule in self.rules: | |
| if not rule.enabled: | |
| continue | |
| if rule.metric not in df.columns: | |
| continue | |
| # Get current value (use latest or aggregate) | |
| if df[rule.metric].dtype in ['int64', 'float64']: | |
| current_value = df[rule.metric].iloc[-1] if len(df) > 0 else 0 | |
| else: | |
| current_value = df[rule.metric].iloc[-1] if len(df) > 0 else None | |
| # Evaluate condition | |
| triggered = self._evaluate_condition( | |
| current_value, | |
| rule.condition, | |
| rule.threshold, | |
| context.get(f'{rule.metric}_previous') | |
| ) | |
| if triggered: | |
| alert = self._create_alert(rule, current_value, context) | |
| alerts.append(alert) | |
| self.alert_history.append(alert) | |
| # Run automatic anomaly detection | |
| anomaly_alerts = self._detect_anomalies(df, context) | |
| alerts.extend(anomaly_alerts) | |
| # Run trend alerts | |
| trend_alerts = self._detect_concerning_trends(df, context) | |
| alerts.extend(trend_alerts) | |
| # Sort by priority | |
| priority_order = { | |
| AlertPriority.CRITICAL: 0, | |
| AlertPriority.HIGH: 1, | |
| AlertPriority.MEDIUM: 2, | |
| AlertPriority.LOW: 3 | |
| } | |
| alerts.sort(key=lambda a: priority_order[a.priority]) | |
| return { | |
| "success": True, | |
| "alerts": [self._alert_to_dict(a) for a in alerts], | |
| "alert_count": len(alerts), | |
| "critical_count": sum(1 for a in alerts if a.priority == AlertPriority.CRITICAL), | |
| "high_count": sum(1 for a in alerts if a.priority == AlertPriority.HIGH), | |
| "summary": self._generate_summary(alerts) | |
| } | |
| except Exception as e: | |
| return {"success": False, "error": str(e)} | |
| def _evaluate_condition( | |
| self, | |
| current: Any, | |
| condition: str, | |
| threshold: Any, | |
| previous: Optional[Any] = None | |
| ) -> bool: | |
| """Evaluate a single condition""" | |
| try: | |
| if current is None: | |
| return False | |
| if condition == 'gt': | |
| return current > threshold | |
| elif condition == 'lt': | |
| return current < threshold | |
| elif condition == 'eq': | |
| return current == threshold | |
| elif condition == 'gte': | |
| return current >= threshold | |
| elif condition == 'lte': | |
| return current <= threshold | |
| elif condition == 'change_pct' and previous is not None: | |
| if previous == 0: | |
| return False | |
| change = ((current - previous) / abs(previous)) * 100 | |
| return abs(change) > abs(threshold) | |
| elif condition == 'drop_pct' and previous is not None: | |
| if previous == 0: | |
| return False | |
| change = ((current - previous) / abs(previous)) * 100 | |
| return change < -abs(threshold) | |
| elif condition == 'increase_pct' and previous is not None: | |
| if previous == 0: | |
| return False | |
| change = ((current - previous) / abs(previous)) * 100 | |
| return change > abs(threshold) | |
| return False | |
| except Exception: | |
| return False | |
| def _create_alert( | |
| self, | |
| rule: AlertRule, | |
| current_value: Any, | |
| context: Dict | |
| ) -> Alert: | |
| """Create an alert from a triggered rule""" | |
| self._alert_counter += 1 | |
| message = rule.message_template or f"{rule.metric} triggered: {current_value} {rule.condition} {rule.threshold}" | |
| # Generate suggested action | |
| action = self._suggest_action(rule, current_value) | |
| return Alert( | |
| id=f"alert_{self._alert_counter}", | |
| type=AlertType.THRESHOLD, | |
| priority=rule.priority, | |
| title=rule.name, | |
| message=message, | |
| metric=rule.metric, | |
| current_value=current_value, | |
| threshold_value=rule.threshold, | |
| suggested_action=action, | |
| metadata={"rule_condition": rule.condition} | |
| ) | |
| def _suggest_action(self, rule: AlertRule, current_value: Any) -> str: | |
| """Generate suggested action for an alert""" | |
| actions = { | |
| 'revenue': "Review sales pipeline and customer acquisition strategies", | |
| 'churn': "Analyze customer feedback and implement retention campaigns", | |
| 'cost': "Review expense categories and identify optimization opportunities", | |
| 'inventory': "Check supply chain status and reorder thresholds", | |
| 'performance': "Schedule performance review meeting with stakeholders", | |
| 'error': "Check system logs and contact technical support", | |
| } | |
| metric_lower = rule.metric.lower() | |
| for key, action in actions.items(): | |
| if key in metric_lower: | |
| return action | |
| return f"Monitor {rule.metric} closely and investigate root cause" | |
| def _detect_anomalies(self, df, context: Dict) -> List[Alert]: | |
| """Automatically detect anomalies in numeric columns""" | |
| import numpy as np | |
| alerts = [] | |
| for col in df.select_dtypes(include=['int64', 'float64']).columns: | |
| if len(df) < 5: # Need minimum data points | |
| continue | |
| values = df[col].dropna().values | |
| if len(values) < 5: | |
| continue | |
| mean = np.mean(values) | |
| std = np.std(values) | |
| if std == 0: | |
| continue | |
| # Check latest value | |
| latest = values[-1] | |
| z_score = abs((latest - mean) / std) | |
| if z_score > 3: # Significant anomaly | |
| self._alert_counter += 1 | |
| alerts.append(Alert( | |
| id=f"alert_{self._alert_counter}", | |
| type=AlertType.ANOMALY, | |
| priority=AlertPriority.HIGH, | |
| title=f"Anomaly Detected: {col}", | |
| message=f"{col} value {latest:,.2f} is {z_score:.1f} standard deviations from mean ({mean:,.2f})", | |
| metric=col, | |
| current_value=latest, | |
| threshold_value=mean, | |
| suggested_action="Investigate sudden change in this metric", | |
| metadata={"z_score": z_score, "mean": mean, "std": std} | |
| )) | |
| elif z_score > 2: # Moderate anomaly | |
| self._alert_counter += 1 | |
| alerts.append(Alert( | |
| id=f"alert_{self._alert_counter}", | |
| type=AlertType.ANOMALY, | |
| priority=AlertPriority.MEDIUM, | |
| title=f"Unusual Value: {col}", | |
| message=f"{col} shows unusual value {latest:,.2f} (z-score: {z_score:.1f})", | |
| metric=col, | |
| current_value=latest, | |
| metadata={"z_score": z_score} | |
| )) | |
| return alerts | |
| def _detect_concerning_trends(self, df, context: Dict) -> List[Alert]: | |
| """Detect concerning trends in time series data""" | |
| import numpy as np | |
| alerts = [] | |
| for col in df.select_dtypes(include=['int64', 'float64']).columns: | |
| if len(df) < 3: | |
| continue | |
| values = df[col].dropna().values | |
| if len(values) < 3: | |
| continue | |
| # Calculate trend (simple linear regression slope) | |
| x = np.arange(len(values)) | |
| slope = np.polyfit(x, values, 1)[0] | |
| # Calculate percentage change over series | |
| if values[0] != 0: | |
| total_change = ((values[-1] - values[0]) / abs(values[0])) * 100 | |
| else: | |
| continue | |
| # Alert on significant negative trends | |
| if total_change < -20 and slope < 0: | |
| self._alert_counter += 1 | |
| alerts.append(Alert( | |
| id=f"alert_{self._alert_counter}", | |
| type=AlertType.TREND, | |
| priority=AlertPriority.HIGH if total_change < -30 else AlertPriority.MEDIUM, | |
| title=f"Declining Trend: {col}", | |
| message=f"{col} has declined {abs(total_change):.1f}% over the period", | |
| metric=col, | |
| current_value=values[-1], | |
| threshold_value=values[0], | |
| suggested_action="Analyze factors contributing to the decline", | |
| metadata={"change_pct": total_change, "slope": slope} | |
| )) | |
| return alerts | |
| def _alert_to_dict(self, alert: Alert) -> Dict: | |
| """Convert alert to dictionary""" | |
| return { | |
| "id": alert.id, | |
| "type": alert.type.value, | |
| "priority": alert.priority.value, | |
| "title": alert.title, | |
| "message": alert.message, | |
| "metric": alert.metric, | |
| "current_value": alert.current_value, | |
| "threshold_value": alert.threshold_value, | |
| "timestamp": alert.timestamp.isoformat(), | |
| "suggested_action": alert.suggested_action, | |
| "metadata": alert.metadata | |
| } | |
| def _generate_summary(self, alerts: List[Alert]) -> str: | |
| """Generate human-readable summary""" | |
| if not alerts: | |
| return "✅ No alerts - all metrics within normal ranges" | |
| critical = sum(1 for a in alerts if a.priority == AlertPriority.CRITICAL) | |
| high = sum(1 for a in alerts if a.priority == AlertPriority.HIGH) | |
| if critical > 0: | |
| return f"🚨 {critical} critical alert(s) require immediate attention" | |
| elif high > 0: | |
| return f"⚠️ {high} high-priority alert(s) detected" | |
| else: | |
| return f"ℹ️ {len(alerts)} alert(s) for review" | |
| # Convenience functions for direct MCP calls | |
| def evaluate_alerts(data, rules=None, context=None): | |
| """Evaluate data and generate alerts""" | |
| engine = AlertEngine() | |
| if rules: | |
| engine.add_rules(rules) | |
| return engine.evaluate(data, context) | |
| def detect_anomalies(data): | |
| """Quick anomaly detection""" | |
| engine = AlertEngine() | |
| result = engine.evaluate(data) | |
| return { | |
| "anomalies": [a for a in result.get("alerts", []) if a.get("type") == "anomaly"], | |
| "count": sum(1 for a in result.get("alerts", []) if a.get("type") == "anomaly") | |
| } | |
| def create_threshold_alert(metric, condition, threshold, priority="medium"): | |
| """Create a simple threshold alert rule""" | |
| return { | |
| "name": f"{metric} {condition} {threshold}", | |
| "metric": metric, | |
| "condition": condition, | |
| "threshold": threshold, | |
| "priority": priority | |
| } | |
| # Quick test | |
| if __name__ == "__main__": | |
| import pandas as pd | |
| # Test data with anomaly | |
| test_data = pd.DataFrame({ | |
| "revenue": [100000, 105000, 98000, 102000, 150000], # Last value is anomaly | |
| "customers": [500, 520, 510, 505, 515], | |
| "churn_rate": [0.05, 0.06, 0.07, 0.08, 0.12] # Increasing trend | |
| }) | |
| # Define rules | |
| rules = [ | |
| {"name": "High Revenue", "metric": "revenue", "condition": "gt", "threshold": 120000, "priority": "high"}, | |
| {"name": "High Churn", "metric": "churn_rate", "condition": "gt", "threshold": 0.10, "priority": "critical"}, | |
| ] | |
| result = evaluate_alerts(test_data, rules) | |
| print("Alert Evaluation Result:") | |
| print(f" Summary: {result['summary']}") | |
| print(f" Total Alerts: {result['alert_count']}") | |
| print(f"\nAlerts:") | |
| for alert in result['alerts']: | |
| print(f" [{alert['priority']}] {alert['title']}: {alert['message']}") | |
| if alert.get('suggested_action'): | |
| print(f" → Action: {alert['suggested_action']}") | |