Beancount plugin to handle transactions geting delayed and not being included in the next bank statement

delayed_balance.py

#!/usr/bin/env python3

# Copyright (C) 2021 Xidorn Quan
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.

__plugins__ = ['delayed_balance']

from collections import defaultdict
from typing import Any, Optional, Tuple

from beancount.core.data import Account, Balance, Directive, \
        Entries, Options, Transaction
from beancount.ops.validation import ValidationError


def delayed_balance(entries: Entries, options_map: Options) \
        -> Tuple[Entries, list[Any]]:
    """Delay certain transaction until after the next balance entry.

    This is for handling the case where certain transactions are not included
    in the next statement due to delays.

    Example:

        2021-06-22 * "Amazon.co.jp" "Purchase some books"
            Expenses:Shopping:Books                     10.00 AUD
            Liabilities:CBA:CreditCard                 -10.00 AUD
                delayed_balance: TRUE
        
        2021-06-24 balance Liabilities:CBA:CreditCard -100.00 AUD

    will be transformed to:

        2021-06-24 balance Liabilities:CBA:CreditCard -100.00 AUD
        
        2021-06-24 * "Amazon.co.jp" "Purchase some books"
            delayed_from: 2021-06-22
            Expenses:Shopping:Books                     10.00 AUD
            Liabilities:CBA:CreditCard                 -10.00 AUD
    """
    result: Entries = []
    errors: list[Any] = []
    pending: dict[Account, list[Transaction]] = defaultdict(list)
    for entry in entries:
        if isinstance(entry, Transaction):
            # Find transactions need to be delayed
            delayed = False
            for posting in entry.postings:
                if not posting.meta:
                    continue
                if 'delayed_balance' not in posting.meta:
                    continue
                if entry not in pending[posting.account]:
                    pending[posting.account].append(entry)
                delayed = True
            if delayed:
                continue

        result.append(entry)

        if isinstance(entry, Balance) and entry.account in pending:
            for txn in pending[entry.account]:
                remaining_delayed = 0
                # Iterate over postings with delayed balance
                for posting in txn.postings:
                    if not posting.meta:
                        continue
                    if 'delayed_balance' not in posting.meta:
                        continue
                    if posting.account == entry.account:
                        del posting.meta['delayed_balance']
                    else:
                        remaining_delayed += 1
                if remaining_delayed > 0:
                    continue

                # Add delayed transactions after the balance entry
                txn.meta['delayed_from'] = txn.date
                result.append(Transaction(
                    txn.meta,
                    entry.date,
                    txn.flag,
                    txn.payee,
                    txn.narration,
                    txn.tags,
                    txn.links,
                    txn.postings,
                ))
            del pending[entry.account]

    for account, txns in pending.items():
        for txn in txns:
            for posting in txn.postings:
                if not posting.meta:
                    continue
                if 'delayed_balance' not in posting.meta:
                    continue
                errors.append(ValidationError(
                    posting.meta,
                    "Unhandled delayed balance",
                    txn,
                ))

    return result, errors


if __name__ == '__main__':
    import sys
    from beancount import loader

    entries, errors, options = loader.load_file(sys.argv[1], log_errors=sys.stderr)
    delayed_balance(entries, {})