feat(breeze-buddy): add global pickup rate alert background task

app/core/config/dynamic.py

@@ -46,6 +46,22 @@ async def ENABLE_BACKGROUND_TASKS() -> bool:
     return await get_config("ENABLE_BACKGROUND_TASKS", "false", bool)
 
 
+# --- Pickup Rate Alert Configuration ---
+async def ENABLE_PICKUP_RATE_ALERT() -> bool:
+    """Returns ENABLE_PICKUP_RATE_ALERT from Redis"""
+    return await get_config("ENABLE_PICKUP_RATE_ALERT", "false", bool)
+
+
+async def PICKUP_RATE_ALERT_INTERVAL_SECONDS() -> int:
+    """Returns PICKUP_RATE_ALERT_INTERVAL_SECONDS from Redis (default: 86400 = 24 h)"""
+    return await get_config("PICKUP_RATE_ALERT_INTERVAL_SECONDS", 86400, int)
+
+
+async def PICKUP_RATE_ALERT_THRESHOLD() -> float:
+    """Returns PICKUP_RATE_ALERT_THRESHOLD from Redis (default: 40.0 %)"""
+    return await get_config("PICKUP_RATE_ALERT_THRESHOLD", 40.0, float)
+
+
 async def DAILY_SUMMARY_HOUR() -> int:
     """Returns DAILY_SUMMARY_HOUR from Redis (24-hour format: 0-23)"""
     return await get_config("DAILY_SUMMARY_HOUR", 21, int)

app/main.py

@@ -66,6 +66,7 @@
     AutomaticVoiceUserConnectRequest,
 )
 from app.services.langfuse.tasks.task import initialize_langfuse_tasks
+from app.services.pickup_rate.task import initialize_pickup_rate_tasks
 from app.services.redis import (
     close_redis_connections,
     get_redis_service,
@@ -168,6 +169,7 @@ async def lifespan(_app: FastAPI):
             await initialize_langfuse_tasks(_background_scheduler)
 
             ### Register new tasks here
+            await initialize_pickup_rate_tasks(_background_scheduler)
 
             # Start the scheduler only if tasks are registered
             if _background_scheduler.tasks:

app/services/langfuse/tasks/score_monitor/score.py

@@ -440,24 +440,22 @@ async def _get_daily_call_stats(self) -> Dict[str, Any]:
 
             # Process each call tracker for call-based stats
             for tracker, calling_provider in call_trackers:
-                # Count attempted calls (FINISHED status)
+                # Count attempted calls and outcomes only for FINISHED calls
                 if tracker.status and tracker.status.value == "FINISHED":
                     calls_attempted += 1
-
-                # Count by outcome
-                outcome_value = tracker.outcome if tracker.outcome else None
-                if outcome_value == "NO_ANSWER":
-                    calls_no_answer += 1
-                elif outcome_value == "CONFIRM":
-                    calls_confirm += 1
-                elif outcome_value == "CANCEL":
-                    calls_cancel += 1
-                elif outcome_value == "ADDRESS_UPDATED":
-                    calls_address_updated += 1
-                elif outcome_value == "BUSY":
-                    calls_busy += 1
-
-                # Count by provider
+                    outcome_value = tracker.outcome if tracker.outcome else None
+                    if outcome_value == "NO_ANSWER":
+                        calls_no_answer += 1
+                    elif outcome_value == "CONFIRM":
+                        calls_confirm += 1
+                    elif outcome_value == "CANCEL":
+                        calls_cancel += 1
+                    elif outcome_value == "ADDRESS_UPDATED":
+                        calls_address_updated += 1
+                    elif outcome_value == "BUSY":
+                        calls_busy += 1
+
+                # Count by provider regardless of status
                 if calling_provider:
                     provider_upper = calling_provider.upper()
                     if provider_upper in provider_counts:

app/services/pickup_rate/__init__.py

@@ -0,0 +1,15 @@
+"""
+Pickup Rate Alert Service
+
+Monitors call pickup rates and sends Slack warning alerts when rates drop
+below configurable thresholds.
+"""
+
+from app.services.pickup_rate.monitor import PickupRateMonitor, pickup_rate_monitor
+from app.services.pickup_rate.task import initialize_pickup_rate_tasks
+
+__all__ = [
+    "PickupRateMonitor",
+    "pickup_rate_monitor",
+    "initialize_pickup_rate_tasks",
+]

app/services/pickup_rate/calculator.py

@@ -0,0 +1,121 @@
+"""
+Pickup Rate Calculator
+
+Computes call-based and lead-based pickup rates for a given time window,
+mirroring the logic in ScoreMonitor._get_daily_call_stats().
+
+Phase 1: merchant_id / reseller_id params are accepted but ignored.
+Phase 2: pass them through to the DB accessor filters once those are wired up.
+"""
+
+from datetime import datetime
+from typing import Any, Dict, Optional
+
+from app.core.logger import logger
+from app.database.accessor.breeze_buddy.lead_call_tracker import (
+    get_all_lead_call_trackers,
+    get_lead_based_analytics,
+)
+
+# Sentinel value returned on DB error – callers must check for None
+_ERROR_RESULT = None
+
+
+async def compute_pickup_rates(
+    start_date: datetime,
+    end_date: datetime,
+    reseller_id: Optional[str] = None,  # Phase 2: filter by reseller
+    merchant_id: Optional[str] = None,  # Phase 2: filter by merchant
+) -> Optional[Dict[str, Any]]:
+    """
+    Calculate call-based and lead-based pickup rates for the given window.
+
+    Args:
+        start_date:   Start of the rolling window (UTC-aware).
+        end_date:     End of the rolling window (UTC-aware).
+        reseller_id:  (Phase 2) Restrict results to this reseller.
+        merchant_id:  (Phase 2) Restrict results to this merchant.
+
+    Returns:
+        Dictionary with the following keys on success::
+
+            {
+                "calls_attempted":    int,   # FINISHED calls in window
+                "calls_picked":       int,   # attempted - NO_ANSWER
+                "calls_no_answer":    int,
+                "call_pickup_rate":   float, # (picked / attempted) * 100
+                "total_leads":        int,   # unique request_ids
+                "leads_picked":       int,   # leads where finished > no_answer
+                "lead_pickup_rate":   float, # (leads_picked / total_leads) * 100
+            }
+
+        Returns ``None`` if a DB error occurs (caller should skip alerting).
+    """
+    try:
+        # ------------------------------------------------------------------
+        # 1. Call-based metrics
+        # ------------------------------------------------------------------
+        # Phase 2 NOTE: pass reseller_id / merchant_id to the accessor once
+        # those optional params are supported by get_all_lead_call_trackers().
+        call_trackers = await get_all_lead_call_trackers(
+            start_date=start_date,
+            end_date=end_date,
+        )
+
+        calls_attempted = 0
+        calls_no_answer = 0
+
+        if call_trackers:
+            for tracker, _provider in call_trackers:
+                if tracker.status and tracker.status.value == "FINISHED":
+                    calls_attempted += 1
+                    if tracker.outcome == "NO_ANSWER":
+                        calls_no_answer += 1
+
+        calls_picked = calls_attempted - calls_no_answer
+        call_pickup_rate = (
+            calls_picked / calls_attempted * 100 if calls_attempted > 0 else 0.0
+        )
+
+        # ------------------------------------------------------------------
+        # 2. Lead-based metrics
+        # ------------------------------------------------------------------
+        # Phase 2 NOTE: pass reseller_id / merchant_id once supported.
+        lead_data = await get_lead_based_analytics(
+            start_date=start_date,
+            end_date=end_date,
+        )
+
+        total_leads = len(lead_data) if lead_data else 0
+        leads_picked = 0
+
+        if lead_data:
+            for lead in lead_data:
+                # A lead is "picked" when more calls were finished than went unanswered
+                if lead["finished_calls"] > lead["no_answer_calls"]:
+                    leads_picked += 1
+
+        lead_pickup_rate = leads_picked / total_leads * 100 if total_leads > 0 else 0.0
+
+        result = {
+            "calls_attempted": calls_attempted,
+            "calls_picked": calls_picked,
+            "calls_no_answer": calls_no_answer,
+            "call_pickup_rate": call_pickup_rate,
+            "total_leads": total_leads,
+            "leads_picked": leads_picked,
+            "lead_pickup_rate": lead_pickup_rate,
+        }
+
+        logger.debug(
+            f"compute_pickup_rates [{start_date} → {end_date}] "
+            f"call_rate={call_pickup_rate}% lead_rate={lead_pickup_rate}%"
+        )
+        return result
+
+    except Exception as e:
+        logger.error(
+            f"compute_pickup_rates failed for window [{start_date} → {end_date}]: {e}",
+            exc_info=True,
+        )
+        return _ERROR_RESULT

app/services/pickup_rate/config.py

@@ -0,0 +1,55 @@
+"""
+AlertConfig dataclass for the Pickup Rate Alert system.
+
+Phase 1: global alerting only.
+Phase 2 will extend this with optional merchant_id / reseller_id for per-merchant scope.
+"""
+
+from dataclasses import dataclass, field
+from typing import Optional
+
+
+@dataclass
+class AlertConfig:
+    """Configuration for a single pickup-rate alert scope.
+
+    Fields:
+        enabled:            Whether alerting is active for this scope.
+        interval_seconds:   How often (in seconds) to check and potentially alert.
+                            Also used as the Redis TTL for the dedup key.
+        threshold_percent:  Alert fires when pickup rate drops below this value (0-100).
+        alert_type:         Which rate(s) to evaluate.
+                            "CALL_BASED"  – only call-based pickup rate.
+                            "LEAD_BASED"  – only lead-based pickup rate.
+                            "BOTH"        – alert if *either* rate is below threshold.
+        scope:              Human-readable label used in Slack messages and Redis keys
+                            (e.g. "global", "merchant:acme/store-1").
+        lookback_hours:     Rolling window of call data to evaluate (default: 24 h).
+
+    Phase 2 extension fields (unused in Phase 1 – kept for forward-compatibility):
+        merchant_id:   Merchant identifier to scope DB queries (None = no filter).
+        reseller_id:   Reseller identifier to scope DB queries (None = no filter).
+    """
+
+    enabled: bool
+    interval_seconds: int
+    threshold_percent: float
+    alert_type: str = "BOTH"
+    scope: str = "global"
+    lookback_hours: int = 24
+
+    # Phase 2: per-merchant fields – not used yet, wired through to calculator
+    merchant_id: Optional[str] = None
+    reseller_id: Optional[str] = None
+
+    # Computed Redis dedup key – derived from scope, not configurable directly
+    _redis_key: str = field(init=False, repr=False, default="")
+
+    def __post_init__(self) -> None:
+        safe_scope = self.scope.replace(":", "_").replace("/", "_")
+        self._redis_key = f"pickup_rate:{safe_scope}:last_alert"
+
+    @property
+    def redis_dedup_key(self) -> str:
+        """Redis key used to track last-alert timestamp for this scope."""
+        return self._redis_key