"""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, ]