订阅收入追踪器

追踪SaaS和订阅业务的MRR/ARR、流失率、NRR、群组留存率和LTV。生成投资者级指标和清晰的会计分录。

核心指标定义

工作流

1. 从Stripe提取MRR

bash

列出所有活跃订阅及其金额

通过Stripe API获取MRR摘要(无需CLI):
bash
curl https://api.stripe.com/v1/subscriptions?status=active&limit=100&expand[]=data.items.data \
-u skliveYOUR_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: 基于Stripe事件的完整MRR瀑布图:
python
import stripe
from datetime import datetime, timezone
from collections import defaultdict
from dateutil.relativedelta import relativedelta

stripe.apikey = skliveYOURKEY

def getmrrwaterfall(year: int, month: int) -> dict:

计算指定月份的MRR瀑布图。
返回: 新增、扩展、收缩、流失、净新增MRR。

# 期间边界
period_start = datetime(year, month, 1, tzinfo=timezone.utc)
periodend = periodstart + relativedelta(months=1)
prevstart = periodstart - relativedelta(months=1)

# 获取期间开始时活跃的订阅(分母)
beginningsubs = getactivesubscriptionsat(prevstart)
endingsubs = getactivesubscriptionsat(periodend)

# 按客户分类
beginningcustomers = {s.customer: getmrr(s) for s in beginningsubs}
endingcustomers = {s.customer: getmrr(s) for s in endingsubs}

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

allcustomers = set(beginningcustomers) | set(ending_customers)

for custid in allcustomers:
beginval = beginningcustomers.get(cust_id, 0.0)
endval = endingcustomers.get(cust_id, 0.0)
delta = endval - beginval

if beginval == 0 and endval > 0:
newmrr += endval
elif beginval > 0 and endval == 0:
churnedmrr += beginval
elif delta > 0:
expansion_mrr += delta
elif delta < 0:
contraction_mrr += abs(delta)

beginningmrr = sum(beginningcustomers.values())

return {
period: f{year}-{month:02d},
beginningmrr: beginningmrr,
newmrr: newmrr,
expansionmrr: expansionmrr,
contractionmrr: contractionmrr,
churnedmrr: churnedmrr,
netnewmrr: newmrr + expansionmrr - contractionmrr - churnedmrr,
endingmrr: beginningmrr + newmrr + expansionmrr - contractionmrr - churnedmrr,
grosschurnrate: churnedmrr / beginningmrr if beginning_mrr else 0,
nrr: (beginningmrr + expansionmrr - contractionmrr - churnedmrr) / beginningmrr if beginningmrr else 0,
}

def getmrr(subscription) -> float:
从Stripe订阅中提取标准化月度值。
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 getactivesubscriptionsat(timestamp: datetime) -> list:
获取在指定时间戳活跃的订阅。
ts = int(timestamp.timestamp())
subs = stripe.Subscription.list(
status=all,
created={lte: ts},
limit=100
)
return [
s for s in subs.autopagingiter()
if s.currentperiodstart <= ts <= (s.canceled_at or ts + 1)
]

2. 群组分析

按注册群组追踪留存率——理解留存质量的黄金标准:

python
import pandas as pd
import numpy as np

def buildcohorttable(subscription_events: pd.DataFrame) -> pd.DataFrame:

构建月度群组留存表。

输入列: customerid, eventtype (started/churned), event_month (YYYY-MM)
输出: 群组 × 起始后月数 → 留存率矩阵

输出示例:
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% | -- | --

# 分配群组(首次订阅月份)
firstsub = (subscriptionevents[subscriptionevents.eventtype == started]
.groupby(customerid)[eventmonth]
.min()
.reset_index()
.rename(columns={event_month: cohort}))

df = subscriptionevents.merge(firstsub, on=customer_id)
df[cohort] = pd.to_datetime(df[cohort])
df[eventmonth] = pd.todatetime(df[event_month])
df[monthssincestart] = (
(df[event_month].dt.year - df[cohort].dt.year) * 12 +
(df[event_month].dt.month - df[cohort].dt.month)
)

# 每个群组每月活跃客户数
active = (df[df.event_type != churned]
.groupby([cohort, monthssincestart])[customer_id]
.nunique()
.reset_index()
.rename(columns={customerid: activecustomers}))

cohort_table = active.pivot(
index=cohort,
columns=monthssincestart,
values=active_customers
)