Merge branch 'main' into issue-453-isolation-forest

Author: David-patrick-chuks

apps/data-processing/src/alert_notifier.py

@@ -1,122 +1,83 @@
-"""
-Alert Notifier module - Sends notifications for detected anomalies
-via multiple channels (Telegram, Webhooks).
-"""
-
 import os
-import json
-import logging
+import time
 import requests
-from typing import List, Dict, Any, Optional
-from datetime import datetime
-from src.anomaly_detector import AnomalyResult
 
-logger = logging.getLogger(__name__)
 
 class AlertNotifier:
-    """
-    Multichannel notification manager for system anomalies.
-    Supports Telegram Bot API and generic JSON webhooks.
-    """
-
-    def __init__(self, min_severity: float = 0.5):
-        """
-        Initialize the notifier.
-        Args:
-            min_severity: Minimum severity score (0.0-1.0) to trigger alerts.
-        """
-        self.min_severity = min_severity
-        # Telegram Config
-        self.tg_token = os.getenv("TELEGRAM_BOT_TOKEN")
-        self.tg_chat_id = os.getenv("TELEGRAM_CHANNEL_ID")
-        
-        # Webhook Config
-        self.webhook_url = os.getenv("ALERT_WEBHOOK_URL")
-        
-        # Status
-        self.has_telegram = bool(self.tg_token and self.tg_chat_id)
-        self.has_webhook = bool(self.webhook_url)
-        
-        if not (self.has_telegram or self.has_webhook):
-            logger.warning("AlertNotifier initialized with no active channels. Alerts will only be logged.")
-
-    def _send_telegram(self, message: str):
-        """Send message via Telegram Bot API."""
-        if not self.has_telegram:
+    def __init__(self):
+        self.telegram_bot_token = os.getenv('TELEGRAM_BOT_TOKEN')
+        self.telegram_channel_id = os.getenv('TELEGRAM_CHANNEL_ID')
+        self.webhook_urls = self._load_webhook_urls()
+        self.max_retries = int(os.getenv('WEBHOOK_MAX_RETRIES', '3'))
+        self.base_backoff_seconds = float(os.getenv('WEBHOOK_BACKOFF_SECONDS', '1'))
+
+    def _load_webhook_urls(self):
+        urls = []
+
+        single_url = os.getenv('ALERT_WEBHOOK_URL')
+        if single_url:
+            urls.append(single_url)
+
+        registry = os.getenv('ALERT_WEBHOOK_URLS', '')
+        if registry:
+            urls.extend([url.strip() for url in registry.split(',') if url.strip()])
+
+        return list(dict.fromkeys(urls))
+
+    def notify_anomaly(self, result):
+        if not getattr(result, 'is_anomaly', False):
             return
-            
-        url = f"https://api.telegram.org/bot{self.tg_token}/sendMessage"
+
         payload = {
-            "chat_id": self.tg_chat_id,
-            "text": message,
-            "parse_mode": "HTML"
+            'event': 'high_priority_insight',
+            'type': 'anomaly',
+            'metric_name': result.metric_name,
+            'severity_score': result.severity_score,
+            'current_value': result.current_value,
+            'baseline_mean': result.baseline_mean,
+            'baseline_std': result.baseline_std,
+            'z_score': result.z_score,
+            'timestamp': result.timestamp.isoformat() if result.timestamp else None,
         }
-        
-        try:
-            response = requests.post(url, json=payload, timeout=10)
-            response.raise_for_status()
-            logger.info("Telegram alert sent successfully")
-        except Exception as e:
-            logger.error(f"Failed to send Telegram alert: {e}")
-
-    def _send_webhook(self, data: Dict[str, Any]):
-        """Send JSON payload to generic webhook."""
-        if not self.has_webhook:
+
+        self._send_telegram(payload)
+        self._send_webhooks(payload)
+
+    def _send_telegram(self, payload):
+        if not self.telegram_bot_token or not self.telegram_channel_id:
             return
-            
-        try:
-            response = requests.post(
-                self.webhook_url, 
-                json=data,
-                headers={"Content-Type": "application/json"},
-                timeout=10
-            )
-            response.raise_for_status()
-            logger.info("Webhook alert sent successfully")
-        except Exception as e:
-            logger.error(f"Failed to send Webhook alert: {e}")
-
-    def format_anomaly_message(self, result: AnomalyResult) -> str:
-        """Format AnomalyResult for Telegram (HTML)."""
-        severity_emoji = "🔴" if result.severity_score > 0.8 else "🟠"
-        
-        return (
-            f"🚨 <b>{result.metric_name.upper()} ANOMALY DETECTED</b> {severity_emoji}\n\n"
-            f"<b>Severity:</b> {result.severity_score:.2f}\n"
-            f"<b>Value:</b> {result.current_value:,.4f}\n"
-            f"<b>Z-Score:</b> {result.z_score:.2f}\n"
-            f"<b>Baseline Mean:</b> {result.baseline_mean:,.4f}\n"
-            f"<b>Timestamp:</b> {result.timestamp.strftime('%Y-%m-%d %H:%M:%S')} UTC\n\n"
-            f"<i>#LumenPulse #Anomalies #Alert</i>"
+
+        text = (
+            '🚨 High-Priority Insight\n'
+            f"Metric: {payload['metric_name']}\n"
+            f"Severity: {payload['severity_score']}\n"
+            f"Current: {payload['current_value']}\n"
+            f"Z-Score: {payload['z_score']}"
         )
 
-    def notify_anomaly(self, result: AnomalyResult):
-        """Send notifications for a single anomaly if it meets severity threshold."""
-        if not result.is_anomaly or result.severity_score < self.min_severity:
-            return
+        requests.post(
+            f"https://api.telegram.org/bot{self.telegram_bot_token}/sendMessage",
+            json={
+                'chat_id': self.telegram_channel_id,
+                'text': text,
+            },
+            timeout=10,
+        )
+
+    def _send_webhooks(self, payload):
+        for url in self.webhook_urls:
+            self._post_with_retry(url, payload)
+
+    def _post_with_retry(self, url, payload):
+        for attempt in range(self.max_retries):
+            try:
+                response = requests.post(url, json=payload, timeout=10)
+                if response.status_code < 400:
+                    return True
+            except requests.RequestException:
+                pass
+
+            if attempt < self.max_retries - 1:
+                time.sleep(self.base_backoff_seconds * (2 ** attempt))
 
-        logger.info(f"Notifying anomaly: {result.metric_name} (severity={result.severity_score:.2f})")
-        
-        # 1. Telegram
-        if self.has_telegram:
-            msg = self.format_anomaly_message(result)
-            self._send_telegram(msg)
-            
-        # 2. Webhook
-        if self.has_webhook:
-            self._send_webhook({
-                "event": "anomaly_detected",
-                "severity": "high" if result.severity_score > 0.8 else "medium",
-                "data": result.to_dict()
-            })
-
-    def notify_batch(self, results: List[AnomalyResult]):
-        """Filter and notify for a list of results."""
-        for result in results:
-            if result.is_anomaly:
-                self.notify_anomaly(result)
-
-# Singleton instance for easy import
-notifier = AlertNotifier(
-    min_severity=float(os.getenv("ALERT_MIN_SEVERITY", "0.5"))
-)
+        return False

apps/data-processing/tests/test_alert_notifier.py

@@ -72,3 +72,39 @@ def test_no_notification_if_not_anomaly(self, mock_post):
 
 if __name__ == '__main__':
     unittest.main()
+
+
+    @patch('requests.post')
+    def test_webhook_retries_then_succeeds(self, mock_post):
+        # Telegram succeeds immediately; webhook fails twice then succeeds.
+        telegram_ok = MagicMock()
+        telegram_ok.status_code = 200
+
+        webhook_fail_1 = MagicMock()
+        webhook_fail_1.status_code = 500
+        webhook_fail_2 = MagicMock()
+        webhook_fail_2.status_code = 502
+        webhook_ok = MagicMock()
+        webhook_ok.status_code = 200
+
+        mock_post.side_effect = [telegram_ok, webhook_fail_1, webhook_fail_2, webhook_ok]
+
+        result = AnomalyResult(
+            is_anomaly=True,
+            severity_score=0.95,
+            metric_name="volume",
+            current_value=175000.0,
+            baseline_mean=50000.0,
+            baseline_std=10000.0,
+            z_score=12.5,
+            timestamp=datetime.now()
+        )
+
+        with patch('time.sleep') as mock_sleep:
+            self.notifier.notify_anomaly(result)
+
+        webhook_calls = [c for c in mock_post.call_args_list if 'test.webhook.com' in c.args[0]]
+        self.assertEqual(len(webhook_calls), 3, "Webhook delivery should retry twice before succeeding")
+
+        # Exponential backoff should sleep between retries (2 retries -> 2 sleeps)
+        self.assertEqual(mock_sleep.call_count, 2)

apps/onchain/contracts/crowdfund_vault/src/events.rs

@@ -106,6 +106,14 @@ pub struct ContributionRefundedEvent {
     pub amount: i128,
 }
 
+#[contractevent]
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ProtocolFeeDeductedEvent {
+    #[topic]
+    pub project_id: u64,
+    pub amount: i128,
+}
+
 #[contractevent]
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct MilestoneVoteStartedEvent {
@@ -115,6 +123,15 @@ pub struct MilestoneVoteStartedEvent {
     pub end_time: u64,
 }
 
+#[contractevent]
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct FeeConfigChangedEvent {
+    #[topic]
+    pub admin: Address,
+    pub fee_bps: u32,
+    pub treasury: Address,
+}
+
 #[contractevent]
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct VoteCastEvent {