Invoice & Billing

  1  from __future__ import annotations
  2  from abc import ABC, abstractmethod
  3  from dataclasses import dataclass, field
  4  from datetime import datetime, timedelta
  5  from enum import Enum
  6  from threading import RLock
  7  import uuid
  8  
  9  
 10  @dataclass(frozen=True)
 11  class Money:
 12      """Integer cents. Currency tag prevents accidental USD+EUR addition."""
 13      cents: int
 14      currency: str = "USD"
 15  
 16      def add(self, other: "Money") -> "Money":
 17          if self.currency != other.currency:
 18              raise ValueError(f"currency mismatch: {self.currency} vs {other.currency}")
 19          return Money(self.cents + other.cents, self.currency)
 20  
 21      def subtract(self, other: "Money") -> "Money":
 22          if self.currency != other.currency:
 23              raise ValueError("currency mismatch")
 24          return Money(self.cents - other.cents, self.currency)
 25  
 26      def multiply(self, factor: float) -> "Money":
 27          # Banker's rounding would be ideal; half-up is a decent default.
 28          return Money(int(round(self.cents * factor)), self.currency)
 29  
 30      def is_zero(self) -> bool:
 31          return self.cents == 0
 32  
 33      def __str__(self) -> str:
 34          return f"{self.cents / 100:.2f} {self.currency}"
 35  
 36  
 37  ZERO_USD = Money(0, "USD")
 38  
 39  
 40  class InvoiceStatus(Enum):
 41      DRAFT = "draft"
 42      ISSUED = "issued"
 43      PARTIALLY_PAID = "partially_paid"
 44      PAID = "paid"
 45      VOIDED = "voided"
 46      OVERDUE = "overdue"
 47  
 48  
 49  class PaymentMethod(Enum):
 50      CREDIT_CARD = "credit_card"
 51      BANK_TRANSFER = "bank_transfer"
 52      CHECK = "check"
 53      CREDIT_NOTE = "credit_note"
 54  
 55  
 56  # ---- Strategies ----
 57  
 58  class TaxRule(ABC):
 59      @abstractmethod
 60      def tax_for(self, taxable_amount: Money) -> Money: ...
 61  
 62  
 63  class NoTaxRule(TaxRule):
 64      def tax_for(self, amount: Money) -> Money:
 65          return Money(0, amount.currency)
 66  
 67  
 68  class SalesTaxRule(TaxRule):
 69      def __init__(self, rate: float):
 70          if rate < 0:
 71              raise ValueError("rate must be non-negative")
 72          self._rate = rate
 73  
 74      def tax_for(self, amount: Money) -> Money:
 75          return amount.multiply(self._rate)
 76  
 77  
 78  class CompoundTaxRule(TaxRule):
 79      """Multiple taxes applied sequentially on the previous base+tax (e.g., Quebec QST on GST)."""
 80  
 81      def __init__(self, rules: list[TaxRule]):
 82          self._rules = rules
 83  
 84      def tax_for(self, amount: Money) -> Money:
 85          total_tax = Money(0, amount.currency)
 86          running = amount
 87          for rule in self._rules:
 88              tax = rule.tax_for(running)
 89              total_tax = total_tax.add(tax)
 90              running = running.add(tax)
 91          return total_tax
 92  
 93  
 94  class DiscountRule(ABC):
 95      @abstractmethod
 96      def apply(self, amount: Money) -> Money: ...
 97  
 98  
 99  class NoDiscount(DiscountRule):
100      def apply(self, amount: Money) -> Money:
101          return Money(0, amount.currency)
102  
103  
104  class PercentDiscount(DiscountRule):
105      def __init__(self, percent: float):
106          if not 0 <= percent <= 100:
107              raise ValueError("percent must be 0..100")
108          self._percent = percent
109  
110      def apply(self, amount: Money) -> Money:
111          return amount.multiply(self._percent / 100.0)
112  
113  
114  class FixedDiscount(DiscountRule):
115      def __init__(self, amount: Money):
116          if amount.cents < 0:
117              raise ValueError("discount must be non-negative")
118          self._amount = amount
119  
120      def apply(self, base: Money) -> Money:
121          # Clamp: never discount more than the base.
122          return self._amount if self._amount.cents <= base.cents else base
123  
124  
125  # ---- Line item + invoice ----
126  
127  @dataclass
128  class LineItem:
129      description: str
130      quantity: int
131      unit_price: Money
132      tax_rule: TaxRule = field(default_factory=NoTaxRule)
133      discount: DiscountRule = field(default_factory=NoDiscount)
134  
135      def __post_init__(self):
136          if self.quantity < 1:
137              raise ValueError("quantity must be >= 1")
138          # Note: unit_price can be negative — that's how credit memos and refunds model reversals.
139  
140      def gross(self) -> Money:
141          return self.unit_price.multiply(self.quantity)
142  
143      def discount_amount(self) -> Money:
144          return self.discount.apply(self.gross())
145  
146      def net(self) -> Money:
147          return self.gross().subtract(self.discount_amount())
148  
149      def tax(self) -> Money:
150          return self.tax_rule.tax_for(self.net())
151  
152      def total(self) -> Money:
153          return self.net().add(self.tax())
154  
155  
156  @dataclass
157  class Payment:
158      id: str
159      invoice_id: str
160      amount: Money
161      method: PaymentMethod
162      received_at: datetime
163  
164  
165  class Invoice:
166      def __init__(self, invoice_id: str, customer_id: str, currency: str = "USD",
167                   due_in: timedelta = timedelta(days=30)):
168          self.id = invoice_id
169          self.customer_id = customer_id
170          self.currency = currency
171          self._items: list[LineItem] = []
172          self._status = InvoiceStatus.DRAFT
173          self._issued_at: datetime | None = None
174          self._due_at: datetime | None = None
175          self._due_in = due_in
176          self._subtotal = Money(0, currency)
177          self._tax_total = Money(0, currency)
178          self._total = Money(0, currency)
179          self._payments: list[Payment] = []
180          self._lock = RLock()
181  
182      def add_item(self, item: LineItem) -> None:
183          with self._lock:
184              if self._status != InvoiceStatus.DRAFT:
185                  raise RuntimeError("cannot modify an issued invoice")
186              if item.unit_price.currency != self.currency:
187                  raise ValueError("line item currency must match invoice")
188              self._items.append(item)
189  
190      def finalize(self) -> None:
191          """Freeze the invoice — compute totals once and mark ISSUED."""
192          with self._lock:
193              if self._status != InvoiceStatus.DRAFT:
194                  raise RuntimeError("already finalized")
195              if not self._items:
196                  raise RuntimeError("invoice has no line items")
197              subtotal = Money(0, self.currency)
198              tax_total = Money(0, self.currency)
199              for item in self._items:
200                  subtotal = subtotal.add(item.net())
201                  tax_total = tax_total.add(item.tax())
202              self._subtotal = subtotal
203              self._tax_total = tax_total
204              self._total = subtotal.add(tax_total)
205              self._status = InvoiceStatus.ISSUED
206              self._issued_at = datetime.utcnow()
207              self._due_at = self._issued_at + self._due_in
208  
209      def apply_payment(self, amount: Money, method: PaymentMethod) -> Payment:
210          with self._lock:
211              if self._status not in (InvoiceStatus.ISSUED, InvoiceStatus.PARTIALLY_PAID, InvoiceStatus.OVERDUE):
212                  raise RuntimeError(f"cannot apply payment to a {self._status.value} invoice")
213              if amount.currency != self.currency:
214                  raise ValueError("payment currency must match invoice")
215              if amount.cents <= 0:
216                  raise ValueError("payment must be positive")
217              remaining = self.amount_due()
218              if amount.cents > remaining.cents:
219                  raise ValueError(f"payment exceeds amount due ({remaining})")
220  
221              payment = Payment(
222                  id=str(uuid.uuid4())[:8],
223                  invoice_id=self.id,
224                  amount=amount, method=method,
225                  received_at=datetime.utcnow(),
226              )
227              self._payments.append(payment)
228  
229              if self.amount_paid().cents >= self._total.cents:
230                  self._status = InvoiceStatus.PAID
231              else:
232                  self._status = InvoiceStatus.PARTIALLY_PAID
233              return payment
234  
235      def void(self) -> None:
236          with self._lock:
237              if self._status == InvoiceStatus.PAID:
238                  raise RuntimeError("cannot void a paid invoice — issue a credit memo instead")
239              if self._status == InvoiceStatus.VOIDED:
240                  return
241              self._status = InvoiceStatus.VOIDED
242  
243      def check_overdue(self, now: datetime | None = None) -> bool:
244          """Transition to OVERDUE if past due and still unpaid. Returns True if changed."""
245          with self._lock:
246              if self._status not in (InvoiceStatus.ISSUED, InvoiceStatus.PARTIALLY_PAID):
247                  return False
248              if self._due_at is None:
249                  return False
250              check_at = now or datetime.utcnow()
251              if check_at >= self._due_at:
252                  self._status = InvoiceStatus.OVERDUE
253                  return True
254              return False
255  
256      def amount_paid(self) -> Money:
257          total = Money(0, self.currency)
258          for p in self._payments:
259              total = total.add(p.amount)
260          return total
261  
262      def amount_due(self) -> Money:
263          return self._total.subtract(self.amount_paid())
264  
265      @property
266      def status(self) -> InvoiceStatus: return self._status
267      @property
268      def subtotal(self) -> Money: return self._subtotal
269      @property
270      def tax_total(self) -> Money: return self._tax_total
271      @property
272      def total(self) -> Money: return self._total
273  
274  
275  class BillingService:
276      def __init__(self):
277          self._invoices: dict[str, Invoice] = {}
278  
279      def create_invoice(self, customer_id: str, currency: str = "USD") -> Invoice:
280          inv = Invoice(invoice_id=str(uuid.uuid4())[:8], customer_id=customer_id, currency=currency)
281          self._invoices[inv.id] = inv
282          return inv
283  
284      def issue(self, invoice_id: str) -> None:
285          self._invoices[invoice_id].finalize()
286  
287      def record_payment(self, invoice_id: str, amount: Money, method: PaymentMethod) -> Payment:
288          return self._invoices[invoice_id].apply_payment(amount, method)
289  
290      def void(self, invoice_id: str) -> None:
291          self._invoices[invoice_id].void()
292  
293      def credit_memo(self, original_invoice_id: str, amount: Money) -> Invoice:
294          """Compensating invoice for refunds or corrections. Never mutates the original."""
295          original = self._invoices[original_invoice_id]
296          memo = self.create_invoice(original.customer_id, original.currency)
297          memo.add_item(LineItem(
298              description=f"Credit for invoice {original_invoice_id}",
299              quantity=1,
300              unit_price=Money(-amount.cents, original.currency),  # negative line
301          ))
302          memo.finalize()
303          return memo
304  
305      def sweep_overdue(self, now: datetime | None = None) -> int:
306          """Mark every issued/partially-paid invoice past its due date as OVERDUE.
307          Run this on a daily cadence; dunning reads OVERDUE invoices downstream."""
308          count = 0
309          for inv in self._invoices.values():
310              if inv.check_overdue(now):
311                  count += 1
312          return count
313  
314  
315  if __name__ == "__main__":
316      svc = BillingService()
317      inv = svc.create_invoice("cust-001", currency="USD")
318  
319      # Two items, 10% discount on the first, 8.25% sales tax on both.
320      tax = SalesTaxRule(0.0825)
321      inv.add_item(LineItem(
322          description="Enterprise license",
323          quantity=1,
324          unit_price=Money(50_000),
325          discount=PercentDiscount(10),
326          tax_rule=tax,
327      ))
328      inv.add_item(LineItem(
329          description="Support add-on",
330          quantity=2,
331          unit_price=Money(5_000),
332          tax_rule=tax,
333      ))
334  
335      svc.issue(inv.id)
336      print(f"Invoice {inv.id} — status {inv.status.value}")
337      print(f"  Subtotal: {inv.subtotal}")
338      print(f"  Tax:      {inv.tax_total}")
339      print(f"  Total:    {inv.total}")
340  
341      # Partial payment.
342      p1 = svc.record_payment(inv.id, Money(30_000), PaymentMethod.CREDIT_CARD)
343      print(f"\nAfter payment of {p1.amount}: status {inv.status.value}, due {inv.amount_due()}")
344  
345      # Pay the rest.
346      p2 = svc.record_payment(inv.id, inv.amount_due(), PaymentMethod.BANK_TRANSFER)
347      print(f"After final payment: status {inv.status.value}, due {inv.amount_due()}")
348      assert inv.status == InvoiceStatus.PAID
349  
350      # Cannot modify an issued invoice.
351      try:
352          inv.add_item(LineItem("late addition", 1, Money(100)))
353      except RuntimeError as e:
354          print(f"Post-issue modification blocked: {e}")
355  
356      # Credit memo for a refund.
357      memo = svc.credit_memo(inv.id, Money(10_000))
358      print(f"\nCredit memo {memo.id}: total {memo.total}")
359  
360      # Overdue sweep: issue a second invoice with a short due window and jump time forward.
361      late = svc.create_invoice("cust-002", currency="USD")
362      late.add_item(LineItem("Late thing", 1, Money(10_000), tax_rule=NoTaxRule()))
363      svc.issue(late.id)
364      assert late.status == InvoiceStatus.ISSUED
365      future = datetime.utcnow() + timedelta(days=60)
366      swept = svc.sweep_overdue(now=future)
367      print(f"\nSwept {swept} overdue invoice(s); late invoice status: {late.status.value}")
368      assert late.status == InvoiceStatus.OVERDUE
369      print("All operations passed.")