Why Custom Alerts Matter
Off-track indicators require immediate attention. This tutorial shows you how to implement a comprehensive alert system that automatically detects and notifies stakeholders when indicators fall below thresholds, enabling timely interventions.

# alert_config.py
from datetime import datetime, timedelta
class AlertConfig:
    """Configuration for alert triggers and rules."""
    
    # Default thresholds
    THRESHOLDS = {
        'critical': {
            'achievement_below': 50,
            'status': 'off_track',
            'days_overdue': 30,
            'priority': 1
        },
        'warning': {
            'achievement_below': 70,
            'status': 'at_risk',
            'days_overdue': 14,
            'priority': 2
        },
        'info': {
            'achievement_below': 90,
            'status': 'on_track',
            'days_overdue': 7,
            'priority': 3
        }
    }
    
    # Alert escalation rules
    ESCALATION = {
        'critical': {
            'escalate_after_hours': 24,
            'escalate_to': ['manager', 'director'],
            'auto_assign': True
        },
        'warning': {
            'escalate_after_hours': 48,
            'escalate_to': ['coordinator'],
            'auto_assign': False
        },
        'info': {
            'escalate_after_hours': 72,
            'escalate_to': ['analyst'],
            'auto_assign': False
        }
    }
    
    # Alert delivery channels
    DELIVERY_CHANNELS = {
        'email': True,
        'dashboard': True,
        'slack': False,  # Can be enabled with Slack API
        'sms': False,    # Can be enabled with SMS API
        'teams': False   # Can be enabled with Teams API
    }
    
    # Alert templates
    TEMPLATES = {
        'critical': {
            'subject': "🚨 CRITICAL ALERT: {indicator_name} - {achievement}% Achievement",
            'body': """

Critical Alert: {indicator_name}

Status: {status}

Achievement: {achievement}% (Target: {target})

Category: {category}

Latest Data: {date}

Days Overdue: {days_overdue}

Suggested Actions:

• • Review implementation plan for {indicator_name}

• • Identify barriers to achievement

• • Develop corrective action plan

• • Schedule emergency review meeting

Assigned To: {assigned_to}

Due Date: {due_date}

“”” }, ‘warning’: { ‘subject’: “⚠️ Warning: {indicator_name} – {achievement}% Achievement”, ‘body’: “””

Warning: {indicator_name}

Status: {status}

Achievement: {achievement}% (Target: {target})

Category: {category}

Latest Data: {date}

Recommended Actions:

• • Monitor {indicator_name} more closely

• • Conduct performance review

• • Adjust implementation approach if needed

“”” }, ‘info’: { ‘subject’: “ℹ️ Info: {indicator_name} – {achievement}% Achievement”, ‘body’: “””

Information: {indicator_name}

Status: {status}

Achievement: {achievement}% (Target: {target})

Category: {category}

This indicator is performing well. Continue current implementation.

“”” } }

import pandas as pd
from datetime import datetime, timedelta
import json
import hashlib
class AlertEngine:
    """Core engine for detecting and generating alerts."""
    
    def __init__(self, config=AlertConfig):
        self.config = config
        self.alerts_history = []
        self.alert_log = []
    
    def check_indicators(self, df, previous_data=None):
        """Check all indicators for alert conditions."""
        alerts = []
        
        for _, row in df.iterrows():
            # Check each indicator against thresholds
            indicator_alerts = self._check_indicator(row, previous_data)
            if indicator_alerts:
                alerts.extend(indicator_alerts)
        
        return alerts
    
    def _check_indicator(self, row, previous_data=None):
        """Check a single indicator for alert conditions."""
        alerts = []
        
        # Get achievement and status
        achievement = row.get('achievement', 0)
        status = row.get('status', 'unknown')
        indicator_name = row.get('indicator_name', 'Unknown')
        category = row.get('category', 'General')
        target = row.get('target', 0)
        date = row.get('date', datetime.now())
        
        # Check critical threshold
        if achievement < self.config.THRESHOLDS['critical']['achievement_below']:
            alerts.append(self._create_alert(
                indicator_name, achievement, target, status, 
                category, date, 'critical', row
            ))
        
        # Check warning threshold
        elif achievement < self.config.THRESHOLDS['warning']['achievement_below']:
            alerts.append(self._create_alert(
                indicator_name, achievement, target, status,
                category, date, 'warning', row
            ))
        
        # Check info threshold
        elif achievement < self.config.THRESHOLDS['info']['achievement_below']: alerts.append(self._create_alert( indicator_name, achievement, target, status, category, date, 'info', row )) # Check for degradation (if previous data available) if previous_data is not None: degradation_alert = self._check_degradation(row, previous_data) if degradation_alert: alerts.append(degradation_alert) # Check for overdue (based on date) if 'date' in row and isinstance(date, pd.Timestamp): days_overdue = self._calculate_days_overdue(date) if days_overdue > self.config.THRESHOLDS['warning']['days_overdue']:
                alerts.append(self._create_overdue_alert(indicator_name, days_overdue, row))
        
        return alerts
    
    def _create_alert(self, indicator_name, achievement, target, status, 
                     category, date, level, row):
        """Create an alert dictionary."""
        alert_id = hashlib.md5(f"{indicator_name}_{datetime.now().isoformat()}".encode()).hexdigest()
        
        return {
            'id': alert_id,
            'indicator': indicator_name,
            'achievement': achievement,
            'target': target,
            'status': status,
            'category': category,
            'date': date.strftime('%Y-%m-%d') if isinstance(date, pd.Timestamp) else str(date),
            'level': level,
            'priority': self.config.THRESHOLDS[level]['priority'],
            'timestamp': datetime.now().isoformat(),
            'message': self._generate_alert_message(indicator_name, achievement, level),
            'details': row.to_dict() if hasattr(row, 'to_dict') else dict(row),
            'assigned_to': self._get_assignment(level),
            'due_date': self._calculate_due_date(level)
        }
    
    def _generate_alert_message(self, indicator_name, achievement, level):
        """Generate human-readable alert message."""
        messages = {
            'critical': f"🚨 {indicator_name} is critically off track at {achievement:.1f}%",
            'warning': f"⚠️ {indicator_name} needs attention at {achievement:.1f}%",
            'info': f"ℹ️ {indicator_name} is approaching target at {achievement:.1f}%"
        }
        return messages.get(level, f"Alert for {indicator_name}")
    
    def _get_assignment(self, level):
        """Get assignment based on alert level."""
        assignments = {
            'critical': 'Program Manager',
            'warning': 'M&E Coordinator',
            'info': 'Data Analyst'
        }
        return assignments.get(level, 'M&E Team')
    
    def _calculate_due_date(self, level):
        """Calculate due date for action."""
        hours = self.config.ESCALATION.get(level, {}).get('escalate_after_hours', 48)
        due_date = datetime.now() + timedelta(hours=hours)
        return due_date.strftime('%Y-%m-%d %H:%M')
    
    def _check_degradation(self, current_row, previous_data):
        """Check if indicator has degraded significantly."""
        # Implementation for comparing with previous data
        return None
    
    def _calculate_days_overdue(self, date):
        """Calculate days since last update."""
        if isinstance(date, pd.Timestamp):
            date = date.to_pydatetime()
        days = (datetime.now() - date).days
        return max(0, days)
    
    def _create_overdue_alert(self, indicator_name, days_overdue, row):
        """Create overdue alert."""
        return {
            'id': hashlib.md5(f"overdue_{indicator_name}".encode()).hexdigest(),
            'indicator': indicator_name,
            'days_overdue': days_overdue,
            'level': 'warning',
            'priority': 2,
            'timestamp': datetime.now().isoformat(),
            'message': f"📅 {indicator_name} has not been updated in {days_overdue} days",
            'details': row.to_dict() if hasattr(row, 'to_dict') else dict(row),
            'type': 'overdue'
        }
    
    def deduplicate_alerts(self, alerts):
        """Remove duplicate alerts."""
        seen = set()
        unique_alerts = []
        for alert in alerts:
            key = f"{alert['indicator']}_{alert['level']}"
            if key not in seen:
                seen.add(key)
                unique_alerts.append(alert)
        return unique_alerts
    
    def categorize_alerts(self, alerts):
        """Categorize alerts by severity level."""
        categorized = {
            'critical': [],
            'warning': [],
            'info': []
        }
        
        for alert in alerts:
            level = alert.get('level', 'info')
            if level in categorized:
                categorized[level].append(alert)
        
        return categorized
    
    def log_alert(self, alert):
        """Log alert to history."""
        self.alerts_history.append(alert)
        self.alert_log.append({
            'timestamp': datetime.now().isoformat(),
            'alert': alert,
            'action': 'created'
        })

class AlertNotifier:
    """Handles alert delivery through multiple channels."""
    
    def __init__(self, config=AlertConfig):
        self.config = config
        self.sent_alerts = []
    
    def send_alert(self, alert, channel='email'):
        """Send alert through specified channel."""
        if channel == 'email':
            return self._send_email_alert(alert)
        elif channel == 'dashboard':
            return self._send_dashboard_alert(alert)
        elif channel == 'slack':
            return self._send_slack_alert(alert)
        else:
            print(f"Unsupported channel: {channel}")
            return False
    
    def _send_email_alert(self, alert):
        """Send alert via email."""
        # Import email sender from previous module
        from email_sender import EmailSender
        
        level = alert.get('level', 'info')
        template = self.config.TEMPLATES.get(level, {})
        indicator = alert.get('indicator', 'Unknown')
        achievement = alert.get('achievement', 0)
        target = alert.get('target', 0)
        status = alert.get('status', 'unknown')
        category = alert.get('category', 'General')
        date = alert.get('date', 'N/A')
        
        # Format email content
        subject = template.get('subject', f"Alert: {indicator}").format(
            indicator_name=indicator,
            achievement=achievement,
            target=target,
            status=status,
            category=category,
            date=date
        )
        
        body = template.get('body', '').format(
            indicator_name=indicator,
            achievement=achievement,
            target=target,
            status=status,
            category=category,
            date=date,
            days_overdue=alert.get('days_overdue', 0),
            assigned_to=alert.get('assigned_to', 'M&E Team'),
            due_date=alert.get('due_date', 'N/A')
        )
        
        # Determine recipients based on alert level
        recipients = self._get_recipients(level)
        
        # Send email
        sender = EmailSender()
        return sender.send_report(
            to_emails=recipients,
            subject=subject,
            attachments=None,
            recipient_name="M&E Team"
        )
    
    def _send_dashboard_alert(self, alert):
        """Log alert to dashboard."""
        # Store in memory for dashboard display
        self.sent_alerts.append(alert)
        return True
    
    def _send_slack_alert(self, alert):
        """Send alert to Slack."""
        # Implementation with Slack API
        # Requires slack_sdk library and webhook URL
        return True
    
    def _get_recipients(self, level):
        """Get email recipients based on alert level."""
        # This should be configured based on your organization
        recipients = {
            'critical': ['director@ngo.org', 'manager@ngo.org'],
            'warning': ['coordinator@ngo.org'],
            'info': ['analyst@ngo.org']
        }
        return recipients.get(level, ['me_team@ngo.org'])
    
    def send_batch_alerts(self, alerts):
        """Send multiple alerts as a summary."""
        if not alerts:
            return
        
        # Group by level
        levels = {'critical': [], 'warning': [], 'info': []}
        for alert in alerts:
            level = alert.get('level', 'info')
            levels[level].append(alert)
        
        # Send summary for each level
        for level, level_alerts in levels.items():
            if level_alerts:
                summary = self._create_alert_summary(level_alerts, level)
                self._send_email_alert(summary)
    
    def _create_alert_summary(self, alerts, level):
        """Create summary alert for multiple alerts."""
        summary = {
            'level': level,
            'indicator': f"{len(alerts)} {level} alerts",
            'achievement': 0,
            'target': 0,
            'status': level,
            'category': 'Multiple',
            'date': datetime.now().strftime('%Y-%m-%d'),
            'assigned_to': 'M&E Team',
            'due_date': (datetime.now() + timedelta(hours=24)).strftime('%Y-%m-%d %H:%M')
        }
        
        # Build alert list
        alert_list = []
        for alert in alerts:
            alert_list.append(f"- {alert['indicator']}: {alert['achievement']:.1f}%")
        
        summary['message'] = f"Summary of {level} alerts:\n" + "\n".join(alert_list)
        return summary

def create_alert_dashboard(alerts):
    """Create HTML widget for dashboard alerts."""
    
    if not alerts:
        return """

“”” categorized = AlertEngine().categorize_alerts(alerts) html = “””

“”” return html

class CompleteAlertSystem:
    """Complete system for monitoring and alerting."""
    
    def __init__(self, data_connector, config=AlertConfig):
        self.data_connector = data_connector
        self.engine = AlertEngine(config)
        self.notifier = AlertNotifier(config)
        self.alert_history = []
        self.alert_summary = {}
    
    def run_check(self, send_notifications=True):
        """Run complete alert check."""
        # Get data from connector
        df = self.data_connector.get_data()
        
        # Check alerts
        alerts = self.engine.check_indicators(df)
        
        # Deduplicate
        unique_alerts = self.engine.deduplicate_alerts(alerts)
        
        # Store history
        for alert in unique_alerts:
            self.alert_history.append(alert)
            self.engine.log_alert(alert)
        
        # Generate summary
        self.alert_summary = {
            'total': len(unique_alerts),
            'critical': len([a for a in unique_alerts if a.get('level') == 'critical']),
            'warning': len([a for a in unique_alerts if a.get('level') == 'warning']),
            'info': len([a for a in unique_alerts if a.get('level') == 'info']),
            'timestamp': datetime.now().isoformat()
        }
        
        # Send notifications
        if send_notifications and unique_alerts:
            self.notifier.send_batch_alerts(unique_alerts)
            
            # Send individual critical alerts
            for alert in unique_alerts:
                if alert.get('level') == 'critical':
                    self.notifier.send_alert(alert, 'email')
                    self.notifier.send_alert(alert, 'dashboard')
        
        return unique_alerts
    
    def get_active_alerts(self):
        """Get currently active alerts."""
        return self.alert_history[-20:]  # Last 20 alerts
    
    def get_alert_summary(self):
        """Get alert summary statistics."""
        return self.alert_summary
    
    def get_alert_history(self, days=7):
        """Get alert history for specified days."""
        cutoff = datetime.now() - timedelta(days=days)
        return [a for a in self.alert_history 
                if datetime.fromisoformat(a.get('timestamp', '')).replace(tzinfo=None) > cutoff]
    
    def clear_resolved_alerts(self):
        """Clear resolved alerts (manual intervention)."""
        # In a real system, this would mark alerts as resolved
        pass
    
    def generate_alert_report(self):
        """Generate alert summary report."""
        alerts = self.alert_history[-30:]  # Last 30 days
        
        report = {
            'timestamp': datetime.now().isoformat(),
            'total_alerts': len(alerts),
            'active_alerts': len([a for a in alerts if not a.get('resolved', False)]),
            'by_level': {
                'critical': len([a for a in alerts if a.get('level') == 'critical']),
                'warning': len([a for a in alerts if a.get('level') == 'warning']),
                'info': len([a for a in alerts if a.get('level') == 'info'])
            },
            'by_indicator': self._count_by_indicator(alerts),
            'trend': self._calculate_trend(alerts)
        }
        
        return report
    
    def _count_by_indicator(self, alerts):
        """Count alerts by indicator."""
        counts = {}
        for alert in alerts:
            indicator = alert.get('indicator', 'Unknown')
            counts[indicator] = counts.get(indicator, 0) + 1
        return sorted(counts.items(), key=lambda x: x[1], reverse=True)[:10]
    
    def _calculate_trend(self, alerts):
        """Calculate alert trend."""
        if len(alerts) < 7: return "insufficient_data" # Compare last 7 days vs previous 7 days now = datetime.now() last_week = sum(1 for a in alerts if datetime.fromisoformat(a.get('timestamp', '')).replace(tzinfo=None) > now - timedelta(days=7))
        prev_week = sum(1 for a in alerts 
                       if datetime.fromisoformat(a.get('timestamp', '')).replace(tzinfo=None) < now - timedelta(days=7) and datetime.fromisoformat(a.get('timestamp', '')).replace(tzinfo=None) > now - timedelta(days=14))
        
        if last_week < prev_week * 0.8: return "decreasing" elif last_week > prev_week * 1.2:
            return "increasing"
        else:
            return "stable"

# main_alert_system.py
import pandas as pd
from datetime import datetime, timedelta
import random
def create_sample_data():
    """Create sample data with varied achievement levels."""
    dates = [(datetime.now() - timedelta(days=i)).strftime('%Y-%m-%d') for i in range(30)]
    
    indicators = [
        ('Children Vaccinated', 5000, 'Health'),
        ('Teachers Trained', 200, 'Education'),
        ('Schools Reached', 100, 'Education'),
        ('Community Events', 50, 'Community'),
        ('Health Centers', 30, 'Health'),
        ('Water Access', 80, 'WASH')
    ]
    
    data = []
    for date in dates:
        for name, target, category in indicators:
            # Random achievement between 40% and 100%
            achievement_factor = random.uniform(0.4, 1.0)
            actual = round(target * achievement_factor)
            
            # Assign status based on achievement
            if achievement_factor >= 0.9:
                status = 'on_track'
            elif achievement_factor >= 0.7:
                status = 'at_risk'
            else:
                status = 'off_track'
            
            data.append({
                'indicator_name': name,
                'target': target,
                'actual': actual,
                'date': date,
                'status': status,
                'category': category,
                'achievement': round(achievement_factor * 100, 1)
            })
    
    df = pd.DataFrame(data)
    df.to_excel('indicator_data.xlsx', index=False)
    return df
def main():
    # Step 1: Create sample data
    print("Creating sample data...")
    df = create_sample_data()
    print(f"Created {len(df)} records")
    
    # Step 2: Initialize data connector
    from data_connector import DashboardDataConnector
    connector = DashboardDataConnector('excel', 'indicator_data.xlsx', refresh_interval=30)
    
    # Step 3: Initialize alert system
    alert_system = CompleteAlertSystem(connector)
    
    # Step 4: Run alert check
    print("\nRunning alert check...")
    alerts = alert_system.run_check(send_notifications=False)  # Set to True for actual email
    
    # Step 5: Display results
    print(f"\nFound {len(alerts)} alerts:")
    summary = alert_system.get_alert_summary()
    print(f"  - Critical: {summary.get('critical', 0)}")
    print(f"  - Warning: {summary.get('warning', 0)}")
    print(f"  - Info: {summary.get('info', 0)}")
    
    # Step 6: Display detailed alerts
    if alerts:
        print("\nDetailed Alerts:")
        for alert in alerts[:5]:  # Show first 5
            print(f"  {alert.get('message')}")
            if alert.get('level') == 'critical':
                print(f"    → Immediate action required for {alert.get('indicator')}")
                print(f"    → Assigned to: {alert.get('assigned_to')}")
                print(f"    → Due: {alert.get('due_date')}")
    
    # Step 7: Generate alert report
    print("\nGenerating alert report...")
    report = alert_system.generate_alert_report()
    print(f"  Total alerts (30 days): {report['total_alerts']}")
    print(f"  Active alerts: {report['active_alerts']}")
    print(f"  Trend: {report['trend']}")
    
    # Step 8: Create dashboard widget
    from alert_dashboard_widget import create_alert_dashboard
    widget_html = create_alert_dashboard(alerts)
    
    # Save widget to file
    with open('alert_widget.html', 'w') as f:
        f.write(widget_html)
    print("\nAlert widget saved to alert_widget.html")
if __name__ == "__main__":
    main()

Master Alert Systems for M&E

The AI Agents for Evaluators Certificate teaches you to build complete M&E solutions including smart alert systems, automated reporting, and stakeholder management.

Course Features: 32 lectures · Lifetime access · Certificate included · 241 students enrolled

Enroll Now – $249 Lifetime Access

Bundle with AI in M&E course and save 30%