Build automated billing systems for recurring payments, invoicing, subscription lifecycle, and dunning management. Use when implementing subscription billing, automating invoicing, or managing recurring payment systems.
apm install @wshobson/billing-automation
[](https://apm-p1ls2dz87-atlamors-projects.vercel.app/packages/@wshobson/billing-automation)---
name: billing-automation
description: Build automated billing systems for recurring payments, invoicing, subscription lifecycle, and dunning management. Use when implementing subscription billing, automating invoicing, or managing recurring payment systems.
---
# Billing Automation
Master automated billing systems including recurring billing, invoice generation, dunning management, proration, and tax calculation.
## When to Use This Skill
- Implementing SaaS subscription billing
- Automating invoice generation and delivery
- Managing failed payment recovery (dunning)
- Calculating prorated charges for plan changes
- Handling sales tax, VAT, and GST
- Processing usage-based billing
- Managing billing cycles and renewals
## Core Concepts
### 1. Billing Cycles
**Common Intervals:**
- Monthly (most common for SaaS)
- Annual (discounted long-term)
- Quarterly
- Weekly
- Custom (usage-based, per-seat)
### 2. Subscription States
```
trial → active → past_due → canceled
→ paused → resumed
```
### 3. Dunning Management
Automated process to recover failed payments through:
- Retry schedules
- Customer notifications
- Grace periods
- Account restrictions
### 4. Proration
Adjusting charges when:
- Upgrading/downgrading mid-cycle
- Adding/removing seats
- Changing billing frequency
## Quick Start
```python
from billing import BillingEngine, Subscription
# Initialize billing engine
billing = BillingEngine()
# Create subscription
subscription = billing.create_subscription(
customer_id="cus_123",
plan_id="plan_pro_monthly",
billing_cycle_anchor=datetime.now(),
trial_days=14
)
# Process billing cycle
billing.process_billing_cycle(subscription.id)
```
## Subscription Lifecycle Management
```python
from datetime import datetime, timedelta
from enum import Enum
class SubscriptionStatus(Enum):
TRIAL = "trial"
ACTIVE = "active"
PAST_DUE = "past_due"
CANCELED = "canceled"
PAUSED = "paused"
class Subscription:
def __init__(self, customer_id, plan, billing_cycle_day=None):
self.id = generate_id()
self.customer_id = customer_id
self.plan = plan
self.status = SubscriptionStatus.TRIAL
self.current_period_start = datetime.now()
self.current_period_end = self.current_period_start + timedelta(days=plan.trial_days or 30)
self.billing_cycle_day = billing_cycle_day or self.current_period_start.day
self.trial_end = datetime.now() + timedelta(days=plan.trial_days) if plan.trial_days else None
def start_trial(self, trial_days):
"""Start trial period."""
self.status = SubscriptionStatus.TRIAL
self.trial_end = datetime.now() + timedelta(days=trial_days)
self.current_period_end = self.trial_end
def activate(self):
"""Activate subscription after trial or immediately."""
self.status = SubscriptionStatus.ACTIVE
self.current_period_start = datetime.now()
self.current_period_end = self.calculate_next_billing_date()
def mark_past_due(self):
"""Mark subscription as past due after failed payment."""
self.status = SubscriptionStatus.PAST_DUE
# Trigger dunning workflow
def cancel(self, at_period_end=True):
"""Cancel subscription."""
if at_period_end:
self.cancel_at_period_end = True
# Will cancel when current period ends
else:
self.status = SubscriptionStatus.CANCELED
self.canceled_at = datetime.now()
def calculate_next_billing_date(self):
"""Calculate next billing date based on interval."""
if self.plan.interval == 'month':
return self.current_period_start + timedelta(days=30)
elif self.plan.interval == 'year':
return self.current_period_start + timedelta(days=365)
elif self.plan.interval == 'week':
return self.current_period_start + timedelta(days=7)
```
## Billing Cycle Processing
```python
class BillingEngine:
def process_billing_cycle(self, subscription_id):
"""Process billing for a subscription."""
subscription = self.get_subscription(subscription_id)
# Check if billing is due
if datetime.now() < subscription.current_period_end:
return
# Generate invoice
invoice = self.generate_invoice(subscription)
# Attempt payment
payment_result = self.charge_customer(
subscription.customer_id,
invoice.total
)
if payment_result.success:
# Payment successful
invoice.mark_paid()
subscription.advance_billing_period()
self.send_invoice(invoice)
else:
# Payment failed
subscription.mark_past_due()
self.start_dunning_process(subscription, invoice)
def generate_invoice(self, subscription):
"""Generate invoice for billing period."""
invoice = Invoice(
customer_id=subscription.customer_id,
subscription_id=subscription.id,
period_start=subscription.current_period_start,
period_end=subscription.current_period_end
)
# Add subscription line item
invoice.add_line_item(
description=subscription.plan.name,
amount=subscription.plan.amount,
quantity=subscription.quantity or 1
)
# Add usage-based charges if applicable
if subscription.has_usage_billing:
usage_charges = self.calculate_usage_charges(subscription)
invoice.add_line_item(
description="Usage charges",
amount=usage_charges
)
# Calculate tax
tax = self.calculate_tax(invoice.subtotal, subscription.customer)
invoice.tax = tax
invoice.finalize()
return invoice
def charge_customer(self, customer_id, amount):
"""Charge customer using saved payment method."""
customer = self.get_customer(customer_id)
try:
# Charge using payment processor
charge = stripe.Charge.create(
customer=customer.stripe_id,
amount=int(amount * 100), # Convert to cents
currency='usd'
)
return PaymentResult(success=True, transaction_id=charge.id)
except stripe.error.CardError as e:
return PaymentResult(success=False, error=str(e))
```
## Dunning Management
```python
class DunningManager:
"""Manage failed payment recovery."""
def __init__(self):
self.retry_schedule = [
{'days': 3, 'email_template': 'payment_failed_first'},
{'days': 7, 'email_template': 'payment_failed_reminder'},
{'days': 14, 'email_template': 'payment_failed_final'}
]
def start_dunning_process(self, subscription, invoice):
"""Start dunning process for failed payment."""
dunning_attempt = DunningAttempt(
subscription_id=subscription.id,
invoice_id=invoice.id,
attempt_number=1,
next_retry=datetime.now() + timedelta(days=3)
)
# Send initial failure notification
self.send_dunning_email(subscription, 'payment_failed_first')
# Schedule retries
self.schedule_retries(dunning_attempt)
def retry_payment(self, dunning_attempt):
"""Retry failed payment."""
subscription = self.get_subscription(dunning_attempt.subscription_id)
invoice = self.get_invoice(dunning_attempt.invoice_id)
# Attempt payment again
result = self.charge_customer(subscription.customer_id, invoice.total)
if result.success:
# Payment succeeded
invoice.mark_paid()
subscription.status = SubscriptionStatus.ACTIVE
self.send_dunning_email(subscription, 'payment_recovered')
dunning_attempt.mark_resolved()
else:
# Still failing
dunning_attempt.attempt_number += 1
if dunning_attempt.attempt_number < len(self.retry_schedule):
# Schedule next retry
next_retry_config = self.retry_schedule[dunning_attempt.attempt_number]
dunning_attempt.next_retry = datetime.now() + timedelta(days=next_retry_config['days'])
self.send_dunning_email(subscription, next_retry_config['email_template'])
else:
# Exhausted retries, cancel subscription
subscription.cancel(at_period_end=False)
self.send_dunning_email(subscription, 'subscription_canceled')
def send_dunning_email(self, subscription, template):
"""Send dunning notification to customer."""
customer = self.get_customer(subscription.customer_id)
email_content = self.render_template(template, {
'customer_name': customer.name,
'amount_due': subscription.plan.amount,
'update_payment_url': f"https://app.example.com/billing"
})
send_email(
to=customer.email,
subject=email_content['subject'],
body=email_content['body']
)
```
## Proration
```python
class ProrationCalculator:
"""Calculate prorated charges for plan changes."""
@staticmethod
def calculate_proration(old_plan, new_plan, period_start, period_end, change_date):
"""Calculate proration for plan change."""
# Days in current period
total_days = (period_end - period_start).days
# Days used on old plan
days_used = (change_date - period_start).days
# Days remaining on new plan
days_remaining = (period_end - change_date).days
# Calculate prorated amounts
unused_amount = (old_plan.amount / total_days) * days_remaining
new_plan_amount = (new_plan.amount / total_days) * days_remaining
# Net charge/credit
proration = new_plan_amount - unused_amount
return {
'old_plan_credit': -unused_amount,
'new_plan_charge': new_plan_amount,
'net_proration': proration,
'days_used': days_used,
'days_remaining': days_remaining
}
@staticmethod
def calculate_seat_proration(current_seats, new_seats, price_per_seat, period_start, period_end, change_date):
"""Calculate proration for seat changes."""
total_days = (period_end - period_start).days
days_remaining = (period_end - change_date).days
# Additional seats charge
additional_seats = new_seats - current_seats
prorated_amount = (additional_seats * price_per_seat / total_days) * days_remaining
return {
'additional_seats': additional_seats,
'prorated_charge': max(0, prorated_amount), # No refund for removing seats mid-cycle
'effective_date': change_date
}
```
## Tax Calculation
```python
class TaxCalculator:
"""Calculate sales tax, VAT, GST."""
def __init__(self):
# Tax rates by region
self.tax_rates = {
'US_CA': 0.0725, # California sales tax
'US_NY': 0.04, # New York sales tax
'GB': 0.20, # UK VAT
'DE': 0.19, # Germany VAT
'FR': 0.20, # France VAT
'AU': 0.10, # Australia GST
}
def calculate_tax(self, amount, customer):
"""Calculate applicable tax."""
# Determine tax jurisdiction
jurisdiction = self.get_tax_jurisdiction(customer)
if not jurisdiction:
return 0
# Get tax rate
tax_rate = self.tax_rates.get(jurisdiction, 0)
# Calculate tax
tax = amount * tax_rate
return {
'tax_amount': tax,
'tax_rate': tax_rate,
'jurisdiction': jurisdiction,
'tax_type': self.get_tax_type(jurisdiction)
}
def get_tax_jurisdiction(self, customer):
"""Determine tax jurisdiction based on customer location."""
if customer.country == 'US':
# US: Tax based on customer state
return f"US_{customer.state}"
elif customer.country in ['GB', 'DE', 'FR']:
# EU: VAT
return customer.country
elif customer.country == 'AU':
# Australia: GST
return 'AU'
else:
return None
def get_tax_type(self, jurisdiction):
"""Get type of tax for jurisdiction."""
if jurisdiction.startswith('US_'):
return 'Sales Tax'
elif jurisdiction in ['GB', 'DE', 'FR']:
return 'VAT'
elif jurisdiction == 'AU':
return 'GST'
return 'Tax'
def validate_vat_number(self, vat_number, country):
"""Validate EU VAT number."""
# Use VIES API for validation
# Returns True if valid, False otherwise
pass
```
## Invoice Generation
```python
class Invoice:
def __init__(self, customer_id, subscription_id=None):
self.id = generate_invoice_number()
self.customer_id = customer_id
self.subscription_id = subscription_id
self.status = 'draft'
self.line_items = []
self.subtotal = 0
self.tax = 0
self.total = 0
self.created_at = datetime.now()
def add_line_item(self, description, amount, quantity=1):
"""Add line item to invoice."""
line_item = {
'description': description,
'unit_amount': amount,
'quantity': quantity,
'total': amount * quantity
}
self.line_items.append(line_item)
self.subtotal += line_item['total']
def finalize(self):
"""Finalize invoice and calculate total."""
self.total = self.subtotal + self.tax
self.status = 'open'
self.finalized_at = datetime.now()
def mark_paid(self):
"""Mark invoice as paid."""
self.status = 'paid'
self.paid_at = datetime.now()
def to_pdf(self):
"""Generate PDF invoice."""
from reportlab.pdfgen import canvas
# Generate PDF
# Include: company info, customer info, line items, tax, total
pass
def to_html(self):
"""Generate HTML invoice."""
template = """
<!DOCTYPE html>
<html>
<head><title>Invoice #{invoice_number}</title></head>
<body>
<h1>Invoice #{invoice_number}</h1>
<p>Date: {date}</p>
<h2>Bill To:</h2>
<p>{customer_name}<br>{customer_address}</p>
<table>
<tr><th>Description</th><th>Quantity</th><th>Amount</th></tr>
{line_items}
</table>
<p>Subtotal: ${subtotal}</p>
<p>Tax: ${tax}</p>
<h3>Total: ${total}</h3>
</body>
</html>
"""
return template.format(
invoice_number=self.id,
date=self.created_at.strftime('%Y-%m-%d'),
customer_name=self.customer.name,
customer_address=self.customer.address,
line_items=self.render_line_items(),
subtotal=self.subtotal,
tax=self.tax,
total=self.total
)
```
## Usage-Based Billing
```python
class UsageBillingEngine:
"""Track and bill for usage."""
def track_usage(self, customer_id, metric, quantity):
"""Track usage event."""
UsageRecord.create(
customer_id=customer_id,
metric=metric,
quantity=quantity,
timestamp=datetime.now()
)
def calculate_usage_charges(self, subscription, period_start, period_end):
"""Calculate charges for usage in billing period."""
usage_records = UsageRecord.get_for_period(
subscription.customer_id,
period_start,
period_end
)
total_usage = sum(record.quantity for record in usage_records)
# Tiered pricing
if subscription.plan.pricing_model == 'tiered':
charge = self.calculate_tiered_pricing(total_usage, subscription.plan.tiers)
# Per-unit pricing
elif subscription.plan.pricing_model == 'per_unit':
charge = total_usage * subscription.plan.unit_price
# Volume pricing
elif subscription.plan.pricing_model == 'volume':
charge = self.calculate_volume_pricing(total_usage, subscription.plan.tiers)
return charge
def calculate_tiered_pricing(self, total_usage, tiers):
"""Calculate cost using tiered pricing."""
charge = 0
remaining = total_usage
for tier in sorted(tiers, key=lambda x: x['up_to']):
tier_usage = min(remaining, tier['up_to'] - tier['from'])
charge += tier_usage * tier['unit_price']
remaining -= tier_usage
if remaining <= 0:
break
return charge
```