""" Settlement engine for TripTalley. Computes who owes whom after tallying all expenses in a trip. Uses debt simplification to minimize the number of transfers. """ from dataclasses import dataclass from decimal import Decimal from typing import Dict, List, Tuple from .money import Money, convert_currency, from_minor @dataclass class Balance: """A user's net balance in a specific currency.""" user_id: int amount_minor: int # positive = is owed, negative = owes currency: str @property def is_positive(self) -> bool: return self.amount_minor > 0 @property def is_negative(self) -> bool: return self.amount_minor < 0 @property def abs_amount(self) -> int: return abs(self.amount_minor) def to_decimal(self) -> Decimal: return from_minor(self.amount_minor, self.currency) def __repr__(self) -> str: sign = "+" if self.amount_minor >= 0 else "" return f"{sign}{from_minor(self.amount_minor, self.currency)}" @dataclass class Payment: """A single payment in the settlement.""" from_user_id: int to_user_id: int amount_minor: int currency: str def to_decimal(self) -> Decimal: return from_minor(self.amount_minor, self.currency) def __repr__(self) -> str: return f"User {self.from_user_id} → User {self.to_user_id}: " \ f"{from_minor(self.amount_minor, self.currency)}" class SettlementEngine: """ Computes trip settlement from expense balances. """ @staticmethod def compute_balances( expenses_data: List[Dict], participants_data: List[Dict], fx_rates: Dict[Tuple[str, str, str], float], # (date, from, to) -> rate settlement_currency: str, ) -> Dict[int, int]: """ Compute net balances for all participants in the settlement currency. Args: expenses_data: List of expense records with: - expense_id, trip_id, payer_id, amount_minor, original_currency, spent_on participants_data: List of participant records with: - expense_id, user_id, share_minor fx_rates: Historical FX rates keyed by (spent_on_date, currency_from, currency_to) settlement_currency: The canonical currency for netting Returns: Dict mapping user_id to net balance (positive = owed, negative = owes) in the settlement currency """ balances: Dict[int, int] = {} # Build a lookup for expense participants expense_participants: Dict[int, List[Tuple[int, int]]] = {} for p in participants_data: exp_id = p["expense_id"] if exp_id not in expense_participants: expense_participants[exp_id] = [] expense_participants[exp_id].append((p["user_id"], p["share_minor"])) # Process each expense for expense in expenses_data: payer_id = expense["payer_id"] amount_minor = expense["amount_minor"] original_currency = expense["original_currency"] spent_on = expense["spent_on"] # Convert total expense to settlement currency fx_key = (spent_on, original_currency, settlement_currency) if fx_key not in fx_rates: raise ValueError( f"No FX rate for {original_currency}→{settlement_currency} on {spent_on}" ) amount_settlement = convert_currency( amount_minor, original_currency, settlement_currency, fx_rates[fx_key] ) # Get participants' shares in original currency participants = expense_participants.get(expense["expense_id"], []) # Payer is credited the full converted amount balances[payer_id] = balances.get(payer_id, 0) + amount_settlement # Each participant is debited their share for user_id, share_minor in participants: # Convert share to settlement currency fx_key_share = (spent_on, original_currency, settlement_currency) share_settlement = convert_currency( share_minor, original_currency, settlement_currency, fx_rates[fx_key_share] ) balances[user_id] = balances.get(user_id, 0) - share_settlement return balances @staticmethod def simplify_debt(balances: Dict[int, int]) -> List[Payment]: """ Simplify a set of balances into minimal payments using a greedy algorithm. The algorithm: 1. Separate creditors (positive balance) and debtors (negative balance) 2. Greedily match the largest creditor with the largest debtor 3. Settle the smaller amount, adjust balances, repeat until all settled This minimizes the number of transfers (at most N-1 for N participants). Args: balances: Dict mapping user_id to net balance Returns: List of Payment objects representing minimal transfer set """ if not balances: return [] # Filter out zero balances and separate creditors/debtors creditors = [] # (user_id, amount_owed_us) debtors = [] # (user_id, amount_owed_them) for user_id, balance in balances.items(): if balance > 0: creditors.append((user_id, balance)) elif balance < 0: debtors.append((user_id, -balance)) # store as positive # Sort by amount descending for greedy matching creditors.sort(key=lambda x: x[1], reverse=True) debtors.sort(key=lambda x: x[1], reverse=True) payments = [] c_idx, d_idx = 0, 0 while c_idx < len(creditors) and d_idx < len(debtors): creditor_id, creditor_amt = creditors[c_idx] debtor_id, debtor_amt = debtors[d_idx] # Settle the smaller amount settle_amount = min(creditor_amt, debtor_amt) if settle_amount > 0: payments.append(Payment( from_user_id=debtor_id, to_user_id=creditor_id, amount_minor=settle_amount, currency="SETTLEMENT" # Will be replaced in actual implementation )) # Update remaining amounts creditors[c_idx] = (creditor_id, creditor_amt - settle_amount) debtors[d_idx] = (debtor_id, debtor_amt - settle_amount) # Move to next if settled if creditors[c_idx][1] == 0: c_idx += 1 if debtors[d_idx][1] == 0: d_idx += 1 return payments @staticmethod def convert_for_display( payments: List[Payment], from_currency: str, to_currency: str, fx_rate: float ) -> List["DisplayPayment"]: """ Convert canonical settlement payments to a viewer's home currency. Args: payments: Payments in the settlement currency from_currency: The settlement currency to_currency: The viewer's home currency fx_rate: FX rate from settlement to home currency Returns: List of DisplayPayment with both canonical and converted amounts """ from dataclasses import dataclass @dataclass class DisplayPayment: from_user_id: int to_user_id: int amount_minor: int # canonical (settlement) currency: str amount_display_minor: int # converted (home currency) display_currency: str def __repr__(self) -> str: return (f"User {self.from_user_id} → User {self.to_user_id}: " f"{from_minor(self.amount_minor, self.currency)} " f"({from_minor(self.amount_display_minor, self.display_currency)})") display_payments = [] for payment in payments: amount_display = convert_currency( payment.amount_minor, payment.currency, to_currency, fx_rate ) display_payments.append(DisplayPayment( from_user_id=payment.from_user_id, to_user_id=payment.to_user_id, amount_minor=payment.amount_minor, currency=payment.currency, amount_display_minor=amount_display, display_currency=to_currency )) return display_payments # Example usage if __name__ == "__main__": # Example: $100 expense split 50/50 between Alice and Bob, paid by Charlie # Settlement currency: USD expenses = [ { "expense_id": 1, "trip_id": 1, "payer_id": 3, # Charlie "amount_minor": 10000, # $100.00 "original_currency": "USD", "spent_on": "2024-01-15" } ] participants = [ {"expense_id": 1, "user_id": 1, "share_minor": 5000}, # Alice 50% {"expense_id": 1, "user_id": 2, "share_minor": 5000}, # Bob 50% {"expense_id": 1, "user_id": 3, "share_minor": 0} # Charlie paid, doesn't share ] fx_rates = { ("2024-01-15", "USD", "USD"): 1.0 } settlement_currency = "USD" # Compute balances balances = SettlementEngine.compute_balances( expenses, participants, fx_rates, settlement_currency ) print("Balances:") for user_id, balance in balances.items(): sign = "+" if balance >= 0 else "" print(f" User {user_id}: {sign}${balance/100:.2f}") # Simplify debt payments = SettlementEngine.simplify_debt(balances) print("\nPayments:") for p in payments: print(f" User {p.from_user_id} → User {p.to_user_id}: ${p.amount_minor/100:.2f}")