Hi everyone! Just discovered beancount, having a great time with it.

I was reading through the documentation, and was surprised when the Importing Process section said that merging has to be done manually. Not sure if I've misunderstood this part, or if I'm missing some tool that makes this easier. Anyway I wrote a plugin to do what I wanted. Sharing it here in case it's useful to others, or more likely in case someone has a better way to do this.

Situation: I'm importing transaction data from multiple bank accounts, which include many transactions that transfer money between my own accounts. In this case the transaction would be logged twice, since it is seen in the logs of both bank accounts (as the documentation mentions).

Assuming every transaction was logged, any transfer that goes to another imported bank account is guaranteed to be logged twice, making it very easy to remove duplicates. Here's the whole plugin:

from datetime import timedelta
from beancount.core.data import Transaction, entry_sortkey

__plugins__ = ('auto_merger',)

# Account pairs between which transfers would be logged twice
duplicate_pairs = [
    frozenset(x) for x in
    [
        ('Assets:Banks:RBS:Main', 'Assets:Banks:RBS:Savings'),
        ('Assets:Banks:RBS:Main', 'Assets:Banks:Revolut'),
    ]
]


def auto_merger(entries, unused_options_map):
    new_entries = []
    waiting = {}

    for entry in entries:
        if not isinstance(entry, Transaction):
            # Not a transaction
            new_entries.append(entry)
            continue

        pair = frozenset([
            str(x.account) for x in
            entry.postings
        ])

        if pair not in duplicate_pairs:
            # Not in the set of duplicate pairs
            new_entries.append(entry)
            continue

        uuid = tuple(
            f"{x.account} {float(x.units.number):.2f} {x.units.currency}" for x in
            entry.postings
        )
        uuid_inv = (uuid[1], uuid[0])

        if uuid_inv in waiting:
            # Found a duplicate entry! Skip it
            del waiting[uuid_inv]
            continue

        # This entry should have a duplicate in the future
        waiting[uuid] = len(new_entries)
        new_entries.append(entry)
        continue

    if len(waiting) > 0:
        print(f"WARNING: Missing {len(waiting)} entries")

    for x, i in waiting.items():
        print(x, new_entries[i].date)

    return new_entries, []

This is iterating over all transactions with the following logic:

1. If a transaction is moving money between two accounts that are both imported, we expect it to have a duplicate. Note down the name of the two accounts, and how much money was transferred. This is added to the "waiting" set.
2. If a transaction matches something in the "waiting" list, we have found a duplicate. A match is made if: the transaction has the same accounts, but in the opposite order; and the amount of money is the same. This duplicate transaction gets ignored, and is removed from the "waiting" set.

If there's anything left in the "waiting" set by the end, a warning is printed to say that transactions are missing.

I'm ignoring transaction dates because there was often a few days of delay between sending and receiving. If theoretically there were two sets of transactions happening around the same time, it shouldn't matter which one cancels which anyway, since the net transfer amount would still match.

This feature was an absolute must-have for me as I'm importing several years of transaction data, and wasn't about to delete duplicates by hand. It seems to be working, but is there a better way? Any feedback welcome.