diff --git a/beancount_format.py b/beancount_format.py new file mode 100644 index 0000000..df25d17 --- /dev/null +++ b/beancount_format.py @@ -0,0 +1,464 @@ +""" +Format Castle entries as Beancount transactions for Fava API. + +All entries submitted to Fava must follow Beancount syntax. +This module converts Castle data models to Fava API format. + +Key concepts: +- Amounts are strings: "200000 SATS" or "100.00 EUR" +- Cost basis syntax: "200000 SATS {100.00 EUR}" +- Flags: "*" (cleared), "!" (pending), "#" (flagged), "?" (unknown) +- Entry type: "t": "Transaction" (required by Fava) +""" + +from datetime import date, datetime +from decimal import Decimal +from typing import Any, Dict, List, Optional + + +def format_transaction( + date_val: date, + flag: str, + narration: str, + postings: List[Dict[str, Any]], + payee: str = "", + tags: Optional[List[str]] = None, + links: Optional[List[str]] = None, + meta: Optional[Dict[str, Any]] = None +) -> Dict[str, Any]: + """ + Format a transaction for Fava's add_entries API. + + Args: + date_val: Transaction date + flag: Beancount flag (* = cleared, ! = pending, # = flagged) + narration: Description + postings: List of posting dicts (formatted by format_posting) + payee: Optional payee + tags: Optional tags (e.g., ["expense-entry", "approved"]) + links: Optional links (e.g., ["castle-abc123", "^invoice-xyz"]) + meta: Optional transaction metadata + + Returns: + Fava API entry dict + + Example: + entry = format_transaction( + date_val=date.today(), + flag="*", + narration="Grocery shopping", + postings=[ + format_posting_with_cost( + account="Expenses:Food", + amount_sats=36930, + fiat_currency="EUR", + fiat_amount=Decimal("36.93") + ), + format_posting_with_cost( + account="Liabilities:Payable:User-abc", + amount_sats=-36930, + fiat_currency="EUR", + fiat_amount=Decimal("36.93") + ) + ], + tags=["expense-entry"], + links=["castle-abc123"], + meta={"user-id": "abc123", "source": "castle-expense-entry"} + ) + """ + return { + "t": "Transaction", # REQUIRED by Fava API + "date": str(date_val), + "flag": flag, + "payee": payee or "", # Empty string, not None + "narration": narration, + "tags": tags or [], + "links": links or [], + "postings": postings, + "meta": meta or {} + } + + +def format_posting_with_cost( + account: str, + amount_sats: int, + fiat_currency: Optional[str] = None, + fiat_amount: Optional[Decimal] = None, + metadata: Optional[Dict[str, Any]] = None +) -> Dict[str, Any]: + """ + Format a posting with cost basis for Fava API. + + This is the RECOMMENDED format for all Castle transactions. + Uses Beancount's cost basis syntax to preserve exchange rates. + + Args: + account: Account name (e.g., "Expenses:Food:Groceries") + amount_sats: Amount in satoshis (signed: positive = debit, negative = credit) + fiat_currency: Fiat currency (EUR, USD, etc.) + fiat_amount: Fiat amount (Decimal, unsigned) + metadata: Optional posting metadata + + Returns: + Fava API posting dict + + Example: + posting = format_posting_with_cost( + account="Expenses:Food", + amount_sats=200000, + fiat_currency="EUR", + fiat_amount=Decimal("100.00") + ) + # Returns: { + # "account": "Expenses:Food", + # "amount": "200000 SATS {100.00 EUR}", + # "meta": { + # "fiat-currency": "EUR", + # "fiat-amount": "100.00", + # "sats-equivalent": "200000", + # "exchange-rate": "2000.00" + # } + # } + """ + # Build amount string with cost basis + if fiat_currency and fiat_amount and fiat_amount > 0: + # Cost basis syntax: "200000 SATS {100.00 EUR}" + # Sign is on the sats amount, fiat amount in cost basis is always positive + amount_str = f"{amount_sats} SATS {{{abs(fiat_amount):.2f} {fiat_currency}}}" + else: + # No cost basis: "200000 SATS" + amount_str = f"{amount_sats} SATS" + + # Build metadata + posting_meta = metadata or {} + + if fiat_currency and fiat_amount and fiat_amount > 0: + # Store fiat information in metadata for easy access + posting_meta["fiat-currency"] = fiat_currency + posting_meta["fiat-amount"] = str(abs(fiat_amount)) + posting_meta["sats-equivalent"] = str(abs(amount_sats)) + + # Calculate exchange rate (sats per fiat unit) + exchange_rate = abs(amount_sats) / abs(fiat_amount) + posting_meta["exchange-rate"] = f"{exchange_rate:.2f}" + + # Calculate BTC rate (fiat per BTC) + btc_rate = abs(fiat_amount) / abs(amount_sats) * 100_000_000 + posting_meta["btc-rate"] = f"{btc_rate:.2f}" + + return { + "account": account, + "amount": amount_str, + "meta": posting_meta + } + + +def format_posting_simple( + account: str, + amount_sats: int, + metadata: Optional[Dict[str, Any]] = None +) -> Dict[str, Any]: + """ + Format a simple posting (SATS only, no cost basis). + + Use this for: + - Lightning payments (no fiat conversion) + - SATS-only transactions + - Internal transfers + + Args: + account: Account name + amount_sats: Amount in satoshis (signed) + metadata: Optional posting metadata + + Returns: + Fava API posting dict + + Example: + posting = format_posting_simple( + account="Assets:Bitcoin:Lightning", + amount_sats=200000 + ) + # Returns: { + # "account": "Assets:Bitcoin:Lightning", + # "amount": "200000 SATS", + # "meta": {} + # } + """ + return { + "account": account, + "amount": f"{amount_sats} SATS", + "meta": metadata or {} + } + + +def format_expense_entry( + user_id: str, + expense_account: str, + user_account: str, + amount_sats: int, + description: str, + entry_date: date, + is_equity: bool = False, + fiat_currency: Optional[str] = None, + fiat_amount: Optional[Decimal] = None, + reference: Optional[str] = None +) -> Dict[str, Any]: + """ + Format an expense entry for submission to Fava. + + Creates a pending transaction (flag="!") that requires admin approval. + + Args: + user_id: User ID + expense_account: Expense account name (e.g., "Expenses:Food:Groceries") + user_account: User's liability/equity account name + amount_sats: Amount in satoshis (unsigned, will be signed correctly) + description: Entry description + entry_date: Date of entry + is_equity: Whether this is an equity contribution + fiat_currency: Optional fiat currency (EUR, USD) + fiat_amount: Optional fiat amount (unsigned) + reference: Optional reference (invoice ID, etc.) + + Returns: + Fava API entry dict + + Example: + entry = format_expense_entry( + user_id="abc123", + expense_account="Expenses:Food:Groceries", + user_account="Liabilities:Payable:User-abc123", + amount_sats=200000, + description="Grocery shopping", + entry_date=date.today(), + fiat_currency="EUR", + fiat_amount=Decimal("100.00") + ) + """ + # Ensure amounts are unsigned for cost basis + amount_sats_abs = abs(amount_sats) + fiat_amount_abs = abs(fiat_amount) if fiat_amount else None + + # Build narration + narration = description + if fiat_currency and fiat_amount_abs: + narration += f" ({fiat_amount_abs:.2f} {fiat_currency})" + + # Build postings with cost basis + postings = [ + format_posting_with_cost( + account=expense_account, + amount_sats=amount_sats_abs, # Positive = debit (expense increase) + fiat_currency=fiat_currency, + fiat_amount=fiat_amount_abs + ), + format_posting_with_cost( + account=user_account, + amount_sats=-amount_sats_abs, # Negative = credit (liability/equity increase) + fiat_currency=fiat_currency, + fiat_amount=fiat_amount_abs + ) + ] + + # Build entry metadata + entry_meta = { + "user-id": user_id, + "source": "castle-api", + "created-via": "expense_entry", + "is-equity": "true" if is_equity else "false" + } + + # Build links + links = [] + if reference: + links.append(reference) + + # Build tags + tags = ["expense-entry"] + if is_equity: + tags.append("equity-contribution") + + return format_transaction( + date_val=entry_date, + flag="!", # Pending approval + narration=narration, + postings=postings, + tags=tags, + links=links, + meta=entry_meta + ) + + +def format_receivable_entry( + user_id: str, + revenue_account: str, + receivable_account: str, + amount_sats: int, + description: str, + entry_date: date, + fiat_currency: Optional[str] = None, + fiat_amount: Optional[Decimal] = None, + reference: Optional[str] = None +) -> Dict[str, Any]: + """ + Format a receivable entry (user owes castle). + + Creates a pending transaction that starts as receivable. + + Args: + user_id: User ID + revenue_account: Revenue account name + receivable_account: User's receivable account name (Assets:Receivable:User-{id}) + amount_sats: Amount in satoshis (unsigned) + description: Entry description + entry_date: Date of entry + fiat_currency: Optional fiat currency + fiat_amount: Optional fiat amount (unsigned) + reference: Optional reference + + Returns: + Fava API entry dict + """ + amount_sats_abs = abs(amount_sats) + fiat_amount_abs = abs(fiat_amount) if fiat_amount else None + + narration = description + if fiat_currency and fiat_amount_abs: + narration += f" ({fiat_amount_abs:.2f} {fiat_currency})" + + postings = [ + format_posting_with_cost( + account=receivable_account, + amount_sats=amount_sats_abs, # Positive = debit (asset increase - user owes) + fiat_currency=fiat_currency, + fiat_amount=fiat_amount_abs + ), + format_posting_with_cost( + account=revenue_account, + amount_sats=-amount_sats_abs, # Negative = credit (revenue increase) + fiat_currency=fiat_currency, + fiat_amount=fiat_amount_abs + ) + ] + + entry_meta = { + "user-id": user_id, + "source": "castle-api", + "created-via": "receivable_entry", + "debtor-user-id": user_id + } + + links = [] + if reference: + links.append(reference) + + return format_transaction( + date_val=entry_date, + flag="!", # Pending until paid + narration=narration, + postings=postings, + tags=["receivable-entry"], + links=links, + meta=entry_meta + ) + + +def format_payment_entry( + user_id: str, + payment_account: str, + payable_or_receivable_account: str, + amount_sats: int, + description: str, + entry_date: date, + is_payable: bool = True, + fiat_currency: Optional[str] = None, + fiat_amount: Optional[Decimal] = None, + payment_hash: Optional[str] = None, + reference: Optional[str] = None +) -> Dict[str, Any]: + """ + Format a payment entry (Lightning payment recorded). + + Creates a cleared transaction (flag="*") since payment already happened. + + Args: + user_id: User ID + payment_account: Payment method account (e.g., "Assets:Bitcoin:Lightning") + payable_or_receivable_account: User's account being settled + amount_sats: Amount in satoshis (unsigned) + description: Payment description + entry_date: Date of payment + is_payable: True if castle paying user (payable), False if user paying castle (receivable) + fiat_currency: Optional fiat currency + fiat_amount: Optional fiat amount (unsigned) + payment_hash: Lightning payment hash + reference: Optional reference + + Returns: + Fava API entry dict + """ + amount_sats_abs = abs(amount_sats) + fiat_amount_abs = abs(fiat_amount) if fiat_amount else None + + if is_payable: + # Castle paying user: DR Payable, CR Lightning + postings = [ + format_posting_with_cost( + account=payable_or_receivable_account, + amount_sats=amount_sats_abs, # Positive = debit (liability decrease) + fiat_currency=fiat_currency, + fiat_amount=fiat_amount_abs + ), + format_posting_with_cost( + account=payment_account, + amount_sats=-amount_sats_abs, # Negative = credit (asset decrease) + fiat_currency=fiat_currency, + fiat_amount=fiat_amount_abs, + metadata={"payment-hash": payment_hash} if payment_hash else None + ) + ] + else: + # User paying castle: DR Lightning, CR Receivable + postings = [ + format_posting_with_cost( + account=payment_account, + amount_sats=amount_sats_abs, # Positive = debit (asset increase) + fiat_currency=fiat_currency, + fiat_amount=fiat_amount_abs, + metadata={"payment-hash": payment_hash} if payment_hash else None + ), + format_posting_with_cost( + account=payable_or_receivable_account, + amount_sats=-amount_sats_abs, # Negative = credit (asset decrease) + fiat_currency=fiat_currency, + fiat_amount=fiat_amount_abs + ) + ] + + entry_meta = { + "user-id": user_id, + "source": "lightning_payment", + "created-via": "payment_entry", + "payer-user-id": user_id if not is_payable else "castle", + "payee-user-id": user_id if is_payable else "castle" + } + + if payment_hash: + entry_meta["payment-hash"] = payment_hash + + links = [] + if reference: + links.append(reference) + if payment_hash: + links.append(f"ln-{payment_hash[:16]}") + + return format_transaction( + date_val=entry_date, + flag="*", # Cleared (payment already happened) + narration=description, + postings=postings, + tags=["lightning-payment"], + links=links, + meta=entry_meta + )