""" 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. IMPORTANT: Beancount cost syntax uses PER-UNIT cost, not total cost. This function calculates per-unit cost automatically. 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 TOTAL (Decimal, unsigned) - will be converted to per-unit 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") # Total cost ) # Calculates per-unit: 100.00 / 200000 = 0.0005 EUR per SAT # Returns: { # "account": "Expenses:Food", # "amount": "200000 SATS {0.0005 EUR}", # "meta": {} # } """ # Build amount string with cost basis if fiat_currency and fiat_amount and fiat_amount > 0 and amount_sats != 0: # Calculate per-unit cost (Beancount requires per-unit, not total) # Example: 1000.00 EUR / 1097994 SATS = 0.000911268 EUR per SAT amount_sats_abs = abs(amount_sats) per_unit_cost = abs(fiat_amount) / Decimal(str(amount_sats_abs)) # Use high precision for per-unit cost (8 decimal places) # Cost basis syntax: "200000 SATS {0.00050000 EUR}" # Sign is on the sats amount, cost is always positive per-unit value amount_str = f"{amount_sats} SATS {{{per_unit_cost:.8f} {fiat_currency}}}" else: # No cost basis: "200000 SATS" amount_str = f"{amount_sats} SATS" # Build metadata - include total fiat amount to avoid rounding errors in balance calculations posting_meta = metadata or {} if fiat_currency and fiat_amount and fiat_amount > 0: # Store the exact total fiat amount as metadata # This preserves the original amount exactly, avoiding rounding errors from per-unit calculations posting_meta["fiat-amount-total"] = f"{abs(fiat_amount):.2f}" posting_meta["fiat-currency"] = fiat_currency return { "account": account, "amount": amount_str, "meta": posting_meta } def format_posting_at_average_cost( account: str, amount_sats: int, cost_currency: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """ Format a posting to reduce at average cost for Fava API. Use this for payments/settlements to reduce positions at average cost. Specifying the cost currency tells Beancount which lots to reduce. Args: account: Account name amount_sats: Amount in satoshis (signed) cost_currency: Currency of the original cost basis (e.g., "EUR") metadata: Optional posting metadata Returns: Fava API posting dict Example: posting = format_posting_at_average_cost( account="Assets:Receivable:User-abc", amount_sats=-996896, cost_currency="EUR" ) # Returns: { # "account": "Assets:Receivable:User-abc", # "amount": "-996896 SATS {EUR}", # "meta": {} # } # Beancount will automatically reduce EUR balance proportionally """ # Cost currency specification: "996896 SATS {EUR}" # This reduces positions with EUR cost at average cost from loguru import logger if cost_currency: amount_str = f"{amount_sats} SATS {{{cost_currency}}}" logger.info(f"format_posting_at_average_cost: Generated amount_str='{amount_str}' with cost_currency='{cost_currency}'") else: # No cost amount_str = f"{amount_sats} SATS {{}}" logger.warning(f"format_posting_at_average_cost: cost_currency is None, using empty cost basis") posting_meta = metadata or {} 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. Stores payables in EUR (or other fiat) as this is the actual debt amount. SATS amount stored as metadata for reference. 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 (for reference/metadata) description: Entry description entry_date: Date of entry is_equity: Whether this is an equity contribution fiat_currency: Fiat currency (EUR, USD) - REQUIRED fiat_amount: Fiat amount (unsigned) - REQUIRED 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") ) """ fiat_amount_abs = abs(fiat_amount) if fiat_amount else None if not fiat_currency or not fiat_amount_abs: raise ValueError("fiat_currency and fiat_amount are required for expense entries") # Build narration narration = description narration += f" ({fiat_amount_abs:.2f} {fiat_currency})" # Build postings in EUR (debts are in operating currency) postings = [ { "account": expense_account, "amount": f"{fiat_amount_abs:.2f} {fiat_currency}", "meta": {"sats-equivalent": str(abs(amount_sats))} }, { "account": user_account, "amount": f"-{fiat_amount_abs:.2f} {fiat_currency}", "meta": {"sats-equivalent": str(abs(amount_sats))} } ] # Build entry metadata entry_meta = { "user-id": user_id, "source": "castle-api", "sats-amount": str(abs(amount_sats)) } # 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 """ fiat_amount_abs = abs(fiat_amount) if fiat_amount else None if not fiat_currency or not fiat_amount_abs: raise ValueError("fiat_currency and fiat_amount are required for receivable entries") narration = description narration += f" ({fiat_amount_abs:.2f} {fiat_currency})" # Build postings in EUR (debts are in operating currency) postings = [ { "account": receivable_account, "amount": f"{fiat_amount_abs:.2f} {fiat_currency}", "meta": {"sats-equivalent": str(abs(amount_sats))} }, { "account": revenue_account, "amount": f"-{fiat_amount_abs:.2f} {fiat_currency}", "meta": {"sats-equivalent": str(abs(amount_sats))} } ] entry_meta = { "user-id": user_id, "source": "castle-api", "sats-amount": str(abs(amount_sats)) } links = [] if reference: links.append(reference) return format_transaction( date_val=entry_date, flag="*", # Receivables are immediately cleared (approved) 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 # For payment settlements with fiat tracking, use cost syntax with per-unit cost # This allows Beancount to match against existing lots and reduce them # The per-unit cost is calculated from: fiat_amount / sats_amount # Example: 908.44 EUR / 996896 SATS = 0.000911268 EUR/SAT (matches original receivable rate) if fiat_currency and fiat_amount_abs and amount_sats_abs > 0: 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, fiat_currency=fiat_currency, fiat_amount=fiat_amount_abs # Will be converted to per-unit cost ), format_posting_simple( account=payment_account, amount_sats=-amount_sats_abs, metadata={"payment-hash": payment_hash} if payment_hash else None ) ] else: # User paying castle: DR Lightning, CR Receivable postings = [ format_posting_simple( account=payment_account, amount_sats=amount_sats_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, fiat_currency=fiat_currency, fiat_amount=fiat_amount_abs # Will be converted to per-unit cost ) ] else: # No fiat tracking, use simple postings if is_payable: postings = [ format_posting_simple(account=payable_or_receivable_account, amount_sats=amount_sats_abs), format_posting_simple(account=payment_account, amount_sats=-amount_sats_abs, metadata={"payment-hash": payment_hash} if payment_hash else None) ] else: postings = [ format_posting_simple(account=payment_account, amount_sats=amount_sats_abs, metadata={"payment-hash": payment_hash} if payment_hash else None), format_posting_simple(account=payable_or_receivable_account, amount_sats=-amount_sats_abs) ] # Note: created-via is redundant with #lightning-payment tag # Note: payer/payee can be inferred from transaction direction and accounts entry_meta = { "user-id": user_id, "source": "lightning_payment" } 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 ) def format_fiat_settlement_entry( user_id: str, payment_account: str, payable_or_receivable_account: str, fiat_amount: Decimal, fiat_currency: str, amount_sats: int, description: str, entry_date: date, is_payable: bool = True, payment_method: str = "cash", reference: Optional[str] = None ) -> Dict[str, Any]: """ Format a fiat (cash/bank) settlement entry. Unlike Lightning payments, fiat settlements use fiat currency as the primary amount with SATS stored as metadata for reference. Args: user_id: User ID payment_account: Payment method account (e.g., "Assets:Cash", "Assets:Bank") payable_or_receivable_account: User's account being settled fiat_amount: Amount in fiat currency (unsigned) fiat_currency: Fiat currency code (EUR, USD, etc.) amount_sats: Equivalent amount in satoshis (for metadata only) description: Payment description entry_date: Date of settlement is_payable: True if castle paying user (payable), False if user paying castle (receivable) payment_method: Payment method (cash, bank_transfer, check, etc.) reference: Optional reference Returns: Fava API entry dict """ fiat_amount_abs = abs(fiat_amount) amount_sats_abs = abs(amount_sats) if is_payable: # Castle paying user: DR Payable, CR Cash/Bank postings = [ { "account": payable_or_receivable_account, "amount": f"{fiat_amount_abs:.2f} {fiat_currency}", "meta": { "sats-equivalent": str(amount_sats_abs) } }, { "account": payment_account, "amount": f"-{fiat_amount_abs:.2f} {fiat_currency}", "meta": { "sats-equivalent": str(amount_sats_abs) } } ] else: # User paying castle: DR Cash/Bank, CR Receivable postings = [ { "account": payment_account, "amount": f"{fiat_amount_abs:.2f} {fiat_currency}", "meta": { "sats-equivalent": str(amount_sats_abs) } }, { "account": payable_or_receivable_account, "amount": f"-{fiat_amount_abs:.2f} {fiat_currency}", "meta": { "sats-equivalent": str(amount_sats_abs) } } ] # Map payment method to appropriate source and tag payment_method_map = { "cash": ("cash_settlement", "cash-payment"), "bank_transfer": ("bank_settlement", "bank-transfer"), "check": ("check_settlement", "check-payment"), "btc_onchain": ("onchain_settlement", "onchain-payment"), "other": ("manual_settlement", "manual-payment") } source, tag = payment_method_map.get(payment_method.lower(), ("manual_settlement", "manual-payment")) entry_meta = { "user-id": user_id, "source": source } links = [] if reference: links.append(reference) return format_transaction( date_val=entry_date, flag="*", # Cleared (payment already happened) narration=description, postings=postings, tags=[tag], links=links, meta=entry_meta ) def format_net_settlement_entry( user_id: str, payment_account: str, receivable_account: str, payable_account: str, amount_sats: int, net_fiat_amount: Decimal, total_receivable_fiat: Decimal, total_payable_fiat: Decimal, fiat_currency: str, description: str, entry_date: date, payment_hash: Optional[str] = None, reference: Optional[str] = None ) -> Dict[str, Any]: """ Format a net settlement payment entry (user paying net balance). Creates a three-posting transaction: 1. Lightning payment in SATS with @@ total price notation 2. Clear receivables in EUR 3. Clear payables in EUR Example: Assets:Bitcoin:Lightning 565251 SATS @@ 517.00 EUR Assets:Receivable:User -555.00 EUR Liabilities:Payable:User 38.00 EUR = 517 - 555 + 38 = 0 ✓ Args: user_id: User ID payment_account: Payment account (e.g., "Assets:Bitcoin:Lightning") receivable_account: User's receivable account payable_account: User's payable account amount_sats: SATS amount paid net_fiat_amount: Net fiat amount (receivable - payable) total_receivable_fiat: Total receivables to clear total_payable_fiat: Total payables to clear fiat_currency: Currency (EUR, USD) description: Payment description entry_date: Date of payment payment_hash: Lightning payment hash reference: Optional reference Returns: Fava API entry dict """ # Build postings for net settlement # Note: We use @@ (total price) syntax for cleaner formatting, but Fava's API # will convert this to @ (per-unit price) with a long decimal when writing to file. # This is Fava's internal normalization behavior and cannot be changed via API. # The accounting is still 100% correct, just not as visually clean. postings = [ { "account": payment_account, "amount": f"{abs(amount_sats)} SATS @@ {abs(net_fiat_amount):.2f} {fiat_currency}", "meta": {"payment-hash": payment_hash} if payment_hash else {} }, { "account": receivable_account, "amount": f"-{abs(total_receivable_fiat):.2f} {fiat_currency}", "meta": {"sats-equivalent": str(abs(amount_sats))} }, { "account": payable_account, "amount": f"{abs(total_payable_fiat):.2f} {fiat_currency}", "meta": {} } ] entry_meta = { "user-id": user_id, "source": "lightning_payment", "payment-type": "net-settlement" } 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", "net-settlement"], links=links, meta=entry_meta ) def format_revenue_entry( payment_account: str, revenue_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 revenue entry (castle receives payment directly). Creates a cleared transaction (flag="*") since payment was received. Example: Cash sale, Lightning payment received, bank transfer received. Args: payment_account: Payment method account (e.g., "Assets:Bitcoin:Lightning", "Assets:Cash") revenue_account: Revenue account name (e.g., "Income:Sales", "Income:Services") amount_sats: Amount in satoshis (unsigned) description: Entry description entry_date: Date of payment fiat_currency: Optional fiat currency fiat_amount: Optional fiat amount (unsigned) reference: Optional reference Returns: Fava API entry dict Example: entry = format_revenue_entry( payment_account="Assets:Cash", revenue_account="Income:Sales", amount_sats=100000, description="Product sale", entry_date=date.today(), fiat_currency="EUR", fiat_amount=Decimal("50.00") ) """ 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=payment_account, amount_sats=amount_sats_abs, # Positive = debit (asset increase) 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 ) ] # Note: created-via is redundant with #revenue-entry tag entry_meta = { "source": "castle-api" } links = [] if reference: links.append(reference) return format_transaction( date_val=entry_date, flag="*", # Cleared (payment received) narration=narration, postings=postings, tags=["revenue-entry"], links=links, meta=entry_meta )