"""Phase 3 leak detectors — cross-connector intelligence.
Detectors that join data from multiple integrations for higher-value
insights:
channel_quality — ad channels with conversion rate below account average
sms_response_time — inbound SMS with no reply within 2 hours (Marlimar)
forecast_gap — pipeline weighted value below 3× monthly goal
customer_churn_risk — customers with declining invoice frequency
budget_allocation — enabled campaigns past end date, paused high-ROAS, etc.
All inherit BaseDetector and are exported via DETECTORS for registration.
"""
from __future__ import annotations
import logging
from datetime import date, datetime, timedelta, timezone
from typing import Any, Dict, List, Optional
from .base import BaseDetector, LeakCandidate
from ._common import (
as_aware,
connector_available,
utcnow,
)
logger = logging.getLogger(__name__)
ADS_SERVICES = ("google_ads", "facebook_ads")
def _ads_connectors_available(company_id: str) -> List[str]:
return [s for s in ADS_SERVICES if connector_available(company_id, s)]
# =============================================================================
# Detector 1: channel_quality — Channel Quality Mismatch
# =============================================================================
class ChannelQualityDetector(BaseDetector):
id = "channel_quality"
name = "Ads: Low-Quality Channel"
connector = ["google_ads", "hubspot"] # ads + crm
enabled_by_default = True
description = (
"Cross-references ad spend with CRM lead conversion rates. Flags "
"ad channels whose lead-to-deal conversion is below the account average."
)
default_params = {"days": 30, "min_spend": 500, "low_rate_threshold": 0.5}
def check(self, company_id: str) -> List[LeakCandidate]:
from app.models import AdCampaign, AdMetric, CrmContact, CrmDeal
services = _ads_connectors_available(company_id)
if not services:
return []
if not connector_available(company_id, "hubspot"):
return []
days = int(self.default_params.get("days", 30))
min_spend = float(self.default_params.get("min_spend", 500))
threshold = float(self.default_params.get("low_rate_threshold", 0.5))
cutoff = (utcnow() - timedelta(days=days)).date()
# --- Per-channel ad spend ---
metrics = AdMetric.query.filter(
AdMetric.company_id == company_id,
AdMetric.source_service.in_(services),
AdMetric.metric_date >= cutoff,
).all()
per_channel: Dict[str, Dict[str, Any]] = {}
for m in metrics:
cid = m.external_campaign_id or ""
if not cid:
continue
channel = m.channel or m.source_service or "unknown"
spend = m.spend or 0.0
conversions = m.conversions or 0.0
ch = per_channel.setdefault(channel, {
"spend": 0.0,
"conversions": 0.0,
"clicks": 0.0,
"campaign_ids": [],
})
ch["spend"] += spend
ch["conversions"] += conversions
ch["clicks"] += m.clicks or 0
if cid not in ch["campaign_ids"]:
ch["campaign_ids"].append(cid)
# --- Per-channel lead conversion rates from CRM ---
# CrmContact has source tracking in properties_json; try to attribute
contacts = CrmContact.query.filter(
CrmContact.company_id == company_id,
CrmContact.created_at > utcnow() - timedelta(days=days),
).all()
per_channel_leads: Dict[str, Dict[str, int]] = {}
for c in contacts:
props = c.properties_json or {}
channel = (props.get("lead_source") or props.get("channel")
or props.get("utm_source") or "").strip()
if not channel:
channel = "organic"
bucket = per_channel_leads.setdefault(channel, {
"leads": 0,
"deals": 0,
})
bucket["leads"] += 1
# Count deals attributed to each channel (via contact)
deals = CrmDeal.query.filter(
CrmDeal.company_id == company_id,
CrmDeal.created_at > utcnow() - timedelta(days=days),
).all()
for deal in deals:
if deal.contact_id:
contact = CrmContact.query.get(deal.contact_id)
if contact:
props = contact.properties_json or {}
channel = (props.get("lead_source") or props.get("channel")
or props.get("utm_source") or "").strip()
if channel:
bucket = per_channel_leads.get(channel)
if bucket:
bucket["deals"] += 1
# --- Calculate conversion rates ---
all_channels = set(list(per_channel.keys()) + list(per_channel_leads.keys()))
if not all_channels:
return []
rates = []
for ch in all_channels:
lead_data = per_channel_leads.get(ch)
if lead_data and lead_data["leads"] > 0:
rate = lead_data["deals"] / lead_data["leads"]
rates.append(rate)
if not rates:
return []
avg_rate = sum(rates) / len(rates) if rates else 0
# --- Flag underperforming channels ---
candidates: List[LeakCandidate] = []
for channel, ch_data in per_channel.items():
spend = ch_data["spend"]
if spend < min_spend:
continue
lead_data = per_channel_leads.get(channel, {"leads": 0, "deals": 0})
actual_rate = (lead_data["deals"] / lead_data["leads"]
if lead_data["leads"] > 0 else 0)
if avg_rate > 0 and actual_rate < (threshold * avg_rate):
loss = spend * (1 - actual_rate / avg_rate) if avg_rate > 0 else spend
severity = "high" if actual_rate < (0.5 * avg_rate) else "medium"
candidates.append(
LeakCandidate(
detector_id=self.id,
source="Ads: Low-Quality Channel",
description=(
f"Channel '{channel}' converts at "
f"{actual_rate:.0%} vs account avg "
f"{avg_rate:.0%}%. Spent ${spend:,.0f} in {days} days"
),
estimated_loss=round(loss, 2),
severity=severity,
metadata_json={
"dedupe_key": f"{self.id}:{channel}",
"channel": channel,
"campaign_ids": ch_data["campaign_ids"],
"actual_rate": round(actual_rate, 4),
"avg_rate": round(avg_rate, 4),
"spend": round(spend, 2),
"leads": lead_data.get("leads", 0),
"deals": lead_data.get("deals", 0),
},
source_connector="google_ads",
rule_params={
"days": days,
"min_spend": min_spend,
"low_rate_threshold": threshold,
},
)
)
# Flag high-spend channels with 0 deals
for channel, ch_data in per_channel.items():
spend = ch_data["spend"]
if spend < min_spend:
continue
lead_data = per_channel_leads.get(channel, {"deals": 0})
if lead_data["deals"] == 0 and channel not in [
c.metadata_json.get("channel") for c in candidates
]:
candidates.append(
LeakCandidate(
detector_id=self.id,
source="Ads: Zero-Deal Channel",
description=(
f"Channel '{channel}' has spent ${spend:,.0f} in "
f"{days} days with 0 attributed deals"
),
estimated_loss=round(spend, 2),
severity="high",
metadata_json={
"dedupe_key": f"{self.id}:zero:{channel}",
"channel": channel,
"campaign_ids": ch_data["campaign_ids"],
"actual_rate": 0.0,
"avg_rate": round(avg_rate, 4),
"spend": round(spend, 2),
},
source_connector="google_ads",
rule_params={
"days": days,
"min_spend": min_spend,
},
)
)
return candidates
# =============================================================================
# Detector 2: sms_response_time — SMS Response Delay
# =============================================================================
class SmsResponseTimeDetector(BaseDetector):
id = "sms_response_time"
name = "Marlimar: SMS Response Delay"
connector = "marlimar"
enabled_by_default = True
description = (
"Flags inbound SMS messages that went unanswered for more than "
"2 hours via Marlimar."
)
default_params = {"response_hours": 2, "window_days": 7}
def check(self, company_id: str) -> List[LeakCandidate]:
if not connector_available(company_id, self.connector):
return []
from app.models import ExternalSyncRecord
window_days = int(self.default_params.get("window_days", 7))
cutoff = utcnow() - timedelta(days=window_days)
# Get inbound messages
inbound = ExternalSyncRecord.query.filter(
ExternalSyncRecord.company_id == company_id,
ExternalSyncRecord.source_service == "marlimar",
ExternalSyncRecord.entity_type == "message",
ExternalSyncRecord.created_at > cutoff,
).all()
if not inbound:
return []
# Index outbound messages by contact
outbound = ExternalSyncRecord.query.filter(
ExternalSyncRecord.company_id == company_id,
ExternalSyncRecord.source_service == "marlimar",
ExternalSyncRecord.entity_type == "message",
ExternalSyncRecord.created_at > cutoff,
).all()
outbound_by_contact: Dict[str, List[datetime]] = {}
for rec in outbound:
props = rec.properties_json or {}
direction = str(props.get("direction", "")).lower()
if direction == "inbound":
continue
contact_key = (
props.get("contact_id")
or props.get("from")
or props.get("phone")
or ""
)
if contact_key:
ts = (
_parse_dt(props.get("sent_at"))
or _parse_dt(props.get("timestamp"))
or as_aware(rec.created_at)
)
if ts:
outbound_by_contact.setdefault(contact_key, []).append(ts)
for key in outbound_by_contact:
outbound_by_contact[key].sort()
candidates: List[LeakCandidate] = []
seen_contacts: set = set()
for rec in inbound:
props = rec.properties_json or {}
direction = str(props.get("direction", "")).lower()
if direction != "inbound":
continue
contact_key = (
props.get("contact_id")
or props.get("from")
or props.get("phone")
or ""
)
if not contact_key:
continue
if contact_key in seen_contacts:
continue
inbound_ts = (
_parse_dt(props.get("timestamp"))
or _parse_dt(props.get("received_at"))
or as_aware(rec.created_at)
)
if not inbound_ts:
continue
# Find first outbound reply after this inbound
replies = outbound_by_contact.get(contact_key, [])
reply_ts = None
for ts in replies:
if ts > inbound_ts:
reply_ts = ts
break
if reply_ts is None:
hours_unanswered = (utcnow() - inbound_ts).total_seconds() / 3600
else:
hours_unanswered = (reply_ts - inbound_ts).total_seconds() / 3600
if hours_unanswered < 2:
continue
seen_contacts.add(contact_key)
severity = "critical" if hours_unanswered > 24 else (
"high" if hours_unanswered > 6 else "medium"
)
candidates.append(
LeakCandidate(
detector_id=self.id,
source="Marlimar: SMS Response Delay",
description=(
f"Inbound message from '{contact_key}' went unanswered "
f"for {hours_unanswered:.0f} hours"
),
estimated_loss=None,
severity=severity,
metadata_json={
"dedupe_key": f"{self.id}:{contact_key}:{rec.id}",
"external_id": rec.id,
"contact_id": contact_key,
"hours_unanswered": round(hours_unanswered, 1),
"message_timestamp": str(inbound_ts.isoformat()),
},
source_connector="marlimar",
rule_params={"response_hours": 2, "window_days": window_days},
)
)
return candidates
# =============================================================================
# Detector 3: forecast_gap — Pipeline Coverage Gap
# =============================================================================
class ForecastGapDetector(BaseDetector):
id = "forecast_gap"
name = "CRM: Pipeline Coverage Gap"
connector = "hubspot"
enabled_by_default = True
description = (
"Checks if weighted pipeline value is sufficient for 3-month coverage "
"based on revenue goals or historical win rates."
)
default_params = {"coverage_months": 3, "min_coverage_pct": 80}
def check(self, company_id: str) -> List[LeakCandidate]:
if not connector_available(company_id, self.connector):
return []
from app.models import CrmDeal, RevenueForecast
coverage_months = int(self.default_params.get("coverage_months", 3))
min_coverage = float(self.default_params.get("min_coverage_pct", 80))
now = utcnow()
three_months_ago = now - timedelta(days=coverage_months * 30)
# Get open deals (non-closed stages)
open_deals = CrmDeal.query.filter(
CrmDeal.company_id == company_id,
CrmDeal.stage.notin_([
"Closed Won", "Closed Lost",
"closed_won", "closed_lost",
"won", "lost",
]),
).all()
weighted_pipeline = 0.0
for deal in open_deals:
amount = deal.amount or 0.0
probability = deal.probability if deal.probability is not None else 0.5
weighted_pipeline += amount * min(max(probability, 0.0), 1.0)
# Get average monthly won revenue
won_deals = CrmDeal.query.filter(
CrmDeal.company_id == company_id,
CrmDeal.stage.in_(["Closed Won", "closed_won", "won"]),
CrmDeal.updated_at > three_months_ago,
).all()
total_won = sum(d.amount or 0.0 for d in won_deals)
avg_monthly_won = total_won / coverage_months if coverage_months > 0 else 0
# Check for revenue goal
goal_monthly = None
forecasts = RevenueForecast.query.filter(
RevenueForecast.company_id == company_id,
).order_by(RevenueForecast.forecast_date.desc()).limit(1).all()
if forecasts:
goal_monthly = forecasts[0].forecast_amount
# Calculate target pipeline
if goal_monthly and goal_monthly > 0:
target_pipeline = goal_monthly * coverage_months
else:
target_pipeline = avg_monthly_won * coverage_months
if target_pipeline <= 0:
return [] # No goal and no history to compare against
coverage_pct = (weighted_pipeline / target_pipeline * 100)
if coverage_pct >= min_coverage:
return [] # Coverage is adequate
if coverage_pct < 40:
severity = "critical"
elif coverage_pct < 60:
severity = "high"
else:
severity = "medium"
gap = round(target_pipeline - weighted_pipeline, 2)
return [
LeakCandidate(
detector_id=self.id,
source="CRM: Pipeline Coverage Gap",
description=(
f"Weighted pipeline (${weighted_pipeline:,.0f}) is only "
f"{coverage_pct:.0f}% of the ${target_pipeline:,.0f} "
f"needed for {coverage_months}-month coverage"
),
estimated_loss=gap,
severity=severity,
metadata_json={
"dedupe_key": f"{self.id}:pipeline_coverage",
"weighted_pipeline": round(weighted_pipeline, 2),
"target_pipeline": round(target_pipeline, 2),
"coverage_pct": round(coverage_pct, 1),
"avg_monthly_won": round(avg_monthly_won, 2),
"goal_monthly": round(goal_monthly, 2) if goal_monthly else None,
},
source_connector="hubspot",
rule_params={
"coverage_months": coverage_months,
"min_coverage_pct": min_coverage,
},
)
]
# =============================================================================
# Detector 4: customer_churn_risk — Declining Customer Activity
# =============================================================================
class CustomerChurnRiskDetector(BaseDetector):
id = "customer_churn_risk"
name = "QuickBooks: Churn Risk"
connector = "quickbooks"
enabled_by_default = True
description = (
"Identifies customers with declining invoice frequency month-over-month "
"and unpaid balances, indicating churn risk."
)
default_params = {
"decline_pct_threshold": 30,
"months_lookback": 3,
"min_invoices_baseline": 2,
}
def check(self, company_id: str) -> List[LeakCandidate]:
if not connector_available(company_id, self.connector):
return []
from app.models import QuickbooksInvoice, QuickbooksCustomer
decline_threshold = float(self.default_params.get("decline_pct_threshold", 30))
months_lookback = int(self.default_params.get("months_lookback", 3))
now = utcnow()
# Build month buckets
month_boundaries: List[datetime] = []
cur = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
for i in range(months_lookback + 1):
month_boundaries.append(cur)
# Go to previous month
prev_month = cur.month - 1
prev_year = cur.year
if prev_month == 0:
prev_month = 12
prev_year -= 1
cur = cur.replace(year=prev_year, month=prev_month)
month_boundaries.reverse()
# Group invoices by customer and month
invoices = QuickbooksInvoice.query.filter(
QuickbooksInvoice.company_id == company_id,
QuickbooksInvoice.tx_date >= month_boundaries[0].replace(tzinfo=None),
QuickbooksInvoice.status.notin_(["Void"]),
).all()
customer_months: Dict[str, Dict[str, Any]] = {}
for inv in invoices:
cust_name = inv.customer_name or "Unknown"
tx_date = as_aware(inv.tx_date)
if not tx_date:
continue
# Find which month bucket this falls into
month_key = None
for i in range(len(month_boundaries) - 1):
if month_boundaries[i] <= tx_date < month_boundaries[i + 1]:
month_key = month_boundaries[i].strftime("%Y-%m")
break
if not month_key:
month_key = "current"
cust = customer_months.setdefault(cust_name, {
"monthly_counts": {},
"monthly_totals": {},
"open_invoices": 0,
"unpaid_balance": 0.0,
})
cust["monthly_counts"][month_key] = (
cust["monthly_counts"].get(month_key, 0) + 1
)
cust["monthly_totals"][month_key] = (
cust["monthly_totals"].get(month_key, 0.0)
+ (inv.total_amount or 0.0)
)
if inv.status not in ("Paid", "paid"):
cust["open_invoices"] += 1
cust["unpaid_balance"] += inv.total_amount or 0.0
candidates: List[LeakCandidate] = []
min_baseline = int(self.default_params.get("min_invoices_baseline", 2))
for cust_name, data in customer_months.items():
# Compare most recent month to previous month
sorted_months = sorted(data["monthly_counts"].keys())
if len(sorted_months) < 2:
continue
last_month = sorted_months[-1]
prev_month = sorted_months[-2]
last_count = data["monthly_counts"][last_month]
prev_count = data["monthly_counts"][prev_month]
last_total = data["monthly_totals"].get(last_month, 0.0)
prev_total = data["monthly_totals"].get(prev_month, 0.0)
if prev_count < min_baseline:
continue # No baseline to compare against
decline_pct = ((prev_count - last_count) / prev_count) * 100
if decline_pct < decline_threshold:
continue
if decline_pct > 75:
severity = "critical"
elif decline_pct > 50:
severity = "high"
else:
severity = "medium"
# Projected 3-month loss if churned
projected_loss = round(last_total * months_lookback, 2)
unpaid_str = ""
if data["open_invoices"] > 0:
unpaid_str = (
f" {data['open_invoices']} open invoices unpaid "
f"(${data['unpaid_balance']:,.0f})"
)
candidates.append(
LeakCandidate(
detector_id=self.id,
source=f"QuickBooks: Churn Risk — {cust_name}",
description=(
f"Customer '{cust_name}' invoicing dropped "
f"{decline_pct:.0f}% (was {prev_count}, now "
f"{last_count}){unpaid_str}"
),
estimated_loss=projected_loss,
severity=severity,
metadata_json={
"dedupe_key": f"{self.id}:{cust_name}",
"customer_name": cust_name,
"invoice_decline_pct": round(decline_pct, 1),
"last_month_total": round(last_total, 2),
"prev_month_total": round(prev_total, 2),
"open_invoices": data["open_invoices"],
"unpaid_balance": round(data["unpaid_balance"], 2),
},
source_connector="quickbooks",
rule_params={
"decline_pct_threshold": decline_threshold,
"months_lookback": months_lookback,
},
)
)
return candidates
# =============================================================================
# Detector 5: budget_allocation — Budget Misallocation
# =============================================================================
class BudgetAllocationDetector(BaseDetector):
id = "budget_allocation"
name = "Ads: Budget Misallocation"
connector = "google_ads"
enabled_by_default = True
description = (
"Identifies campaigns spending past end dates, paused high-ROAS "
"campaigns, and over-budget low-ROAS campaigns."
)
default_params = {"days": 30, "overbudget_pct": 20, "roas_quartile": 0.25}
def check(self, company_id: str) -> List[LeakCandidate]:
from app.models import AdCampaign, AdMetric
services = _ads_connectors_available(company_id)
if not services:
return []
days = int(self.default_params.get("days", 30))
cutoff = (utcnow() - timedelta(days=days)).date()
today = utcnow().date()
metrics = AdMetric.query.filter(
AdMetric.company_id == company_id,
AdMetric.source_service.in_(services),
AdMetric.metric_date >= cutoff,
).all()
campaigns = AdCampaign.query.filter(
AdCampaign.company_id == company_id,
AdCampaign.source_service.in_(services),
).all()
# Aggregate spend and ROAS per campaign
per_campaign: Dict[str, Dict[str, Any]] = {}
for m in metrics:
cid = m.external_campaign_id or ""
if not cid:
continue
agg = per_campaign.setdefault(cid, {
"spend": 0.0,
"revenue": 0.0,
"roas_values": [],
})
spend = m.spend or 0.0
agg["spend"] += spend
if m.roas is not None:
revenue = spend * m.roas
agg["revenue"] += revenue
agg["roas_values"].append(m.roas)
# Calculate ROAS quartile for "top/bottom 25%" classification
all_roas = [
agg["revenue"] / agg["spend"]
for agg in per_campaign.values()
if agg["spend"] > 0 and agg["revenue"] > 0
]
all_roas.sort()
def _quartile_value(q: float) -> float:
if not all_roas:
return 0
idx = max(0, int(len(all_roas) * q) - 1)
return all_roas[idx]
bottom_25_roas = _quartile_value(self.default_params.get("roas_quartile", 0.25))
top_25_roas = _quartile_value(0.75)
total_spend = sum(a["spend"] for a in per_campaign.values())
issues = []
for camp in campaigns:
cid = camp.external_id
agg = per_campaign.get(cid)
if not agg or agg["spend"] <= 0:
continue
avg_roas = agg["revenue"] / agg["spend"] if agg["spend"] > 0 else 0
# Issue 1: Enabled campaigns past end_date
if (camp.status in ("ENABLED", "enabled", "active")
and camp.end_date and camp.end_date < today):
daily_spend = agg["spend"] / days
issues.append({
"type": "past_end_date",
"campaign_id": cid,
"campaign_name": camp.name,
"daily_spend": daily_spend,
"days_past": (today - camp.end_date).days,
})
# Issue 2: Paused campaigns with high ROAS (top 25%)
if (camp.status in ("PAUSED", "paused", "inactive")
and top_25_roas > 0
and avg_roas >= top_25_roas):
issues.append({
"type": "paused_high_roas",
"campaign_id": cid,
"campaign_name": camp.name,
"daily_spend": agg["spend"] / days,
"roas": round(avg_roas, 2),
})
# Issue 3: Over-budget low-ROAS
overbudget_pct = self.default_params.get("overbudget_pct", 20)
if (total_spend > 0
and (agg["spend"] / total_spend * 100) > overbudget_pct
and bottom_25_roas > 0
and avg_roas <= bottom_25_roas):
issues.append({
"type": "overbudget_low_roas",
"campaign_id": cid,
"campaign_name": camp.name,
"daily_spend": agg["spend"] / days,
"budget_share_pct": round(agg["spend"] / total_spend * 100, 1),
"roas": round(avg_roas, 2),
})
if not issues:
return []
# Calculate monthly waste
monthly_waste = 0.0
for issue in issues:
daily = issue.get("daily_spend", 0)
if issue["type"] == "past_end_date":
monthly_waste += daily * issue.get("days_past", 0)
elif issue["type"] == "paused_high_roas":
monthly_waste += daily * 30 # missed opportunity
elif issue["type"] == "overbudget_low_roas":
monthly_waste += daily * 30 * 0.5 # partial waste estimate
severity = "critical" if len(issues) >= 3 else (
"high" if len(issues) >= 2 else "medium"
)
return [
LeakCandidate(
detector_id=self.id,
source="Ads: Budget Misallocation",
description=(
f"{len(issues)} budget issues detected: "
+ "; ".join(
f"{i['type']} — {i['campaign_name']}"
for i in issues
)
+ f". Estimated monthly waste: ${monthly_waste:,.0f}"
),
estimated_loss=round(monthly_waste, 2),
severity=severity,
metadata_json={
"dedupe_key": f"{self.id}:budget_allocation",
"campaign_issues": issues,
"monthly_waste": round(monthly_waste, 2),
"total_spend": round(total_spend, 2),
"num_issues": len(issues),
},
source_connector=services[0],
rule_params={
"days": days,
"overbudget_pct": self.default_params.get("overbudget_pct", 20),
},
)
]
# =============================================================================
# Helper
# =============================================================================
def _parse_dt(value: Any) -> Optional[datetime]:
if value is None or value == "":
return None
if isinstance(value, datetime):
return as_aware(value)
if isinstance(value, (int, float)):
try:
ts = float(value)
if ts > 1e12:
ts /= 1000.0
return datetime.fromtimestamp(ts, tz=timezone.utc)
except (ValueError, OSError, OverflowError):
return None
if isinstance(value, str):
raw = value.strip()
if not raw:
return None
try:
return as_aware(datetime.fromisoformat(raw.replace("Z", "+00:00")))
except ValueError:
pass
for fmt in ("%m/%d/%Y %H:%M:%S", "%m/%d/%Y %H:%M", "%m/%d/%Y",
"%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
try:
return datetime.strptime(raw, fmt).replace(tzinfo=timezone.utc)
except ValueError:
continue
return None
# =============================================================================
# Phase 3 detector classes — imported and registered in __init__.py
# =============================================================================
DETECTORS = [
ChannelQualityDetector,
SmsResponseTimeDetector,
ForecastGapDetector,
CustomerChurnRiskDetector,
BudgetAllocationDetector,
]