SaaS Metrics skill for Codex

Gives the Codex app a display name, a one-line description and a starter prompt for this skill.

interface:
  display_name: "SaaS Metrics"
  short_description: "MRR bridges that reconcile to the penny"
  default_prompt: "$saas-metrics Compute our MRR, churn and NRR for the last 12 months from this Stripe subscriptions export"

#!/usr/bin/env python3
"""MRR movement bridge and retention from a per-customer monthly MRR CSV.

Input CSV columns (header required): customer_id, month, mrr
  month is YYYY-MM; mrr is the normalised monthly recurring revenue for that
  customer in that month (annual plans divided by 12, discounts applied, no
  one-off fees). A customer missing from a month counts as 0 MRR that month.

Usage:
  python3 mrr_movements.py subscriptions.csv            # movement table
  python3 mrr_movements.py subscriptions.csv --cohorts  # plus logo retention cohorts

Exits 1 on malformed input or if the bridge fails to reconcile.
"""
import argparse
import csv
import sys
from collections import defaultdict
from decimal import Decimal, InvalidOperation


def month_range(first, last):
    y, m = map(int, first.split("-"))
    ly, lm = map(int, last.split("-"))
    out = []
    while (y, m) <= (ly, lm):
        out.append(f"{y:04d}-{m:02d}")
        y, m = (y + 1, 1) if m == 12 else (y, m + 1)
    return out


def load(path):
    data = defaultdict(lambda: defaultdict(Decimal))
    with open(path, newline="", encoding="utf-8-sig") as fh:
        reader = csv.DictReader(fh)
        need = {"customer_id", "month", "mrr"}
        if not reader.fieldnames or not need <= {f.strip() for f in reader.fieldnames}:
            sys.exit(f"error: CSV needs columns {sorted(need)}, got {reader.fieldnames}")
        for n, row in enumerate(reader, start=2):
            month = row["month"].strip()[:7]
            if len(month) != 7 or month[4] != "-":
                sys.exit(f"error: line {n}: month must be YYYY-MM, got {row['month']!r}")
            try:
                data[row["customer_id"].strip()][month] += Decimal(row["mrr"].strip() or "0")
            except InvalidOperation:
                sys.exit(f"error: line {n}: mrr is not a number: {row['mrr']!r}")
    if not data:
        sys.exit("error: no rows")
    return data


def pct(num, den):
    return f"{(num / den * 100):9.1f}%" if den else "       n/a"


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("csv")
    ap.add_argument("--cohorts", action="store_true", help="print logo retention by first paid month")
    args = ap.parse_args()

    data = load(args.csv)
    months = month_range(min(m for c in data.values() for m in c), max(m for c in data.values() for m in c))
    ever_paid = set()
    cols = ["month", "start", "new", "reactiv", "expand", "contract", "churn", "end",
            "logos", "logo_churn", "GRR_m", "NRR_m"]
    print("  ".join(f"{c:>10}" for c in cols))
    ok = True
    prev = {c: Decimal(0) for c in data}
    for i, month in enumerate(months):
        cur = {c: data[c].get(month, Decimal(0)) for c in data}
        mv = defaultdict(Decimal)
        lost_logos = 0
        for c in data:
            p, n = prev[c], cur[c]
            if p == 0 and n > 0:
                mv["reactiv" if c in ever_paid else "new"] += n
            elif p > 0 and n == 0:
                mv["churn"] += p
                lost_logos += 1
            elif n > p:
                mv["expand"] += n - p
            elif n < p:
                mv["contract"] += p - n
            if n > 0:
                ever_paid.add(c)
        start, end = sum(prev.values()), sum(cur.values())
        bridge = start + mv["new"] + mv["reactiv"] + mv["expand"] - mv["contract"] - mv["churn"]
        if bridge != end:
            ok = False
        start_logos = sum(1 for v in prev.values() if v > 0)
        grr = start - mv["contract"] - mv["churn"]
        nrr = grr + mv["expand"]
        row = [month, start, mv["new"], mv["reactiv"], mv["expand"], mv["contract"], mv["churn"], end]
        out = [f"{row[0]:>10}"] + [f"{v:>10.2f}" for v in row[1:]]
        out += [f"{sum(1 for v in cur.values() if v > 0):>10}",
                pct(Decimal(lost_logos), Decimal(start_logos)) if i else "       n/a",
                pct(grr, start) if i else "       n/a", pct(nrr, start) if i else "       n/a"]
        print("  ".join(out))
        prev = cur

    if args.cohorts:
        first = {c: min(m for m, v in data[c].items() if v > 0) for c in data if any(v > 0 for v in data[c].values())}
        print("\nLogo retention by first paid month (share of cohort still paying N months later)")
        for cohort in months:
            members = [c for c, f in first.items() if f == cohort]
            if not members:
                continue
            later = months[months.index(cohort):]
            cells = [pct(Decimal(sum(1 for c in members if data[c].get(m, 0) > 0)), Decimal(len(members))) for m in later[:13]]
            print(f"{cohort} n={len(members):<4} " + " ".join(cells))

    if not ok:
        sys.exit("error: MRR bridge did not reconcile for at least one month")
    print("\nbridge reconciles: start + new + reactivation + expansion - contraction - churn = end")


if __name__ == "__main__":
    main()