padreug
2e862d0ebd
Introduces utilities to format Castle data models into Beancount transactions for Fava API compatibility. Provides functions to format transactions, postings with cost basis, expense entries, receivable entries, and payment entries. These functions ensure data is correctly formatted for Fava's add_entries API, including cost basis, flags, and metadata.
464 lines
14 KiB
Python
464 lines
14 KiB
Python
"""
|
|
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
|
|
)
|