Subscription Billing Systems: Recurring Payments & Dunning

# Database models for subscription billing
from sqlalchemy import Column, Integer, String, DateTime, Numeric, Boolean, ForeignKey, Enum as SQLEnum
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from datetime import datetime
from enum import Enum

Base = declarative_base()

class BillingInterval(Enum):
    DAILY = "daily"
    WEEKLY = "weekly"
    MONTHLY = "monthly"
    YEARLY = "yearly"

class SubscriptionStatus(Enum):
    ACTIVE = "active"
    PAST_DUE = "past_due"
    CANCELED = "canceled"
    UNPAID = "unpaid"
    TRIALING = "trialing"

class Customer(Base):
    __tablename__ = "customers"
    
    id = Column(Integer, primary_key=True)
    email = Column(String(255), unique=True, nullable=False)
    stripe_customer_id = Column(String(255), unique=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    subscriptions = relationship("Subscription", back_populates="customer")
    invoices = relationship("Invoice", back_populates="customer")
    payment_methods = relationship("PaymentMethod", back_populates="customer(Base):
    __")

class Subscriptiontablename__ = "subscriptions"
    
    id = Column(Integer, primary_key=True)
    customer_id = Column(Integer, ForeignKey("customers.id"), nullable=False)
    plan_id = Column(Integer, ForeignKey("plans.id"), nullable=False)
    
    status = Column(SQLEnum(SubscriptionStatus), default=SubscriptionStatus.ACTIVE)
    
    # Billing dates
    current_period_start = Column(DateTime, nullable=False)
    current_period_end = Column(DateTime, nullable=False)
    trial_start = Column(DateTime)
    trial_end = Column(DateTime)
    canceled_at = Column(DateTime)
    cancel_at_period_end = Column(Boolean, default=False)
    
    # Usage tracking
    usage_records = relationship("UsageRecord", back_populates="subscription")
    invoices = relationship("Invoice", back_populates="subscription")
    
    customer = relationship("Customer", back_populates="subscriptions")
    plan = relationship("Plan")

class Plan(Base):
    __tablename__ = "plans"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    stripe_price_id = Column(String(255), unique=True)
    
    # Pricing
    amount = Column(Numeric(10, 2), nullable=False)
    currency = Column(String(3), default="USD")
    billing_interval = Column(SQLEnum(BillingInterval), default=BillingInterval.MONTHLY)
    
    # Features
    is_active = Column(Boolean, default=True)
    metadata = Column(String(1000))  # JSON string for extra features
    
    subscriptions = relationship("Subscription", back_populates="plan")

class Invoice(Base):
    __tablename__ = "invoices"
    
    id = Column(Integer, primary_key=True)
    customer_id = Column(Integer, ForeignKey("customers.id"), nullable=False)
    subscription_id = Column(Integer, ForeignKey("subscriptions.id"))
    
    invoice_number = Column(String(50), unique=True)
    
    # Amounts
    subtotal = Column(Numeric(10, 2), nullable=False)
    tax = Column(Numeric(10, 2), default=0)
    total = Column(Numeric(10, 2), nullable=False)
    amount_paid = Column(Numeric(10, 2), default=0)
    amount_due = Column(Numeric(10, 2), nullable=False)
    
    # Status
    status = Column(String(50), default="draft")  # draft, open, paid, void
    stripe_invoice_id = Column(String(255))
    
    # Dates
    invoice_date = Column(DateTime, nullable=False)
    due_date = Column(DateTime)
    paid_at = Column(DateTime)
    
    line_items = relationship("InvoiceLineItem", back_populates="invoice")
    customer = relationship("Customer", back_populates="invoices")
    subscription = relationship("Subscription", back_populates="invoices")

class InvoiceLineItem(Base):
    __tablename__ = "invoice_line_items"
    
    id = Column(Integer, primary_key=True)
    invoice_id = Column(Integer, ForeignKey("invoices.id"), nullable=False)
    
    description = Column(String(500))
    quantity = Column(Integer, default=1)
    unit_amount = Column(Numeric(10, 2))
    amount = Column(Numeric(10, 2), nullable=False)
    
    invoice = relationship("Invoice", back_populates="line_items")

class UsageRecord(Base):
    __tablename__ = "usage_records"
    
    id = Column(Integer, primary_key=True)
    subscription_id = Column(Integer, ForeignKey("subscriptions.id"), nullable=False)
    
    feature_name = Column(String(100), nullable=False)
    quantity = Column(Integer, default=0)
    period_start = Column(DateTime, nullable=False)
    period_end = Column(DateTime, nullable=False)
    timestamp = Column(DateTime, default=datetime.utcnow)
    
    subscription = relationship("Subscription", back_populates="usage_records")

# Stripe subscription management
import stripe
from datetime import datetime, timedelta

stripe.api_key = "sk_test_..."

class StripeBilling:
    def __init__(self):
        self.webhook_events = {
            'customer.subscription.created': self.handle_subscription_created,
            'customer.subscription.updated': self.handle_subscription_updated,
            'customer.subscription.deleted': self.handle_subscription_deleted,
            'invoice.paid': self.handle_invoice_paid,
            'invoice.payment_failed': self.handle_payment_failed,
            'invoice.payment_succeeded': self.handle_payment_succeeded,
        }
    
    def create_customer(self, email, name, metadata=None):
        """Create Stripe customer"""
        
        customer = stripe.Customer.create(
            email=email,
            name=name,
            metadata=metadata or {},
            stripe_version='2023-10-16'
        )
        
        return customer
    
    def create_subscription(self, customer_id, price_id, trial_days=None):
        """Create a new subscription"""
        
        subscription_data = {
            'customer': customer_id,
            'items': [{'price': price_id}],
            'payment_behavior': 'default_incomplete',
            'expand': ['latest_invoice.payment_intent'],
        }
        
        if trial_days:
            subscription_data['trial_period_days'] = trial_days
        
        subscription = stripe.Subscription.create(**subscription_data)
        
        return {
            'subscription_id': subscription.id,
            'status': subscription.status,
            'client_secret': subscription.latest_invoice.payment_intent.client_secret
        }
    
    def cancel_subscription(self, subscription_id, at_period_end=True):
        """Cancel subscription"""
        
        if at_period_end:
            subscription = stripe.Subscription.modify(
                subscription_id,
                cancel_at_period_end=True
            )
        else:
            subscription = stripe.Subscription.cancel(subscription_id)
        
        return subscription
    
    def change_plan(self, subscription_id, new_price_id):
        """Change subscription plan"""
        
        subscription = stripe.Subscription.retrieve(subscription_id)
        
        # Update subscription items
        updated_subscription = stripe.Subscription.modify(
            subscription_id,
            items=[{
                'id': subscription['items']['data'][0].id,
                'price': new_price_id,
            }],
            proration_behavior='always_invoice'
        )
        
        return updated_subscription
    
    def create_usage_record(self, subscription_item_id, quantity, timestamp=None):
        """Report usage for metered billing"""
        
        record = stripe.SubscriptionItem.create_usage_record(
            subscription_item_id,
            quantity=quantity,
            timestamp=int(timestamp.timestamp()) if timestamp else None,
            action='increment'
        )
        
        return record
    
    def get_usage_summary(self, subscription_item_id):
        """Get usage summary for billing period"""
        
        usage_records = stripe.SubscriptionItem.list_usage_record_summaries(
            subscription_item_id
        )
        
        return {
            'total_usage': sum(r.quantity for r in usage_records.data),
            'period_start': usage_records.data[0].period.start if usage_records.data else None,
            'period_end': usage_records.data[0].period.end if usage_records.data else None,
        }
    
    def handle_webhook(self, payload, signature):
        """Handle Stripe webhook"""
        
        webhook_secret = "whsec_..."
        
        try:
            event = stripe.Webhook.construct_event(
                payload, signature, webhook_secret
            )
            
            handler = self.webhook_events.get(event.type)
            if handler:
                handler(event.data.object)
            
            return {'status': 'success'}
            
        except stripe.error.SignatureVerificationError:
            return {'status': 'error', 'message': 'Invalid signature'}
    
    def handle_payment_failed(self, invoice):
        """Handle failed payment"""
        # Send notification, update status, etc.
        pass

# Usage tracking and billing

from datetime import datetime
from collections import defaultdict
import threading

class UsageTracker:
    def __init__(self):
        self.usage = defaultdict(lambda: defaultdict(int))
        self.lock = threading.Lock()
    
    def record_usage(self, customer_id, feature, quantity=1):
        """Record usage event"""
        
        period_key = self._get_current_period_key()
        
        with self.lock:
            self.usage[customer_id][f"{feature}:{period_key}"] += quantity
    
    def get_usage(self, customer_id, feature, period_start, period_end):
        """Get usage for a specific period"""
        
        total = 0
        period_key = self._get_period_key(period_start)
        
        with self.lock:
            total = self.usage[customer_id].get(f"{feature}:{period_key}", 0)
        
        return total
    
    def get_usage_by_plan(self, customer_id, plan_features):
        """Get usage breakdown by feature with plan limits"""
        
        usage_summary = []
        
        for feature, limit in plan_features.items():
            current_usage = self.get_usage(
                customer_id,
                feature,
                datetime.utcnow().replace(day=1),
                datetime.utcnow()
            )
            
            usage_summary.append({
                'feature': feature,
                'usage': current_usage,
                'limit': limit,
                'percent_used': (current_usage / limit * 100) if limit > 0 else 0,
                'remaining': max(0, limit - current_usage),
                'overage': max(0, current_usage - limit)
            })
        
        return usage_summary
    
    def reset_usage(self, customer_id):
        """Reset usage at billing period end"""
        
        with self.lock:
            self.usage[customer_id].clear()
    
    def _get_current_period_key(self):
        """Get current billing period key (monthly)"""
        now = datetime.utcnow()
        return f"{now.year}-{now.month:02d}"
    
    def _get_period_key(self, date):
        return f"{date.year}-{date.month:02d}"

# Dunning management system
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import List, Optional
import smtplib
from email.mime.text import MIMEText

@dataclass
class DunningConfig:
    max_retries: int = 3
    retry_days: List[int] = None  # Days between retries
    
    def __post_init__(self):
        if self.retry_days is None:
            self.retry_days = [3, 3, 7]  # 3 retries at days 3, 6, 13

@dataclass
class FailedPayment:
    customer_id: int
    subscription_id: str
    invoice_id: str
    amount: float
    currency: str
    failure_reason: str
    last_attempt: datetime
    retry_count: int = 0
    status: str = "pending"

class DunningManager:
    def __init__(self, config: DunningConfig, db):
        self.config = config
        self.db = db
        self.failed_payments: List[FailedPayment] = []
    
    def handle_payment_failure(self, customer_id, subscription_id, 
                              invoice_id, amount, currency, reason):
        """Handle a failed payment"""
        
        failed_payment = FailedPayment(
            customer_id=customer_id,
            subscription_id=subscription_id,
            invoice_id=invoice_id,
            amount=amount,
            currency=currency,
            failure_reason=reason,
            last_attempt=datetime.utcnow()
        )
        
        self.failed_payments.append(failed_payment)
        
        # Update subscription status
        self._update_subscription_status(subscription_id, "past_due")
        
        # Send first notification
        self._send_dunning_email(failed_payment, step=0)
        
        # Schedule retry
        self._schedule_retry(failed_payment)
    
    def process_retries(self):
        """Process pending retries"""
        
        now = datetime.utcnow()
        to_retry = []
        
        for payment in self.failed_payments:
            if payment.status != "pending":
                continue
            
            days_since_failure = (now - payment.last_attempt).days
            
            if payment.retry_count < len(self.config.retry_days):
                retry_day = self.config.retry_days[payment.retry_count]
                
                if days_since_failure >= retry_day:
                    to_retry.append(payment)
        
        for payment in to_retry:
            self._retry_payment(payment)
    
    def _retry_payment(self, payment: FailedPayment):
        """Retry a failed payment"""
        
        print(f"Retrying payment {payment.invoice_id} "
              f"(attempt {payment.retry_count + 1})")
        
        try:
            # Retry via payment provider
            success = self._charge_customer(
                payment.customer_id,
                payment.amount,
                payment.currency
            )
            
            if success:
                payment.status = "resolved"
                self._update_subscription_status(
                    payment.subscription_id, "active"
                )
                self._send_recovery_email(payment)
            else:
                payment.retry_count += 1
                payment.last_attempt = datetime.utcnow()
                
                if payment.retry_count >= self.config.max_retries:
                    payment.status = "exhausted"
                    self._handle_max_retries(payment)
                else:
                    # Schedule next retry
                    self._send_dunning_email(payment, step=payment.retry_count)
                    self._schedule_retry(payment)
                    
        except Exception as e:
            print(f"Retry failed: {e}")
            payment.retry_count += 1
    
    def _handle_max_retries(self, payment: FailedPayment):
        """Handle when all retries exhausted"""
        
        # Suspend account
        self._update_subscription_status(payment.subscription_id, "unpaid")
        self._suspend_account(payment.customer_id)
        
        # Send final warning
        self._send_final_warning_email(payment)
        
        # Schedule cancellation
        self._schedule_cancellation(payment)
    
    def _charge_customer(self, customer_id, amount, currency):
        """Charge customer via payment provider"""
        # Implementation with Stripe or other provider
        pass
    
    def _update_subscription_status(self, subscription_id, status):
        """Update subscription in database"""
        pass
    
    def _suspend_account(self, customer_id):
        """Suspend customer account"""
        pass
    
    def _schedule_retry(self, payment):
        """Schedule payment retry"""
        # Could use Celery, RQ, or similar
        pass
    
    def _schedule_cancellation(self, payment):
        """Schedule account cancellation"""
        # Cancel after grace period
        pass
    
    def _send_dunning_email(self, payment: FailedPayment, step: int):
        """Send dunning email"""
        
        templates = {
            0: "payment_failed_first",
            1: "payment_failed_second",
            2: "payment_failed_final"
        }
        
        template = templates.get(step, "payment_failed_generic")
        
        print(f"Sending dunning email: {template} to customer {payment.customer_id}")
    
    def _send_recovery_email(self, payment: FailedPayment):
        """Send payment recovered email"""
        print(f"Payment recovered for customer {payment.customer_id}")
    
    def _send_final_warning_email(self, payment: FailedPayment):
        """Send final warning before cancellation"""
        print(f"Final warning to customer {payment.customer_id}")

# Invoice generation with line items

from datetime import datetime
from decimal import Decimal

class InvoiceBuilder:
    def __init__(self, tax_rate=0.0):
        self.tax_rate = Decimal(str(tax_rate))
        self.line_items = []
    
    def add_subscription_line(self, plan_name, quantity, unit_price, 
                              start_date, end_date):
        """Add subscription line item"""
        
        days = (end_date - start_date).days
        
        # Prorate for partial months
        prorate_factor = Decimal(days) / Decimal(30)
        
        line_total = Decimal(str(unit_price)) * quantity * prorate_factor
        
        self.line_items.append({
            'type': 'subscription',
            'description': f"{plan_name} ({start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')})",
            'quantity': quantity,
            'unit_price': Decimal(str(unit_price)),
            'prorate_factor': prorate_factor,
            'subtotal': line_total
        })
        
        return self
    
    def add_usage_line(self, feature, quantity, unit_price):
        """Add usage-based line item"""
        
        line_total = Decimal(str(unit_price)) * Decimal(quantity)
        
        self.line_items.append({
            'type': 'usage',
            'description': f"{feature} - {quantity} units",
            'quantity': quantity,
            'unit_price': Decimal(str(unit_price)),
            'subtotal': line_total
        })
        
        return self
    
    def add_discount_line(self, description, amount):
        """Add discount line item"""
        
        self.line_items.append({
            'type': 'discount',
            'description': description,
            'quantity': 1,
            'unit_price': -Decimal(str(amount)),
            'subtotal': -Decimal(str(amount))
        })
        
        return self
    
    def add_tax_line(self, description="Sales Tax"):
        """Add tax line item"""
        
        subtotal = self._get_subtotal()
        tax_amount = subtotal * self.tax_rate
        
        if tax_amount > 0:
            self.line_items.append({
                'type': 'tax',
                'description': description,
                'quantity': 1,
                'unit_price': tax_amount,
                'subtotal': tax_amount
            })
        
        return self
    
    def _get_subtotal(self):
        """Calculate subtotal"""
        return sum(item['subtotal'] for item in self.line_items 
                   if item['type'] != 'tax')
    
    def build(self, invoice_number, customer, due_date=None):
        """Build final invoice"""
        
        subtotal = self._get_subtotal()
        total = subtotal + sum(item['subtotal'] for item in self.line_items 
                               if item['type'] == 'tax')
        
        return {
            'invoice_number': invoice_number,
            'customer': customer,
            'invoice_date': datetime.utcnow(),
            'due_date': due_date,
            'line_items': self.line_items,
            'subtotal': subtotal,
            'tax': subtotal * self.tax_rate,
            'total': total,
            'currency': 'USD'
        }

# Usage
invoice = (InvoiceBuilder(tax_rate=0.08)
    .add_subscription_line(
        plan_name="Pro Plan",
        quantity=1,
        unit_price=29.00,
        start_date=datetime(2026, 3, 15),
        end_date=datetime(2026, 3, 31)
    )
    .add_usage_line(
        feature="Additional API Calls",
        quantity=15000,
        unit_price=0.001
    )
    .add_discount_line(
        description="New Customer Discount (10%)",
        amount=2.90
    )
    .add_tax_line()
    .build(
        invoice_number="INV-2026-001",
        customer={"id": 1, "name": "Acme Corp"}
    ))

print(invoice)

# Intelligent payment retry scheduler

from datetime import datetime, timedelta
from typing import List
import random

class RetryScheduler:
    """
    Intelligent retry scheduling based on payment failure patterns
    """
    
    # Retry schedule: (delay_hours, probability_weight)
    BASE_SCHEDULE = [
        (4, 10),    # 4 hours: 10% of failures may work
        (12, 20),   # 12 hours: 20%
        (24, 30),   # 24 hours: 30%
        (48, 25),   # 48 hours: 25%
        (72, 15),   # 72 hours: 15%
    ]
    
    # Additional factors
    RETRY_FACTORS = {
        'high_value': 1.5,  # Retry more often for high value
        'new_customer': 0.8,  # Less aggressive for new customers
        'previous_failures': 1.2,  # More aggressive if multiple failures
        'card_expiring_soon': 1.3,  # More aggressive if card expiring
        'good_payment_history': 0.9,  # Less aggressive for good history
    }
    
    def __init__(self, db):
        self.db = db
    
    def calculate_retry_schedule(self, customer_id, invoice_amount):
        """Calculate optimal retry schedule"""
        
        customer = self._get_customer(customer_id)
        
        # Get customer factors
        factor = 1.0
        
        if invoice_amount > 500:
            factor *= self.RETRY_FACTORS['high_value']
        
        if self._is_new_customer(customer):
            factor *= self.RETRY_FACTORS['new_customer']
        
        failure_count = self._get_failure_count(customer_id)
        if failure_count > 2:
            factor *= self.RETRY_FACTORS['previous_failures']
        
        if self._card_expiring_soon(customer):
            factor *= self.RETRY_FACTORS['card_expiring_soon']
        
        payment_history = self._get_payment_success_rate(customer_id)
        if payment_history > 0.95:
            factor *= self.RETRY_FACTORS['good_payment_history']
        
        # Adjust schedule by factor
        schedule = []
        cumulative = 0
        
        for hours, weight in self.BASE_SCHEDULE:
            adjusted_hours = int(hours / factor)
            adjusted_hours = max(2, adjusted_hours)  # Minimum 2 hours
            
            cumulative += adjusted_hours
            schedule.append(datetime.utcnow() + timedelta(hours=cumulative))
        
        return schedule
    
    def _get_customer(self, customer_id):
        pass
    
    def _is_new_customer(self, customer):
        return (datetime.utcnow() - customer['created_at']).days < 90
    
    def _get_failure_count(self, customer_id):
        return 0
    
    def _card_expiring_soon(self, customer):
        return False
    
    def _get_payment_success_rate(self, customer_id):
        return 0.9

# Key subscription metrics

import pandas as pd
from datetime import datetime, timedelta

class SubscriptionAnalytics:
    def __init__(self, db):
        self.db = db
    
    def calculate_mrr(self):
        """Calculate Monthly Recurring Revenue"""
        
        subscriptions = self.db.get_active_subscriptions()
        
        mrr = 0
        for sub in subscriptions:
            if sub.billing_interval == 'monthly':
                mrr += sub.amount
            elif sub.billing_interval == 'yearly':
                mrr += sub.amount / 12
            elif sub.billing_interval == 'daily':
                mrr += sub.amount * 30
        
        return mrr
    
    def calculate_arr(self):
        """Calculate Annual Recurring Revenue"""
        return self.calculate_mrr() * 12
    
    def calculate_churn_rate(self, period_days=30):
        """Calculate customer churn rate"""
        
        period_start = datetime.utcnow() - timedelta(days=period_days)
        
        # Customers at start
        customers_start = self.db.count_customers(before=period_start)
        
        # Churned customers in period
        churned = self.db.count_churned(period_start, datetime.utcnow())
        
        if customers_start == 0:
            return 0
        
        return (churned / customers_start) * 100
    
    def calculate_net_revenue_retention(self, period_days=30):
        """
        NRR = (MRR at end + Expansion - Churn - Contraction) / MRR at start
        """
        
        period_start = datetime.utcnow() - timedelta(days=period_days)
        
        mrr_start = self.db.get_mrr(period_start)
        mrr_end = self.db.get_mrr(datetime.utcnow())
        
        expansion = self.db.get_expansion_revenue(period_start, datetime.utcnow())
        contraction = self.db.get_contraction_revenue(period_start, datetime.utcnow())
        churn = self.db.get_churn_revenue(period_start, datetime.utcnow())
        
        if mrr_start == 0:
            return 0
        
        nrr = ((mrr_end + expansion) - churn - contraction) / mrr_start * 100
        
        return nrr
    
    def calculate_ltv(self, arpu, churn_rate):
        """
        LTV = ARPU / Churn Rate
        """
        
        if churn_rate == 0:
            return float('inf')
        
        return arpu / (churn_rate / 100)
    
    def get_cohort_analysis(self):
        """Get cohort retention analysis"""
        
        cohorts = self.db.get_cohorts()
        
        retention_matrix = []
        
        for cohort in cohorts:
            months = []
            
            for month_num in range(13):
                retention = self.db.get_retention(
                    cohort['month'], month_num
                )
                months.append(retention)
            
            retention_matrix.append({
                'cohort': cohort['month'],
                'months': months
            })
        
        return retention_matrix

# Essential billing system features

essential:
  - Automated recurring billing
  - Multiple payment methods
  - Invoice generation and delivery
  - Tax calculation
  - Proration for plan changes
  - Webhook handling for payment events
  
dunning:
  - Automatic retry on failure
  - Email notifications
  - Account suspension/termination
  - Grace periods before cancellation
  - Recovery tracking
  
reporting:
  - MRR/ARR tracking
  - Churn analysis
  - Revenue cohort analysis
  - LTV calculation
  - Cash flow forecasting
  
compliance:
  - PCI DSS compliance
  - Tax compliance (Stripe Tax, Avalara)
  - GDPR for customer data
  - Invoice archival (7-10 years)