Saas Metrics Coach

# cohorts.csv — User cohort retention data for SaaS retention analysis
# Fields: cohort_month, users_acquired, month_0, month_1, ..., month_11
# Values represent active users remaining in each month after acquisition
# Use with the saas-metrics-coach retention analysis tools
cohort_month,users_acquired,month_0,month_1,month_2,month_3,month_4,month_5,month_6,month_7,month_8,month_9,month_10,month_11
2025-01,48,48,42,38,35,32,30,28,26,25,24,23,22
2025-02,55,55,47,41,37,34,31,28,26,24,23,22,
2025-03,62,62,54,48,43,39,35,32,30,28,26,,
2025-04,58,58,49,42,37,33,30,27,25,23,,,
2025-05,71,71,60,52,46,41,37,33,31,,,,
2025-06,65,65,55,47,41,36,32,29,,,,,
2025-07,78,78,66,57,50,44,39,,,,,,
2025-08,84,84,71,61,53,46,,,,,,,
2025-09,73,73,61,52,45,,,,,,,,
2025-10,90,90,76,65,,,,,,,,,
2025-11,82,82,69,,,,,,,,,,
2025-12,95,95,,,,,,,,,,,

# subscriptions.csv — 12 months of SaaS subscription data for MRR calculator
# Fields: customer_id, customer_name, plan, amount_cents, start_date, end_date, status
# Includes upgrades, downgrades, churns, and new business for MRR analysis
customer_id,customer_name,plan,amount_cents,start_date,end_date,status
C001,Globex Corporation,enterprise,49900,2025-01-15,,active
C002,Initech LLC,pro,2990,2025-02-01,,active
C003,Hooli Inc,enterprise,49900,2025-01-08,2025-09-30,churned
C004,Pied Piper,starter,990,2025-03-12,,active
C005,Massive Dynamic,pro,2990,2025-01-22,,active
C006,Umbrella Corp,enterprise,49900,2025-04-01,,active
C007,Wayne Enterprises,pro,2990,2025-02-18,2025-08-15,churned
C008,Stark Industries,enterprise,99900,2025-01-05,,active
C009,Cyberdyne Systems,starter,990,2025-05-20,2025-11-30,churned
C010,Soylent Corp,pro,2990,2025-03-01,,active
C011,Tyrell Corporation,enterprise,49900,2025-06-15,,active
C012,Weyland-Yutani,pro,2990,2025-04-22,,active
C013,Oscorp Industries,starter,990,2025-07-01,,active
C014,LexCorp,enterprise,49900,2025-05-10,,active
C015,Wonka Industries,pro,2990,2025-08-01,,active
C016,Acme Corp,starter,990,2025-01-30,2025-06-28,churned
C017,Bluth Company,starter,990,2025-09-14,,active
C018,Sterling Cooper,pro,2990,2025-06-01,,active
C019,Dunder Mifflin,starter,990,2025-10-05,,active
C020,Prestige Worldwide,pro,2990,2025-07-18,,active
C021,Vandelay Industries,enterprise,49900,2025-11-01,,active
C022,Aperture Science,pro,2990,2025-08-22,,active
C023,Black Mesa,starter,990,2025-12-01,,active
C024,Nakatomi Trading,pro,2990,2025-09-10,,active
C025,Rekall Inc,enterprise,49900,2025-10-15,,active
C002,Initech LLC,enterprise,49900,2025-08-01,,active
C004,Pied Piper,pro,2990,2025-09-01,,active
C010,Soylent Corp,enterprise,49900,2025-11-01,,active
C013,Oscorp Industries,pro,2990,2025-12-01,,active
C005,Massive Dynamic,starter,990,2025-10-01,,active
C015,Wonka Industries,starter,990,2025-12-15,,active

#!/usr/bin/env python3
"""
Cohort Retention Analyzer

Performs cohort retention analysis from user signup and activity data.
Groups users by signup period and tracks retention over subsequent periods.

Expected CSV columns: user_id, signup_date, activity_date (or last_active_date)
Each row represents one activity event or the user record with last active date.

Usage:
    python cohort_analyzer.py users.csv
    python cohort_analyzer.py users.csv --cohort-period monthly
    python cohort_analyzer.py users.csv --cohort-period weekly --format json
    python cohort_analyzer.py users.csv --max-periods 12
"""

import argparse
import csv
import json
import sys
from collections import defaultdict
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional, Set, Tuple


def parse_date(date_str: str) -> Optional[datetime]:
    """Parse date string in common formats."""
    if not date_str or date_str.strip() == "":
        return None
    for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%m/%d/%Y", "%d/%m/%Y"):
        try:
            return datetime.strptime(date_str.strip(), fmt)
        except ValueError:
            continue
    return None


def month_key(dt: datetime) -> str:
    """Return YYYY-MM key."""
    return dt.strftime("%Y-%m")


def week_key(dt: datetime) -> str:
    """Return ISO week key YYYY-Www."""
    iso = dt.isocalendar()
    return f"{iso[0]}-W{iso[1]:02d}"


def quarter_key(dt: datetime) -> str:
    """Return YYYY-Qq key."""
    q = (dt.month - 1) // 3 + 1
    return f"{dt.year}-Q{q}"


def get_period_key(dt: datetime, period_type: str) -> str:
    """Get period key based on period type."""
    if period_type == "weekly":
        return week_key(dt)
    elif period_type == "quarterly":
        return quarter_key(dt)
    return month_key(dt)


def load_user_data(filepath: str) -> Tuple[Dict[str, datetime], Dict[str, Set[str]]]:
    """
    Load user data from CSV.
    Returns: (user_signups, user_activity_periods)
    - user_signups: {user_id: signup_date}
    - user_activity_periods: {user_id: set of period_keys where user was active}
    """
    user_signups = {}
    user_activities = defaultdict(list)

    with open(filepath, "r", newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        fields = set(reader.fieldnames or [])

        has_signup = "signup_date" in fields
        activity_col = None
        for col in ["activity_date", "last_active_date", "event_date", "active_date"]:
            if col in fields:
                activity_col = col
                break

        if "user_id" not in fields:
            print("Error: CSV must have 'user_id' column", file=sys.stderr)
            sys.exit(1)

        if not has_signup:
            print("Error: CSV must have 'signup_date' column", file=sys.stderr)
            sys.exit(1)

        for row in reader:
            uid = row["user_id"].strip()
            signup = parse_date(row.get("signup_date", ""))
            if signup and uid not in user_signups:
                user_signups[uid] = signup

            if activity_col:
                act_date = parse_date(row.get(activity_col, ""))
                if act_date:
                    user_activities[uid].append(act_date)
            elif signup:
                # If no activity column, treat signup as the only known activity
                user_activities[uid].append(signup)

    return user_signups, user_activities


def build_cohort_table(
    user_signups: Dict[str, datetime],
    user_activities: Dict[str, List[datetime]],
    period_type: str,
    max_periods: int,
) -> Dict[str, Any]:
    """Build cohort retention table."""
    # Group users into cohorts by signup period
    cohorts = defaultdict(set)
    for uid, signup in user_signups.items():
        cohort = get_period_key(signup, period_type)
        cohorts[cohort].add(uid)

    # Build activity period sets per user
    user_period_sets = {}
    for uid, activities in user_activities.items():
        user_period_sets[uid] = {get_period_key(a, period_type) for a in activities}

    # Get all periods in order
    all_periods = set()
    for uid, pset in user_period_sets.items():
        all_periods.update(pset)
    for cohort in cohorts:
        all_periods.add(cohort)
    sorted_periods = sorted(all_periods)

    period_index = {p: i for i, p in enumerate(sorted_periods)}

    # Build retention matrix
    cohort_names = sorted(cohorts.keys())
    retention_matrix = {}
    cohort_sizes = {}

    for cohort in cohort_names:
        users = cohorts[cohort]
        cohort_sizes[cohort] = len(users)
        cohort_idx = period_index.get(cohort, 0)
        retention = {}

        for offset in range(max_periods + 1):
            target_idx = cohort_idx + offset
            target_period = None
            for p, idx in period_index.items():
                if idx == target_idx:
                    target_period = p
                    break
            if target_period is None:
                break

            active_count = 0
            for uid in users:
                if uid in user_period_sets and target_period in user_period_sets[uid]:
                    active_count += 1

            rate = active_count / len(users) if users else 0
            retention[offset] = {
                "active": active_count,
                "total": len(users),
                "rate": round(rate, 4),
            }

        retention_matrix[cohort] = retention

    # Calculate average retention by period offset
    avg_retention = {}
    for offset in range(max_periods + 1):
        rates = []
        for cohort in cohort_names:
            if offset in retention_matrix.get(cohort, {}):
                rates.append(retention_matrix[cohort][offset]["rate"])
        if rates:
            avg_retention[offset] = round(sum(rates) / len(rates), 4)

    return {
        "period_type": period_type,
        "total_users": len(user_signups),
        "total_cohorts": len(cohort_names),
        "cohort_sizes": {c: cohort_sizes[c] for c in cohort_names},
        "retention_matrix": retention_matrix,
        "average_retention": avg_retention,
    }


def assess_retention(avg_retention: Dict[int, float]) -> List[str]:
    """Generate retention health assessment."""
    findings = []

    if 1 in avg_retention:
        r1 = avg_retention[1]
        if r1 < 0.5:
            findings.append(f"CRITICAL: Period 1 retention is {r1*100:.1f}% - severe activation problem")
        elif r1 < 0.7:
            findings.append(f"WARNING: Period 1 retention is {r1*100:.1f}% - activation needs improvement")
        else:
            findings.append(f"Period 1 retention is {r1*100:.1f}% - healthy activation")

    if 3 in avg_retention:
        r3 = avg_retention[3]
        if r3 < 0.3:
            findings.append(f"CRITICAL: Period 3 retention is {r3*100:.1f}% - weak product-market fit signal")
        elif r3 < 0.5:
            findings.append(f"WARNING: Period 3 retention is {r3*100:.1f}% - moderate product-market fit")
        else:
            findings.append(f"Period 3 retention is {r3*100:.1f}% - strong product-market fit signal")

    # Check for stabilization
    if len(avg_retention) >= 4:
        recent = [avg_retention.get(i, 0) for i in range(max(0, len(avg_retention) - 3), len(avg_retention))]
        if len(recent) >= 2:
            deltas = [abs(recent[i] - recent[i - 1]) for i in range(1, len(recent))]
            avg_delta = sum(deltas) / len(deltas) if deltas else 0
            if avg_delta < 0.02:
                findings.append("Retention curve has stabilized - healthy flattening pattern")
            elif avg_delta > 0.05:
                findings.append("WARNING: Retention continues declining without stabilization")

    # Check if later cohorts improve
    return findings


def print_human(data: Dict[str, Any]) -> None:
    """Print cohort analysis in human-readable format."""
    print("=" * 70)
    print(f"  Cohort Retention Analysis ({data['period_type']})")
    print("=" * 70)
    print(f"\n  Total Users: {data['total_users']}")
    print(f"  Total Cohorts: {data['total_cohorts']}")

    # Retention matrix
    matrix = data["retention_matrix"]
    cohorts = sorted(matrix.keys())

    if not cohorts:
        print("\n  No cohort data available.")
        return

    max_offset = max(max(matrix[c].keys()) for c in cohorts if matrix[c])

    # Header
    print(f"\n  {'Cohort':<12} {'Size':>6}", end="")
    for i in range(min(max_offset + 1, 13)):
        label = f"P{i}"
        print(f" {label:>7}", end="")
    print()
    print(f"  {'-' * 12} {'-' * 6}", end="")
    for i in range(min(max_offset + 1, 13)):
        print(f" {'-' * 7}", end="")
    print()

    for cohort in cohorts:
        size = data["cohort_sizes"][cohort]
        print(f"  {cohort:<12} {size:>6}", end="")
        for offset in range(min(max_offset + 1, 13)):
            if offset in matrix[cohort]:
                rate = matrix[cohort][offset]["rate"]
                print(f" {rate * 100:>6.1f}%", end="")
            else:
                print(f" {'':>7}", end="")
        print()

    # Average retention
    avg = data["average_retention"]
    print(f"\n  {'Average':<12} {'':>6}", end="")
    for offset in range(min(max_offset + 1, 13)):
        if offset in avg:
            print(f" {avg[offset] * 100:>6.1f}%", end="")
        else:
            print(f" {'':>7}", end="")
    print()

    # Assessment
    findings = assess_retention(avg)
    if findings:
        print(f"\n  --- Assessment ---")
        for f in findings:
            print(f"  {f}")
    print()


def main():
    parser = argparse.ArgumentParser(
        description="Cohort retention analysis from user signup and activity data"
    )
    parser.add_argument("file", help="CSV file with user data")
    parser.add_argument("--format", choices=["human", "json"], default="human", help="Output format")
    parser.add_argument(
        "--cohort-period",
        choices=["weekly", "monthly", "quarterly"],
        default="monthly",
        help="Cohort grouping period (default: monthly)",
    )
    parser.add_argument(
        "--max-periods", type=int, default=12, help="Maximum number of periods to track (default: 12)"
    )
    args = parser.parse_args()

    user_signups, user_activities = load_user_data(args.file)
    if not user_signups:
        print("Error: No valid user records found", file=sys.stderr)
        sys.exit(1)

    data = build_cohort_table(user_signups, user_activities, args.cohort_period, args.max_periods)

    if args.format == "json":
        # Convert integer keys to strings for JSON
        json_data = dict(data)
        json_data["average_retention"] = {str(k): v for k, v in data["average_retention"].items()}
        for cohort in json_data["retention_matrix"]:
            json_data["retention_matrix"][cohort] = {
                str(k): v for k, v in json_data["retention_matrix"][cohort].items()
            }
        print(json.dumps(json_data, indent=2))
    else:
        print_human(data)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
MRR Calculator

Calculates MRR, ARR, growth rate, churn rate, net new MRR, and SaaS quick ratio
from subscription data in CSV format.

Expected CSV columns: customer_id, plan, mrr, start_date, end_date, status
- end_date can be empty for active subscriptions
- status: active, churned, or cancelled

Usage:
    python mrr_calculator.py subscriptions.csv
    python mrr_calculator.py subscriptions.csv --format json
    python mrr_calculator.py subscriptions.csv --breakdown
    python mrr_calculator.py subscriptions.csv --period 2025-06
"""

import argparse
import csv
import json
import sys
from collections import defaultdict
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional, Tuple


def parse_date(date_str: str) -> Optional[datetime]:
    """Parse date string in common formats."""
    if not date_str or date_str.strip() == "":
        return None
    for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%m/%d/%Y", "%d/%m/%Y"):
        try:
            return datetime.strptime(date_str.strip(), fmt)
        except ValueError:
            continue
    return None


def month_key(dt: datetime) -> str:
    """Return YYYY-MM string from datetime."""
    return dt.strftime("%Y-%m")


def load_subscriptions(filepath: str) -> List[Dict[str, Any]]:
    """Load subscription data from CSV."""
    subscriptions = []
    with open(filepath, "r", newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        required = {"customer_id", "plan", "mrr", "start_date"}
        if not required.issubset(set(reader.fieldnames or [])):
            missing = required - set(reader.fieldnames or [])
            print(f"Error: Missing required columns: {missing}", file=sys.stderr)
            sys.exit(1)
        for row in reader:
            try:
                sub = {
                    "customer_id": row["customer_id"].strip(),
                    "plan": row.get("plan", "unknown").strip(),
                    "mrr": float(row["mrr"]),
                    "start_date": parse_date(row["start_date"]),
                    "end_date": parse_date(row.get("end_date", "")),
                    "status": row.get("status", "active").strip().lower(),
                }
                if sub["start_date"] is None:
                    continue
                subscriptions.append(sub)
            except (ValueError, KeyError):
                continue
    return subscriptions


def get_active_mrr_at(subscriptions: List[Dict], target: datetime) -> Tuple[float, int]:
    """Calculate total MRR and active customer count at a given date."""
    total_mrr = 0.0
    active_count = 0
    seen = set()
    for sub in subscriptions:
        if sub["start_date"] <= target:
            if sub["end_date"] is None or sub["end_date"] > target:
                if sub["customer_id"] not in seen:
                    total_mrr += sub["mrr"]
                    active_count += 1
                    seen.add(sub["customer_id"])
    return total_mrr, active_count


def calculate_mrr_components(subscriptions: List[Dict], period: str) -> Dict[str, Any]:
    """Calculate MRR components for a given YYYY-MM period."""
    year, month = int(period[:4]), int(period[5:7])
    period_start = datetime(year, month, 1)
    if month == 12:
        period_end = datetime(year + 1, 1, 1) - timedelta(days=1)
    else:
        period_end = datetime(year, month + 1, 1) - timedelta(days=1)

    # Previous period
    if month == 1:
        prev_start = datetime(year - 1, 12, 1)
    else:
        prev_start = datetime(year, month - 1, 1)

    beginning_mrr, beginning_customers = get_active_mrr_at(subscriptions, period_start - timedelta(days=1))
    ending_mrr, ending_customers = get_active_mrr_at(subscriptions, period_end)

    new_mrr = 0.0
    new_count = 0
    churned_mrr = 0.0
    churned_count = 0

    customer_mrr_start = {}
    customer_mrr_end = {}

    for sub in subscriptions:
        cid = sub["customer_id"]
        if sub["start_date"] <= period_start - timedelta(days=1):
            if sub["end_date"] is None or sub["end_date"] > period_start - timedelta(days=1):
                customer_mrr_start[cid] = sub["mrr"]

        if sub["start_date"] <= period_end:
            if sub["end_date"] is None or sub["end_date"] > period_end:
                customer_mrr_end[cid] = sub["mrr"]

    # New customers: in end but not in start
    for cid, mrr in customer_mrr_end.items():
        if cid not in customer_mrr_start:
            new_mrr += mrr
            new_count += 1

    # Churned customers: in start but not in end
    for cid, mrr in customer_mrr_start.items():
        if cid not in customer_mrr_end:
            churned_mrr += mrr
            churned_count += 1

    # Expansion and contraction
    expansion_mrr = 0.0
    contraction_mrr = 0.0
    for cid in customer_mrr_start:
        if cid in customer_mrr_end:
            diff = customer_mrr_end[cid] - customer_mrr_start[cid]
            if diff > 0:
                expansion_mrr += diff
            elif diff < 0:
                contraction_mrr += abs(diff)

    net_new_mrr = new_mrr + expansion_mrr - churned_mrr - contraction_mrr
    gross_churn_rate = (churned_mrr + contraction_mrr) / beginning_mrr if beginning_mrr > 0 else 0.0
    net_churn_rate = (churned_mrr + contraction_mrr - expansion_mrr) / beginning_mrr if beginning_mrr > 0 else 0.0
    nrr = (beginning_mrr + expansion_mrr - churned_mrr - contraction_mrr) / beginning_mrr if beginning_mrr > 0 else 0.0
    growth_rate = (ending_mrr - beginning_mrr) / beginning_mrr if beginning_mrr > 0 else 0.0

    # SaaS Quick Ratio
    inflows = new_mrr + expansion_mrr
    outflows = churned_mrr + contraction_mrr
    quick_ratio = inflows / outflows if outflows > 0 else float("inf")

    return {
        "period": period,
        "beginning_mrr": round(beginning_mrr, 2),
        "ending_mrr": round(ending_mrr, 2),
        "arr": round(ending_mrr * 12, 2),
        "new_mrr": round(new_mrr, 2),
        "expansion_mrr": round(expansion_mrr, 2),
        "contraction_mrr": round(contraction_mrr, 2),
        "churned_mrr": round(churned_mrr, 2),
        "net_new_mrr": round(net_new_mrr, 2),
        "beginning_customers": beginning_customers,
        "ending_customers": ending_customers,
        "new_customers": new_count,
        "churned_customers": churned_count,
        "gross_churn_rate": round(gross_churn_rate, 4),
        "net_churn_rate": round(net_churn_rate, 4),
        "net_revenue_retention": round(nrr, 4),
        "mom_growth_rate": round(growth_rate, 4),
        "saas_quick_ratio": round(quick_ratio, 2) if quick_ratio != float("inf") else "infinite",
    }


def breakdown_by_plan(subscriptions: List[Dict], period: str) -> Dict[str, Dict]:
    """Break down MRR by plan tier."""
    year, month = int(period[:4]), int(period[5:7])
    if month == 12:
        period_end = datetime(year + 1, 1, 1) - timedelta(days=1)
    else:
        period_end = datetime(year, month + 1, 1) - timedelta(days=1)

    plan_data = defaultdict(lambda: {"mrr": 0.0, "customers": 0})
    for sub in subscriptions:
        if sub["start_date"] <= period_end:
            if sub["end_date"] is None or sub["end_date"] > period_end:
                plan = sub["plan"]
                plan_data[plan]["mrr"] += sub["mrr"]
                plan_data[plan]["customers"] += 1

    total_mrr = sum(p["mrr"] for p in plan_data.values())
    result = {}
    for plan, data in sorted(plan_data.items()):
        result[plan] = {
            "mrr": round(data["mrr"], 2),
            "customers": data["customers"],
            "arpu": round(data["mrr"] / data["customers"], 2) if data["customers"] > 0 else 0,
            "mix_pct": round(data["mrr"] / total_mrr * 100, 1) if total_mrr > 0 else 0,
        }
    return result


def detect_periods(subscriptions: List[Dict]) -> List[str]:
    """Detect all months with subscription activity."""
    months = set()
    for sub in subscriptions:
        if sub["start_date"]:
            months.add(month_key(sub["start_date"]))
        if sub["end_date"]:
            months.add(month_key(sub["end_date"]))
    return sorted(months)


def format_currency(amount: float) -> str:
    """Format number as currency."""
    return f"${amount:,.2f}"


def format_pct(value: float) -> str:
    """Format decimal as percentage."""
    return f"{value * 100:.1f}%"


def print_human(results: Dict, breakdown: Optional[Dict] = None) -> None:
    """Print results in human-readable format."""
    print("=" * 60)
    print(f"  SaaS MRR Report - {results['period']}")
    print("=" * 60)

    print(f"\n  MRR:                {format_currency(results['ending_mrr'])}")
    print(f"  ARR:                {format_currency(results['arr'])}")
    print(f"  Beginning MRR:      {format_currency(results['beginning_mrr'])}")
    print(f"  Net New MRR:        {format_currency(results['net_new_mrr'])}")

    print(f"\n  --- MRR Components ---")
    print(f"  New MRR:            {format_currency(results['new_mrr'])} ({results['new_customers']} customers)")
    print(f"  Expansion MRR:      {format_currency(results['expansion_mrr'])}")
    print(f"  Contraction MRR:   -{format_currency(results['contraction_mrr'])}")
    print(f"  Churned MRR:       -{format_currency(results['churned_mrr'])} ({results['churned_customers']} customers)")

    print(f"\n  --- Health Metrics ---")
    print(f"  Gross Churn Rate:   {format_pct(results['gross_churn_rate'])}")
    print(f"  Net Churn Rate:     {format_pct(results['net_churn_rate'])}")
    print(f"  Net Revenue Retention: {format_pct(results['net_revenue_retention'])}")
    print(f"  MoM Growth Rate:    {format_pct(results['mom_growth_rate'])}")
    qr = results['saas_quick_ratio']
    print(f"  SaaS Quick Ratio:   {qr}")

    print(f"\n  --- Customers ---")
    print(f"  Active Customers:   {results['ending_customers']}")
    arpu = results['ending_mrr'] / results['ending_customers'] if results['ending_customers'] > 0 else 0
    print(f"  ARPU:               {format_currency(arpu)}")

    # Health assessment
    print(f"\n  --- Assessment ---")
    warnings = []
    if results['gross_churn_rate'] > 0.05:
        warnings.append("Gross churn rate exceeds 5% monthly threshold")
    if results['net_revenue_retention'] < 0.90:
        warnings.append("Net revenue retention below 90% - significant revenue leakage")
    elif results['net_revenue_retention'] < 1.00:
        warnings.append("Net revenue retention below 100% - expansion not offsetting churn")
    if isinstance(qr, (int, float)) and qr < 2:
        warnings.append("SaaS Quick Ratio below 2 - growth efficiency is low")

    if warnings:
        for w in warnings:
            print(f"  WARNING: {w}")
    else:
        print("  All metrics within healthy ranges")

    if breakdown:
        print(f"\n  --- Plan Breakdown ---")
        print(f"  {'Plan':<15} {'MRR':>12} {'Customers':>10} {'ARPU':>10} {'Mix':>8}")
        print(f"  {'-'*55}")
        for plan, data in breakdown.items():
            print(f"  {plan:<15} {format_currency(data['mrr']):>12} {data['customers']:>10} {format_currency(data['arpu']):>10} {data['mix_pct']:>7.1f}%")

    print()


def main():
    parser = argparse.ArgumentParser(
        description="Calculate MRR, ARR, growth rate, and churn from subscription CSV data"
    )
    parser.add_argument("file", help="CSV file with subscription data")
    parser.add_argument("--format", choices=["human", "json"], default="human", help="Output format")
    parser.add_argument("--period", help="Period to analyze (YYYY-MM). Default: latest detected period")
    parser.add_argument("--breakdown", action="store_true", help="Include breakdown by plan tier")
    parser.add_argument("--all-periods", action="store_true", help="Show metrics for all detected periods")
    args = parser.parse_args()

    subscriptions = load_subscriptions(args.file)
    if not subscriptions:
        print("Error: No valid subscription records found", file=sys.stderr)
        sys.exit(1)

    periods = detect_periods(subscriptions)
    if not periods:
        print("Error: No periods detected in data", file=sys.stderr)
        sys.exit(1)

    if args.all_periods:
        target_periods = periods
    elif args.period:
        target_periods = [args.period]
    else:
        target_periods = [periods[-1]]

    all_results = []
    for period in target_periods:
        results = calculate_mrr_components(subscriptions, period)
        bd = breakdown_by_plan(subscriptions, period) if args.breakdown else None
        all_results.append({"metrics": results, "breakdown": bd})

    if args.format == "json":
        output = all_results if len(all_results) > 1 else all_results[0]
        print(json.dumps(output, indent=2, default=str))
    else:
        for item in all_results:
            print_human(item["metrics"], item.get("breakdown"))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Unit Economics Calculator

Calculates LTV, CAC, LTV:CAC ratio, payback period, and related unit economics
metrics from SaaS business data.

Expected JSON input with fields:
  total_customers, new_customers, churned_customers, total_mrr,
  arpu, gross_margin, sales_marketing_spend, monthly_churn_rate

Usage:
    python unit_economics.py metrics.json
    python unit_economics.py metrics.json --format json
    python unit_economics.py metrics.json --discount-rate 0.10
    python unit_economics.py --interactive
"""

import argparse
import json
import math
import sys
from typing import Any, Dict, List, Optional, Tuple


REQUIRED_FIELDS = [
    "total_customers",
    "new_customers",
    "total_mrr",
    "sales_marketing_spend",
]


def validate_input(data: Dict[str, Any]) -> List[str]:
    """Validate input data and return list of errors."""
    errors = []
    for field in REQUIRED_FIELDS:
        if field not in data:
            errors.append(f"Missing required field: {field}")

    if "total_customers" in data and data["total_customers"] <= 0:
        errors.append("total_customers must be positive")
    if "new_customers" in data and data["new_customers"] < 0:
        errors.append("new_customers cannot be negative")
    if "sales_marketing_spend" in data and data["sales_marketing_spend"] < 0:
        errors.append("sales_marketing_spend cannot be negative")

    return errors


def derive_missing_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Derive fields that can be calculated from other fields."""
    d = dict(data)

    # Derive ARPU if not provided
    if "arpu" not in d and "total_mrr" in d and "total_customers" in d:
        d["arpu"] = d["total_mrr"] / d["total_customers"] if d["total_customers"] > 0 else 0

    # Derive monthly churn rate if not provided
    if "monthly_churn_rate" not in d:
        if "churned_customers" in d and "total_customers" in d and d["total_customers"] > 0:
            d["monthly_churn_rate"] = d["churned_customers"] / d["total_customers"]
        else:
            d["monthly_churn_rate"] = 0.05  # Default assumption

    # Default gross margin if not provided
    if "gross_margin" not in d:
        d["gross_margin"] = 0.80  # SaaS typical default

    return d


def calculate_ltv_simple(arpu: float, churn_rate: float) -> float:
    """Simple LTV = ARPU / churn rate."""
    if churn_rate <= 0:
        return arpu * 120  # Cap at 10 years if no churn
    return arpu / churn_rate


def calculate_ltv_gross_margin(arpu: float, gross_margin: float, churn_rate: float) -> float:
    """Gross margin adjusted LTV = (ARPU * GM) / churn rate."""
    if churn_rate <= 0:
        return arpu * gross_margin * 120
    return (arpu * gross_margin) / churn_rate


def calculate_ltv_dcf(
    arpu: float, gross_margin: float, churn_rate: float, monthly_discount_rate: float, months: int = 60
) -> float:
    """DCF-based LTV calculation over projected lifetime."""
    ltv = 0.0
    survival_rate = 1.0
    for month in range(1, months + 1):
        survival_rate *= (1 - churn_rate)
        monthly_value = arpu * gross_margin * survival_rate
        discount_factor = 1 / ((1 + monthly_discount_rate) ** month)
        ltv += monthly_value * discount_factor
    return ltv


def calculate_cac(sales_marketing_spend: float, new_customers: int) -> float:
    """Calculate Customer Acquisition Cost."""
    if new_customers <= 0:
        return 0.0
    return sales_marketing_spend / new_customers


def calculate_payback_months(cac: float, arpu: float, gross_margin: float) -> float:
    """Calculate CAC payback period in months."""
    monthly_contribution = arpu * gross_margin
    if monthly_contribution <= 0:
        return float("inf")
    return cac / monthly_contribution


def calculate_unit_economics(data: Dict[str, Any], annual_discount_rate: float = 0.10) -> Dict[str, Any]:
    """Calculate comprehensive unit economics."""
    d = derive_missing_fields(data)

    arpu = d["arpu"]
    churn = d["monthly_churn_rate"]
    gm = d["gross_margin"]
    monthly_dr = (1 + annual_discount_rate) ** (1 / 12) - 1

    # LTV calculations (three methods)
    ltv_simple = calculate_ltv_simple(arpu, churn)
    ltv_gm = calculate_ltv_gross_margin(arpu, gm, churn)
    ltv_dcf = calculate_ltv_dcf(arpu, gm, churn, monthly_dr)

    # CAC
    cac = calculate_cac(d["sales_marketing_spend"], d["new_customers"])

    # Ratios
    ltv_cac_simple = ltv_simple / cac if cac > 0 else float("inf")
    ltv_cac_gm = ltv_gm / cac if cac > 0 else float("inf")
    ltv_cac_dcf = ltv_dcf / cac if cac > 0 else float("inf")

    # Payback
    payback = calculate_payback_months(cac, arpu, gm)

    # Expected lifetime in months
    avg_lifetime = 1 / churn if churn > 0 else 120

    # Monthly contribution margin per customer
    monthly_contribution = arpu * gm

    # Annual unit profit (LTV GM - CAC, annualized)
    annual_unit_profit = (ltv_gm - cac) / (avg_lifetime / 12) if avg_lifetime > 0 else 0

    # Magic number: Net New ARR / Prior Quarter S&M Spend
    magic_number = None
    if "net_new_mrr" in d and d["sales_marketing_spend"] > 0:
        magic_number = (d["net_new_mrr"] * 12) / (d["sales_marketing_spend"] * 3)

    return {
        "period": d.get("period", "current"),
        "inputs": {
            "total_customers": d["total_customers"],
            "new_customers": d["new_customers"],
            "churned_customers": d.get("churned_customers", "N/A"),
            "total_mrr": round(d["total_mrr"], 2),
            "arpu": round(arpu, 2),
            "gross_margin": round(gm, 4),
            "monthly_churn_rate": round(churn, 4),
            "sales_marketing_spend": round(d["sales_marketing_spend"], 2),
            "annual_discount_rate": annual_discount_rate,
        },
        "ltv": {
            "simple": round(ltv_simple, 2),
            "gross_margin_adjusted": round(ltv_gm, 2),
            "dcf": round(ltv_dcf, 2),
        },
        "cac": round(cac, 2),
        "ltv_cac_ratio": {
            "simple": round(ltv_cac_simple, 2) if ltv_cac_simple != float("inf") else "infinite",
            "gross_margin_adjusted": round(ltv_cac_gm, 2) if ltv_cac_gm != float("inf") else "infinite",
            "dcf": round(ltv_cac_dcf, 2) if ltv_cac_dcf != float("inf") else "infinite",
        },
        "payback_months": round(payback, 1) if payback != float("inf") else "infinite",
        "avg_customer_lifetime_months": round(avg_lifetime, 1),
        "monthly_contribution_margin": round(monthly_contribution, 2),
        "magic_number": round(magic_number, 2) if magic_number is not None else "N/A",
    }


def assess_health(results: Dict[str, Any]) -> List[Dict[str, str]]:
    """Generate health assessment from unit economics."""
    findings = []

    # LTV:CAC assessment
    ratio = results["ltv_cac_ratio"]["gross_margin_adjusted"]
    if isinstance(ratio, (int, float)):
        if ratio < 1:
            findings.append({
                "metric": "LTV:CAC",
                "status": "CRITICAL",
                "message": f"LTV:CAC is {ratio:.1f}x - losing money on every customer acquired",
                "action": "Immediately reduce CAC or improve retention/pricing",
            })
        elif ratio < 3:
            findings.append({
                "metric": "LTV:CAC",
                "status": "WARNING",
                "message": f"LTV:CAC is {ratio:.1f}x - below the 3x healthy threshold",
                "action": "Focus on reducing churn and optimizing acquisition spend",
            })
        elif ratio > 5:
            findings.append({
                "metric": "LTV:CAC",
                "status": "OPPORTUNITY",
                "message": f"LTV:CAC is {ratio:.1f}x - potentially under-investing in growth",
                "action": "Consider increasing marketing spend to accelerate growth",
            })
        else:
            findings.append({
                "metric": "LTV:CAC",
                "status": "HEALTHY",
                "message": f"LTV:CAC is {ratio:.1f}x - within healthy 3-5x range",
                "action": "Maintain current balance, optimize incrementally",
            })

    # Payback assessment
    payback = results["payback_months"]
    if isinstance(payback, (int, float)):
        if payback > 24:
            findings.append({
                "metric": "CAC Payback",
                "status": "CRITICAL",
                "message": f"Payback period is {payback:.0f} months - strains cash flow severely",
                "action": "Reduce CAC, increase prices, or improve gross margin",
            })
        elif payback > 18:
            findings.append({
                "metric": "CAC Payback",
                "status": "WARNING",
                "message": f"Payback period is {payback:.0f} months - above ideal range",
                "action": "Target sub-18-month payback through pricing or efficiency",
            })
        elif payback <= 12:
            findings.append({
                "metric": "CAC Payback",
                "status": "HEALTHY",
                "message": f"Payback period is {payback:.0f} months - strong cash efficiency",
                "action": "Healthy position, consider scaling spend",
            })

    # Churn assessment
    churn = results["inputs"]["monthly_churn_rate"]
    if churn > 0.05:
        findings.append({
            "metric": "Monthly Churn",
            "status": "CRITICAL",
            "message": f"Monthly churn is {churn*100:.1f}% - losing >5% of customers monthly",
            "action": "Prioritize retention: onboarding, engagement, customer success",
        })
    elif churn > 0.03:
        findings.append({
            "metric": "Monthly Churn",
            "status": "WARNING",
            "message": f"Monthly churn is {churn*100:.1f}% - above target range",
            "action": "Investigate churn reasons, improve activation and engagement",
        })

    return findings


def format_currency(amount: float) -> str:
    return f"${amount:,.2f}"


def print_human(results: Dict[str, Any]) -> None:
    """Print results in human-readable format."""
    print("=" * 60)
    print(f"  Unit Economics Dashboard - {results['period']}")
    print("=" * 60)

    inp = results["inputs"]
    print(f"\n  --- Inputs ---")
    print(f"  Total Customers:      {inp['total_customers']:,}")
    print(f"  New Customers:        {inp['new_customers']:,}")
    print(f"  ARPU:                 {format_currency(inp['arpu'])}")
    print(f"  Gross Margin:         {inp['gross_margin']*100:.1f}%")
    print(f"  Monthly Churn Rate:   {inp['monthly_churn_rate']*100:.2f}%")
    print(f"  S&M Spend:            {format_currency(inp['sales_marketing_spend'])}")

    print(f"\n  --- Lifetime Value ---")
    ltv = results["ltv"]
    print(f"  LTV (Simple):              {format_currency(ltv['simple'])}")
    print(f"  LTV (Gross Margin Adj):    {format_currency(ltv['gross_margin_adjusted'])}")
    print(f"  LTV (DCF, {inp['annual_discount_rate']*100:.0f}% discount):  {format_currency(ltv['dcf'])}")

    print(f"\n  --- Acquisition ---")
    print(f"  CAC:                  {format_currency(results['cac'])}")
    print(f"  LTV:CAC (GM Adj):     {results['ltv_cac_ratio']['gross_margin_adjusted']}x")
    pb = results['payback_months']
    print(f"  Payback Period:       {pb} months")
    print(f"  Avg Lifetime:         {results['avg_customer_lifetime_months']} months")
    print(f"  Monthly Contribution: {format_currency(results['monthly_contribution_margin'])}")

    if results["magic_number"] != "N/A":
        print(f"  Magic Number:         {results['magic_number']}")

    # Health assessment
    findings = assess_health(results)
    if findings:
        print(f"\n  --- Health Assessment ---")
        for f in findings:
            status_icon = {"CRITICAL": "!!!", "WARNING": " ! ", "OPPORTUNITY": " * ", "HEALTHY": " + "}
            icon = status_icon.get(f["status"], "   ")
            print(f"  [{icon}] {f['metric']}: {f['message']}")
            print(f"        Action: {f['action']}")

    print()


def interactive_mode() -> Dict[str, Any]:
    """Collect metrics interactively from stdin."""
    print("Unit Economics Calculator - Interactive Mode")
    print("-" * 40)
    data = {}
    data["total_customers"] = int(input("Total active customers: "))
    data["new_customers"] = int(input("New customers this period: "))
    data["churned_customers"] = int(input("Churned customers this period: "))
    data["total_mrr"] = float(input("Total MRR ($): "))
    data["gross_margin"] = float(input("Gross margin (0.0-1.0): "))
    data["sales_marketing_spend"] = float(input("Sales & marketing spend ($): "))
    data["period"] = input("Period label (e.g., 2025-Q4): ") or "current"
    return data


def main():
    parser = argparse.ArgumentParser(
        description="Calculate LTV, CAC, LTV:CAC ratio, and payback period"
    )
    parser.add_argument("file", nargs="?", help="JSON file with unit economics data")
    parser.add_argument("--format", choices=["human", "json"], default="human", help="Output format")
    parser.add_argument(
        "--discount-rate", type=float, default=0.10, help="Annual discount rate for DCF LTV (default: 0.10)"
    )
    parser.add_argument("--interactive", action="store_true", help="Enter data interactively")
    args = parser.parse_args()

    if args.interactive:
        data = interactive_mode()
    elif args.file:
        with open(args.file, "r", encoding="utf-8") as f:
            data = json.load(f)
    else:
        print("Error: Provide a JSON file or use --interactive mode", file=sys.stderr)
        sys.exit(1)

    errors = validate_input(data)
    if errors:
        for e in errors:
            print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)

    results = calculate_unit_economics(data, args.discount_rate)

    if args.format == "json":
        output = {"unit_economics": results, "assessment": assess_health(results)}
        print(json.dumps(output, indent=2, default=str))
    else:
        print_human(results)


if __name__ == "__main__":
    main()