scripts/funnel_analyzer.py
10.9 KB| 1 | #!/usr/bin/env python3 |
| 2 | """ |
| 3 | Funnel Analyzer - Conversion funnel analysis with bottleneck detection. |
| 4 | |
| 5 | Analyzes marketing/sales funnels to identify: |
| 6 | - Stage-to-stage conversion rates and drop-off percentages |
| 7 | - Biggest bottleneck (largest absolute and relative drops) |
| 8 | - Overall funnel conversion rate |
| 9 | - Segment comparison when multiple segments are provided |
| 10 | |
| 11 | Usage: |
| 12 | python funnel_analyzer.py funnel_data.json |
| 13 | python funnel_analyzer.py funnel_data.json --format json |
| 14 | """ |
| 15 | |
| 16 | import argparse |
| 17 | import json |
| 18 | import sys |
| 19 | from typing import Any, Dict, List, Optional |
| 20 | |
| 21 | |
| 22 | def safe_divide(numerator: float, denominator: float, default: float = 0.0) -> float: |
| 23 | """Safely divide two numbers, returning default if denominator is zero.""" |
| 24 | if denominator == 0: |
| 25 | return default |
| 26 | return numerator / denominator |
| 27 | |
| 28 | |
| 29 | def analyze_funnel(stages: List[str], counts: List[int]) -> Dict[str, Any]: |
| 30 | """Analyze a single funnel and return stage-by-stage metrics. |
| 31 | |
| 32 | Args: |
| 33 | stages: Ordered list of funnel stage names (top to bottom). |
| 34 | counts: Corresponding counts at each stage. |
| 35 | |
| 36 | Returns: |
| 37 | Dictionary with stage metrics, bottleneck info, and overall conversion. |
| 38 | """ |
| 39 | if len(stages) != len(counts): |
| 40 | raise ValueError("Number of stages must match number of counts.") |
| 41 | if not stages: |
| 42 | raise ValueError("Funnel must have at least one stage.") |
| 43 | |
| 44 | stage_metrics: List[Dict[str, Any]] = [] |
| 45 | max_dropoff_abs = 0 |
| 46 | max_dropoff_rel = 0.0 |
| 47 | bottleneck_abs: Optional[str] = None |
| 48 | bottleneck_rel: Optional[str] = None |
| 49 | |
| 50 | for i, (stage, count) in enumerate(zip(stages, counts)): |
| 51 | metric: Dict[str, Any] = { |
| 52 | "stage": stage, |
| 53 | "count": count, |
| 54 | "cumulative_conversion": round(safe_divide(count, counts[0]) * 100, 2), |
| 55 | } |
| 56 | |
| 57 | if i > 0: |
| 58 | prev_count = counts[i - 1] |
| 59 | dropoff = prev_count - count |
| 60 | conversion_rate = safe_divide(count, prev_count) * 100 |
| 61 | dropoff_rate = 100 - conversion_rate |
| 62 | |
| 63 | metric["from_previous"] = stages[i - 1] |
| 64 | metric["conversion_rate"] = round(conversion_rate, 2) |
| 65 | metric["dropoff_count"] = dropoff |
| 66 | metric["dropoff_rate"] = round(dropoff_rate, 2) |
| 67 | |
| 68 | # Track biggest absolute drop-off |
| 69 | if dropoff > max_dropoff_abs: |
| 70 | max_dropoff_abs = dropoff |
| 71 | bottleneck_abs = f"{stages[i-1]} -> {stage}" |
| 72 | |
| 73 | # Track biggest relative drop-off |
| 74 | if dropoff_rate > max_dropoff_rel: |
| 75 | max_dropoff_rel = dropoff_rate |
| 76 | bottleneck_rel = f"{stages[i-1]} -> {stage}" |
| 77 | else: |
| 78 | metric["conversion_rate"] = 100.0 |
| 79 | metric["dropoff_count"] = 0 |
| 80 | metric["dropoff_rate"] = 0.0 |
| 81 | |
| 82 | stage_metrics.append(metric) |
| 83 | |
| 84 | overall_conversion = safe_divide(counts[-1], counts[0]) * 100 |
| 85 | |
| 86 | return { |
| 87 | "stage_metrics": stage_metrics, |
| 88 | "overall_conversion_rate": round(overall_conversion, 2), |
| 89 | "total_entries": counts[0], |
| 90 | "total_conversions": counts[-1], |
| 91 | "total_lost": counts[0] - counts[-1], |
| 92 | "bottleneck_absolute": { |
| 93 | "transition": bottleneck_abs, |
| 94 | "dropoff_count": max_dropoff_abs, |
| 95 | }, |
| 96 | "bottleneck_relative": { |
| 97 | "transition": bottleneck_rel, |
| 98 | "dropoff_rate": round(max_dropoff_rel, 2), |
| 99 | }, |
| 100 | } |
| 101 | |
| 102 | |
| 103 | def compare_segments(segments: Dict[str, Dict[str, Any]], stages: List[str]) -> Dict[str, Any]: |
| 104 | """Compare funnel performance across segments. |
| 105 | |
| 106 | Args: |
| 107 | segments: Dict mapping segment name to {"counts": [...]}. |
| 108 | stages: Shared stage names for all segments. |
| 109 | |
| 110 | Returns: |
| 111 | Comparison data with per-segment analysis and relative rankings. |
| 112 | """ |
| 113 | segment_results: Dict[str, Dict[str, Any]] = {} |
| 114 | |
| 115 | for seg_name, seg_data in segments.items(): |
| 116 | counts = seg_data.get("counts", []) |
| 117 | if len(counts) != len(stages): |
| 118 | raise ValueError( |
| 119 | f"Segment '{seg_name}' has {len(counts)} counts but {len(stages)} stages." |
| 120 | ) |
| 121 | segment_results[seg_name] = analyze_funnel(stages, counts) |
| 122 | |
| 123 | # Rank segments by overall conversion rate |
| 124 | ranked = sorted( |
| 125 | segment_results.items(), |
| 126 | key=lambda x: x[1]["overall_conversion_rate"], |
| 127 | reverse=True, |
| 128 | ) |
| 129 | rankings = [ |
| 130 | { |
| 131 | "rank": i + 1, |
| 132 | "segment": name, |
| 133 | "overall_conversion_rate": result["overall_conversion_rate"], |
| 134 | "total_entries": result["total_entries"], |
| 135 | "total_conversions": result["total_conversions"], |
| 136 | } |
| 137 | for i, (name, result) in enumerate(ranked) |
| 138 | ] |
| 139 | |
| 140 | # Stage-by-stage comparison |
| 141 | stage_comparison: List[Dict[str, Any]] = [] |
| 142 | for i, stage in enumerate(stages): |
| 143 | stage_data: Dict[str, Any] = {"stage": stage} |
| 144 | for seg_name in segments: |
| 145 | metrics = segment_results[seg_name]["stage_metrics"][i] |
| 146 | stage_data[seg_name] = { |
| 147 | "count": metrics["count"], |
| 148 | "conversion_rate": metrics["conversion_rate"], |
| 149 | } |
| 150 | stage_comparison.append(stage_data) |
| 151 | |
| 152 | return { |
| 153 | "segment_results": segment_results, |
| 154 | "rankings": rankings, |
| 155 | "stage_comparison": stage_comparison, |
| 156 | } |
| 157 | |
| 158 | |
| 159 | def format_single_funnel_text(analysis: Dict[str, Any], title: str = "FUNNEL") -> str: |
| 160 | """Format a single funnel analysis as human-readable text.""" |
| 161 | lines: List[str] = [] |
| 162 | lines.append(f" {title}") |
| 163 | lines.append(f" {'='*60}") |
| 164 | lines.append(f" Total Entries: {analysis['total_entries']:,}") |
| 165 | lines.append(f" Total Conversions: {analysis['total_conversions']:,}") |
| 166 | lines.append(f" Total Lost: {analysis['total_lost']:,}") |
| 167 | lines.append(f" Overall Conversion: {analysis['overall_conversion_rate']}%") |
| 168 | lines.append("") |
| 169 | |
| 170 | lines.append(f" {'Stage':<20} {'Count':>10} {'Conv Rate':>12} {'Drop-off':>12} {'Cumulative':>12}") |
| 171 | lines.append(f" {'-'*20} {'-'*10} {'-'*12} {'-'*12} {'-'*12}") |
| 172 | |
| 173 | for m in analysis["stage_metrics"]: |
| 174 | stage = m["stage"] |
| 175 | count = m["count"] |
| 176 | conv = f"{m['conversion_rate']:.1f}%" |
| 177 | drop = f"-{m['dropoff_count']:,} ({m['dropoff_rate']:.1f}%)" if m["dropoff_count"] > 0 else "-" |
| 178 | cumul = f"{m['cumulative_conversion']:.1f}%" |
| 179 | lines.append(f" {stage:<20} {count:>10,} {conv:>12} {drop:>12} {cumul:>12}") |
| 180 | |
| 181 | lines.append("") |
| 182 | bn_abs = analysis["bottleneck_absolute"] |
| 183 | bn_rel = analysis["bottleneck_relative"] |
| 184 | lines.append(f" BOTTLENECK (Absolute): {bn_abs['transition']} (lost {bn_abs['dropoff_count']:,})") |
| 185 | lines.append(f" BOTTLENECK (Relative): {bn_rel['transition']} ({bn_rel['dropoff_rate']}% drop-off)") |
| 186 | |
| 187 | return "\n".join(lines) |
| 188 | |
| 189 | |
| 190 | def format_text(results: Dict[str, Any]) -> str: |
| 191 | """Format full results as human-readable text output.""" |
| 192 | lines: List[str] = [] |
| 193 | lines.append("=" * 70) |
| 194 | lines.append("FUNNEL CONVERSION ANALYSIS") |
| 195 | lines.append("=" * 70) |
| 196 | |
| 197 | if "stage_comparison" in results: |
| 198 | # Multi-segment output |
| 199 | lines.append("") |
| 200 | lines.append("SEGMENT RANKINGS") |
| 201 | lines.append(f" {'Rank':>4} {'Segment':<25} {'Conversion':>12} {'Entries':>10} {'Conversions':>12}") |
| 202 | lines.append(f" {'-'*4} {'-'*25} {'-'*12} {'-'*10} {'-'*12}") |
| 203 | for r in results["rankings"]: |
| 204 | lines.append( |
| 205 | f" {r['rank']:>4} {r['segment']:<25} {r['overall_conversion_rate']:>11.2f}% " |
| 206 | f"{r['total_entries']:>10,} {r['total_conversions']:>12,}" |
| 207 | ) |
| 208 | |
| 209 | lines.append("") |
| 210 | for seg_name, seg_result in results["segment_results"].items(): |
| 211 | lines.append("") |
| 212 | lines.append(format_single_funnel_text(seg_result, title=f"SEGMENT: {seg_name.upper()}")) |
| 213 | |
| 214 | # Stage comparison table |
| 215 | lines.append("") |
| 216 | lines.append("-" * 70) |
| 217 | lines.append("STAGE-BY-STAGE COMPARISON") |
| 218 | lines.append("-" * 70) |
| 219 | seg_names = list(results["segment_results"].keys()) |
| 220 | header = f" {'Stage':<20}" |
| 221 | for sn in seg_names: |
| 222 | header += f" {sn:>20}" |
| 223 | lines.append(header) |
| 224 | lines.append(f" {'-'*20}" + f" {'-'*20}" * len(seg_names)) |
| 225 | |
| 226 | for sc in results["stage_comparison"]: |
| 227 | row = f" {sc['stage']:<20}" |
| 228 | for sn in seg_names: |
| 229 | data = sc[sn] |
| 230 | row += f" {data['count']:>8,} ({data['conversion_rate']:>5.1f}%)" |
| 231 | lines.append(row) |
| 232 | |
| 233 | else: |
| 234 | # Single funnel output |
| 235 | lines.append("") |
| 236 | lines.append(format_single_funnel_text(results)) |
| 237 | |
| 238 | lines.append("") |
| 239 | return "\n".join(lines) |
| 240 | |
| 241 | |
| 242 | def main() -> None: |
| 243 | """Main entry point for the funnel analyzer.""" |
| 244 | parser = argparse.ArgumentParser( |
| 245 | description="Analyze conversion funnels with bottleneck detection and segment comparison.", |
| 246 | epilog="Example: python funnel_analyzer.py funnel_data.json --format json", |
| 247 | ) |
| 248 | parser.add_argument( |
| 249 | "input_file", |
| 250 | help="Path to JSON file containing funnel data", |
| 251 | ) |
| 252 | parser.add_argument( |
| 253 | "--format", |
| 254 | choices=["json", "text"], |
| 255 | default="text", |
| 256 | dest="output_format", |
| 257 | help="Output format (default: text)", |
| 258 | ) |
| 259 | |
| 260 | args = parser.parse_args() |
| 261 | |
| 262 | # Load input data |
| 263 | try: |
| 264 | with open(args.input_file, "r") as f: |
| 265 | data = json.load(f) |
| 266 | except FileNotFoundError: |
| 267 | print(f"Error: File not found: {args.input_file}", file=sys.stderr) |
| 268 | sys.exit(1) |
| 269 | except json.JSONDecodeError as e: |
| 270 | print(f"Error: Invalid JSON in {args.input_file}: {e}", file=sys.stderr) |
| 271 | sys.exit(1) |
| 272 | |
| 273 | # Determine mode: single funnel vs. segment comparison |
| 274 | if "segments" in data: |
| 275 | # Multi-segment mode |
| 276 | stages = data.get("funnel", {}).get("stages", data.get("stages", [])) |
| 277 | if not stages: |
| 278 | print("Error: 'stages' list required for segment comparison.", file=sys.stderr) |
| 279 | sys.exit(1) |
| 280 | segments = data["segments"] |
| 281 | if not segments: |
| 282 | print("Error: 'segments' dict is empty.", file=sys.stderr) |
| 283 | sys.exit(1) |
| 284 | results = compare_segments(segments, stages) |
| 285 | elif "funnel" in data: |
| 286 | # Single funnel mode |
| 287 | funnel = data["funnel"] |
| 288 | stages = funnel.get("stages", []) |
| 289 | counts = funnel.get("counts", []) |
| 290 | if not stages or not counts: |
| 291 | print("Error: 'funnel' must contain 'stages' and 'counts' arrays.", file=sys.stderr) |
| 292 | sys.exit(1) |
| 293 | results = analyze_funnel(stages, counts) |
| 294 | else: |
| 295 | print("Error: Input must contain 'funnel' or 'segments' key.", file=sys.stderr) |
| 296 | sys.exit(1) |
| 297 | |
| 298 | if args.output_format == "json": |
| 299 | print(json.dumps(results, indent=2)) |
| 300 | else: |
| 301 | print(format_text(results)) |
| 302 | |
| 303 | |
| 304 | if __name__ == "__main__": |
| 305 | main() |
文件内容是平台检查时读出的文本，只供查看。平台不会运行包里的任何脚本或程序。
