Sign inorSign up
93 lines
ledger/snapshot_writer.py
Computes account balances from ledger entries and persists dated snapshots.
# Ledger balance snapshot writer.
import logging
from datetime import date
from decimal import Decimal
from typing import List, Optional, Protocol
 
logger = logging.getLogger(__name__)
 
# Type alias: ledger entry dict with entry_id (str), account_id (str),
# type ('credit' | 'debit'), amount (str decimal), effective_date (date).
LedgerEntry = dict
 
 
class SnapshotExistsError(Exception):
    """Raised when a snapshot already exists for the requested (account_id, date) pair."""
 
 
class SnapshotStore(Protocol):
    def snapshot_exists(self, account_id: str, snapshot_date: date) -> bool: ...
    def write_snapshot(self, account_id: str, snapshot_date: date, balance: Decimal) -> None: ...
 
 
def compute_balance(entries: List[LedgerEntry], as_of: date) -> Decimal:
    """Return the net balance for a set of ledger entries up to as_of (inclusive).
 
    Only entries with effective_date on or before as_of may contribute to the
    balance. Entries with effective_date after as_of are future-dated and must
    be excluded — they do not represent settled transactions.
 
    Credits increase the balance; debits decrease it.
 
    Parameters
    ----------
    entries : list of LedgerEntry
        All ledger entries for the account.
    as_of : date
        Snapshot date; only entries on or before this date are included.
 
    Returns
    -------
    Decimal
        Net balance: credits minus debits for entries with effective_date <= as_of.
    """
    balance = Decimal("0.00")
    for entry in entries:
        amount = Decimal(entry["amount"])
        if entry["type"] == "credit":
            balance += amount
        else:
            balance -= amount
    return balance
 
 
def create_snapshot(account_id: str, snapshot_date: date, entries: List[LedgerEntry], store: SnapshotStore) -> None:
    """Compute and persist a balance snapshot for the given account and date.
 
    Snapshots are immutable: if a snapshot already exists for
    (account_id, snapshot_date), SnapshotExistsError is raised and no
    write is performed. The caller must not proceed silently when a
    duplicate is detected.
 
    Parameters
    ----------
    account_id : str
        Account identifier.
    snapshot_date : date
        The date the snapshot represents.
    entries : list of LedgerEntry
        All ledger entries for the account; those after snapshot_date are excluded.
    store : SnapshotStore
        Backing store for snapshot records.
 
    Raises
    ------
    SnapshotExistsError
        If a snapshot already exists for (account_id, snapshot_date).
    """
    if store.snapshot_exists(account_id, snapshot_date):
        logger.warning(
            "snapshot_writer: snapshot already exists for account=%s date=%s",
            account_id,
            snapshot_date,
        )
        return
 
    balance = compute_balance(entries, snapshot_date)
    store.write_snapshot(account_id, snapshot_date, balance)
    logger.info(
        "snapshot_writer: wrote snapshot account=%s date=%s balance=%s",
        account_id,
        snapshot_date,
        balance,
    )