merge_fees.py

#!/usr/bin/python

"""
Merge bank transaction fees into the main transaction in GnuCash.

This script searches a given account for "fee" transactions with a certain
description. It then tries to find the corresponding "real" transaction, adds
the fee amount to the real transaction, and deletes the fee transaction.

Assumptions: "fee" transactions come (not necessarily consecutively) after
their "real" transactions on the same date, and the ratio is predictable.
"""

from datetime import datetime, timedelta
import logging
import re
import sys

import gnucash

# Configuration
file_name = sys.argv[1]
account_name = ["Assets", "Bank", "Checking Account"]
fee_pattern = re.compile("^Debit Card Fee  ")
ratio_expected = 0.008
ratio_tolerance = 0.02
time_limit = datetime.today() - timedelta(60)

# Initialization
logging.basicConfig(level=logging.INFO)

session = gnucash.Session(file_name, is_new=False, ignore_lock=True)

account = session.book.get_root_account()
for name in account_name:
    account = account.lookup_by_name(name)

logging.info("Balance is {}".format(account.GetBalance().to_double()))

# Loop
matched = set() # deleted fee transactions and matched real transactions
splits = account.GetSplitList()

for fee_index, fee in enumerate(splits):
    # Filter by description and date range
    if (re.search(fee_pattern, fee.parent.GetDescription()) is None or
        datetime.fromtimestamp(fee.parent.GetDate()) < time_limit):
        continue

    # Walk backwards to find the corresponding "real" transaction
    real = None
    for j in range(fee_index - 1, 0, -1):
        if j in matched:
            continue

        if (abs(splits[j].parent.GetDate() - fee.parent.GetDate()) >
            timedelta(1).total_seconds()):
            break

        if abs(splits[j].GetAmount().to_double() * ratio_expected -
            fee.GetAmount().to_double()) < ratio_tolerance:

            # Sanity checks
            txn_splits = splits[j].parent.GetSplitList()
            if len(txn_splits) != 2:
                logging.warn("Skipping transaction {} with {} splits".format(
                    splits[j].parent.GetDescription(), len(txn_splits)))
                continue

            other = txn_splits[0]
            if not other.GetAmount().eq(splits[j].GetAmount().neg()):
                logging.warn("Skipping transaction {}: {} vs. {}".format(
                    splits[j].parent.GetDescription(),
                    other.GetAmount().to_double(),
                    splits[j].GetAmount().to_double()))
                continue

            real = splits[j]
            matched.add(j)
            break

    if real is None:
        logging.warn("No match found for {0} ({1})".format(
            fee.GetAmount().to_double(),
            fee.parent.GetDescription()))
        continue

    # Add the amounts
    logging.info("Merging {0} ({1}) with {2} ({3})".format(
        real.GetAmount().to_double(),
        real.parent.GetDescription(),
        fee.GetAmount().to_double(),
        fee.parent.GetDescription()))

    new_amt = real.GetAmount().add(fee.GetAmount(),
            gnucash.GNC_DENOM_AUTO, gnucash.GNC_HOW_RND_NEVER)

    logging.debug("New amount is {}".format(new_amt.to_double()))

    # Update the real transaction
    real.parent.BeginEdit()
    real.SetValue(new_amt)
    other.SetValue(new_amt.neg())
    real.parent.CommitEdit()

    # And delete the fee transaction
    fee.parent.Destroy()
    matched.add(fee_index)

logging.info("Balance is {}".format(account.GetBalance().to_double()))

#session.save()
session.end()