castle/docs/BEANCOUNT_PATTERNS.md

Beancount Patterns Analysis for Castle Extension

Overview

After analyzing the Beancount repository, I've identified several excellent design patterns and architectural decisions that we should adopt or consider for the Castle Accounting extension.

Key Patterns to Adopt

1. Immutable Data Structures with NamedTuple

Beancount Pattern:

class Transaction(NamedTuple):
    """A transaction with immutable attributes"""
    meta: Meta
    date: datetime.date
    flag: Flag
    payee: Optional[str]
    narration: str
    tags: frozenset[str]
    links: frozenset[str]
    postings: list[Posting]

class Posting(NamedTuple):
    """Individual posting within a transaction"""
    account: Account
    units: Optional[Amount]
    cost: Optional[Union[Cost, CostSpec]]
    price: Optional[Amount]
    flag: Optional[Flag]
    meta: Optional[Meta]

Benefits:

• Immutability prevents accidental modifications
• NamedTuple provides free __repr__, __eq__, __hash__
• Type hints improve IDE support and catch errors
• More memory efficient than regular classes
• Thread-safe by design

Castle Application:

# In models.py
from typing import NamedTuple, Optional
from datetime import datetime
from decimal import Decimal

class ImmutableJournalEntry(NamedTuple):
    """Immutable representation of a journal entry"""
    id: str
    description: str
    entry_date: datetime
    created_by: str
    created_at: datetime
    reference: Optional[str]
    lines: tuple[ImmutableEntryLine, ...]  # Tuple, not list!
    meta: dict[str, Any]  # Could store filename, lineno for audit trail

class ImmutableEntryLine(NamedTuple):
    """Immutable representation of an entry line"""
    id: str
    journal_entry_id: str
    account_id: str
    debit: int
    credit: int
    description: Optional[str]
    metadata: dict[str, Any]
    flag: Optional[str]  # Like Beancount: '!', '*', etc.

# Conversion functions
def entry_to_immutable(entry: JournalEntry) -> ImmutableJournalEntry:
    """Convert mutable Pydantic model to immutable NamedTuple"""
    return ImmutableJournalEntry(
        id=entry.id,
        description=entry.description,
        entry_date=entry.entry_date,
        created_by=entry.created_by,
        created_at=entry.created_at,
        reference=entry.reference,
        lines=tuple(line_to_immutable(line) for line in entry.lines),
        meta={}
    )

Recommendation: Use immutable NamedTuples for internal processing, Pydantic models for API I/O.

---

2. Plugin Architecture

Beancount Pattern:

# In plugins/check_commodity.py
__plugins__ = ('validate_commodity_directives',)

def validate_commodity_directives(entries, options_map, config):
    """
    Plugin that validates all commodities have declarations.

    Args:
        entries: List of directive entries
        options_map: Parser options
        config: Plugin-specific configuration

    Returns:
        (entries, errors) tuple
    """
    errors = []
    # ... validation logic ...
    return entries, errors

Castle Application:

# Create plugins/ directory
# lnbits/extensions/castle/plugins/__init__.py

from typing import Protocol, Tuple, List, Any

class CastlePlugin(Protocol):
    """Protocol for Castle plugins"""

    def __call__(
        self,
        entries: List[JournalEntry],
        settings: dict[str, Any],
        config: dict[str, Any]
    ) -> Tuple[List[JournalEntry], List[dict]]:
        """
        Process entries and return modified entries + errors.

        Args:
            entries: Journal entries to process
            settings: Castle settings
            config: Plugin-specific configuration

        Returns:
            (modified_entries, errors) tuple
        """
        ...

# Example plugins:

# plugins/check_balanced.py
__plugins__ = ('check_all_balanced',)

def check_all_balanced(entries, settings, config):
    """Verify all journal entries have debits = credits"""
    errors = []
    for entry in entries:
        total_debits = sum(line.debit for line in entry.lines)
        total_credits = sum(line.credit for line in entry.lines)
        if total_debits != total_credits:
            errors.append({
                'entry_id': entry.id,
                'message': f'Unbalanced entry: debits={total_debits}, credits={total_credits}',
                'severity': 'error'
            })
    return entries, errors

# plugins/auto_tags.py
__plugins__ = ('auto_add_tags',)

def auto_add_tags(entries, settings, config):
    """Automatically add tags based on rules"""
    tag_rules = config.get('rules', {})
    for entry in entries:
        for pattern, tag in tag_rules.items():
            if pattern in entry.description.lower():
                if 'tags' not in entry.meta:
                    entry.meta['tags'] = set()
                entry.meta['tags'].add(tag)
    return entries, []

# plugins/check_receivables.py
def check_receivable_limits(entries, settings, config):
    """Warn if receivables exceed configured limits"""
    errors = []
    max_per_user = config.get('max_receivable_per_user', 1_000_000)

    # Calculate current receivables per user
    receivables = {}
    for entry in entries:
        for line in entry.lines:
            if 'Accounts Receivable' in line.account_name:
                user_id = extract_user_from_account(line.account_name)
                receivables[user_id] = receivables.get(user_id, 0) + line.debit - line.credit

    for user_id, amount in receivables.items():
        if amount > max_per_user:
            errors.append({
                'user_id': user_id,
                'message': f'Receivable {amount} sats exceeds limit {max_per_user}',
                'severity': 'warning'
            })

    return entries, errors

Plugin Manager:

# plugins/manager.py
import importlib
from pathlib import Path

class PluginManager:
    def __init__(self, plugin_dir: Path):
        self.plugins = []
        self.load_plugins(plugin_dir)

    def load_plugins(self, plugin_dir: Path):
        """Discover and load all plugins"""
        for plugin_file in plugin_dir.glob('*.py'):
            if plugin_file.name.startswith('_'):
                continue

            module_name = f"castle.plugins.{plugin_file.stem}"
            module = importlib.import_module(module_name)

            if hasattr(module, '__plugins__'):
                for plugin_name in module.__plugins__:
                    plugin_func = getattr(module, plugin_name)
                    self.plugins.append((plugin_name, plugin_func))

    def run_plugins(
        self,
        entries: List[JournalEntry],
        settings: dict,
        plugin_configs: dict[str, dict]
    ) -> Tuple[List[JournalEntry], List[dict]]:
        """Run all plugins in sequence"""
        all_errors = []

        for plugin_name, plugin_func in self.plugins:
            config = plugin_configs.get(plugin_name, {})
            entries, errors = plugin_func(entries, settings, config)
            all_errors.extend(errors)

        return entries, all_errors

Benefits:

• Extensibility without modifying core code
• Easy to enable/disable validation rules
• Community can contribute plugins
• Each plugin is independently testable

---

3. Inventory System for Position Tracking

Beancount Pattern:

class Inventory(dict[tuple[str, Optional[Cost]], Position]):
    """
    Tracks positions keyed by (currency, cost).
    Handles lot matching with FIFO, LIFO, HIFO strategies.
    """

    def add_position(self, position: Position):
        """Add or merge a position"""
        key = (position.units.currency, position.cost)
        if key in self:
            self[key] = self[key].add(position)
        else:
            self[key] = position

    def reduce(self, reducer_func):
        """Convert inventory using a reduction function"""
        return sum((reducer_func(pos) for pos in self), Inventory())

    def get_currency_units(self, currency: str) -> Decimal:
        """Get total units of a specific currency"""
        return sum(
            pos.units.number
            for (curr, cost), pos in self.items()
            if curr == currency
        )

Castle Application:

# core/inventory.py
from decimal import Decimal
from typing import Optional, Dict, Tuple
from dataclasses import dataclass

@dataclass(frozen=True)
class CastlePosition:
    """A position in the Castle inventory"""
    currency: str  # "SATS", "EUR", "USD"
    amount: Decimal
    cost_currency: Optional[str] = None  # Original currency if converted
    cost_amount: Optional[Decimal] = None  # Original amount
    date: Optional[datetime] = None
    metadata: Dict[str, Any] = None

class CastleInventory:
    """
    Track user balances across multiple currencies with conversion tracking.
    Similar to Beancount's Inventory but optimized for Castle's use case.
    """

    def __init__(self):
        self.positions: Dict[Tuple[str, Optional[str]], CastlePosition] = {}

    def add_position(self, position: CastlePosition):
        """Add or merge a position"""
        key = (position.currency, position.cost_currency)

        if key in self.positions:
            existing = self.positions[key]
            self.positions[key] = CastlePosition(
                currency=position.currency,
                amount=existing.amount + position.amount,
                cost_currency=position.cost_currency,
                cost_amount=(
                    (existing.cost_amount or Decimal(0)) +
                    (position.cost_amount or Decimal(0))
                ),
                date=position.date,
                metadata={**(existing.metadata or {}), **(position.metadata or {})}
            )
        else:
            self.positions[key] = position

    def get_balance_sats(self) -> Decimal:
        """Get total balance in satoshis"""
        return sum(
            pos.amount
            for (curr, _), pos in self.positions.items()
            if curr == "SATS"
        )

    def get_balance_fiat(self, currency: str) -> Decimal:
        """Get balance in specific fiat currency from cost metadata"""
        return sum(
            pos.cost_amount or Decimal(0)
            for (curr, cost_curr), pos in self.positions.items()
            if cost_curr == currency
        )

    def to_dict(self) -> dict:
        """Export to dictionary"""
        return {
            "sats": float(self.get_balance_sats()),
            "fiat": {
                curr: float(self.get_balance_fiat(curr))
                for curr in set(
                    cost_curr
                    for _, cost_curr in self.positions.keys()
                    if cost_curr
                )
            }
        }

# Usage in balance calculation:
async def get_user_inventory(user_id: str) -> CastleInventory:
    """Calculate user's inventory from journal entries"""
    inventory = CastleInventory()

    user_accounts = await get_user_accounts(user_id)
    for account in user_accounts:
        entry_lines = await get_entry_lines_for_account(account.id)

        for line in entry_lines:
            # Add as position
            metadata = json.loads(line.metadata) if line.metadata else {}

            if line.debit > 0:
                inventory.add_position(CastlePosition(
                    currency="SATS",
                    amount=Decimal(line.debit),
                    cost_currency=metadata.get("fiat_currency"),
                    cost_amount=Decimal(metadata.get("fiat_amount", 0)),
                    date=line.created_at,
                    metadata=metadata
                ))

            if line.credit > 0:
                inventory.add_position(CastlePosition(
                    currency="SATS",
                    amount=-Decimal(line.credit),
                    cost_currency=metadata.get("fiat_currency"),
                    cost_amount=-Decimal(metadata.get("fiat_amount", 0)),
                    date=line.created_at,
                    metadata=metadata
                ))

    return inventory

Benefits:

• Cleaner abstraction for balance calculations
• Separates position tracking from database queries
• Easier to add lot tracking if needed later
• Consistent with accounting standards

---

4. Meta Attribute Pattern

Beancount Pattern: Every directive has a meta: dict[str, Any] attribute that stores:

• filename: Source file
• lineno: Line number
• Custom metadata like tags, links, notes

Castle Application:

class JournalEntryMeta(BaseModel):
    """Metadata for journal entries"""
    filename: Optional[str] = None  # For imported entries
    lineno: Optional[int] = None
    source: str = "api"  # "api", "import", "plugin", "migration"
    created_via: str = "web_ui"  # "web_ui", "cli", "script"
    tags: set[str] = set()  # e.g., {"recurring", "automated"}
    links: set[str] = set()  # Link to other entries
    reviewed: bool = False
    reviewed_by: Optional[str] = None
    reviewed_at: Optional[datetime] = None
    notes: Optional[str] = None

# Add to journal_entries table:
"""
ALTER TABLE journal_entries ADD COLUMN meta TEXT DEFAULT '{}';
"""

# Usage:
entry = await create_journal_entry(
    data=entry_data,
    created_by=user_id,
    meta={
        "source": "api",
        "created_via": "web_ui",
        "tags": ["manual", "receivable"],
        "notes": "User requested invoice #1234"
    }
)

Benefits:

• Audit trail for compliance
• Better debugging ("where did this entry come from?")
• Support for tags and categorization
• Can track review/approval workflow

---

5. Balance Assertion Pattern

Beancount Pattern:

2025-10-22 balance Assets:Lightning:Balance  268548 SATS

This asserts that the account balance should be exactly 268,548 sats on that date. If it's not, Beancount throws an error.

Castle Application:

# models.py
class BalanceAssertion(BaseModel):
    """Assert expected balance at a specific date"""
    id: str
    date: datetime
    account_id: str
    expected_balance: int  # in satoshis
    tolerance: int = 0  # Allow +/- this much difference
    checked_balance: Optional[int] = None
    difference: Optional[int] = None
    status: str = "pending"  # pending, passed, failed
    created_by: str
    created_at: datetime

# API endpoint
@castle_api_router.post("/api/v1/assertions/balance")
async def create_balance_assertion(
    data: CreateBalanceAssertion,
    wallet: WalletTypeInfo = Depends(require_admin_key),
) -> BalanceAssertion:
    """Create a balance assertion for reconciliation"""
    assertion = BalanceAssertion(
        id=urlsafe_short_hash(),
        date=data.date,
        account_id=data.account_id,
        expected_balance=data.expected_balance,
        tolerance=data.tolerance,
        created_by=wallet.wallet.user,
        created_at=datetime.now()
    )

    # Check balance immediately
    actual_balance = await get_account_balance_at_date(
        data.account_id,
        data.date
    )

    difference = abs(actual_balance - data.expected_balance)

    assertion.checked_balance = actual_balance
    assertion.difference = actual_balance - data.expected_balance
    assertion.status = "passed" if difference <= data.tolerance else "failed"

    await db.insert("balance_assertions", assertion)

    if assertion.status == "failed":
        raise HTTPException(
            status_code=HTTPStatus.CONFLICT,
            detail=f"Balance assertion failed: expected {data.expected_balance}, "
                   f"got {actual_balance}, difference {assertion.difference}"
        )

    return assertion

# Periodic check job
async def check_all_balance_assertions():
    """Recheck all balance assertions (run daily)"""
    assertions = await db.fetchall(
        "SELECT * FROM balance_assertions WHERE status = 'pending'"
    )

    for assertion in assertions:
        actual_balance = await get_account_balance_at_date(
            assertion.account_id,
            assertion.date
        )

        difference = abs(actual_balance - assertion.expected_balance)
        status = "passed" if difference <= assertion.tolerance else "failed"

        await db.execute(
            """
            UPDATE balance_assertions
            SET checked_balance = :actual, difference = :diff, status = :status
            WHERE id = :id
            """,
            {
                "actual": actual_balance,
                "diff": actual_balance - assertion.expected_balance,
                "status": status,
                "id": assertion.id
            }
        )

Benefits:

• Catch data entry errors early
• Reconciliation checkpoints
• Build confidence in accounting accuracy
• Required for external audits

---

6. Hierarchical Account Names

Beancount Pattern:

Assets:US:BofA:Checking
Assets:US:ETrade:Cash
Assets:US:ETrade:ITOT
Liabilities:US:CreditCard:Amex

Accounts are organized hierarchically with : separator.

Castle Application:

# Currently: "Accounts Receivable - af983632"
# Better:    "Assets:Receivable:User-af983632"

# Update account naming:
ACCOUNT_HIERARCHY = {
    "asset": "Assets",
    "liability": "Liabilities",
    "equity": "Equity",
    "revenue": "Income",  # Beancount uses "Income" not "Revenue"
    "expense": "Expenses"
}

def format_account_name(
    account_type: AccountType,
    base_name: str,
    user_id: Optional[str] = None
) -> str:
    """Format account name in hierarchical style"""
    root = ACCOUNT_HIERARCHY[account_type.value]

    # Convert "Accounts Receivable" -> "Receivable"
    # Convert "Food & Supplies" -> "Food:Supplies"
    base_clean = base_name.replace("Accounts ", "").replace(" & ", ":")

    if user_id:
        return f"{root}:{base_clean}:User-{user_id[:8]}"
    else:
        return f"{root}:{base_clean}"

# Examples:
# "Assets:Receivable:User-af983632"
# "Liabilities:Payable:User-af983632"
# "Equity:MemberEquity:User-af983632"
# "Income:Accommodation"
# "Expenses:Food:Supplies"
# "Assets:Lightning:Balance"

Benefits:

• Standard accounting hierarchy
• Better organization for reporting
• Easier to generate balance sheet (all Assets:, all Liabilities:, etc.)
• Compatible with Beancount import

---

7. Flags for Transaction Status

Beancount Pattern:

2025-10-22 * "Confirmed transaction"
  Assets:Checking  -100 USD
  Expenses:Food     100 USD

2025-10-23 ! "Pending/uncleared transaction"
  Assets:Checking  -50 USD
  Expenses:Gas      50 USD

Flags: * = cleared, ! = pending, # = flagged for review

Castle Application:

# Add flag field to journal_entries
class JournalEntryFlag(str, Enum):
    CLEARED = "*"      # Fully reconciled
    PENDING = "!"      # Not yet confirmed
    FLAGGED = "#"      # Needs review
    VOID = "x"         # Voided entry

class JournalEntry(BaseModel):
    # ... existing fields ...
    flag: JournalEntryFlag = JournalEntryFlag.CLEARED

# Usage:
# Lightning payment: flag="*" (immediately cleared)
# Manual payment request: flag="!" (pending approval)
# Suspicious entry: flag="#" (flagged for review)
# Reversed entry: flag="x" (voided)

# UI: Show icon based on flag
# * = green checkmark
# ! = yellow warning
# # = red flag
# x = gray strikethrough

Benefits:

• Visual indication of entry status
• Filter by flag in queries
• Standard accounting practice
• Supports reconciliation workflow

---

8. Decimal for Money, Not Float

Beancount Pattern:

from decimal import Decimal

# All amounts use Decimal, never float
amount = Decimal("19.99")

Castle Current Issue: We're using int for satoshis (good!) but float for fiat amounts (bad!).

Fix:

from decimal import Decimal

class ExpenseEntry(BaseModel):
    amount: Decimal  # Not float!
    # ...

class ReceivableEntry(BaseModel):
    amount: Decimal  # Not float!
    # ...

# Metadata storage
metadata = {
    "fiat_amount": str(Decimal("250.00")),  # Store as string to preserve precision
    "fiat_rate": str(Decimal("1074.192")),
}

# When reading:
fiat_amount = Decimal(metadata["fiat_amount"])

Benefits:

• Prevents floating point rounding errors
• Required for financial applications
• Exact decimal arithmetic
• Decimal("0.1") + Decimal("0.2") == Decimal("0.3") (float doesn't!)

---

9. Query Language (Future Enhancement)

Beancount Pattern:

-- Beancount Query Language (BQL)
SELECT account, sum(position) AS balance
WHERE account ~ 'Assets:'
GROUP BY account
ORDER BY balance DESC;

SELECT date, narration, position
WHERE account = 'Assets:Checking'
  AND date >= 2025-01-01;

Castle Application (Future):

# Add query endpoint
@castle_api_router.post("/api/v1/query")
async def execute_query(
    query: str,
    wallet: WalletTypeInfo = Depends(require_invoice_key),
) -> dict:
    """
    Execute a query against accounting data.
    Supports SQL-like syntax for reporting.
    """
    # Parse query (could use SQLAlchemy for this)
    # Apply user permissions (filter to user's accounts)
    # Execute and return results
    pass

# Example queries users could run:
# - "Show all expenses in October"
# - "What's my average monthly spending on food?"
# - "Show all receivables older than 30 days"

Benefits:

• Power users can build custom reports
• Reduces need for custom endpoints
• Flexible analytics

---

Structural Improvements

1. Separate Core from API

Beancount Structure:

beancount/
  core/           # Pure data structures and logic
    data.py       # Immutable data types
    inventory.py  # Position tracking
    account.py    # Account operations
  parser/         # Text file parsing
  plugins/        # Validation and transformation
  scripts/        # CLI tools
  tools/          # Reporting and analysis

Castle Should Adopt:

castle/
  core/              # NEW: Pure accounting logic
    __init__.py
    inventory.py     # CastleInventory for position tracking
    balance.py       # Balance calculation logic
    validation.py    # Entry validation (debits=credits, etc)
    account.py       # Account hierarchy and naming
  models.py          # Pydantic models (API I/O)
  crud.py            # Database operations
  views_api.py       # API endpoints
  plugins/           # NEW: Validation plugins
    __init__.py
    check_balanced.py
    check_receivables.py
  services.py        # Business logic
  migrations.py
  templates/
  static/

Benefits:

• Core logic is independent of database
• Easier to test (no DB mocking needed)
• Could swap DB backend
• Core logic can be reused in CLI tools

---

2. Error Handling with Typed Errors

Beancount Pattern:

ConfigError = collections.namedtuple('ConfigError', 'source message entry')
ValidationError = collections.namedtuple('ValidationError', 'source message entry')

def validate_entries(entries):
    errors = []
    for entry in entries:
        if not is_valid(entry):
            errors.append(ValidationError(
                source={'filename': entry.meta['filename'], 'lineno': entry.meta['lineno']},
                message="Entry is invalid",
                entry=entry
            ))
    return errors

Castle Application:

from typing import NamedTuple, Optional

class CastleError(NamedTuple):
    """Base error type"""
    source: dict  # {'endpoint': '...', 'user_id': '...'}
    message: str
    entry: Optional[dict]  # Related entry data
    severity: str  # 'error', 'warning', 'info'
    code: str  # 'UNBALANCED_ENTRY', 'MISSING_ACCOUNT', etc.

class UnbalancedEntryError(NamedTuple):
    source: dict
    message: str
    entry: dict
    severity: str = 'error'
    code: str = 'UNBALANCED_ENTRY'
    total_debits: int
    total_credits: int
    difference: int

# Return errors from validation
async def validate_journal_entry(entry: CreateJournalEntry) -> list[CastleError]:
    errors = []

    total_debits = sum(line.debit for line in entry.lines)
    total_credits = sum(line.credit for line in entry.lines)

    if total_debits != total_credits:
        errors.append(UnbalancedEntryError(
            source={'created_by': entry.created_by},
            message=f"Entry does not balance: debits={total_debits}, credits={total_credits}",
            entry=entry.dict(),
            total_debits=total_debits,
            total_credits=total_credits,
            difference=total_debits - total_credits
        ))

    return errors

Benefits:

• Structured error information
• Easier to log and track
• Can return multiple errors at once
• Better error messages for users

---

Implementation Roadmap

Phase 1: Foundation (High Priority) ✅ COMPLETE

1. ✅ Switch from float to Decimal for fiat amounts
2. ✅ Add meta field to journal entries for audit trail
3. ✅ Add flag field for transaction status
4. ✅ Implement hierarchical account naming

Phase 2: Reconciliation (High Priority) ✅ COMPLETE

5. ✅ Implement balance assertions
6. ✅ Add reconciliation API endpoints
7. ✅ Build reconciliation UI
8. ✅ Add automated daily balance checks

Phase 3: Core Logic Refactoring (Medium Priority) ✅ COMPLETE

9. ✅ Create core/ module with pure accounting logic
10. ✅ Implement CastleInventory for position tracking
11. ✅ Move balance calculation to core/balance.py
12. ✅ Add comprehensive validation in core/validation.py

Phase 4: Validation Plugins (Medium Priority) - Works better after Phase 3

13. Create plugin system architecture
14. Implement check_balanced plugin
15. Implement check_receivables plugin
16. Add plugin configuration UI

Phase 5: Advanced Features (Low Priority)

17. Add tags and links to entries
18. Implement query language
19. Add lot tracking to inventory
20. Support multi-currency in single entry

---

Key Takeaways

What Beancount Does Right:

1. ✅ Immutable data structures
2. ✅ Plugin architecture for extensibility
3. ✅ Rich metadata on every directive
4. ✅ Balance assertions for reconciliation
5. ✅ Hierarchical account naming
6. ✅ Use of Decimal for precision
7. ✅ Separation of core logic from I/O
8. ✅ Comprehensive validation

What Castle Should Adopt First:

1. Decimal for fiat amounts (prevent rounding errors)
2. Meta field (audit trail, source tracking)
3. Flag field (transaction status)
4. Balance assertions (reconciliation)
5. Plugin system (extensible validation)

What Can Wait:

• Query language (nice to have)
• Lot tracking (not needed for current use case)
• Full immutability (Pydantic models are fine for now)

---

Conclusion

Beancount's architecture is battle-tested for serious accounting work. By adopting these patterns, Castle can:

• Prevent financial calculation errors (Decimal)
• Support complex workflows (plugins)
• Build user trust (balance assertions, audit trail)
• Scale to larger deployments (clean architecture)
• Export to professional tools (hierarchical accounts)

The most important immediate change is switching from float to Decimal for fiat amounts. This alone will prevent serious bugs. The plugin system and balance assertions should follow shortly after as they're critical for financial correctness.