""" 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 )