scripts/campaign_roi_calculator.py
18.2 KB| 1 | #!/usr/bin/env python3 |
| 2 | """ |
| 3 | Campaign ROI Calculator - Comprehensive campaign ROI and performance metrics. |
| 4 | |
| 5 | Calculates: |
| 6 | - ROI (Return on Investment) |
| 7 | - ROAS (Return on Ad Spend) |
| 8 | - CPA (Cost per Acquisition/Customer) |
| 9 | - CPL (Cost per Lead) |
| 10 | - CAC (Customer Acquisition Cost) |
| 11 | - CTR (Click-Through Rate) |
| 12 | - CVR (Conversion Rate - Leads to Customers) |
| 13 | |
| 14 | Includes industry benchmarking and underperformance flagging. |
| 15 | |
| 16 | Usage: |
| 17 | python campaign_roi_calculator.py campaign_data.json |
| 18 | python campaign_roi_calculator.py campaign_data.json --format json |
| 19 | """ |
| 20 | |
| 21 | import argparse |
| 22 | import json |
| 23 | import sys |
| 24 | from typing import Any, Dict, List, Optional |
| 25 | |
| 26 | |
| 27 | # Industry benchmark ranges by channel |
| 28 | # Format: {metric: {channel: (low, target, high)}} |
| 29 | BENCHMARKS: Dict[str, Dict[str, tuple]] = { |
| 30 | "ctr": { |
| 31 | "email": (1.0, 2.5, 5.0), |
| 32 | "paid_search": (1.5, 3.5, 7.0), |
| 33 | "paid_social": (0.5, 1.2, 3.0), |
| 34 | "display": (0.05, 0.1, 0.5), |
| 35 | "organic_search": (1.5, 3.0, 8.0), |
| 36 | "organic_social": (0.5, 1.5, 4.0), |
| 37 | "referral": (1.0, 3.0, 6.0), |
| 38 | "direct": (2.0, 4.0, 8.0), |
| 39 | "default": (0.5, 2.0, 5.0), |
| 40 | }, |
| 41 | "roas": { |
| 42 | "email": (30.0, 42.0, 60.0), |
| 43 | "paid_search": (2.0, 4.0, 8.0), |
| 44 | "paid_social": (1.5, 3.0, 6.0), |
| 45 | "display": (0.5, 1.5, 3.0), |
| 46 | "organic_search": (5.0, 10.0, 20.0), |
| 47 | "organic_social": (3.0, 6.0, 12.0), |
| 48 | "referral": (3.0, 5.0, 10.0), |
| 49 | "direct": (4.0, 8.0, 15.0), |
| 50 | "default": (2.0, 4.0, 8.0), |
| 51 | }, |
| 52 | "cpa": { |
| 53 | "email": (5.0, 15.0, 40.0), |
| 54 | "paid_search": (20.0, 50.0, 150.0), |
| 55 | "paid_social": (15.0, 40.0, 100.0), |
| 56 | "display": (30.0, 75.0, 200.0), |
| 57 | "organic_search": (5.0, 20.0, 60.0), |
| 58 | "organic_social": (10.0, 30.0, 80.0), |
| 59 | "referral": (10.0, 25.0, 70.0), |
| 60 | "direct": (5.0, 15.0, 50.0), |
| 61 | "default": (15.0, 45.0, 120.0), |
| 62 | }, |
| 63 | } |
| 64 | |
| 65 | |
| 66 | def safe_divide(numerator: float, denominator: float, default: float = 0.0) -> float: |
| 67 | """Safely divide two numbers, returning default if denominator is zero.""" |
| 68 | if denominator == 0: |
| 69 | return default |
| 70 | return numerator / denominator |
| 71 | |
| 72 | |
| 73 | def get_benchmark(metric: str, channel: str) -> tuple: |
| 74 | """Get benchmark range for a metric and channel. |
| 75 | |
| 76 | Returns: |
| 77 | Tuple of (low, target, high) for the given metric and channel. |
| 78 | """ |
| 79 | metric_benchmarks = BENCHMARKS.get(metric, {}) |
| 80 | return metric_benchmarks.get(channel, metric_benchmarks.get("default", (0, 0, 0))) |
| 81 | |
| 82 | |
| 83 | def assess_performance(value: float, benchmark: tuple, higher_is_better: bool = True) -> str: |
| 84 | """Assess a metric value against its benchmark range. |
| 85 | |
| 86 | Args: |
| 87 | value: The metric value to assess. |
| 88 | benchmark: Tuple of (low, target, high). |
| 89 | higher_is_better: Whether higher values are better (True for CTR, ROAS; False for CPA). |
| 90 | |
| 91 | Returns: |
| 92 | Performance assessment string. |
| 93 | """ |
| 94 | low, target, high = benchmark |
| 95 | |
| 96 | if higher_is_better: |
| 97 | if value >= high: |
| 98 | return "excellent" |
| 99 | elif value >= target: |
| 100 | return "good" |
| 101 | elif value >= low: |
| 102 | return "below_target" |
| 103 | else: |
| 104 | return "underperforming" |
| 105 | else: |
| 106 | # For cost metrics, lower is better |
| 107 | if value <= low: |
| 108 | return "excellent" |
| 109 | elif value <= target: |
| 110 | return "good" |
| 111 | elif value <= high: |
| 112 | return "below_target" |
| 113 | else: |
| 114 | return "underperforming" |
| 115 | |
| 116 | |
| 117 | def calculate_campaign_metrics(campaign: Dict[str, Any]) -> Dict[str, Any]: |
| 118 | """Calculate all ROI metrics for a single campaign. |
| 119 | |
| 120 | Args: |
| 121 | campaign: Dict with keys: name, channel, spend, revenue, impressions, clicks, leads, customers. |
| 122 | |
| 123 | Returns: |
| 124 | Dict with all calculated metrics, benchmarks, and assessments. |
| 125 | """ |
| 126 | name = campaign.get("name", "Unnamed Campaign") |
| 127 | channel = campaign.get("channel", "default") |
| 128 | spend = campaign.get("spend", 0.0) |
| 129 | revenue = campaign.get("revenue", 0.0) |
| 130 | impressions = campaign.get("impressions", 0) |
| 131 | clicks = campaign.get("clicks", 0) |
| 132 | leads = campaign.get("leads", 0) |
| 133 | customers = campaign.get("customers", 0) |
| 134 | |
| 135 | # Core metrics |
| 136 | roi = safe_divide(revenue - spend, spend) * 100 |
| 137 | roas = safe_divide(revenue, spend) |
| 138 | cpa = safe_divide(spend, customers) if customers > 0 else None |
| 139 | cpl = safe_divide(spend, leads) if leads > 0 else None |
| 140 | cac = safe_divide(spend, customers) if customers > 0 else None |
| 141 | ctr = safe_divide(clicks, impressions) * 100 if impressions > 0 else None |
| 142 | cvr = safe_divide(customers, leads) * 100 if leads > 0 else None |
| 143 | cpc = safe_divide(spend, clicks) if clicks > 0 else None |
| 144 | cpm = safe_divide(spend, impressions) * 1000 if impressions > 0 else None |
| 145 | lead_conversion_rate = safe_divide(leads, clicks) * 100 if clicks > 0 else None |
| 146 | |
| 147 | # Profit |
| 148 | profit = revenue - spend |
| 149 | |
| 150 | # Benchmark assessments |
| 151 | assessments: Dict[str, Any] = {} |
| 152 | flags: List[str] = [] |
| 153 | |
| 154 | if ctr is not None: |
| 155 | benchmark = get_benchmark("ctr", channel) |
| 156 | assessment = assess_performance(ctr, benchmark, higher_is_better=True) |
| 157 | assessments["ctr"] = { |
| 158 | "value": round(ctr, 2), |
| 159 | "benchmark_range": {"low": benchmark[0], "target": benchmark[1], "high": benchmark[2]}, |
| 160 | "assessment": assessment, |
| 161 | } |
| 162 | if assessment == "underperforming": |
| 163 | flags.append(f"CTR ({ctr:.2f}%) is below industry low ({benchmark[0]}%) for {channel}") |
| 164 | |
| 165 | if roas > 0: |
| 166 | benchmark = get_benchmark("roas", channel) |
| 167 | assessment = assess_performance(roas, benchmark, higher_is_better=True) |
| 168 | assessments["roas"] = { |
| 169 | "value": round(roas, 2), |
| 170 | "benchmark_range": {"low": benchmark[0], "target": benchmark[1], "high": benchmark[2]}, |
| 171 | "assessment": assessment, |
| 172 | } |
| 173 | if assessment == "underperforming": |
| 174 | flags.append(f"ROAS ({roas:.2f}x) is below industry low ({benchmark[0]}x) for {channel}") |
| 175 | |
| 176 | if cpa is not None: |
| 177 | benchmark = get_benchmark("cpa", channel) |
| 178 | assessment = assess_performance(cpa, benchmark, higher_is_better=False) |
| 179 | assessments["cpa"] = { |
| 180 | "value": round(cpa, 2), |
| 181 | "benchmark_range": {"low": benchmark[0], "target": benchmark[1], "high": benchmark[2]}, |
| 182 | "assessment": assessment, |
| 183 | } |
| 184 | if assessment == "underperforming": |
| 185 | flags.append(f"CPA (${cpa:.2f}) exceeds industry high (${benchmark[2]:.2f}) for {channel}") |
| 186 | |
| 187 | if profit < 0: |
| 188 | flags.append(f"Campaign is unprofitable: ${profit:,.2f} net loss") |
| 189 | |
| 190 | # Recommendations |
| 191 | recommendations: List[str] = [] |
| 192 | if ctr is not None and assessments.get("ctr", {}).get("assessment") in ("below_target", "underperforming"): |
| 193 | recommendations.append("Improve ad creative and targeting to increase CTR") |
| 194 | if assessments.get("roas", {}).get("assessment") in ("below_target", "underperforming"): |
| 195 | recommendations.append("Review targeting and bid strategy to improve ROAS") |
| 196 | if assessments.get("cpa", {}).get("assessment") in ("below_target", "underperforming"): |
| 197 | recommendations.append("Optimize landing pages and conversion flow to reduce CPA") |
| 198 | if cvr is not None and cvr < 10: |
| 199 | recommendations.append("Lead-to-customer conversion is low; review sales process and lead quality") |
| 200 | if lead_conversion_rate is not None and lead_conversion_rate < 2: |
| 201 | recommendations.append("Click-to-lead rate is low; improve landing page relevance and form experience") |
| 202 | if profit > 0 and assessments.get("roas", {}).get("assessment") in ("good", "excellent"): |
| 203 | recommendations.append("Campaign performing well; consider scaling budget") |
| 204 | |
| 205 | return { |
| 206 | "name": name, |
| 207 | "channel": channel, |
| 208 | "metrics": { |
| 209 | "spend": round(spend, 2), |
| 210 | "revenue": round(revenue, 2), |
| 211 | "profit": round(profit, 2), |
| 212 | "roi_pct": round(roi, 2), |
| 213 | "roas": round(roas, 2), |
| 214 | "cpa": round(cpa, 2) if cpa is not None else None, |
| 215 | "cpl": round(cpl, 2) if cpl is not None else None, |
| 216 | "cac": round(cac, 2) if cac is not None else None, |
| 217 | "ctr_pct": round(ctr, 2) if ctr is not None else None, |
| 218 | "cvr_pct": round(cvr, 2) if cvr is not None else None, |
| 219 | "cpc": round(cpc, 2) if cpc is not None else None, |
| 220 | "cpm": round(cpm, 2) if cpm is not None else None, |
| 221 | "lead_conversion_rate_pct": round(lead_conversion_rate, 2) if lead_conversion_rate is not None else None, |
| 222 | "impressions": impressions, |
| 223 | "clicks": clicks, |
| 224 | "leads": leads, |
| 225 | "customers": customers, |
| 226 | }, |
| 227 | "assessments": assessments, |
| 228 | "flags": flags, |
| 229 | "recommendations": recommendations, |
| 230 | } |
| 231 | |
| 232 | |
| 233 | def calculate_portfolio_summary(campaign_results: List[Dict[str, Any]]) -> Dict[str, Any]: |
| 234 | """Calculate aggregate metrics across all campaigns. |
| 235 | |
| 236 | Args: |
| 237 | campaign_results: List of individual campaign result dicts. |
| 238 | |
| 239 | Returns: |
| 240 | Portfolio-level summary with totals and weighted averages. |
| 241 | """ |
| 242 | total_spend = sum(c["metrics"]["spend"] for c in campaign_results) |
| 243 | total_revenue = sum(c["metrics"]["revenue"] for c in campaign_results) |
| 244 | total_impressions = sum(c["metrics"]["impressions"] for c in campaign_results) |
| 245 | total_clicks = sum(c["metrics"]["clicks"] for c in campaign_results) |
| 246 | total_leads = sum(c["metrics"]["leads"] for c in campaign_results) |
| 247 | total_customers = sum(c["metrics"]["customers"] for c in campaign_results) |
| 248 | total_profit = total_revenue - total_spend |
| 249 | |
| 250 | underperforming = [c["name"] for c in campaign_results if c["flags"]] |
| 251 | top_performers = sorted( |
| 252 | campaign_results, |
| 253 | key=lambda c: c["metrics"]["roi_pct"], |
| 254 | reverse=True, |
| 255 | ) |
| 256 | |
| 257 | # Channel breakdown |
| 258 | channel_totals: Dict[str, Dict[str, float]] = {} |
| 259 | for c in campaign_results: |
| 260 | ch = c["channel"] |
| 261 | if ch not in channel_totals: |
| 262 | channel_totals[ch] = {"spend": 0, "revenue": 0, "leads": 0, "customers": 0} |
| 263 | channel_totals[ch]["spend"] += c["metrics"]["spend"] |
| 264 | channel_totals[ch]["revenue"] += c["metrics"]["revenue"] |
| 265 | channel_totals[ch]["leads"] += c["metrics"]["leads"] |
| 266 | channel_totals[ch]["customers"] += c["metrics"]["customers"] |
| 267 | |
| 268 | channel_summary = {} |
| 269 | for ch, totals in channel_totals.items(): |
| 270 | channel_summary[ch] = { |
| 271 | "spend": round(totals["spend"], 2), |
| 272 | "revenue": round(totals["revenue"], 2), |
| 273 | "roi_pct": round(safe_divide(totals["revenue"] - totals["spend"], totals["spend"]) * 100, 2), |
| 274 | "roas": round(safe_divide(totals["revenue"], totals["spend"]), 2), |
| 275 | "leads": int(totals["leads"]), |
| 276 | "customers": int(totals["customers"]), |
| 277 | } |
| 278 | |
| 279 | return { |
| 280 | "total_campaigns": len(campaign_results), |
| 281 | "total_spend": round(total_spend, 2), |
| 282 | "total_revenue": round(total_revenue, 2), |
| 283 | "total_profit": round(total_profit, 2), |
| 284 | "portfolio_roi_pct": round(safe_divide(total_profit, total_spend) * 100, 2), |
| 285 | "portfolio_roas": round(safe_divide(total_revenue, total_spend), 2), |
| 286 | "total_impressions": total_impressions, |
| 287 | "total_clicks": total_clicks, |
| 288 | "total_leads": total_leads, |
| 289 | "total_customers": total_customers, |
| 290 | "blended_ctr_pct": round(safe_divide(total_clicks, total_impressions) * 100, 2), |
| 291 | "blended_cpl": round(safe_divide(total_spend, total_leads), 2) if total_leads > 0 else None, |
| 292 | "blended_cpa": round(safe_divide(total_spend, total_customers), 2) if total_customers > 0 else None, |
| 293 | "underperforming_campaigns": underperforming, |
| 294 | "top_performer": top_performers[0]["name"] if top_performers else None, |
| 295 | "channel_summary": channel_summary, |
| 296 | } |
| 297 | |
| 298 | |
| 299 | def format_text(results: Dict[str, Any]) -> str: |
| 300 | """Format full results as human-readable text.""" |
| 301 | lines: List[str] = [] |
| 302 | lines.append("=" * 70) |
| 303 | lines.append("CAMPAIGN ROI ANALYSIS") |
| 304 | lines.append("=" * 70) |
| 305 | |
| 306 | # Portfolio summary |
| 307 | summary = results["portfolio_summary"] |
| 308 | lines.append("") |
| 309 | lines.append("PORTFOLIO SUMMARY") |
| 310 | lines.append(f" Total Campaigns: {summary['total_campaigns']}") |
| 311 | lines.append(f" Total Spend: ${summary['total_spend']:>12,.2f}") |
| 312 | lines.append(f" Total Revenue: ${summary['total_revenue']:>12,.2f}") |
| 313 | lines.append(f" Total Profit: ${summary['total_profit']:>12,.2f}") |
| 314 | lines.append(f" Portfolio ROI: {summary['portfolio_roi_pct']}%") |
| 315 | lines.append(f" Portfolio ROAS: {summary['portfolio_roas']}x") |
| 316 | lines.append(f" Blended CTR: {summary['blended_ctr_pct']}%") |
| 317 | if summary["blended_cpl"] is not None: |
| 318 | lines.append(f" Blended CPL: ${summary['blended_cpl']:>12,.2f}") |
| 319 | if summary["blended_cpa"] is not None: |
| 320 | lines.append(f" Blended CPA: ${summary['blended_cpa']:>12,.2f}") |
| 321 | |
| 322 | if summary["top_performer"]: |
| 323 | lines.append(f" Top Performer: {summary['top_performer']}") |
| 324 | if summary["underperforming_campaigns"]: |
| 325 | lines.append(f" Flagged: {', '.join(summary['underperforming_campaigns'])}") |
| 326 | |
| 327 | # Channel summary |
| 328 | if summary["channel_summary"]: |
| 329 | lines.append("") |
| 330 | lines.append("-" * 70) |
| 331 | lines.append("CHANNEL SUMMARY") |
| 332 | lines.append(f" {'Channel':<20} {'Spend':>12} {'Revenue':>12} {'ROI':>10} {'ROAS':>8}") |
| 333 | lines.append(f" {'-'*20} {'-'*12} {'-'*12} {'-'*10} {'-'*8}") |
| 334 | for ch, cs in sorted(summary["channel_summary"].items()): |
| 335 | lines.append( |
| 336 | f" {ch:<20} ${cs['spend']:>10,.2f} ${cs['revenue']:>10,.2f} " |
| 337 | f"{cs['roi_pct']:>8.1f}% {cs['roas']:>6.2f}x" |
| 338 | ) |
| 339 | |
| 340 | # Individual campaigns |
| 341 | for campaign in results["campaigns"]: |
| 342 | lines.append("") |
| 343 | lines.append("-" * 70) |
| 344 | lines.append(f"CAMPAIGN: {campaign['name']}") |
| 345 | lines.append(f"Channel: {campaign['channel']}") |
| 346 | lines.append("-" * 70) |
| 347 | |
| 348 | m = campaign["metrics"] |
| 349 | lines.append(f" {'Metric':<25} {'Value':>15}") |
| 350 | lines.append(f" {'-'*25} {'-'*15}") |
| 351 | lines.append(f" {'Spend':<25} ${m['spend']:>13,.2f}") |
| 352 | lines.append(f" {'Revenue':<25} ${m['revenue']:>13,.2f}") |
| 353 | lines.append(f" {'Profit':<25} ${m['profit']:>13,.2f}") |
| 354 | lines.append(f" {'ROI':<25} {m['roi_pct']:>13.2f}%") |
| 355 | lines.append(f" {'ROAS':<25} {m['roas']:>13.2f}x") |
| 356 | |
| 357 | if m["cpa"] is not None: |
| 358 | lines.append(f" {'CPA':<25} ${m['cpa']:>13,.2f}") |
| 359 | if m["cpl"] is not None: |
| 360 | lines.append(f" {'CPL':<25} ${m['cpl']:>13,.2f}") |
| 361 | if m["cac"] is not None: |
| 362 | lines.append(f" {'CAC':<25} ${m['cac']:>13,.2f}") |
| 363 | if m["ctr_pct"] is not None: |
| 364 | lines.append(f" {'CTR':<25} {m['ctr_pct']:>13.2f}%") |
| 365 | if m["cpc"] is not None: |
| 366 | lines.append(f" {'CPC':<25} ${m['cpc']:>13,.2f}") |
| 367 | if m["cpm"] is not None: |
| 368 | lines.append(f" {'CPM':<25} ${m['cpm']:>13,.2f}") |
| 369 | if m["cvr_pct"] is not None: |
| 370 | lines.append(f" {'Lead-to-Customer CVR':<25} {m['cvr_pct']:>13.2f}%") |
| 371 | if m["lead_conversion_rate_pct"] is not None: |
| 372 | lines.append(f" {'Click-to-Lead Rate':<25} {m['lead_conversion_rate_pct']:>13.2f}%") |
| 373 | |
| 374 | # Benchmark assessments |
| 375 | if campaign["assessments"]: |
| 376 | lines.append("") |
| 377 | lines.append(" BENCHMARK ASSESSMENT") |
| 378 | for metric_name, a in campaign["assessments"].items(): |
| 379 | br = a["benchmark_range"] |
| 380 | status = a["assessment"].upper().replace("_", " ") |
| 381 | lines.append( |
| 382 | f" {metric_name.upper()}: {a['value']} " |
| 383 | f"[low={br['low']}, target={br['target']}, high={br['high']}] " |
| 384 | f"-> {status}" |
| 385 | ) |
| 386 | |
| 387 | # Flags |
| 388 | if campaign["flags"]: |
| 389 | lines.append("") |
| 390 | lines.append(" WARNING FLAGS") |
| 391 | for flag in campaign["flags"]: |
| 392 | lines.append(f" ! {flag}") |
| 393 | |
| 394 | # Recommendations |
| 395 | if campaign["recommendations"]: |
| 396 | lines.append("") |
| 397 | lines.append(" RECOMMENDATIONS") |
| 398 | for i, rec in enumerate(campaign["recommendations"], 1): |
| 399 | lines.append(f" {i}. {rec}") |
| 400 | |
| 401 | lines.append("") |
| 402 | return "\n".join(lines) |
| 403 | |
| 404 | |
| 405 | def main() -> None: |
| 406 | """Main entry point for the campaign ROI calculator.""" |
| 407 | parser = argparse.ArgumentParser( |
| 408 | description="Calculate campaign ROI, ROAS, CPA, CPL, CAC with industry benchmarking.", |
| 409 | epilog="Example: python campaign_roi_calculator.py campaigns.json --format json", |
| 410 | ) |
| 411 | parser.add_argument( |
| 412 | "input_file", |
| 413 | help="Path to JSON file containing campaign data", |
| 414 | ) |
| 415 | parser.add_argument( |
| 416 | "--format", |
| 417 | choices=["json", "text"], |
| 418 | default="text", |
| 419 | dest="output_format", |
| 420 | help="Output format (default: text)", |
| 421 | ) |
| 422 | |
| 423 | args = parser.parse_args() |
| 424 | |
| 425 | # Load input data |
| 426 | try: |
| 427 | with open(args.input_file, "r") as f: |
| 428 | data = json.load(f) |
| 429 | except FileNotFoundError: |
| 430 | print(f"Error: File not found: {args.input_file}", file=sys.stderr) |
| 431 | sys.exit(1) |
| 432 | except json.JSONDecodeError as e: |
| 433 | print(f"Error: Invalid JSON in {args.input_file}: {e}", file=sys.stderr) |
| 434 | sys.exit(1) |
| 435 | |
| 436 | campaigns = data.get("campaigns", []) |
| 437 | if not campaigns: |
| 438 | print("Error: No 'campaigns' array found in input data.", file=sys.stderr) |
| 439 | sys.exit(1) |
| 440 | |
| 441 | # Calculate metrics for each campaign |
| 442 | campaign_results = [calculate_campaign_metrics(c) for c in campaigns] |
| 443 | |
| 444 | # Calculate portfolio summary |
| 445 | portfolio_summary = calculate_portfolio_summary(campaign_results) |
| 446 | |
| 447 | results = { |
| 448 | "portfolio_summary": portfolio_summary, |
| 449 | "campaigns": campaign_results, |
| 450 | } |
| 451 | |
| 452 | if args.output_format == "json": |
| 453 | print(json.dumps(results, indent=2)) |
| 454 | else: |
| 455 | print(format_text(results)) |
| 456 | |
| 457 | |
| 458 | if __name__ == "__main__": |
| 459 | main() |
文件内容是平台检查时读出的文本，只供查看。平台不会运行包里的任何脚本或程序。
