"""Detector: ad campaigns with spend but zero conversions in the last 7 days."""
from __future__ import annotations
from datetime import timedelta
from typing import List
from sqlalchemy import func
from .base import BaseDetector, LeakCandidate
from ._common import any_connector_available, connector_available, utcnow
AD_SERVICES = ("google_ads", "facebook_ads")
def _spend_severity(spend: float) -> str:
if spend > 2000:
return "critical"
if spend >= 500:
return "high"
return "medium"
class AdsZeroConversionDetector(BaseDetector):
id = "ads_zero_conversion"
name = "Ads: Zero-Conversion Campaigns"
connector = None # google_ads OR facebook_ads
enabled_by_default = True
description = (
"Flags ad campaigns that spent money over the last 7 days without a "
"single conversion."
)
default_params = {"window_days": 7, "high_spend": 500, "critical_spend": 2000}
def check(self, company_id: str) -> List[LeakCandidate]:
if not any_connector_available(company_id, AD_SERVICES):
return []
from app.models import AdCampaign, AdMetric
window_start = (utcnow() - timedelta(days=7)).date()
# Aggregate 7-day spend/conversions per campaign; flag campaigns
# with positive spend and zero total conversions.
rows = (
AdMetric.query.with_entities(
AdMetric.external_campaign_id,
AdMetric.source_service,
func.sum(AdMetric.spend).label("spend"),
func.sum(AdMetric.conversions).label("conversions"),
)
.filter(
AdMetric.company_id == company_id,
AdMetric.metric_date >= window_start,
AdMetric.external_campaign_id.isnot(None),
)
.group_by(AdMetric.external_campaign_id, AdMetric.source_service)
.all()
)
candidates: List[LeakCandidate] = []
for external_campaign_id, source_service, spend, conversions in rows:
spend = round(float(spend or 0.0), 2)
conversions = float(conversions or 0.0)
if spend <= 0 or conversions > 0:
continue
# Only report for connected ad platforms
if source_service in AD_SERVICES and not connector_available(
company_id, source_service
):
continue
campaign = AdCampaign.query.filter_by(
company_id=company_id, external_id=external_campaign_id
).first()
campaign_name = (campaign.name if campaign else "") or external_campaign_id
platform = source_service or (campaign.source_service if campaign else "")
candidates.append(
LeakCandidate(
detector_id=self.id,
source="Ads: Zero-Conversion Campaign",
description=(
f"Campaign '{campaign_name}' spent ${spend:,.2f} with "
f"0 conversions in the last 7 days"
),
estimated_loss=spend,
severity=_spend_severity(spend),
metadata_json={
"dedupe_key": f"{self.id}:{external_campaign_id}",
"campaign_id": campaign.id if campaign else external_campaign_id,
"campaign_name": campaign_name,
"days": 7,
"platform": platform,
},
source_connector=platform if platform in AD_SERVICES else None,
rule_params={
"window_days": 7,
"high_spend": 500,
"critical_spend": 2000,
},
)
)
return candidates