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%
The Billing Architecture Decision
Billing is the highest-stakes system in any SaaS — errors erode customer trust and directly drive churn. A sound billing architecture rests on three pillars: accurate metering, correct invoicing, and reliable reconciliation.
The choice between metered, flat-rate, and tiered billing carries deep architectural implications. Metered billing demands a high-throughput event pipeline (Kafka, Kinesis, or similar) to prevent data loss at scale. Flat-rate billing is simple to implement but leaves money on the table as high-usage customers consume disproportionately more resources.
Tiered pricing strikes a balance between fairness and predictability, though it complicates invoice line-item logic and proration. Among these, reconciliation — matching recorded usage to billed charges — is the most overlooked function yet the most critical for audit compliance and revenue integrity.
Third-party platforms like Stripe, Recurly, and Chargebee abstract payment and subscription logic, but your team still owns the metering pipeline and the reconciliation layer that sits between raw events and final invoices.
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