Revenue Operations

{
  "coverage": {
    "total_pipeline_value": 1105000,
    "quota": 500000,
    "coverage_ratio": 2.21,
    "rating": "At Risk",
    "target": "3.0x - 4.0x"
  },
  "stage_conversions": [
    {
      "from_stage": "Discovery",
      "to_stage": "Qualification",
      "from_count": 17,
      "to_count": 12,
      "conversion_rate_pct": 70.6
    },
    {
      "from_stage": "Qualification",
      "to_stage": "Proposal",
      "from_count": 12,
      "to_count": 9,
      "conversion_rate_pct": 75.0
    },
    {
      "from_stage": "Proposal",
      "to_stage": "Negotiation",
      "from_count": 9,
      "to_count": 5,
      "conversion_rate_pct": 55.6
    },
    {
      "from_stage": "Negotiation",
      "to_stage": "Closed Won",
      "from_count": 5,
      "to_count": 2,
      "conversion_rate_pct": 40.0
    }
  ],
  "velocity": {
    "num_opportunities": 17,
    "avg_deal_size": 74588.24,
    "win_rate_pct": 11.8,
    "avg_cycle_days": 32.5,
    "velocity_per_day": 4594.2,
    "velocity_per_month": 137826.09
  },
  "aging": {
    "global_aging_threshold_days": 90,
    "stage_thresholds": {
      "Discovery": 90,
      "Qualification": 78,
      "Proposal": 67,
      "Negotiation": 56
    },
    "total_open_deals": 15,
    "healthy_deals": 13,
    "at_risk_deals": 2,
    "aging_deals": [
      {
        "id": "D011",
        "name": "Vertex Solutions",
        "stage": "Proposal",
        "age_days": 95,
        "threshold_days": 67,
        "days_over": 28,
        "value": 110000
      },
      {
        "id": "D014",
        "name": "Horizon Telecom",
        "stage": "Negotiation",
        "age_days": 60,
        "threshold_days": 56,
        "days_over": 4,
        "value": 250000
      }
    ]
  },
  "risk": {
    "overall_risk": "MEDIUM",
    "risk_factors_count": 3,
    "concentration_risks": [],
    "has_concentration_risk": false,
    "stage_distribution": {
      "Discovery": {
        "count": 5,
        "value": 194000,
        "pct_of_pipeline": 17.6
      },
      "Qualification": {
        "count": 3,
        "value": 150000,
        "pct_of_pipeline": 13.6
      },
      "Proposal": {
        "count": 4,
        "value": 333000,
        "pct_of_pipeline": 30.1
      },
      "Negotiation": {
        "count": 3,
        "value": 428000,
        "pct_of_pipeline": 38.7
      }
    },
    "empty_stages": [],
    "coverage_gaps": [
      {
        "quarter": "2025-Q2",
        "pipeline_value": 344000,
        "quarterly_target": 125000.0,
        "coverage_ratio": 2.75,
        "gap": "Below 3x target"
      }
    ]
  }
}

{
  "forecast_periods": [
    {"period": "2024-Q1", "forecast": 420000, "actual": 445000},
    {"period": "2024-Q2", "forecast": 480000, "actual": 460000},
    {"period": "2024-Q3", "forecast": 510000, "actual": 525000},
    {"period": "2024-Q4", "forecast": 550000, "actual": 510000},
    {"period": "2025-Q1", "forecast": 520000, "actual": 540000},
    {"period": "2025-Q2", "forecast": 580000, "actual": 560000}
  ],
  "category_breakdowns": {
    "by_rep": [
      {"category": "Sarah Chen", "forecast": 210000, "actual": 225000},
      {"category": "Marcus Johnson", "forecast": 185000, "actual": 160000},
      {"category": "Priya Patel", "forecast": 125000, "actual": 135000},
      {"category": "Alex Rivera", "forecast": 60000, "actual": 40000}
    ],
    "by_segment": [
      {"category": "Enterprise", "forecast": 320000, "actual": 310000},
      {"category": "Mid-Market", "forecast": 180000, "actual": 175000},
      {"category": "SMB", "forecast": 80000, "actual": 75000}
    ]
  }
}

{
  "revenue": {
    "current_arr": 5000000,
    "prior_arr": 3800000,
    "net_new_arr": 1200000,
    "arpa_monthly": 2500,
    "revenue_growth_pct": 31.6
  },
  "costs": {
    "sales_marketing_spend": 1800000,
    "cac": 18000,
    "gross_margin_pct": 78,
    "total_operating_expense": 6500000,
    "net_burn": 1500000,
    "fcf_margin_pct": 8.4
  },
  "customers": {
    "beginning_arr": 3800000,
    "expansion_arr": 600000,
    "contraction_arr": 100000,
    "churned_arr": 300000,
    "annual_churn_rate_pct": 8
  }
}

{
  "quota": 500000,
  "stages": ["Discovery", "Qualification", "Proposal", "Negotiation", "Closed Won"],
  "average_cycle_days": 45,
  "deals": [
    {
      "id": "D001",
      "name": "Acme Corp",
      "stage": "Proposal",
      "value": 85000,
      "age_days": 32,
      "close_date": "2025-03-15",
      "owner": "rep_1"
    },
    {
      "id": "D002",
      "name": "TechFlow Inc",
      "stage": "Discovery",
      "value": 42000,
      "age_days": 8,
      "close_date": "2025-04-30",
      "owner": "rep_2"
    },
    {
      "id": "D003",
      "name": "GlobalData Systems",
      "stage": "Negotiation",
      "value": 120000,
      "age_days": 55,
      "close_date": "2025-02-28",
      "owner": "rep_1"
    },
    {
      "id": "D004",
      "name": "Pinnacle Software",
      "stage": "Qualification",
      "value": 35000,
      "age_days": 18,
      "close_date": "2025-04-15",
      "owner": "rep_3"
    },
    {
      "id": "D005",
      "name": "Meridian Health",
      "stage": "Proposal",
      "value": 95000,
      "age_days": 40,
      "close_date": "2025-03-20",
      "owner": "rep_2"
    },
    {
      "id": "D006",
      "name": "CloudVault",
      "stage": "Discovery",
      "value": 28000,
      "age_days": 5,
      "close_date": "2025-05-15",
      "owner": "rep_1"
    },
    {
      "id": "D007",
      "name": "Nexus Financial",
      "stage": "Closed Won",
      "value": 72000,
      "age_days": 38,
      "close_date": "2025-01-31",
      "owner": "rep_3"
    },
    {
      "id": "D008",
      "name": "Urban Analytics",
      "stage": "Negotiation",
      "value": 58000,
      "age_days": 42,
      "close_date": "2025-03-05",
      "owner": "rep_2"
    },
    {
      "id": "D009",
      "name": "Redwood Logistics",
      "stage": "Discovery",
      "value": 31000,
      "age_days": 12,
      "close_date": "2025-05-01",
      "owner": "rep_3"
    },
    {
      "id": "D010",
      "name": "Summit Enterprises",
      "stage": "Qualification",
      "value": 48000,
      "age_days": 22,
      "close_date": "2025-04-10",
      "owner": "rep_1"
    },
    {
      "id": "D011",
      "name": "Vertex Solutions",
      "stage": "Proposal",
      "value": 110000,
      "age_days": 95,
      "close_date": "2025-03-01",
      "owner": "rep_2"
    },
    {
      "id": "D012",
      "name": "DataBridge AI",
      "stage": "Discovery",
      "value": 55000,
      "age_days": 3,
      "close_date": "2025-06-15",
      "owner": "rep_1"
    },
    {
      "id": "D013",
      "name": "Atlas Manufacturing",
      "stage": "Qualification",
      "value": 67000,
      "age_days": 28,
      "close_date": "2025-04-20",
      "owner": "rep_3"
    },
    {
      "id": "D014",
      "name": "Horizon Telecom",
      "stage": "Negotiation",
      "value": 250000,
      "age_days": 60,
      "close_date": "2025-03-10",
      "owner": "rep_1"
    },
    {
      "id": "D015",
      "name": "BlueShift Labs",
      "stage": "Proposal",
      "value": 43000,
      "age_days": 35,
      "close_date": "2025-03-25",
      "owner": "rep_3"
    },
    {
      "id": "D016",
      "name": "Crestview Partners",
      "stage": "Discovery",
      "value": 38000,
      "age_days": 15,
      "close_date": "2025-05-20",
      "owner": "rep_2"
    },
    {
      "id": "D017",
      "name": "Ironclad Security",
      "stage": "Closed Won",
      "value": 91000,
      "age_days": 44,
      "close_date": "2025-02-10",
      "owner": "rep_1"
    }
  ]
}

#!/usr/bin/env python3
"""Forecast Accuracy Tracker - Measures forecast accuracy and bias for SaaS revenue teams.

Calculates MAPE (Mean Absolute Percentage Error), detects systematic forecasting
bias, analyzes accuracy trends, and provides category-level breakdowns.

Usage:
    python forecast_accuracy_tracker.py forecast_data.json --format text
    python forecast_accuracy_tracker.py forecast_data.json --format json
"""

import argparse
import json
import sys
from typing import Any


def safe_divide(numerator: float, denominator: float, default: float = 0.0) -> float:
    """Safely divide two numbers, returning default if denominator is zero."""
    if denominator == 0:
        return default
    return numerator / denominator


def calculate_mape(periods: list[dict]) -> float:
    """Calculate Mean Absolute Percentage Error.

    Formula: mean(|actual - forecast| / |actual|) x 100

    Args:
        periods: List of dicts with 'forecast' and 'actual' keys.

    Returns:
        MAPE as a percentage.
    """
    if not periods:
        return 0.0

    errors = []
    for p in periods:
        actual = p["actual"]
        forecast = p["forecast"]
        if actual != 0:
            errors.append(abs(actual - forecast) / abs(actual))

    if not errors:
        return 0.0

    return (sum(errors) / len(errors)) * 100


def calculate_weighted_mape(periods: list[dict]) -> float:
    """Calculate value-weighted MAPE.

    Weights each period's error by its actual value, giving more importance
    to larger periods.

    Args:
        periods: List of dicts with 'forecast' and 'actual' keys.

    Returns:
        Weighted MAPE as a percentage.
    """
    if not periods:
        return 0.0

    total_actual = sum(abs(p["actual"]) for p in periods)
    if total_actual == 0:
        return 0.0

    weighted_errors = 0.0
    for p in periods:
        actual = p["actual"]
        forecast = p["forecast"]
        if actual != 0:
            weight = abs(actual) / total_actual
            weighted_errors += weight * (abs(actual - forecast) / abs(actual))

    return weighted_errors * 100


def get_accuracy_rating(mape: float) -> dict[str, str]:
    """Return accuracy rating based on MAPE threshold.

    Ratings:
        Excellent: <10%
        Good: 10-15%
        Fair: 15-25%
        Poor: >25%
    """
    if mape < 10:
        return {"rating": "Excellent", "description": "Highly predictable, data-driven process"}
    elif mape < 15:
        return {"rating": "Good", "description": "Reliable forecasting with minor variance"}
    elif mape < 25:
        return {"rating": "Fair", "description": "Needs process improvement"}
    else:
        return {"rating": "Poor", "description": "Significant forecasting methodology gaps"}


def analyze_bias(periods: list[dict]) -> dict[str, Any]:
    """Analyze systematic forecasting bias.

    Positive bias = over-forecasting (forecast > actual, i.e., actual fell short)
    Negative bias = under-forecasting (forecast < actual, i.e., actual exceeded)

    Args:
        periods: List of dicts with 'forecast' and 'actual' keys.

    Returns:
        Bias analysis with direction, magnitude, and ratio.
    """
    if not periods:
        return {
            "direction": "None",
            "bias_pct": 0.0,
            "over_forecast_count": 0,
            "under_forecast_count": 0,
            "exact_count": 0,
            "bias_ratio": 0.0,
        }

    over_count = 0
    under_count = 0
    exact_count = 0
    total_bias = 0.0

    for p in periods:
        diff = p["forecast"] - p["actual"]
        total_bias += diff
        if diff > 0:
            over_count += 1
        elif diff < 0:
            under_count += 1
        else:
            exact_count += 1

    avg_bias = total_bias / len(periods)
    total_actual = sum(p["actual"] for p in periods)
    bias_pct = safe_divide(total_bias, total_actual) * 100

    if over_count > under_count:
        direction = "Over-forecasting"
    elif under_count > over_count:
        direction = "Under-forecasting"
    else:
        direction = "Balanced"

    bias_ratio = safe_divide(over_count, over_count + under_count)

    return {
        "direction": direction,
        "avg_bias_amount": round(avg_bias, 2),
        "bias_pct": round(bias_pct, 1),
        "over_forecast_count": over_count,
        "under_forecast_count": under_count,
        "exact_count": exact_count,
        "bias_ratio": round(bias_ratio, 2),
    }


def analyze_trend(periods: list[dict]) -> dict[str, Any]:
    """Analyze period-over-period accuracy trend.

    Determines if forecast accuracy is improving, stable, or declining
    by comparing error rates across consecutive periods.

    Args:
        periods: List of dicts with 'period', 'forecast', and 'actual' keys.

    Returns:
        Trend analysis with direction and period details.
    """
    if len(periods) < 2:
        return {
            "trend": "Insufficient data",
            "period_errors": [],
            "improving_periods": 0,
            "declining_periods": 0,
        }

    period_errors = []
    for p in periods:
        actual = p["actual"]
        forecast = p["forecast"]
        if actual != 0:
            error_pct = abs(actual - forecast) / abs(actual) * 100
        else:
            error_pct = 0.0
        period_errors.append({
            "period": p.get("period", "Unknown"),
            "error_pct": round(error_pct, 1),
            "forecast": forecast,
            "actual": actual,
        })

    improving = 0
    declining = 0
    for i in range(1, len(period_errors)):
        if period_errors[i]["error_pct"] < period_errors[i - 1]["error_pct"]:
            improving += 1
        elif period_errors[i]["error_pct"] > period_errors[i - 1]["error_pct"]:
            declining += 1

    if improving > declining:
        trend = "Improving"
    elif declining > improving:
        trend = "Declining"
    else:
        trend = "Stable"

    # Calculate recent vs historical MAPE
    midpoint = len(periods) // 2
    if midpoint > 0:
        early_mape = calculate_mape(periods[:midpoint])
        recent_mape = calculate_mape(periods[midpoint:])
        mape_change = recent_mape - early_mape
    else:
        early_mape = 0.0
        recent_mape = 0.0
        mape_change = 0.0

    return {
        "trend": trend,
        "period_errors": period_errors,
        "improving_periods": improving,
        "declining_periods": declining,
        "early_mape": round(early_mape, 1),
        "recent_mape": round(recent_mape, 1),
        "mape_change": round(mape_change, 1),
    }


def analyze_categories(category_breakdowns: dict) -> dict[str, Any]:
    """Analyze accuracy by category (rep, product, segment, etc.).

    Args:
        category_breakdowns: Dict of category_name -> list of
            {category, forecast, actual} dicts.

    Returns:
        Category-level MAPE and accuracy analysis.
    """
    results = {}

    for category_name, entries in category_breakdowns.items():
        category_results = []
        for entry in entries:
            actual = entry["actual"]
            forecast = entry["forecast"]
            if actual != 0:
                error_pct = abs(actual - forecast) / abs(actual) * 100
            else:
                error_pct = 0.0

            diff = forecast - actual
            if diff > 0:
                bias = "Over"
            elif diff < 0:
                bias = "Under"
            else:
                bias = "Exact"

            rating = get_accuracy_rating(error_pct)

            category_results.append({
                "category": entry["category"],
                "forecast": forecast,
                "actual": actual,
                "error_pct": round(error_pct, 1),
                "bias": bias,
                "variance": round(diff, 2),
                "rating": rating["rating"],
            })

        # Sort by error percentage (worst first)
        category_results.sort(key=lambda x: x["error_pct"], reverse=True)

        overall_mape = calculate_mape(entries)
        results[category_name] = {
            "entries": category_results,
            "overall_mape": round(overall_mape, 1),
            "overall_rating": get_accuracy_rating(overall_mape)["rating"],
        }

    return results


def generate_recommendations(
    mape: float, bias: dict, trend: dict, categories: dict
) -> list[str]:
    """Generate actionable recommendations based on analysis results.

    Args:
        mape: Overall MAPE percentage.
        bias: Bias analysis results.
        trend: Trend analysis results.
        categories: Category analysis results.

    Returns:
        List of recommendation strings.
    """
    recommendations = []

    # MAPE-based recommendations
    if mape > 25:
        recommendations.append(
            "CRITICAL: MAPE exceeds 25%. Implement structured forecasting methodology "
            "(e.g., weighted pipeline with stage-based probabilities)."
        )
    elif mape > 15:
        recommendations.append(
            "Forecast accuracy needs improvement. Consider implementing deal-level "
            "forecasting with commit/upside/pipeline categories."
        )

    # Bias-based recommendations
    if bias["direction"] == "Over-forecasting" and abs(bias["bias_pct"]) > 10:
        recommendations.append(
            f"Systematic over-forecasting detected ({bias['bias_pct']}% bias). "
            "Review deal qualification criteria and apply more conservative "
            "stage probabilities."
        )
    elif bias["direction"] == "Under-forecasting" and abs(bias["bias_pct"]) > 10:
        recommendations.append(
            f"Systematic under-forecasting detected ({bias['bias_pct']}% bias). "
            "Review upside deals more carefully and improve pipeline visibility."
        )

    # Trend-based recommendations
    if trend["trend"] == "Declining":
        recommendations.append(
            "Forecast accuracy is declining over time. Schedule a forecasting "
            "methodology review and retrain the team on forecasting best practices."
        )
    elif trend["trend"] == "Improving":
        recommendations.append(
            "Forecast accuracy is improving. Continue current methodology and "
            "document best practices for consistency."
        )

    # Category-based recommendations
    for cat_name, cat_data in categories.items():
        worst_entries = [
            e for e in cat_data["entries"] if e["error_pct"] > 25
        ]
        if worst_entries:
            names = ", ".join(e["category"] for e in worst_entries[:3])
            recommendations.append(
                f"High error rates in {cat_name}: {names}. "
                f"Provide targeted coaching on forecasting discipline."
            )

    if not recommendations:
        recommendations.append(
            "Forecasting performance is strong. Maintain current processes "
            "and continue monitoring for drift."
        )

    return recommendations


def track_forecast_accuracy(data: dict) -> dict[str, Any]:
    """Run complete forecast accuracy analysis.

    Args:
        data: Forecast data with periods and optional category breakdowns.

    Returns:
        Complete forecast accuracy analysis results.
    """
    periods = data["forecast_periods"]

    mape = calculate_mape(periods)
    weighted_mape = calculate_weighted_mape(periods)
    rating = get_accuracy_rating(mape)
    bias = analyze_bias(periods)
    trend = analyze_trend(periods)

    categories = {}
    if "category_breakdowns" in data:
        categories = analyze_categories(data["category_breakdowns"])

    recommendations = generate_recommendations(mape, bias, trend, categories)

    return {
        "mape": round(mape, 1),
        "weighted_mape": round(weighted_mape, 1),
        "accuracy_rating": rating,
        "bias": bias,
        "trend": trend,
        "category_breakdowns": categories,
        "recommendations": recommendations,
        "periods_analyzed": len(periods),
    }


def format_currency(value: float) -> str:
    """Format a number as currency."""
    if abs(value) >= 1_000_000:
        return f"${value / 1_000_000:,.1f}M"
    elif abs(value) >= 1_000:
        return f"${value / 1_000:,.1f}K"
    return f"${value:,.0f}"


def format_text_report(results: dict) -> str:
    """Format analysis results as a human-readable text report."""
    lines = []
    lines.append("=" * 70)
    lines.append("FORECAST ACCURACY REPORT")
    lines.append("=" * 70)

    # Overall accuracy
    lines.append("")
    lines.append("OVERALL ACCURACY")
    lines.append("-" * 40)
    lines.append(f"  MAPE:              {results['mape']}%")
    lines.append(f"  Weighted MAPE:     {results['weighted_mape']}%")
    lines.append(f"  Rating:            {results['accuracy_rating']['rating']}")
    lines.append(f"  Assessment:        {results['accuracy_rating']['description']}")
    lines.append(f"  Periods Analyzed:  {results['periods_analyzed']}")

    # Bias analysis
    bias = results["bias"]
    lines.append("")
    lines.append("FORECAST BIAS")
    lines.append("-" * 40)
    lines.append(f"  Direction:         {bias['direction']}")
    lines.append(f"  Bias %:            {bias['bias_pct']}%")
    lines.append(f"  Avg Bias Amount:   {format_currency(bias['avg_bias_amount'])}")
    lines.append(f"  Over-forecast:     {bias['over_forecast_count']} periods")
    lines.append(f"  Under-forecast:    {bias['under_forecast_count']} periods")
    lines.append(f"  Bias Ratio:        {bias['bias_ratio']}")

    # Trend analysis
    trend = results["trend"]
    lines.append("")
    lines.append("ACCURACY TREND")
    lines.append("-" * 40)
    lines.append(f"  Trend:             {trend['trend']}")
    lines.append(f"  Improving:         {trend['improving_periods']} periods")
    lines.append(f"  Declining:         {trend['declining_periods']} periods")
    if trend.get("early_mape") is not None and trend["trend"] != "Insufficient data":
        lines.append(f"  Early MAPE:        {trend['early_mape']}%")
        lines.append(f"  Recent MAPE:       {trend['recent_mape']}%")
        lines.append(f"  MAPE Change:       {trend['mape_change']:+.1f}%")

    if trend.get("period_errors"):
        lines.append("")
        lines.append("  PERIOD DETAIL:")
        for pe in trend["period_errors"]:
            lines.append(
                f"    {pe['period']:12s}  "
                f"Forecast: {format_currency(pe['forecast']):>10s}  "
                f"Actual: {format_currency(pe['actual']):>10s}  "
                f"Error: {pe['error_pct']}%"
            )

    # Category breakdowns
    if results["category_breakdowns"]:
        lines.append("")
        lines.append("CATEGORY BREAKDOWN")
        lines.append("-" * 40)
        for cat_name, cat_data in results["category_breakdowns"].items():
            lines.append(
                f"\n  {cat_name.upper()} (Overall MAPE: {cat_data['overall_mape']}% "
                f"- {cat_data['overall_rating']})"
            )
            for entry in cat_data["entries"]:
                lines.append(
                    f"    {entry['category']:20s}  "
                    f"Error: {entry['error_pct']:5.1f}%  "
                    f"Bias: {entry['bias']:5s}  "
                    f"Rating: {entry['rating']}"
                )

    # Recommendations
    lines.append("")
    lines.append("RECOMMENDATIONS")
    lines.append("-" * 40)
    for i, rec in enumerate(results["recommendations"], 1):
        lines.append(f"  {i}. {rec}")

    lines.append("")
    lines.append("=" * 70)
    return "\n".join(lines)


def main() -> None:
    """Main entry point for forecast accuracy tracker CLI."""
    parser = argparse.ArgumentParser(
        description="Track and analyze forecast accuracy for SaaS revenue teams."
    )
    parser.add_argument(
        "input",
        help="Path to JSON file containing forecast data",
    )
    parser.add_argument(
        "--format",
        choices=["json", "text"],
        default="text",
        help="Output format: json or text (default: text)",
    )

    args = parser.parse_args()

    try:
        with open(args.input, "r") as f:
            data = json.load(f)
    except FileNotFoundError:
        print(f"Error: File not found: {args.input}", file=sys.stderr)
        sys.exit(1)
    except json.JSONDecodeError as e:
        print(f"Error: Invalid JSON in {args.input}: {e}", file=sys.stderr)
        sys.exit(1)

    if "forecast_periods" not in data:
        print("Error: Missing required field 'forecast_periods' in input data", file=sys.stderr)
        sys.exit(1)

    results = track_forecast_accuracy(data)

    if args.format == "json":
        print(json.dumps(results, indent=2))
    else:
        print(format_text_report(results))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""GTM Efficiency Calculator - Calculates go-to-market efficiency metrics for SaaS.

Computes Magic Number, LTV:CAC, CAC Payback, Burn Multiple, Rule of 40,
and Net Dollar Retention with industry benchmarking and ratings.

Usage:
    python gtm_efficiency_calculator.py gtm_data.json --format text
    python gtm_efficiency_calculator.py gtm_data.json --format json
"""

import argparse
import json
import sys
from typing import Any


def safe_divide(numerator: float, denominator: float, default: float = 0.0) -> float:
    """Safely divide two numbers, returning default if denominator is zero."""
    if denominator == 0:
        return default
    return numerator / denominator


# --- Benchmark tables ---
# Each benchmark defines green/yellow/red thresholds
# and optional percentile placement guidance

BENCHMARKS = {
    "magic_number": {
        "green": {"min": 0.75, "label": ">0.75 - Efficient GTM spend"},
        "yellow": {"min": 0.50, "max": 0.75, "label": "0.50-0.75 - Acceptable efficiency"},
        "red": {"max": 0.50, "label": "<0.50 - Inefficient GTM spend"},
        "elite": 1.0,
        "description": "Net New ARR / Prior Period S&M Spend",
    },
    "ltv_cac_ratio": {
        "green": {"min": 3.0, "label": ">3:1 - Strong unit economics"},
        "yellow": {"min": 1.0, "max": 3.0, "label": "1:1-3:1 - Marginal unit economics"},
        "red": {"max": 1.0, "label": "<1:1 - Unsustainable unit economics"},
        "elite": 5.0,
        "description": "Customer LTV / Customer Acquisition Cost",
    },
    "cac_payback_months": {
        "green": {"max": 18, "label": "<18 months - Healthy payback"},
        "yellow": {"min": 18, "max": 24, "label": "18-24 months - Acceptable payback"},
        "red": {"min": 24, "label": ">24 months - Capital intensive"},
        "elite": 12,
        "description": "CAC / (ARPA x Gross Margin) in months",
    },
    "burn_multiple": {
        "green": {"max": 2.0, "label": "<2x - Capital efficient growth"},
        "yellow": {"min": 2.0, "max": 4.0, "label": "2-4x - Moderate burn"},
        "red": {"min": 4.0, "label": ">4x - Unsustainable burn"},
        "elite": 1.0,
        "description": "Net Burn / Net New ARR",
    },
    "rule_of_40": {
        "green": {"min": 40, "label": ">40% - Strong balance of growth & profitability"},
        "yellow": {"min": 20, "max": 40, "label": "20-40% - Acceptable balance"},
        "red": {"max": 20, "label": "<20% - Needs improvement"},
        "elite": 60,
        "description": "Revenue Growth % + FCF Margin %",
    },
    "ndr_pct": {
        "green": {"min": 110, "label": ">110% - Strong expansion revenue"},
        "yellow": {"min": 100, "max": 110, "label": "100-110% - Stable base"},
        "red": {"max": 100, "label": "<100% - Net revenue contraction"},
        "elite": 130,
        "description": "(Begin ARR + Expansion - Contraction - Churn) / Begin ARR",
    },
}


def rate_metric(metric_name: str, value: float) -> dict[str, str]:
    """Rate a metric as Green/Yellow/Red based on benchmark thresholds.

    Args:
        metric_name: Key into BENCHMARKS dict.
        value: The metric value to rate.

    Returns:
        Dict with rating color, label, and percentile guidance.
    """
    bench = BENCHMARKS.get(metric_name)
    if not bench:
        return {"rating": "Unknown", "label": "No benchmark available"}

    # For metrics where lower is better (cac_payback, burn_multiple)
    lower_is_better = metric_name in ("cac_payback_months", "burn_multiple")

    if lower_is_better:
        if "max" in bench["green"] and value <= bench["green"]["max"]:
            rating = "Green"
            label = bench["green"]["label"]
        elif "min" in bench.get("yellow", {}) and "max" in bench.get("yellow", {}):
            if bench["yellow"]["min"] <= value <= bench["yellow"]["max"]:
                rating = "Yellow"
                label = bench["yellow"]["label"]
            else:
                rating = "Red"
                label = bench["red"]["label"]
        else:
            rating = "Red"
            label = bench["red"]["label"]
    else:
        if "min" in bench["green"] and value >= bench["green"]["min"]:
            rating = "Green"
            label = bench["green"]["label"]
        elif "min" in bench.get("yellow", {}) and "max" in bench.get("yellow", {}):
            if bench["yellow"]["min"] <= value <= bench["yellow"]["max"]:
                rating = "Yellow"
                label = bench["yellow"]["label"]
            else:
                rating = "Red"
                label = bench["red"]["label"]
        else:
            rating = "Red"
            label = bench["red"]["label"]

    # Percentile placement (simplified)
    elite = bench.get("elite", 0)
    if lower_is_better:
        if elite > 0 and value > 0:
            if value <= elite:
                percentile = "Top 10%"
            elif rating == "Green":
                percentile = "Top 25%"
            elif rating == "Yellow":
                percentile = "Median"
            else:
                percentile = "Below median"
        else:
            percentile = "N/A"
    else:
        if elite > 0:
            if value >= elite:
                percentile = "Top 10%"
            elif rating == "Green":
                percentile = "Top 25%"
            elif rating == "Yellow":
                percentile = "Median"
            else:
                percentile = "Below median"
        else:
            percentile = "N/A"

    return {
        "rating": rating,
        "label": label,
        "percentile": percentile,
    }


def calculate_magic_number(net_new_arr: float, sm_spend: float) -> dict[str, Any]:
    """Calculate Magic Number.

    Formula: Net New ARR / Prior Period S&M Spend
    Target: >0.75

    Args:
        net_new_arr: Net new annual recurring revenue in the period.
        sm_spend: Sales & marketing spend in the prior period.

    Returns:
        Magic number value with rating and benchmark.
    """
    value = safe_divide(net_new_arr, sm_spend)
    benchmark = rate_metric("magic_number", value)

    return {
        "value": round(value, 2),
        "net_new_arr": net_new_arr,
        "sm_spend": sm_spend,
        "formula": "Net New ARR / Prior Period S&M Spend",
        "target": ">0.75",
        **benchmark,
    }


def calculate_ltv_cac(
    arpa_monthly: float,
    gross_margin_pct: float,
    annual_churn_rate_pct: float,
    cac: float,
) -> dict[str, Any]:
    """Calculate LTV:CAC Ratio.

    LTV = ARPA_monthly x 12 x Gross Margin / Annual Churn Rate
    Ratio = LTV / CAC
    Target: >3:1

    Args:
        arpa_monthly: Average revenue per account per month.
        gross_margin_pct: Gross margin as percentage (e.g., 78 for 78%).
        annual_churn_rate_pct: Annual churn rate as percentage (e.g., 8 for 8%).
        cac: Customer acquisition cost.

    Returns:
        LTV:CAC ratio with component values, rating, and benchmark.
    """
    gross_margin = gross_margin_pct / 100
    churn_rate = annual_churn_rate_pct / 100

    arpa_annual = arpa_monthly * 12
    ltv = safe_divide(arpa_annual * gross_margin, churn_rate)
    ratio = safe_divide(ltv, cac)

    benchmark = rate_metric("ltv_cac_ratio", ratio)

    return {
        "ratio": round(ratio, 1),
        "ltv": round(ltv, 2),
        "cac": cac,
        "arpa_monthly": arpa_monthly,
        "arpa_annual": arpa_annual,
        "gross_margin_pct": gross_margin_pct,
        "annual_churn_rate_pct": annual_churn_rate_pct,
        "formula": "LTV (ARPA x Gross Margin / Churn Rate) / CAC",
        "target": ">3:1",
        **benchmark,
    }


def calculate_cac_payback(
    cac: float, arpa_monthly: float, gross_margin_pct: float
) -> dict[str, Any]:
    """Calculate CAC Payback Period.

    Formula: CAC / (ARPA_monthly x Gross Margin) in months
    Target: <18 months

    Args:
        cac: Customer acquisition cost.
        arpa_monthly: Average revenue per account per month.
        gross_margin_pct: Gross margin as percentage.

    Returns:
        CAC payback months with rating and benchmark.
    """
    gross_margin = gross_margin_pct / 100
    monthly_contribution = arpa_monthly * gross_margin
    payback_months = safe_divide(cac, monthly_contribution)

    benchmark = rate_metric("cac_payback_months", payback_months)

    return {
        "months": round(payback_months, 1),
        "cac": cac,
        "arpa_monthly": arpa_monthly,
        "gross_margin_pct": gross_margin_pct,
        "monthly_contribution": round(monthly_contribution, 2),
        "formula": "CAC / (ARPA_monthly x Gross Margin)",
        "target": "<18 months",
        **benchmark,
    }


def calculate_burn_multiple(net_burn: float, net_new_arr: float) -> dict[str, Any]:
    """Calculate Burn Multiple.

    Formula: Net Burn / Net New ARR
    Target: <2x (lower is better)

    Args:
        net_burn: Net cash burn in the period.
        net_new_arr: Net new ARR added in the period.

    Returns:
        Burn multiple with rating and benchmark.
    """
    value = safe_divide(net_burn, net_new_arr)
    benchmark = rate_metric("burn_multiple", value)

    return {
        "value": round(value, 2),
        "net_burn": net_burn,
        "net_new_arr": net_new_arr,
        "formula": "Net Burn / Net New ARR",
        "target": "<2x",
        **benchmark,
    }


def calculate_rule_of_40(
    revenue_growth_pct: float, fcf_margin_pct: float
) -> dict[str, Any]:
    """Calculate Rule of 40.

    Formula: Revenue Growth % + FCF Margin %
    Target: >40%

    Args:
        revenue_growth_pct: Year-over-year revenue growth percentage.
        fcf_margin_pct: Free cash flow margin percentage.

    Returns:
        Rule of 40 score with rating and benchmark.
    """
    value = revenue_growth_pct + fcf_margin_pct
    benchmark = rate_metric("rule_of_40", value)

    return {
        "value": round(value, 1),
        "revenue_growth_pct": revenue_growth_pct,
        "fcf_margin_pct": fcf_margin_pct,
        "formula": "Revenue Growth % + FCF Margin %",
        "target": ">40%",
        **benchmark,
    }


def calculate_ndr(
    beginning_arr: float,
    expansion_arr: float,
    contraction_arr: float,
    churned_arr: float,
) -> dict[str, Any]:
    """Calculate Net Dollar Retention.

    Formula: (Beginning ARR + Expansion - Contraction - Churn) / Beginning ARR
    Target: >110%

    Args:
        beginning_arr: ARR at start of period.
        expansion_arr: Expansion revenue from existing customers.
        contraction_arr: Revenue lost from downgrades.
        churned_arr: Revenue lost from customer churn.

    Returns:
        NDR percentage with rating and benchmark.
    """
    ending_arr = beginning_arr + expansion_arr - contraction_arr - churned_arr
    ndr_pct = safe_divide(ending_arr, beginning_arr) * 100

    benchmark = rate_metric("ndr_pct", ndr_pct)

    return {
        "ndr_pct": round(ndr_pct, 1),
        "beginning_arr": beginning_arr,
        "expansion_arr": expansion_arr,
        "contraction_arr": contraction_arr,
        "churned_arr": churned_arr,
        "ending_arr": round(ending_arr, 2),
        "formula": "(Begin ARR + Expansion - Contraction - Churn) / Begin ARR",
        "target": ">110%",
        **benchmark,
    }


def generate_recommendations(metrics: dict) -> list[str]:
    """Generate strategic recommendations based on GTM efficiency metrics.

    Args:
        metrics: Dict of all calculated metric results.

    Returns:
        List of recommendation strings.
    """
    recs = []

    # Magic Number
    mn = metrics["magic_number"]
    if mn["rating"] == "Red":
        recs.append(
            f"Magic Number is {mn['value']} (target >0.75). GTM spend is inefficient. "
            "Audit channel ROI, optimize sales productivity, and consider reducing "
            "low-performing spend."
        )
    elif mn["rating"] == "Yellow":
        recs.append(
            f"Magic Number is {mn['value']}. GTM efficiency is acceptable but can improve. "
            "Focus on sales enablement and pipeline quality over quantity."
        )

    # LTV:CAC
    lc = metrics["ltv_cac"]
    if lc["rating"] == "Red":
        recs.append(
            f"LTV:CAC ratio is {lc['ratio']}:1 (target >3:1). Unit economics are unsustainable. "
            "Reduce CAC through better targeting, improve retention to increase LTV, "
            "or increase ARPA through pricing optimization."
        )
    elif lc["rating"] == "Yellow":
        recs.append(
            f"LTV:CAC ratio is {lc['ratio']}:1. Unit economics are marginal. "
            "Focus on reducing churn and expanding within existing accounts."
        )

    # CAC Payback
    cp = metrics["cac_payback"]
    if cp["rating"] == "Red":
        recs.append(
            f"CAC payback is {cp['months']} months (target <18). Capital recovery is too slow. "
            "Reduce acquisition costs or increase gross-margin-weighted ARPA."
        )

    # Burn Multiple
    bm = metrics["burn_multiple"]
    if bm["rating"] == "Red":
        recs.append(
            f"Burn multiple is {bm['value']}x (target <2x). Cash consumption relative to "
            "growth is unsustainable. Prioritize operating efficiency and path to profitability."
        )

    # Rule of 40
    r40 = metrics["rule_of_40"]
    if r40["rating"] == "Red":
        recs.append(
            f"Rule of 40 score is {r40['value']}% (target >40%). Balance of growth and "
            "profitability needs improvement. Either accelerate growth or improve margins."
        )

    # NDR
    ndr = metrics["ndr"]
    if ndr["rating"] == "Red":
        recs.append(
            f"NDR is {ndr['ndr_pct']}% (target >110%). Net revenue is contracting from "
            "the existing base. Prioritize churn reduction and expansion playbooks."
        )
    elif ndr["rating"] == "Yellow":
        recs.append(
            f"NDR is {ndr['ndr_pct']}%. Base is stable but not expanding. "
            "Invest in cross-sell/upsell motions and customer success capacity."
        )

    # Positive summary if everything is green
    green_count = sum(
        1 for m in metrics.values()
        if isinstance(m, dict) and m.get("rating") == "Green"
    )
    total_metrics = 6
    if green_count == total_metrics:
        recs.append(
            "All GTM efficiency metrics are in healthy ranges. Maintain current "
            "trajectory and optimize for best-in-class performance."
        )
    elif green_count >= 4:
        recs.append(
            f"{green_count}/{total_metrics} metrics are green. GTM efficiency is generally "
            "healthy. Address the yellow/red areas for continuous improvement."
        )

    return recs


def calculate_all_metrics(data: dict) -> dict[str, Any]:
    """Calculate all GTM efficiency metrics from input data.

    Args:
        data: Input data with revenue, costs, and customers sections.

    Returns:
        Complete GTM efficiency analysis results.
    """
    revenue = data["revenue"]
    costs = data["costs"]
    customers = data["customers"]

    metrics = {
        "magic_number": calculate_magic_number(
            net_new_arr=revenue["net_new_arr"],
            sm_spend=costs["sales_marketing_spend"],
        ),
        "ltv_cac": calculate_ltv_cac(
            arpa_monthly=revenue["arpa_monthly"],
            gross_margin_pct=costs["gross_margin_pct"],
            annual_churn_rate_pct=customers["annual_churn_rate_pct"],
            cac=costs["cac"],
        ),
        "cac_payback": calculate_cac_payback(
            cac=costs["cac"],
            arpa_monthly=revenue["arpa_monthly"],
            gross_margin_pct=costs["gross_margin_pct"],
        ),
        "burn_multiple": calculate_burn_multiple(
            net_burn=costs["net_burn"],
            net_new_arr=revenue["net_new_arr"],
        ),
        "rule_of_40": calculate_rule_of_40(
            revenue_growth_pct=revenue["revenue_growth_pct"],
            fcf_margin_pct=costs["fcf_margin_pct"],
        ),
        "ndr": calculate_ndr(
            beginning_arr=customers["beginning_arr"],
            expansion_arr=customers["expansion_arr"],
            contraction_arr=customers["contraction_arr"],
            churned_arr=customers["churned_arr"],
        ),
    }

    metrics["recommendations"] = generate_recommendations(metrics)

    return metrics


def format_currency(value: float) -> str:
    """Format a number as currency."""
    if abs(value) >= 1_000_000:
        return f"${value / 1_000_000:,.1f}M"
    elif abs(value) >= 1_000:
        return f"${value / 1_000:,.1f}K"
    return f"${value:,.0f}"


def format_text_report(results: dict) -> str:
    """Format analysis results as a human-readable text report."""
    lines = []
    lines.append("=" * 70)
    lines.append("GTM EFFICIENCY REPORT")
    lines.append("=" * 70)

    # Metric summary table
    metrics_order = [
        ("magic_number", "Magic Number", lambda m: f"{m['value']}"),
        ("ltv_cac", "LTV:CAC Ratio", lambda m: f"{m['ratio']}:1"),
        ("cac_payback", "CAC Payback", lambda m: f"{m['months']} months"),
        ("burn_multiple", "Burn Multiple", lambda m: f"{m['value']}x"),
        ("rule_of_40", "Rule of 40", lambda m: f"{m['value']}%"),
        ("ndr", "Net Dollar Retention", lambda m: f"{m['ndr_pct']}%"),
    ]

    lines.append("")
    lines.append("METRICS SUMMARY")
    lines.append("-" * 70)
    lines.append(f"  {'Metric':25s} {'Value':>12s} {'Rating':>8s} {'Target':>15s}")
    lines.append(f"  {'':25s} {'':>12s} {'':>8s} {'':>15s}")

    for key, name, fmt_fn in metrics_order:
        m = results[key]
        lines.append(
            f"  {name:25s} {fmt_fn(m):>12s} {m['rating']:>8s} {m['target']:>15s}"
        )

    # Detailed breakdown
    lines.append("")
    lines.append("DETAILED BREAKDOWN")
    lines.append("-" * 70)

    # Magic Number
    mn = results["magic_number"]
    lines.append("")
    lines.append(f"  MAGIC NUMBER: {mn['value']}")
    lines.append(f"    Net New ARR:         {format_currency(mn['net_new_arr'])}")
    lines.append(f"    S&M Spend:           {format_currency(mn['sm_spend'])}")
    lines.append(f"    Rating:              {mn['rating']} - {mn['label']}")
    lines.append(f"    Percentile:          {mn['percentile']}")

    # LTV:CAC
    lc = results["ltv_cac"]
    lines.append("")
    lines.append(f"  LTV:CAC RATIO: {lc['ratio']}:1")
    lines.append(f"    Customer LTV:        {format_currency(lc['ltv'])}")
    lines.append(f"    CAC:                 {format_currency(lc['cac'])}")
    lines.append(f"    ARPA (Monthly):      {format_currency(lc['arpa_monthly'])}")
    lines.append(f"    Gross Margin:        {lc['gross_margin_pct']}%")
    lines.append(f"    Churn Rate:          {lc['annual_churn_rate_pct']}%")
    lines.append(f"    Rating:              {lc['rating']} - {lc['label']}")
    lines.append(f"    Percentile:          {lc['percentile']}")

    # CAC Payback
    cp = results["cac_payback"]
    lines.append("")
    lines.append(f"  CAC PAYBACK: {cp['months']} months")
    lines.append(f"    CAC:                 {format_currency(cp['cac'])}")
    lines.append(f"    Monthly Contribution:{format_currency(cp['monthly_contribution'])}")
    lines.append(f"    Rating:              {cp['rating']} - {cp['label']}")
    lines.append(f"    Percentile:          {cp['percentile']}")

    # Burn Multiple
    bm = results["burn_multiple"]
    lines.append("")
    lines.append(f"  BURN MULTIPLE: {bm['value']}x")
    lines.append(f"    Net Burn:            {format_currency(bm['net_burn'])}")
    lines.append(f"    Net New ARR:         {format_currency(bm['net_new_arr'])}")
    lines.append(f"    Rating:              {bm['rating']} - {bm['label']}")
    lines.append(f"    Percentile:          {bm['percentile']}")

    # Rule of 40
    r40 = results["rule_of_40"]
    lines.append("")
    lines.append(f"  RULE OF 40: {r40['value']}%")
    lines.append(f"    Revenue Growth:      {r40['revenue_growth_pct']}%")
    lines.append(f"    FCF Margin:          {r40['fcf_margin_pct']}%")
    lines.append(f"    Rating:              {r40['rating']} - {r40['label']}")
    lines.append(f"    Percentile:          {r40['percentile']}")

    # NDR
    ndr = results["ndr"]
    lines.append("")
    lines.append(f"  NET DOLLAR RETENTION: {ndr['ndr_pct']}%")
    lines.append(f"    Beginning ARR:       {format_currency(ndr['beginning_arr'])}")
    lines.append(f"    Expansion:           +{format_currency(ndr['expansion_arr'])}")
    lines.append(f"    Contraction:         -{format_currency(ndr['contraction_arr'])}")
    lines.append(f"    Churn:               -{format_currency(ndr['churned_arr'])}")
    lines.append(f"    Ending ARR:          {format_currency(ndr['ending_arr'])}")
    lines.append(f"    Rating:              {ndr['rating']} - {ndr['label']}")
    lines.append(f"    Percentile:          {ndr['percentile']}")

    # Recommendations
    lines.append("")
    lines.append("RECOMMENDATIONS")
    lines.append("-" * 70)
    for i, rec in enumerate(results["recommendations"], 1):
        lines.append(f"  {i}. {rec}")

    lines.append("")
    lines.append("=" * 70)
    return "\n".join(lines)


def main() -> None:
    """Main entry point for GTM efficiency calculator CLI."""
    parser = argparse.ArgumentParser(
        description="Calculate GTM efficiency metrics for SaaS revenue teams."
    )
    parser.add_argument(
        "input",
        help="Path to JSON file containing GTM data",
    )
    parser.add_argument(
        "--format",
        choices=["json", "text"],
        default="text",
        help="Output format: json or text (default: text)",
    )

    args = parser.parse_args()

    try:
        with open(args.input, "r") as f:
            data = json.load(f)
    except FileNotFoundError:
        print(f"Error: File not found: {args.input}", file=sys.stderr)
        sys.exit(1)
    except json.JSONDecodeError as e:
        print(f"Error: Invalid JSON in {args.input}: {e}", file=sys.stderr)
        sys.exit(1)

    required_sections = ["revenue", "costs", "customers"]
    for section in required_sections:
        if section not in data:
            print(
                f"Error: Missing required section '{section}' in input data",
                file=sys.stderr,
            )
            sys.exit(1)

    results = calculate_all_metrics(data)

    if args.format == "json":
        print(json.dumps(results, indent=2))
    else:
        print(format_text_report(results))


if __name__ == "__main__":
    main()