"""Detector: speed-to-lead gap — new leads with no call/SMS within 5 minutes.
Cross-connector: joins lead sources (HubSpot / LeadPerfection / Angi) with
communication activity (Five9 call logs, Marlimar SMS, Slack alerts).
"""
from __future__ import annotations
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
from .base import BaseDetector, LeakCandidate
from ._common import (
any_connector_available,
as_aware,
normalize_phone,
parse_dt,
utcnow,
)
COMM_SERVICES = ("five9", "marlimar")
LEAD_SERVICES = ("hubspot", "leadperfection", "angi_leads")
SLACK_GRACE_MINUTES = 15
def _minutes_severity(minutes: float) -> str:
if minutes > 120:
return "critical"
if minutes > 30:
return "high"
return "medium"
class SpeedToLeadDetector(BaseDetector):
id = "speed_to_lead"
name = "Speed-to-Lead: Slow Response"
connector = ["five9", "marlimar"] # at least one comm channel + a lead source
enabled_by_default = True
description = (
"Cross-references new leads (HubSpot, LeadPerfection, Angi) with call "
"and SMS activity (Five9, Marlimar, Slack) and flags leads not "
"contacted within 5 minutes. Industry benchmark: 5 minutes."
)
default_params = {"speed_to_lead_minutes": 5, "benchmark_minutes": 5,
"window_days": 7, "slack_grace_minutes": SLACK_GRACE_MINUTES}
def check(self, company_id: str) -> List[LeakCandidate]:
# Need at least one communication channel and at least one lead source.
if not any_connector_available(company_id, COMM_SERVICES):
return []
if not any_connector_available(company_id, LEAD_SERVICES):
return []
from app.models import AngiLead, CrmContact, ExternalSyncRecord
cutoff = utcnow() - timedelta(days=7)
# ------------------------------------------------------------------
# 1. Collect candidate leads from all available sources.
# Each entry: (lead_id, source, name, phone, email, created, cost)
# ------------------------------------------------------------------
leads: List[Tuple[str, str, str, str, str, datetime, Optional[float]]] = []
for contact in CrmContact.query.filter(
CrmContact.company_id == company_id,
CrmContact.lifecycle_stage == "lead",
CrmContact.created_at > cutoff,
).all():
leads.append((
contact.id, "hubspot", contact.full_name or contact.email or "Unknown",
contact.phone or "", contact.email or "",
as_aware(contact.created_at), None,
))
for rec in ExternalSyncRecord.query.filter(
ExternalSyncRecord.company_id == company_id,
ExternalSyncRecord.source_service == "leadperfection",
ExternalSyncRecord.entity_type == "lead",
ExternalSyncRecord.created_at > cutoff,
).all():
props = rec.properties_json or {}
leads.append((
rec.id, "leadperfection", rec.name or "Unknown",
props.get("phone") or "", props.get("email") or "",
as_aware(rec.created_at), None,
))
for angi in AngiLead.query.filter(
AngiLead.company_id == company_id,
AngiLead.created_at > cutoff,
).all():
leads.append((
angi.id, angi.lead_source or "angi", angi.full_name or "Unknown",
angi.phone or "", angi.email or "",
as_aware(angi.received_at or angi.created_at), angi.cost_per_lead,
))
if not leads:
return []
# ------------------------------------------------------------------
# 2. Build contact-event indexes: phone/email -> sorted timestamps.
# Five9 calls + Marlimar messages count as first contact.
# Slack notifications count with a 15-minute grace window.
# ------------------------------------------------------------------
contact_events: Dict[str, List[datetime]] = {}
slack_events: Dict[str, List[datetime]] = {}
def _index(store: Dict[str, List[datetime]], key: str, ts: Optional[datetime]):
if key and ts is not None:
store.setdefault(key, []).append(ts)
comm_records = ExternalSyncRecord.query.filter(
ExternalSyncRecord.company_id == company_id,
ExternalSyncRecord.source_service.in_(["five9", "marlimar", "slack"]),
ExternalSyncRecord.created_at > cutoff - timedelta(days=1),
).all()
for rec in comm_records:
props = rec.properties_json or {}
ts = (
parse_dt(props.get("timestamp"))
or parse_dt(props.get("call_time"))
or parse_dt(props.get("sent_at"))
or as_aware(rec.created_at)
)
phone = normalize_phone(
props.get("phone") or props.get("contact_phone")
or props.get("to") or props.get("dnis") or props.get("ani")
)
email = (props.get("email") or props.get("contact_email") or "").strip().lower()
if rec.source_service == "slack":
_index(slack_events, phone, ts)
_index(slack_events, email, ts)
# Slack messages often only carry a name/text mention
_index(slack_events, (rec.name or "").strip().lower(), ts)
elif rec.source_service == "five9" and rec.entity_type == "call_log":
_index(contact_events, phone, ts)
_index(contact_events, email, ts)
elif rec.source_service == "marlimar" and rec.entity_type == "message":
direction = str(props.get("direction", "")).lower()
if direction != "inbound": # outbound SMS = contact attempt
_index(contact_events, phone, ts)
_index(contact_events, email, ts)
for store in (contact_events, slack_events):
for key in store:
store[key].sort()
def _first_after(store: Dict[str, List[datetime]], keys, after: datetime):
best = None
for key in keys:
for ts in store.get(key, ()):
if ts >= after:
if best is None or ts < best:
best = ts
break
return best
# ------------------------------------------------------------------
# 3. Evaluate each lead's time-to-first-contact.
# ------------------------------------------------------------------
now = utcnow()
candidates: List[LeakCandidate] = []
for lead_id, lead_source, name, phone, email, created, cost in leads:
if created is None:
continue
minutes_to_contact: Optional[float] = None
# Angi leads carry their own response tracking — trust it first.
if lead_source in ("angi", "homeadvisor"):
angi = AngiLead.query.get(lead_id)
if angi is not None:
if angi.response_time_minutes is not None:
minutes_to_contact = float(angi.response_time_minutes)
else:
first = as_aware(angi.first_contact_at or angi.responded_at)
if first is not None:
minutes_to_contact = (first - created).total_seconds() / 60.0
if minutes_to_contact is None:
keys = [k for k in (normalize_phone(phone), (email or "").strip().lower()) if k]
first_contact = _first_after(contact_events, keys, created) if keys else None
first_slack = _first_after(
slack_events, keys + [(name or "").strip().lower()], created
)
if first_contact is not None:
minutes_to_contact = (first_contact - created).total_seconds() / 60.0
elif first_slack is not None:
slack_minutes = (first_slack - created).total_seconds() / 60.0
if slack_minutes <= SLACK_GRACE_MINUTES:
continue # Slack alert acted on fast enough
minutes_to_contact = slack_minutes
else:
# Never contacted — clock is still running.
minutes_to_contact = (now - created).total_seconds() / 60.0
minutes_to_contact = max(round(minutes_to_contact, 1), 0.0)
if minutes_to_contact <= 5:
continue
candidates.append(
LeakCandidate(
detector_id=self.id,
source="Speed-to-Lead: Slow Response",
description=(
f"Lead '{name}' waited {int(minutes_to_contact)} minutes "
f"for first contact. Industry benchmark: 5 min."
),
estimated_loss=cost,
severity=_minutes_severity(minutes_to_contact),
metadata_json={
"dedupe_key": f"{self.id}:{lead_source}:{lead_id}",
"lead_id": lead_id,
"lead_source": lead_source,
"minutes_to_contact": minutes_to_contact,
"lead_phone": phone,
},
source_connector="five9" if any_connector_available(
company_id, ["five9"]
) else "marlimar",
rule_params={"speed_to_lead_minutes": 5, "benchmark_minutes": 5},
)
)
return candidates