"""Phase 2 leak detectors — behavioral patterns (trends, baselines, spikes).
Eight trend-aware detectors that compare current behaviour against
historical baselines:
ads_low_roas — campaign ROAS below target over trailing 14 days
ads_rising_cpc — CPC up >15% week-over-week with flat conversions
qb_refund_spike — CreditMemo volume 2x the trailing 3-month average
qb_margin_erosion — expense category spike while revenue is flat
crm_funnel_dropoff — stage conversion rate below historical baseline
leads_noshow — past LeadPerfection appointments with no result
five9_missed_followup — voicemail/no-answer calls with no 24h follow-up
qb_partial_stall — partially paid invoices with no payment in 30+ days
All detectors inherit BaseDetector and are exported via ``DETECTORS`` for
registration in ``__init__``.
"""
from __future__ import annotations
import json
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, days_since, utcnow
logger = logging.getLogger(__name__)
#: Ads connector services that populate ad_campaigns / ad_metrics
ADS_SERVICES = ("google_ads", "facebook_ads")
def _ads_connectors_available(company_id: str) -> List[str]:
"""Connected ads services for a company."""
return [s for s in ADS_SERVICES if connector_available(company_id, s)]
def _parse_dt(value: Any) -> Optional[datetime]:
"""Best-effort parse of a datetime out of raw connector payload data.
Handles ISO 8601 (with/without Z), epoch seconds/milliseconds, and
common US date formats. Returns a UTC-aware datetime or None.
"""
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: # epoch millis
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
# ISO 8601
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",
"%m-%d-%Y",
):
try:
return datetime.strptime(raw, fmt).replace(tzinfo=timezone.utc)
except ValueError:
continue
return None
def _month_start(dt: datetime) -> datetime:
return dt.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
def _shift_month(dt: datetime, months: int) -> datetime:
"""Shift a month-start datetime by N months (negative = back)."""
total = dt.year * 12 + (dt.month - 1) + months
year, month = divmod(total, 12)
return dt.replace(year=year, month=month + 1)
# =============================================================================
# Detector 1: ads_low_roas — Underperforming Campaigns
# =============================================================================
class AdsLowRoasDetector(BaseDetector):
id = "ads_low_roas"
name = "Ads: Underperforming Campaigns"
connector = "google_ads" # primary; also checks facebook_ads
enabled_by_default = True
description = (
"Flags ad campaigns whose ROAS over the trailing 14 days is below "
"the target (2.0 by default), wasting ad spend."
)
default_params = {"target_roas": 2.0, "days": 14}
def check(self, company_id: str) -> List[LeakCandidate]:
services = _ads_connectors_available(company_id)
if not services:
return []
from app.models import AdCampaign, AdMetric
days = int(self.default_params.get("days", 14))
target = float(self.default_params.get("target_roas", 2.0))
cutoff = (utcnow() - timedelta(days=days)).date()
metrics = AdMetric.query.filter(
AdMetric.company_id == company_id,
AdMetric.source_service.in_(services),
AdMetric.metric_date >= cutoff,
).all()
if not metrics:
return []
# Aggregate per campaign: spend + revenue (spend * roas per row)
per_campaign: Dict[str, Dict[str, float]] = {}
total_spend = 0.0
total_revenue = 0.0
for m in metrics:
cid = m.external_campaign_id or ""
if not cid:
continue
spend = m.spend or 0.0
revenue = spend * (m.roas or 0.0)
agg = per_campaign.setdefault(
cid, {"spend": 0.0, "revenue": 0.0, "roas_rows": 0}
)
agg["spend"] += spend
agg["revenue"] += revenue
if m.roas is not None:
agg["roas_rows"] += 1
total_spend += spend
total_revenue += revenue
account_roas = (total_revenue / total_spend) if total_spend > 0 else None
# Campaign name lookup
campaigns = {
c.external_id: c
for c in AdCampaign.query.filter(
AdCampaign.company_id == company_id,
AdCampaign.source_service.in_(services),
).all()
}
candidates: List[LeakCandidate] = []
for cid, agg in per_campaign.items():
spend = agg["spend"]
if spend <= 0 or agg["roas_rows"] == 0:
continue # no meaningful ROAS data
actual = agg["revenue"] / spend
if actual >= target:
continue
severity = "high" if actual < 1.0 else "medium"
loss = round(spend * (1 - actual / target), 2)
campaign = campaigns.get(cid)
name = (campaign.name if campaign else "") or cid
candidates.append(
LeakCandidate(
detector_id=self.id,
source="Ads: Underperforming Campaign",
description=(
f"Campaign '{name}' ROAS is {actual:.2f} vs target "
f"{target:.1f}. Spent ${spend:,.2f} in {days} days"
),
estimated_loss=loss,
severity=severity,
metadata_json={
"dedupe_key": f"{self.id}:{cid}",
"campaign_id": cid,
"campaign_name": name,
"actual_roas": round(actual, 2),
"target_roas": target,
"account_roas": round(account_roas, 2) if account_roas else None,
"spend": round(spend, 2),
"days": days,
},
source_connector=(campaign.source_service if campaign else services[0]),
rule_params={"target_roas": target, "days": days},
)
)
return candidates
# =============================================================================
# Detector 2: ads_rising_cpc — Rising Acquisition Cost
# =============================================================================
class AdsRisingCpcDetector(BaseDetector):
id = "ads_rising_cpc"
name = "Ads: Rising Acquisition Cost"
connector = "google_ads"
enabled_by_default = True
description = (
"Flags campaigns whose cost-per-click rose more than 15% "
"week-over-week while conversions stayed flat or declined."
)
default_params = {"pct_threshold": 15.0}
def check(self, company_id: str) -> List[LeakCandidate]:
services = _ads_connectors_available(company_id)
if not services:
return []
from app.models import AdCampaign, AdMetric
threshold = float(self.default_params.get("pct_threshold", 15.0))
today = utcnow().date()
cur_start = today - timedelta(days=7)
prev_start = today - timedelta(days=14)
metrics = AdMetric.query.filter(
AdMetric.company_id == company_id,
AdMetric.source_service.in_(services),
AdMetric.metric_date >= prev_start,
).all()
if not metrics:
return []
# Per campaign, split into previous week / current week buckets
buckets: Dict[str, Dict[str, Dict[str, float]]] = {}
earliest: Dict[str, date] = {}
for m in metrics:
cid = m.external_campaign_id or ""
if not cid or m.metric_date is None:
continue
week = "current" if m.metric_date >= cur_start else "previous"
camp = buckets.setdefault(cid, {
"current": {"spend": 0.0, "clicks": 0.0, "conversions": 0.0},
"previous": {"spend": 0.0, "clicks": 0.0, "conversions": 0.0},
})
camp[week]["spend"] += m.spend or 0.0
camp[week]["clicks"] += m.clicks or 0
camp[week]["conversions"] += m.conversions or 0.0
if cid not in earliest or m.metric_date < earliest[cid]:
earliest[cid] = m.metric_date
campaigns = {
c.external_id: c
for c in AdCampaign.query.filter(
AdCampaign.company_id == company_id,
AdCampaign.source_service.in_(services),
).all()
}
candidates: List[LeakCandidate] = []
for cid, camp in buckets.items():
cur, prev = camp["current"], camp["previous"]
# Require a full comparison window of data
if prev["clicks"] <= 0 or cur["clicks"] <= 0:
continue
old_cpc = prev["spend"] / prev["clicks"]
new_cpc = cur["spend"] / cur["clicks"]
if old_cpc <= 0:
continue
pct = (new_cpc - old_cpc) / old_cpc * 100.0
if pct <= threshold:
continue
# Conversions must be flat or declining
if cur["conversions"] > prev["conversions"]:
continue
if cur["conversions"] < prev["conversions"]:
conv_trend = "declined"
else:
conv_trend = "stayed flat"
severity = "high" if pct > 30 else "medium"
loss = round(max(new_cpc - old_cpc, 0.0) * cur["clicks"], 2)
campaign = campaigns.get(cid)
name = (campaign.name if campaign else "") or cid
candidates.append(
LeakCandidate(
detector_id=self.id,
source="Ads: Rising Acquisition Cost",
description=(
f"CPC on '{name}' up {pct:.0f}% (was ${old_cpc:.2f}, "
f"now ${new_cpc:.2f}) while conversions {conv_trend}"
),
estimated_loss=loss,
severity=severity,
metadata_json={
"dedupe_key": f"{self.id}:{cid}",
"campaign_id": cid,
"campaign_name": name,
"old_cpc": round(old_cpc, 2),
"new_cpc": round(new_cpc, 2),
"pct_change": round(pct, 1),
"conversion_trend": conv_trend,
},
source_connector=(campaign.source_service if campaign else services[0]),
rule_params={"pct_threshold": threshold},
)
)
return candidates
# =============================================================================
# Detector 3: qb_refund_spike — Refund/Discount Spike
# =============================================================================
class QbRefundSpikeDetector(BaseDetector):
id = "qb_refund_spike"
name = "QuickBooks: Refund/Discount Spike"
connector = "quickbooks"
enabled_by_default = True
description = (
"Flags months where credit memo (refund/discount) volume exceeds "
"2x the trailing 3-month average."
)
default_params = {"multiplier_threshold": 2.0, "trailing_months": 3}
def check(self, company_id: str) -> List[LeakCandidate]:
if not connector_available(company_id, self.connector):
return []
from app.models import QuickbooksTransaction
threshold = float(self.default_params.get("multiplier_threshold", 2.0))
trailing_months = int(self.default_params.get("trailing_months", 3))
now = utcnow()
cur_month_start = _month_start(now)
window_start = _shift_month(cur_month_start, -trailing_months)
txs = QuickbooksTransaction.query.filter(
QuickbooksTransaction.company_id == company_id,
QuickbooksTransaction.tx_type == "CreditMemo",
QuickbooksTransaction.tx_date.isnot(None),
QuickbooksTransaction.tx_date >= window_start.replace(tzinfo=None),
).all()
if not txs:
return []
current_total = 0.0
trailing_total = 0.0
for tx in txs:
tx_date = as_aware(tx.tx_date)
if tx_date is None:
continue
amt = abs(tx.amount or 0.0)
if tx_date >= cur_month_start:
current_total += amt
else:
trailing_total += amt
average = trailing_total / trailing_months if trailing_months else 0.0
if average <= 0 or current_total <= threshold * average:
return []
multiplier = current_total / average
if multiplier > 5:
severity = "critical"
elif multiplier > 3:
severity = "high"
else:
severity = "medium"
month_label = cur_month_start.strftime("%Y-%m")
return [
LeakCandidate(
detector_id=self.id,
source="QuickBooks: Refund/Discount Spike",
description=(
f"Refunds/discounts this month (${current_total:,.2f}) are "
f"{multiplier:.1f}x the {trailing_months}-month average "
f"(${average:,.2f})"
),
estimated_loss=round(current_total - average, 2),
severity=severity,
metadata_json={
"dedupe_key": f"{self.id}:{month_label}",
"current_month_total": round(current_total, 2),
"trailing_average": round(average, 2),
"multiplier": round(multiplier, 2),
"month": month_label,
},
source_connector="quickbooks",
rule_params={
"multiplier_threshold": threshold,
"trailing_months": trailing_months,
},
)
]
# =============================================================================
# Detector 4: qb_margin_erosion — Margin Erosion
# =============================================================================
class QbMarginErosionDetector(BaseDetector):
id = "qb_margin_erosion"
name = "QuickBooks: Margin Erosion"
connector = "quickbooks"
enabled_by_default = True
description = (
"Flags expense categories with a month-over-month increase above "
"20% while revenue stayed flat — compressed margins."
)
default_params = {"expense_pct_threshold": 20.0, "revenue_flat_pct": 5.0}
def check(self, company_id: str) -> List[LeakCandidate]:
if not connector_available(company_id, self.connector):
return []
from app.models import QuickbooksExpense, QuickbooksInvoice
exp_threshold = float(self.default_params.get("expense_pct_threshold", 20.0))
rev_flat_pct = float(self.default_params.get("revenue_flat_pct", 5.0))
now = utcnow()
cur_start = _month_start(now)
prev_start = _shift_month(cur_start, -1)
window_start_naive = prev_start.replace(tzinfo=None)
# --- Revenue MoM (invoices by tx_date) -------------------------------
invoices = QuickbooksInvoice.query.filter(
QuickbooksInvoice.company_id == company_id,
QuickbooksInvoice.tx_date.isnot(None),
QuickbooksInvoice.tx_date >= window_start_naive,
QuickbooksInvoice.status.notin_(["Void"]),
).all()
cur_rev = prev_rev = 0.0
for inv in invoices:
d = as_aware(inv.tx_date)
if d is None:
continue
if d >= cur_start:
cur_rev += inv.total_amount or 0.0
else:
prev_rev += inv.total_amount or 0.0
if prev_rev <= 0:
return [] # no revenue baseline — can't call it "flat"
rev_pct = (cur_rev - prev_rev) / prev_rev * 100.0
if abs(rev_pct) >= rev_flat_pct:
return [] # revenue moved — not a flat-revenue margin squeeze
# --- Expenses by category MoM -----------------------------------------
expenses = QuickbooksExpense.query.filter(
QuickbooksExpense.company_id == company_id,
QuickbooksExpense.tx_date.isnot(None),
QuickbooksExpense.tx_date >= window_start_naive,
).all()
by_cat: Dict[str, Dict[str, float]] = {}
for exp in expenses:
d = as_aware(exp.tx_date)
if d is None:
continue
cat = (exp.category or exp.account_name or "Uncategorized").strip() or "Uncategorized"
agg = by_cat.setdefault(cat, {"current": 0.0, "previous": 0.0})
key = "current" if d >= cur_start else "previous"
agg[key] += abs(exp.amount or 0.0)
month_label = cur_start.strftime("%Y-%m")
candidates: List[LeakCandidate] = []
for cat, agg in by_cat.items():
prev_amt, cur_amt = agg["previous"], agg["current"]
if prev_amt <= 0:
continue # new category — no baseline
pct = (cur_amt - prev_amt) / prev_amt * 100.0
if pct <= exp_threshold:
continue
increase = round(cur_amt - prev_amt, 2)
severity = "high" if pct > 40 else "medium"
candidates.append(
LeakCandidate(
detector_id=self.id,
source="QuickBooks: Margin Erosion",
description=(
f"Expenses in '{cat}' up ${increase:,.2f} ({pct:.0f}%) "
f"while revenue changed {rev_pct:+.1f}% — margin compressed"
),
estimated_loss=increase,
severity=severity,
metadata_json={
"dedupe_key": f"{self.id}:{cat}:{month_label}",
"category": cat,
"expense_increase": increase,
"pct_change": round(pct, 1),
"revenue_pct_change": round(rev_pct, 1),
"month": month_label,
},
source_connector="quickbooks",
rule_params={
"expense_pct_threshold": exp_threshold,
"revenue_flat_pct": rev_flat_pct,
},
)
)
return candidates
# =============================================================================
# Detector 5: crm_funnel_dropoff — Funnel Stage Drop-off
# =============================================================================
class CrmFunnelDropoffDetector(BaseDetector):
id = "crm_funnel_dropoff"
name = "CRM: Funnel Drop-off"
connector = "hubspot"
enabled_by_default = True
description = (
"Flags funnel stage transitions whose recent conversion rate fell "
"below 60% of the 6-month historical baseline."
)
default_params = {
"baseline_months": 6,
"recent_days": 30,
"dropoff_ratio": 0.6,
"min_baseline_deals": 5,
}
#: Known canonical funnel orderings (lowercased) — deals advance left→right.
_STAGE_ORDER_HINTS = [
"new", "lead", "appointments_set", "appointmentscheduled",
"appointment_set", "qualified", "qualifiedtobuy", "presentation",
"presentationscheduled", "proposal", "quote", "decisionmakerboughtin",
"contractsent", "negotiation", "closedwon", "closed won", "won",
]
def check(self, company_id: str) -> List[LeakCandidate]:
if not connector_available(company_id, self.connector):
return []
from app.models import CrmDeal
params = self.default_params
baseline_months = int(params.get("baseline_months", 6))
recent_days = int(params.get("recent_days", 30))
ratio = float(params.get("dropoff_ratio", 0.6))
min_deals = int(params.get("min_baseline_deals", 5))
now = utcnow()
recent_cutoff = now - timedelta(days=recent_days)
baseline_cutoff = now - timedelta(days=baseline_months * 30)
deals = CrmDeal.query.filter(
CrmDeal.company_id == company_id,
CrmDeal.created_at >= baseline_cutoff.replace(tzinfo=None),
).all()
if not deals:
return []
# Build funnel stage order: hint-based ranking first, then by deal
# count descending (funnels narrow as they progress).
stage_counts: Dict[str, int] = {}
for d in deals:
stage = (d.stage or "").strip()
if stage:
stage_counts[stage] = stage_counts.get(stage, 0) + 1
if len(stage_counts) < 2:
return []
def _rank(stage: str):
low = stage.lower().replace(" ", "_")
for i, hint in enumerate(self._STAGE_ORDER_HINTS):
if hint.replace(" ", "_") == low:
return (0, i)
return (1, -stage_counts[stage]) # unknown: order by count desc
ordered_stages = sorted(stage_counts, key=_rank)
stage_index = {s: i for i, s in enumerate(ordered_stages)}
# Cohorts: baseline = older deals, recent = last N days
def _cohort_rates(cohort: List) -> Dict[int, Dict[str, float]]:
"""For each stage index i: deals at/past i, and conversion i→i+1."""
reached = [0] * len(ordered_stages)
for deal in cohort:
idx = stage_index.get((deal.stage or "").strip())
if idx is None:
continue
for i in range(idx + 1):
reached[i] += 1
rates: Dict[int, Dict[str, float]] = {}
for i in range(len(ordered_stages) - 1):
if reached[i] > 0:
rates[i] = {
"rate": reached[i + 1] / reached[i],
"reached": reached[i],
"advanced": reached[i + 1],
}
return rates
baseline_cohort = [
d for d in deals
if as_aware(d.created_at) and as_aware(d.created_at) < recent_cutoff
]
recent_cohort = [
d for d in deals
if as_aware(d.created_at) and as_aware(d.created_at) >= recent_cutoff
]
if not baseline_cohort or not recent_cohort:
return []
baseline_rates = _cohort_rates(baseline_cohort)
recent_rates = _cohort_rates(recent_cohort)
# Find dropped transitions
dropped = []
for i, base in baseline_rates.items():
if base["reached"] < min_deals or base["rate"] <= 0:
continue
rec = recent_rates.get(i)
if rec is None or rec["reached"] <= 0:
continue
if rec["rate"] < base["rate"] * ratio:
dropped.append((i, base, rec))
if not dropped:
return []
severity = "high" if len(dropped) >= 2 else "medium"
candidates: List[LeakCandidate] = []
for i, base, rec in dropped:
stage_a, stage_b = ordered_stages[i], ordered_stages[i + 1]
stuck_deals = [
d for d in recent_cohort
if (d.stage or "").strip() == stage_a
]
deals_stuck = len(stuck_deals)
amounts = [d.amount or 0.0 for d in stuck_deals]
avg_deal_size = (sum(amounts) / len(amounts)) if amounts else 0.0
est_loss = round(sum(
(d.amount or avg_deal_size or 0.0)
* (d.probability if d.probability is not None else 0.5)
for d in stuck_deals
), 2) or None
cur_pct = rec["rate"] * 100.0
avg_pct = base["rate"] * 100.0
candidates.append(
LeakCandidate(
detector_id=self.id,
source="CRM: Funnel Drop-off",
description=(
f"Conversion from '{stage_a}' → '{stage_b}' dropped to "
f"{cur_pct:.0f}% (avg was {avg_pct:.0f}%). "
f"{deals_stuck} deals stuck."
),
estimated_loss=est_loss,
severity=severity,
metadata_json={
"dedupe_key": f"{self.id}:{stage_a}->{stage_b}",
"stage_from": stage_a,
"stage_to": stage_b,
"current_rate": round(cur_pct, 1),
"avg_rate": round(avg_pct, 1),
"deals_stuck": deals_stuck,
"avg_deal_size": round(avg_deal_size, 2),
},
source_connector="hubspot",
rule_params={
"baseline_months": baseline_months,
"recent_days": recent_days,
"dropoff_ratio": ratio,
},
)
)
return candidates
# =============================================================================
# Detector 6: leads_noshow — No-Show Appointments (LeadPerfection)
# =============================================================================
class LeadsNoshowDetector(BaseDetector):
id = "leads_noshow"
name = "LeadPerfection: No-Show Appointments"
connector = "leadperfection"
enabled_by_default = True
description = (
"Flags past LeadPerfection appointments with no result logged — "
"likely no-shows that never got rescheduled."
)
default_params = {"lookback_days": 90}
def check(self, company_id: str) -> List[LeakCandidate]:
if not connector_available(company_id, self.connector):
return []
from app.models import ExternalSyncRecord
lookback = int(self.default_params.get("lookback_days", 90))
now = utcnow()
lookback_cutoff = now - timedelta(days=lookback)
records = ExternalSyncRecord.query.filter(
ExternalSyncRecord.company_id == company_id,
ExternalSyncRecord.source_service == "leadperfection",
ExternalSyncRecord.entity_type == "lead",
).all()
candidates: List[LeakCandidate] = []
for rec in records:
props = rec.properties_json or {}
appt_dt = _parse_dt(props.get("appt_date"))
if appt_dt is None or appt_dt >= now or appt_dt < lookback_cutoff:
continue
# Result logged? status column carries status/apptresult; also
# check the raw payload for an explicit apptresult.
raw = props.get("raw") or {}
appt_result = str(
raw.get("apptresult") or raw.get("ApptResult") or ""
).strip()
status = (rec.status or "").strip()
if status or appt_result:
continue
name = rec.name or rec.external_id
candidates.append(
LeakCandidate(
detector_id=self.id,
source="LeadPerfection: No-Show Appointment",
description=(
f"Appointment for '{name}' on "
f"{appt_dt.strftime('%Y-%m-%d')} has no result logged "
f"— possible no-show"
),
estimated_loss=None,
severity="medium",
metadata_json={
"dedupe_key": f"{self.id}:{rec.external_id}",
"external_id": rec.external_id,
"appt_date": appt_dt.isoformat(),
"source_service": "leadperfection",
},
source_connector="leadperfection",
rule_params={"lookback_days": lookback},
)
)
return candidates
# =============================================================================
# Detector 7: five9_missed_followup — Missed Call Follow-up
# =============================================================================
class Five9MissedFollowupDetector(BaseDetector):
id = "five9_missed_followup"
name = "Five9: Missed Follow-ups"
connector = "five9"
enabled_by_default = True
description = (
"Flags voicemail/no-answer/busy calls from the last 7 days with no "
"follow-up call to the same contact within 24 hours."
)
default_params = {"lookback_days": 7, "followup_hours": 24}
_MISSED_DISPOSITIONS = {"no-answer", "no answer", "noanswer", "voicemail", "busy"}
def check(self, company_id: str) -> List[LeakCandidate]:
if not connector_available(company_id, self.connector):
return []
from app.models import ExternalSyncRecord
lookback = int(self.default_params.get("lookback_days", 7))
followup_hours = int(self.default_params.get("followup_hours", 24))
now = utcnow()
cutoff = now - timedelta(days=lookback)
followup_window = timedelta(hours=followup_hours)
records = ExternalSyncRecord.query.filter(
ExternalSyncRecord.company_id == company_id,
ExternalSyncRecord.source_service == "five9",
ExternalSyncRecord.entity_type == "call_log",
).all()
# Parse all calls once: (contact_key, started_at, disposition, record)
calls = []
for rec in records:
props = rec.properties_json or {}
started = _parse_dt(props.get("started_at"))
if started is None:
started = as_aware(rec.created_at)
disposition = str(
props.get("disposition") or rec.status or ""
).strip().lower()
# Contact key: explicit contact_id in payload, else the phone
# number (record.name carries ani/dnis).
raw = props.get("raw") or {}
contact_key = str(
raw.get("contactId") or raw.get("contact_id")
or rec.name or ""
).strip()
calls.append((contact_key, started, disposition, rec))
candidates: List[LeakCandidate] = []
for contact_key, started, disposition, rec in calls:
if started is None or started < cutoff or started >= now:
continue
if disposition not in self._MISSED_DISPOSITIONS:
continue
if not contact_key:
continue
# Give calls newer than the follow-up window a chance to be followed up
if now - started < followup_window:
continue
# Any subsequent call to the same contact within the window?
followed_up = any(
other_key == contact_key
and other_started is not None
and started < other_started <= started + followup_window
for other_key, other_started, _d, other_rec in calls
if other_rec.id != rec.id
)
if followed_up:
continue
contact_label = rec.name or contact_key
candidates.append(
LeakCandidate(
detector_id=self.id,
source="Five9: Missed Follow-up",
description=(
f"Call to '{contact_label}' resulted in {disposition} "
f"with no follow-up in {followup_hours} hours"
),
estimated_loss=None,
severity="medium",
metadata_json={
"dedupe_key": f"{self.id}:{rec.external_id}",
"external_id": rec.external_id,
"contact_id": contact_key,
"disposition": disposition,
"call_date": started.isoformat(),
},
source_connector="five9",
rule_params={
"lookback_days": lookback,
"followup_hours": followup_hours,
},
)
)
return candidates
# =============================================================================
# Detector 8: qb_partial_stall — Partial Payment Stall
# =============================================================================
class QbPartialStallDetector(BaseDetector):
id = "qb_partial_stall"
name = "QuickBooks: Partial Payment Stalls"
connector = "quickbooks"
enabled_by_default = True
description = (
"Flags partially paid invoices with no new payment activity in 30+ "
"days — outstanding balances going stale."
)
default_params = {"stall_days": 30}
def check(self, company_id: str) -> List[LeakCandidate]:
if not connector_available(company_id, self.connector):
return []
from app.models import QuickbooksInvoice, QuickbooksTransaction
stall_days = int(self.default_params.get("stall_days", 30))
now = utcnow()
invoices = QuickbooksInvoice.query.filter(
QuickbooksInvoice.company_id == company_id,
QuickbooksInvoice.status == "Partial",
).all()
if not invoices:
return []
# Payments in the stall window — used to prove recent activity.
payment_cutoff = (now - timedelta(days=stall_days)).replace(tzinfo=None)
recent_payments = QuickbooksTransaction.query.filter(
QuickbooksTransaction.company_id == company_id,
QuickbooksTransaction.tx_type == "Payment",
QuickbooksTransaction.tx_date.isnot(None),
QuickbooksTransaction.tx_date >= payment_cutoff,
).all()
def _payment_references_invoice(tx, invoice) -> bool:
"""Heuristic link: raw payment payload mentions the invoice doc id."""
meta = tx.metadata_json or {}
try:
blob = json.dumps(meta, default=str)
except (TypeError, ValueError):
return False
return bool(invoice.qb_doc_id) and invoice.qb_doc_id in blob
candidates: List[LeakCandidate] = []
for inv in invoices:
# Recent payment against this invoice → not stalled
if any(_payment_references_invoice(tx, inv) for tx in recent_payments):
continue
total = inv.total_amount or 0.0
meta = inv.metadata_json or {}
# QBO invoice raw carries Balance = amount still owed
balance = meta.get("Balance")
try:
remaining = float(balance) if balance is not None else None
except (TypeError, ValueError):
remaining = None
if remaining is None or remaining <= 0 or remaining > total:
remaining = total # unknown paid amount — assume full balance
amount_paid = round(max(total - remaining, 0.0), 2)
days_partial = days_since(inv.tx_date or inv.updated_at)
if days_partial < stall_days:
continue
if days_partial > 60:
severity = "critical"
elif days_partial > 30:
severity = "high"
else:
severity = "medium"
candidates.append(
LeakCandidate(
detector_id=self.id,
source="QuickBooks: Partial Payment Stall",
description=(
f"Invoice {inv.invoice_num or inv.qb_doc_id} for "
f"${total:,.2f} is partially paid — "
f"${remaining:,.2f} outstanding for {days_partial} days"
),
estimated_loss=round(remaining, 2),
severity=severity,
metadata_json={
"dedupe_key": f"{self.id}:{inv.qb_doc_id}",
"invoice_id": inv.id,
"invoice_num": inv.invoice_num,
"total_amount": round(total, 2),
"amount_paid": amount_paid,
"remaining": round(remaining, 2),
"days_partial": days_partial,
},
source_connector="quickbooks",
rule_params={"stall_days": stall_days},
)
)
return candidates
#: Phase 2 detector classes — imported and registered in __init__.py
DETECTORS = [
AdsLowRoasDetector,
AdsRisingCpcDetector,
QbRefundSpikeDetector,
QbMarginErosionDetector,
CrmFunnelDropoffDetector,
LeadsNoshowDetector,
Five9MissedFollowupDetector,
QbPartialStallDetector,
]