Introduction
SaaS billing is complex. From subscription management to metered usage tracking, building accurate billing systems requires careful design to ensure revenue integrity and customer trust.
Key Statistics:
- Billing errors cost 2-5% of revenue
- 40% of SaaS companies struggle with usage-based billing
- Revenue recognition errors trigger audit failures
- Automated billing reduces AR by 60%
Billing Models
┌─────────────────────────────────────────────────────────────────┐
│ SaaS Billing Models │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐│
│ │ Subscription │ │ Metered │ │ Hybrid ││
│ │ │ │ │ │ ││
│ │ Monthly/Annual │ │ Pay-as-you-go │ │ Base + Usage ││
│ │ Fixed price │ │ Usage-based │ │ Tiered pricing││
│ │ │ │ Overage │ │ + Overage ││
│ └─────────────────┘ └─────────────────┘ └─────────────────┘│
│ │
│ Pricing Examples │
│ ├── $29/user/month (subscription) │
│ ├── $0.001/API call (metered) │
│ └── $99 + $0.10/GB storage (hybrid) │
│ │
└─────────────────────────────────────────────────────────────────┘
Usage Tracking
Event Collection
#!/usr/bin/env python3
"""Usage tracking system."""
from datetime import datetime
import json
class UsageTracker:
"""Track usage events for billing."""
def __init__(self, db, event_queue):
self.db = db
self.queue = event_queue
def track_event(self, customer_id, resource_type,
quantity, metadata=None):
"""Record usage event."""
event = {
'event_id': generate_uuid(),
'customer_id': customer_id,
'resource_type': resource_type,
'quantity': quantity,
'metadata': metadata or {},
'timestamp': datetime.utcnow().isoformat(),
'processed': False
}
# Store in database
self.db.usage_events.insert(event)
# Queue for processing
self.queue.publish('usage.events', event)
return event['event_id']
def track_api_usage(self, customer_id, endpoint, duration_ms):
"""Track API call usage."""
self.track_event(
customer_id=customer_id,
resource_type='api_call',
quantity=1,
metadata={
'endpoint': endpoint,
'duration_ms': duration_ms,
'tier': self.get_tier(customer_id)
}
)
def track_storage_usage(self, customer_id, bytes_stored):
"""Track storage usage."""
# Convert to GB-hours
gb_hours = (bytes_stored / (1024**3)) * (1/24)
self.track_event(
customer_id=customer_id,
resource_type='storage_gb',
quantity=gb_hours,
metadata={'bytes': bytes_stored}
)
def track_compute_usage(self, customer_id, compute_units):
"""Track compute usage."""
self.track_event(
customer_id=customer_id,
resource_type='compute_unit',
quantity=compute_units,
metadata={'unit_type': 'cpu_hour'}
)
Usage Aggregation
#!/usr/bin/env python3
"""Usage aggregation for billing."""
class UsageAggregator:
"""Aggregate usage for billing periods."""
def __init__(self, db):
self.db = db
def aggregate_daily_usage(self, customer_id, billing_period):
"""Aggregate daily usage for customer."""
pipeline = [
{'$match': {
'customer_id': customer_id,
'timestamp': {
'$gte': billing_period.start,
'$lt': billing_period.end
}
}},
{'$group': {
'_id': {
'resource_type': '$resource_type',
'date': {'$dateToString': {'format': '%Y-%m-%d', 'date': '$timestamp'}}
},
'total_quantity': {'$sum': '$quantity'},
'event_count': {'$sum': 1}
}},
{'$sort': {'_id.date': 1}}
]
return list(self.db.usage_events.aggregate(pipeline))
def calculate_usage_cost(self, customer_id, billing_period):
"""Calculate cost based on usage and pricing."""
daily_usage = self.aggregate_daily_usage(customer_id, billing_period)
# Get pricing rules
pricing = self.get_pricing_rules(customer_id)
total_cost = 0
cost_breakdown = []
for usage in daily_usage:
resource_type = usage['_id']['resource_type']
quantity = usage['total_quantity']
unit_price = pricing.get(resource_type, {}).get('unit_price', 0)
# Apply tiered pricing if applicable
if resource_type in pricing.get('tiered', []):
unit_price = self.calculate_tiered_price(
quantity,
pricing['tiers'][resource_type]
)
cost = quantity * unit_price
total_cost += cost
cost_breakdown.append({
'resource_type': resource_type,
'quantity': quantity,
'unit_price': unit_price,
'cost': cost
})
return {
'customer_id': customer_id,
'billing_period': str(billing_period),
'total_cost': total_cost,
'breakdown': cost_breakdown
}
def calculate_tiered_price(self, quantity, tiers):
"""Calculate tiered pricing."""
tiers = sorted(tiers, key=lambda t: t['threshold'])
total_cost = 0
remaining = quantity
for tier in tiers:
threshold = tier['threshold']
price = tier['price']
if remaining <= 0:
break
# Calculate quantity in this tier
tier_quantity = min(remaining, threshold)
total_cost += tier_quantity * price
remaining -= tier_quantity
return total_cost / quantity if quantity > 0 else 0
Invoice Generation
#!/usr/bin/env python3
"""Invoice generation system."""
from datetime import datetime
import pdfkit
class InvoiceGenerator:
"""Generate invoices for customers."""
def __init__(self, db, template_renderer):
self.db = db
self.template = template_renderer
def generate_invoice(self, customer_id, billing_period):
"""Generate invoice for billing period."""
# Get subscription
subscription = self.db.subscriptions.find_one({
'customer_id': customer_id,
'status': 'active'
})
# Calculate subscription cost
sub_cost = subscription['price']
# Calculate usage cost
usage_cost = self.calculate_usage_cost(customer_id, billing_period)
# Apply discounts
discount = self.apply_discounts(
subscription,
sub_cost + usage_cost['total_cost']
)
# Calculate tax
tax = self.calculate_tax(
customer_id,
sub_cost + usage_cost['total_cost'] - discount
)
total = sub_cost + usage_cost['total_cost'] - discount + tax
invoice = {
'invoice_id': generate_invoice_number(),
'customer_id': customer_id,
'billing_period': str(billing_period),
'line_items': [
{
'description': f"Subscription ({subscription['plan_name']})",
'quantity': 1,
'unit_price': sub_cost,
'total': sub_cost
},
*[
{
'description': f"{item['resource_type']} usage",
'quantity': item['quantity'],
'unit_price': item['unit_price'],
'total': item['cost']
}
for item in usage_cost['breakdown']
]
],
'subtotal': sub_cost + usage_cost['total_cost'],
'discount': discount,
'tax': tax,
'total': total,
'currency': subscription['currency'],
'due_date': billing_period.end + timedelta(days=30),
'status': 'pending'
}
# Save invoice
self.db.invoices.insert(invoice)
return invoice
def generate_pdf(self, invoice):
"""Generate PDF invoice."""
html = self.template.render('invoice.html', invoice=invoice)
pdf = pdfkit.from_string(html, False)
return pdf
def send_invoice(self, invoice):
"""Send invoice to customer."""
customer = self.db.customers.find_one({'_id': invoice['customer_id']})
# Generate PDF
pdf = self.generate_pdf(invoice)
# Send email
email_service.send(
to=customer['email'],
subject=f"Invoice {invoice['invoice_id']}",
body=self.template.render('invoice_email.html', invoice=invoice),
attachments=[('invoice.pdf', pdf, 'application/pdf')]
)
Revenue Recognition
#!/usr/bin/env python3
"""Revenue recognition (ASC 606 / IFRS 15)."""
class RevenueRecognizer:
"""Handle revenue recognition."""
def __init__(self, db):
self.db = db
def recognize_subscription_revenue(self, invoice):
"""Recognize revenue for subscription."""
# Subscription revenue recognized over time
recognition_schedule = []
start_date = datetime.fromisoformat(invoice['billing_period'].start)
end_date = datetime.fromisoformat(invoice['billing_period'].end)
days_in_period = (end_date - start_date).days
daily_rate = invoice['subtotal'] / days_in_period
current_date = start_date
while current_date < end_date:
recognition_schedule.append({
'invoice_id': invoice['invoice_id'],
'recognition_date': current_date,
'amount': daily_rate,
'revenue_type': 'subscription'
})
current_date += timedelta(days=1)
# Store recognition schedule
self.db.revenue_recognition.insert_many(recognition_schedule)
return recognition_schedule
def recognize_usage_revenue(self, usage, event_timestamp):
"""Recognize usage revenue at point in time."""
# Usage revenue recognized when usage occurs
recognition = {
'invoice_id': usage.get('invoice_id'),
'recognition_date': event_timestamp,
'amount': usage['cost'],
'revenue_type': 'usage',
'customer_id': usage['customer_id']
}
self.db.revenue_recognition.insert_one(recognition)
return recognition
def generate_revenue_report(self, start_date, end_date):
"""Generate revenue recognition report."""
pipeline = [
{'$match': {
'recognition_date': {
'$gte': start_date,
'$lt': end_date
}
}},
{'$group': {
'_id': {
'year': {'$year': '$recognition_date'},
'month': {'$month': '$recognition_date'},
'revenue_type': '$revenue_type'
},
'total_revenue': {'$sum': '$amount'}
}},
{'$sort': {'_id.year': 1, '_id.month': 1}}
]
return list(self.db.revenue_recognition.aggregate(pipeline))
Reconciliation
#!/usr/bin/env python3
"""Billing reconciliation."""
class BillingReconciler:
"""Reconcile billing records."""
def __init__(self, db, payment_gateway):
self.db = db
self.gateway = payment_gateway
def reconcile_payments(self, start_date, end_date):
"""Reconcile payments with invoices."""
# Get all payments
payments = list(self.db.payments.find({
'created_at': {'$gte': start_date, '$lt': end_date},
'status': 'succeeded'
}))
# Get all invoices
invoices = list(self.db.invoices.find({
'billing_period.start': {'$gte': start_date},
'billing_period.end': {'$lt': end_date}
}))
# Build payment lookup
payment_lookup = {p['invoice_id']: p for p in payments}
# Reconcile
results = []
for invoice in invoices:
payment = payment_lookup.get(invoice['invoice_id'])
if not payment:
results.append({
'invoice_id': invoice['invoice_id'],
'status': 'missing_payment',
'invoice_amount': invoice['total'],
'payment_amount': 0
})
elif abs(payment['amount'] - invoice['total']) > 0.01:
results.append({
'invoice_id': invoice['invoice_id'],
'status': 'amount_mismatch',
'invoice_amount': invoice['total'],
'payment_amount': payment['amount']
})
else:
results.append({
'invoice_id': invoice['invoice_id'],
'status': 'matched',
'invoice_amount': invoice['total'],
'payment_amount': payment['amount']
})
return results
Comments