subscription-revenue-tracker

What this skill does

# Subscription Revenue Tracker

Track MRR/ARR, churn, NRR, cohort retention, and LTV for SaaS and subscription businesses. Produces investor-grade metrics and clean GL entries.

---

## Core Metrics Defined

| Metric | Formula | Why It Matters |
|--------|---------|----------------|
| MRR | Sum of all active recurring monthly revenue | Pulse of the business |
| ARR | MRR × 12 | Annualized scale metric for investors |
| New MRR | Revenue from new customers this month | Growth engine |
| Expansion MRR | Upgrades / upsells from existing customers | Efficiency signal |
| Contraction MRR | Downgrades from existing customers | Negative signal |
| Churned MRR | Revenue lost from cancellations | Retention health |
| Net New MRR | New + Expansion − Contraction − Churned | Net growth |
| Gross Churn Rate | Churned MRR / Beginning MRR | Revenue decay rate |
| Net Revenue Retention (NRR) | (Beginning + Expansion − Contraction − Churned) / Beginning | Growth from existing base |
| LTV | ARPU / Gross Churn Rate | Customer economic value |
| CAC | Sales + Marketing Spend / New Customers | Acquisition cost |
| LTV:CAC | LTV / CAC | Unit economics health (target: >3x) |
| Payback Period | CAC / (ARPU × Gross Margin) | Months to recover acquisition cost |

---

## Workflows

### 1. Pull MRR from Stripe

```bash
# List all active subscriptions with their amounts
stripe subscriptions list \
  --status=active \
  --limit=100 \
  --expand[]=data.items.data \
  2>&1 | jq '
    .data[] | {
      id: .id,
      customer: .customer,
      status: .status,
      current_period_start: (.current_period_start | strftime("%Y-%m-%d")),
      mrr: (.items.data[0].price.unit_amount / 100 * 
            (if .items.data[0].price.recurring.interval == "year" then 1/12 else 1 end))
    }
  '
```

**Get MRR summary via Stripe API (no CLI):**
```bash
curl "https://api.stripe.com/v1/subscriptions?status=active&limit=100&expand[]=data.items.data" \
  -u sk_live_YOUR_KEY: | jq '
  [.data[] | 
    (.items.data[0].price.unit_amount / 100) * 
    (if .items.data[0].price.recurring.interval == "year" then 1/12 else 1 end)
  ] | add
  '
```

**Python: Full MRR waterfall from Stripe events:**
```python
import stripe
from datetime import datetime, timezone
from collections import defaultdict
from dateutil.relativedelta import relativedelta

stripe.api_key = "sk_live_YOUR_KEY"

def get_mrr_waterfall(year: int, month: int) -> dict:
    """
    Calculate MRR waterfall for a given month.
    Returns: new, expansion, contraction, churned, net_new MRR.
    """
    # Period boundaries
    period_start = datetime(year, month, 1, tzinfo=timezone.utc)
    period_end = period_start + relativedelta(months=1)
    prev_start = period_start - relativedelta(months=1)

    # Get subscriptions active at start of period (denominator)
    beginning_subs = _get_active_subscriptions_at(prev_start)
    ending_subs = _get_active_subscriptions_at(period_end)

    # Categorize by customer
    beginning_customers = {s.customer: _get_mrr(s) for s in beginning_subs}
    ending_customers = {s.customer: _get_mrr(s) for s in ending_subs}

    new_mrr = 0.0
    expansion_mrr = 0.0
    contraction_mrr = 0.0
    churned_mrr = 0.0

    all_customers = set(beginning_customers) | set(ending_customers)

    for cust_id in all_customers:
        begin_val = beginning_customers.get(cust_id, 0.0)
        end_val = ending_customers.get(cust_id, 0.0)
        delta = end_val - begin_val

        if begin_val == 0 and end_val > 0:
            new_mrr += end_val
        elif begin_val > 0 and end_val == 0:
            churned_mrr += begin_val
        elif delta > 0:
            expansion_mrr += delta
        elif delta < 0:
            contraction_mrr += abs(delta)

    beginning_mrr = sum(beginning_customers.values())

    return {
        "period": f"{year}-{month:02d}",
        "beginning_mrr": beginning_mrr,
        "new_mrr": new_mrr,
        "expansion_mrr": expansion_mrr,
        "contraction_mrr": contraction_mrr,
        "churned_mrr": churned_mrr,
        "net_new_mrr": new_mrr + expansion_mrr - contraction_mrr - churned_mrr,
        "ending_mrr": beginning_mrr + new_mrr + expansion_mrr - contraction_mrr - churned_mrr,
        "gross_churn_rate": churned_mrr / beginning_mrr if beginning_mrr else 0,
        "nrr": (beginning_mrr + expansion_mrr - contraction_mrr - churned_mrr) / beginning_mrr if beginning_mrr else 0,
    }

def _get_mrr(subscription) -> float:
    """Extract normalized monthly value from a Stripe subscription."""
    item = subscription.get("items", {}).get("data", [{}])[0]
    price = item.get("price", {})
    amount = price.get("unit_amount", 0) / 100
    qty = item.get("quantity", 1)
    interval = price.get("recurring", {}).get("interval", "month")
    
    if interval == "year":
        return (amount * qty) / 12
    elif interval == "week":
        return (amount * qty) * 4.333
    return amount * qty

def _get_active_subscriptions_at(timestamp: datetime) -> list:
    """Get subscriptions that were active at a given timestamp."""
    ts = int(timestamp.timestamp())
    subs = stripe.Subscription.list(
        status="all",
        created={"lte": ts},
        limit=100
    )
    return [
        s for s in subs.auto_paging_iter()
        if s.current_period_start <= ts <= (s.canceled_at or ts + 1)
    ]
```

### 2. Cohort Analysis

Track retention by signup cohort — the gold standard for understanding retention quality:

```python
import pandas as pd
import numpy as np

def build_cohort_table(subscription_events: pd.DataFrame) -> pd.DataFrame:
    """
    Build monthly cohort retention table.
    
    Input columns: customer_id, event_type (started/churned), event_month (YYYY-MM)
    Output: matrix of cohort × months_since_start → retention percentage
    
    Example output:
    cohort    | M+0  | M+1  | M+2  | M+3  | M+6  | M+12
    2025-01   | 100% | 87%  | 79%  | 74%  | 65%  | 54%
    2025-02   | 100% | 91%  | 83%  | 78%  | --   | --
    """
    # Assign cohort (month of first subscription)
    first_sub = (subscription_events[subscription_events.event_type == "started"]
                 .groupby("customer_id")["event_month"]
                 .min()
                 .reset_index()
                 .rename(columns={"event_month": "cohort"}))
    
    df = subscription_events.merge(first_sub, on="customer_id")
    df["cohort"] = pd.to_datetime(df["cohort"])
    df["event_month"] = pd.to_datetime(df["event_month"])
    df["months_since_start"] = (
        (df["event_month"].dt.year - df["cohort"].dt.year) * 12 +
        (df["event_month"].dt.month - df["cohort"].dt.month)
    )
    
    # Active customers per cohort per month
    active = (df[df.event_type != "churned"]
              .groupby(["cohort", "months_since_start"])["customer_id"]
              .nunique()
              .reset_index()
              .rename(columns={"customer_id": "active_customers"}))
    
    cohort_table = active.pivot(
        index="cohort", 
        columns="months_since_start", 
        values="active_customers"
    )
    
    # Normalize to cohort size (M+0 = 100%)
    cohort_sizes = cohort_table[0]
    retention_table = cohort_table.divide(cohort_sizes, axis=0) * 100
    
    return retention_table.round(1)


def average_retention_curve(cohort_table: pd.DataFrame, min_cohorts: int = 3) -> pd.Series:
    """
    Compute average retention curve across cohorts with enough data.
    Used for LTV projection.
    """
    # Only include cohorts with at least min_cohorts data points per period
    valid_cols = cohort_table.columns[cohort_table.notna().sum() >= min_cohorts]
    return cohort_table[valid_cols].mean()
```

### 3. LTV and Unit Economics

```python
def calculate_ltv(arpu: float, gross_margin: float, monthly_churn_rate: float) -> dict:
    """
    Calculate Customer Lifetime Value and payback metrics.
    
    Args:
        arpu: Average Revenue Per User per month
        gross_margin: Gross margin % (0.0-1.0)
        month