billing-automation · diff
git:20260307.47a5dbc to git:20260522.be57c0b
2 added, 460 removed. Audit A to A.
---
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
+ ## Detailed patterns and worked examples
- if remaining <= 0:
- break
+ Detailed pattern documentation lives in `references/details.md`. Read that file when the navigation tier above is insufficient.
- return charge
- ```