#!/usr/bin/env python3 """ Google Ads Agency Agent - Waste Detection Finds all categories of wasted spend across accounts. """ import sys import argparse from datetime import datetime from typing import List, Dict, Tuple import uuid from core import ( get_client, get_thresholds, ACCOUNTS, run_query, micros_to_currency, format_currency, Issue, Severity, Recommendation, ActionType, add_recommendation, telegram_section ) def find_zombie_keywords(client, account_key: str, days: int = 7) -> Tuple[List[Issue], List[Recommendation]]: """ Find keywords with spend but zero conversions. These are the silent budget killers. """ account = ACCOUNTS[account_key] thresholds = get_thresholds() issues = [] recommendations = [] query = f""" SELECT ad_group_criterion.criterion_id, ad_group_criterion.keyword.text, ad_group_criterion.keyword.match_type, ad_group.name, ad_group.id, campaign.name, campaign.id, metrics.cost_micros, metrics.impressions, metrics.clicks, metrics.conversions FROM keyword_view WHERE segments.date DURING LAST_{days}_DAYS AND ad_group_criterion.status = 'ENABLED' AND campaign.status = 'ENABLED' AND metrics.conversions = 0 AND metrics.cost_micros > 0 ORDER BY metrics.cost_micros DESC LIMIT 50 """ results = run_query(client, account['id'], query) total_waste = 0 for row in results: cost = micros_to_currency(row.metrics.cost_micros) if cost >= thresholds['waste_threshold_usd']: total_waste += cost keyword = row.ad_group_criterion.keyword.text match_type = row.ad_group_criterion.keyword.match_type.name campaign = row.campaign.name clicks = row.metrics.clicks issues.append(Issue( type="ZOMBIE_KEYWORD", severity=Severity.HIGH if cost > 100 else Severity.MEDIUM, account=account['name'], message=f"'{keyword}' [{match_type}] spent {format_currency(cost)} with 0 conversions ({days}d)", details={ "keyword": keyword, "match_type": match_type, "campaign": campaign, "ad_group": row.ad_group.name, "cost": cost, "clicks": clicks, "ctr": f"{(clicks / row.metrics.impressions * 100):.2f}%" if row.metrics.impressions > 0 else "0%", } )) # Generate recommendation to pause recommendations.append(Recommendation( id=f"pause-kw-{uuid.uuid4().hex[:8]}", action=ActionType.PAUSE_KEYWORD, account=account['name'], target=f"{campaign} > {row.ad_group.name} > {keyword}", target_id=str(row.ad_group_criterion.criterion_id), reason=f"Zero conversions with {format_currency(cost)} spend over {days} days", expected_impact=f"Save ~{format_currency(cost / days * 7)}/week", confidence=0.9 if cost > 100 else 0.7, details={ "campaign_id": str(row.campaign.id), "ad_group_id": str(row.ad_group.id), "keyword_id": str(row.ad_group_criterion.criterion_id), "keyword_text": keyword, "current_cost": cost, } )) return issues, recommendations def find_search_term_waste(client, account_key: str, days: int = 7) -> Tuple[List[Issue], List[Recommendation]]: """ Find irrelevant search terms eating budget. """ account = ACCOUNTS[account_key] thresholds = get_thresholds() issues = [] recommendations = [] query = f""" SELECT search_term_view.search_term, campaign.name, campaign.id, ad_group.name, ad_group.id, metrics.cost_micros, metrics.clicks, metrics.conversions FROM search_term_view WHERE segments.date DURING LAST_{days}_DAYS AND metrics.conversions = 0 AND metrics.cost_micros > {int(thresholds['waste_threshold_usd'] * 1_000_000 * 0.5)} ORDER BY metrics.cost_micros DESC LIMIT 50 """ results = run_query(client, account['id'], query) # Keywords that often indicate low-intent traffic low_intent_patterns = ['free', 'cheap', 'discount', 'diy', 'how to', 'reddit', 'review'] for row in results: cost = micros_to_currency(row.metrics.cost_micros) search_term = row.search_term_view.search_term.lower() # Check for low-intent patterns is_low_intent = any(pattern in search_term for pattern in low_intent_patterns) if cost >= thresholds['waste_threshold_usd'] * 0.5: severity = Severity.HIGH if is_low_intent else Severity.MEDIUM issues.append(Issue( type="SEARCH_TERM_WASTE", severity=severity, account=account['name'], message=f"Search term '{row.search_term_view.search_term}' spent {format_currency(cost)}, 0 conv", details={ "search_term": row.search_term_view.search_term, "campaign": row.campaign.name, "cost": cost, "clicks": row.metrics.clicks, "is_low_intent": is_low_intent, } )) # Recommend adding as negative keyword recommendations.append(Recommendation( id=f"neg-kw-{uuid.uuid4().hex[:8]}", action=ActionType.ADD_NEGATIVE, account=account['name'], target=f"{row.campaign.name}", target_id=str(row.campaign.id), reason=f"Search term with {format_currency(cost)} spend, 0 conversions" + (" (low-intent pattern detected)" if is_low_intent else ""), expected_impact=f"Save ~{format_currency(cost / days * 7)}/week", confidence=0.85 if is_low_intent else 0.65, details={ "campaign_id": str(row.campaign.id), "ad_group_id": str(row.ad_group.id), "search_term": row.search_term_view.search_term, "negative_type": "EXACT", } )) return issues, recommendations def find_time_waste(client, account_key: str, days: int = 30) -> Tuple[List[Issue], List[Recommendation]]: """ Find hours of day or days of week with poor performance. """ account = ACCOUNTS[account_key] issues = [] recommendations = [] # Hour of day analysis query = f""" SELECT segments.hour, metrics.cost_micros, metrics.conversions, metrics.clicks FROM campaign WHERE segments.date DURING LAST_{days}_DAYS AND campaign.status = 'ENABLED' AND metrics.cost_micros > 0 """ results = run_query(client, account['id'], query) # Aggregate by hour hour_data = {} for row in results: hour = row.segments.hour if hour not in hour_data: hour_data[hour] = {'cost': 0, 'conversions': 0, 'clicks': 0} hour_data[hour]['cost'] += micros_to_currency(row.metrics.cost_micros) hour_data[hour]['conversions'] += row.metrics.conversions hour_data[hour]['clicks'] += row.metrics.clicks # Find hours with high spend but low/no conversions total_cost = sum(h['cost'] for h in hour_data.values()) total_conv = sum(h['conversions'] for h in hour_data.values()) avg_cpa = total_cost / total_conv if total_conv > 0 else 0 bad_hours = [] for hour, data in hour_data.items(): if data['cost'] > total_cost * 0.03: # At least 3% of budget hour_cpa = data['cost'] / data['conversions'] if data['conversions'] > 0 else float('inf') if data['conversions'] == 0 or hour_cpa > avg_cpa * 2.5: bad_hours.append({ 'hour': hour, 'cost': data['cost'], 'conversions': data['conversions'], 'cpa': hour_cpa }) if bad_hours: bad_hours.sort(key=lambda x: x['cost'], reverse=True) total_bad_cost = sum(h['cost'] for h in bad_hours[:5]) hours_str = ', '.join(f"{h['hour']}:00" for h in bad_hours[:5]) issues.append(Issue( type="TIME_WASTE", severity=Severity.MEDIUM, account=account['name'], message=f"Poor performing hours detected: {hours_str} ({format_currency(total_bad_cost)} combined spend)", details={ "bad_hours": bad_hours[:5], "analysis_period": f"{days} days", "avg_cpa": avg_cpa, } )) return issues, recommendations def find_device_waste(client, account_key: str, days: int = 30) -> Tuple[List[Issue], List[Recommendation]]: """ Find device types with poor performance. """ account = ACCOUNTS[account_key] issues = [] recommendations = [] query = f""" SELECT segments.device, metrics.cost_micros, metrics.conversions, metrics.clicks, metrics.impressions FROM campaign WHERE segments.date DURING LAST_{days}_DAYS AND campaign.status = 'ENABLED' """ results = run_query(client, account['id'], query) # Aggregate by device device_data = {} for row in results: device = row.segments.device.name if device not in device_data: device_data[device] = {'cost': 0, 'conversions': 0, 'clicks': 0} device_data[device]['cost'] += micros_to_currency(row.metrics.cost_micros) device_data[device]['conversions'] += row.metrics.conversions device_data[device]['clicks'] += row.metrics.clicks # Calculate CPA per device total_conv = sum(d['conversions'] for d in device_data.values()) total_cost = sum(d['cost'] for d in device_data.values()) avg_cpa = total_cost / total_conv if total_conv > 0 else 0 for device, data in device_data.items(): if data['cost'] > 0 and data['conversions'] > 0: device_cpa = data['cost'] / data['conversions'] if device_cpa > avg_cpa * 2 and data['cost'] > total_cost * 0.1: issues.append(Issue( type="DEVICE_WASTE", severity=Severity.MEDIUM, account=account['name'], message=f"{device} CPA ({format_currency(device_cpa)}) is {device_cpa/avg_cpa:.1f}x higher than average ({format_currency(avg_cpa)})", details={ "device": device, "cost": data['cost'], "conversions": data['conversions'], "cpa": device_cpa, "vs_avg": f"{(device_cpa/avg_cpa - 1) * 100:.0f}% higher" } )) return issues, recommendations def find_geo_waste(client, account_key: str, days: int = 30) -> Tuple[List[Issue], List[Recommendation]]: """ Find geographic locations with poor performance. """ account = ACCOUNTS[account_key] issues = [] recommendations = [] query = f""" SELECT geographic_view.country_criterion_id, geographic_view.location_type, metrics.cost_micros, metrics.conversions, metrics.clicks FROM geographic_view WHERE segments.date DURING LAST_{days}_DAYS AND metrics.cost_micros > 0 ORDER BY metrics.cost_micros DESC LIMIT 20 """ results = run_query(client, account['id'], query) total_cost = sum(micros_to_currency(r.metrics.cost_micros) for r in results) total_conv = sum(r.metrics.conversions for r in results) avg_cpa = total_cost / total_conv if total_conv > 0 else 0 for row in results: cost = micros_to_currency(row.metrics.cost_micros) conv = row.metrics.conversions if cost > total_cost * 0.05 and conv == 0: # >5% of budget, no conversions issues.append(Issue( type="GEO_WASTE", severity=Severity.MEDIUM, account=account['name'], message=f"Location ID {row.geographic_view.country_criterion_id}: {format_currency(cost)} spent, 0 conversions", details={ "location_id": row.geographic_view.country_criterion_id, "cost": cost, "clicks": row.metrics.clicks, } )) return issues, recommendations def run_full_waste_analysis(days: int = 7, account_filter: str = None) -> Dict: """ Run all waste detection across all accounts. """ client = get_client() all_issues = [] all_recommendations = [] accounts_to_check = ACCOUNTS.keys() if account_filter: accounts_to_check = [k for k in accounts_to_check if account_filter.lower() in k.lower()] for account_key in accounts_to_check: print(f"Analyzing {ACCOUNTS[account_key]['name']}...", file=sys.stderr) # Run all waste finders finders = [ ("Zombie Keywords", find_zombie_keywords), ("Search Term Waste", find_search_term_waste), ("Time Waste", find_time_waste), ("Device Waste", find_device_waste), ("Geo Waste", find_geo_waste), ] for name, finder in finders: try: issues, recs = finder(client, account_key, days) all_issues.extend(issues) all_recommendations.extend(recs) except Exception as e: print(f" {name} error: {e}", file=sys.stderr) return { "issues": all_issues, "recommendations": all_recommendations, "analysis_date": datetime.now().isoformat(), "period_days": days, } def format_waste_report(analysis: Dict) -> str: """Format waste analysis for Telegram.""" issues = analysis['issues'] recommendations = analysis['recommendations'] if not issues: return "✅ **No Significant Waste Detected**\n\nAll accounts look healthy!" # Group by severity critical = [i for i in issues if i.severity == Severity.CRITICAL] high = [i for i in issues if i.severity == Severity.HIGH] medium = [i for i in issues if i.severity == Severity.MEDIUM] # Calculate total waste total_waste = sum(i.details.get('cost', 0) for i in issues) report = f"🔍 **Waste Detection Report**\n" report += f"_Period: Last {analysis['period_days']} days_\n\n" report += f"💸 **Total Waste Identified: {format_currency(total_waste)}**\n\n" if critical or high: report += "🔴 **High Priority**\n" for issue in (critical + high)[:5]: report += f"• {issue.message}\n" report += "\n" if medium: report += "⚠️ **Medium Priority**\n" for issue in medium[:5]: report += f"• {issue.message}\n" report += "\n" # Recommendations summary if recommendations: report += f"📋 **{len(recommendations)} Optimization Recommendations**\n" report += "Use 'Show pending recommendations' to review and approve.\n" # Save recommendations for rec in recommendations: add_recommendation(rec) return report def main(): parser = argparse.ArgumentParser(description="Find wasted Google Ads spend") parser.add_argument("--days", type=int, default=7, help="Analysis period in days") parser.add_argument("--account", type=str, help="Filter to specific account") parser.add_argument("--json", action="store_true", help="Output as JSON") args = parser.parse_args() analysis = run_full_waste_analysis(args.days, args.account) if args.json: import json output = { "issues": [i.to_dict() for i in analysis['issues']], "recommendations": [r.to_dict() for r in analysis['recommendations']], "analysis_date": analysis['analysis_date'], "period_days": analysis['period_days'], } print(json.dumps(output, indent=2)) else: print(format_waste_report(analysis)) if __name__ == "__main__": main()