Beancount importers, scripts, etc.

config.py

import os, sys

# beancount doesn't run from this directory
sys.path.append(os.path.dirname(__file__))

# importers located in the importers directory
from importers import amex, chase, schwab, citi

CONFIG = [
    chase.ChaseCCImporter('Liabilities:CC:Chase:Reserve', '0000'),
    chase.ChaseBankImporter('Assets:Chase:Checking', '0000'),
    schwab.SchwabBankImporter('Assets:Schwab:Checking', '8000'),
]

dedup.py

#!/usr/bin/env python3

"""
Helper script to find merchants close to one another.
"""

from fuzzywuzzy import fuzz
from itertools import combinations
from subprocess import check_output

# tuples to exclude as valid
valid = [
    ('Proper Grounds', 'Higher Grounds'),
]

merchants = check_output(['bean-query', '-q', 'mwt.beancount', 'select distinct(payee)']).decode('utf-8')
merchants = [a.strip() for a in merchants.splitlines()]

for pair in combinations(merchants, 2):
    ratio = fuzz.ratio(*pair)
    if ratio > 70 and pair not in valid:
        print("'{}' is close to '{}'".format(*pair))

importers-amex.py

from beancount.core.number import D
from beancount.ingest import importer
from beancount.core import account
from beancount.core import amount
from beancount.core import flags
from beancount.core import data
from beancount.core.position import Cost

from dateutil.parser import parse

from titlecase import titlecase

import csv
import os
import re

class AmexImporter(importer.ImporterProtocol):
    def identify(self, f):
        # TODO: accept account in the constructor
        if not re.match('Transactions.*\.csv', os.path.basename(f.name)):
            return False

        if 'skymiles' in f.name.lower():
            self.account = 'Liabilities:CC:Amex:SkyMiles-Platinum'
        elif 'reserve' in f.name.lower():
            self.account = 'Liabilities:CC:Amex:SkyMiles-Reserve'
        else:
            raise RuntimeError('Please include SkyMiles or Reserve in the filename')

        return True

    def extract(self, f):
        entries = []

        with open(f.name) as f:
            for index, row in enumerate(csv.reader(f)):
                trans_date = parse(row[0].split(' ')[0]).date()
                trans_desc = titlecase(row[2])
                trans_amt  = row[7]

                if trans_desc == "Online Payment - Thank You":
                    # Record these at the bank level
                    continue

                if trans_desc == "Payment Received - Thank You":
                    # Record these at the bank level
                    continue

                meta = data.new_metadata(f.name, index)

                txn = data.Transaction(
                    meta=meta,
                    date=trans_date,
                    flag=flags.FLAG_OKAY,
                    payee=trans_desc,
                    narration="",
                    tags=set(),
                    links=set(),
                    postings=[],
                )

                txn.postings.append(
                    data.Posting(
                        self.account,
                        amount.Amount(-1*D(trans_amt), 'USD'),
                        None, None, None, None
                    )
                )

                entries.append(txn)

        return entries

importers-chase.py

from beancount.core.number import D
from beancount.ingest import importer
from beancount.core import account
from beancount.core import amount
from beancount.core import flags
from beancount.core import data
from beancount.core.position import Cost

from dateutil.parser import parse

from titlecase import titlecase

import csv
import os
import re

class ChaseCCImporter(importer.ImporterProtocol):
    def __init__(self, account, lastfour):
        self.account = account
        self.lastfour = lastfour

    def identify(self, f):
        return re.match('Chase{}.*\.CSV'.format(self.lastfour), os.path.basename(f.name))

    def extract(self, f):
        entries = []

        with open(f.name) as f:
            for index, row in enumerate(csv.DictReader(f)):
                trans_date = parse(row['Trans Date']).date()
                trans_desc = titlecase(row['Description'])
                trans_amt  = row['Amount']

                meta = data.new_metadata(f.name, index)

                txn = data.Transaction(
                    meta=meta,
                    date=trans_date,
                    flag=flags.FLAG_OKAY,
                    payee=trans_desc,
                    narration="",
                    tags=set(),
                    links=set(),
                    postings=[],
                )

                txn.postings.append(
                    data.Posting(self.account, amount.Amount(D(trans_amt),
                        'USD'), None, None, None, None)
                )

                entries.append(txn)

        return entries

class ChaseBankImporter(importer.ImporterProtocol):
    def __init__(self, account, lastfour):
        self.account = account
        self.lastfour = lastfour

    def identify(self, f):
        return re.match('Chase{}.*\.CSV'.format(self.lastfour), os.path.basename(f.name))

    def extract(self, f):
        entries = []

        with open(f.name) as f:
            for index, row in enumerate(csv.DictReader(f)):
                trans_date = parse(row['Posting Date']).date()
                trans_desc = titlecase(row['Description'])
                trans_amt  = row['Amount']

                meta = data.new_metadata(f.name, index)

                txn = data.Transaction(
                    meta=meta,
                    date=trans_date,
                    flag=flags.FLAG_OKAY,
                    payee=trans_desc,
                    narration="",
                    tags=set(),
                    links=set(),
                    postings=[],
                )

                txn.postings.append(
                    data.Posting(self.account, amount.Amount(D(trans_amt),
                        'USD'), None, None, None, None)
                )

                entries.append(txn)

        return entries

importers-citi.py

from beancount.core.number import D
from beancount.ingest import importer
from beancount.core import amount
from beancount.core import flags
from beancount.core import data

from dateutil.parser import parse

from titlecase import titlecase

import csv
import os
import re

class CitiImporter(importer.ImporterProtocol):
    def __init__(self):
        self.account = 'Liabilities:CC:Citi:ThankYou'

    def identify(self, f):
        return re.match('MC_377_CURRENT_VIEW.CSV', os.path.basename(f.name))

    def file_account(self, f):
        return self.account

    def extract(self, f):
        entries = []

        with open(f.name) as f:
            for index, row in enumerate(csv.DictReader(f)):

                trans_date = parse(row['Date']).date()

                #rstrip removes trailing new line characters
                trans_desc = titlecase(row['Description'].rstrip())

                if bool(row['Debit']):
                    trans_amt = "-" + row['Debit']
                else:
                    trans_amt = row['Credit']

                meta = data.new_metadata(f.name, index)

                txn = data.Transaction(
                    meta=meta,
                    date=trans_date,
                    flag=flags.FLAG_OKAY,
                    payee=trans_desc,
                    narration="",
                    tags=set(),
                    links=set(),
                    postings=[]
                )

                txn.postings.append(
                    data.Posting(self.account, amount.Amount(D(trans_amt),
                        'USD'), None, None, None, None)
                )

                entries.append(txn)

        return entries

importers-schwab.py

#!/usr/bin/env python3

from beancount.core.number import D
from beancount.ingest import importer
from beancount.core import account
from beancount.core import amount
from beancount.core import flags
from beancount.core import data
from beancount.core.position import Cost

from dateutil.parser import parse

from titlecase import titlecase

import csv
import os
import re

class SchwabBankImporter(importer.ImporterProtocol):
    def __init__(self, account, lastfour):
        self.account = account
        self.lastfour = lastfour

    def identify(self, f):
        return re.match('XXXXXX.*{}_Checking_Transactions_.*\.CSV'.format(self.lastfour), os.path.basename(f.name))

    def extract(self, f):
        entries = []

        with open(f.name) as f:
            for _ in range(3): # first 3 lines are garbage
                next(f)

            for index, row in enumerate(csv.reader(f)):
                trans_date = parse(row[0]).date()
                trans_desc = titlecase(row[3])

                if row[4]:
                    trans_amt = float(row[4].strip('$')) * -1.
                elif row[5]:
                    trans_amt = float(row[5].strip('$'))
                else:
                    continue # 0 dollar transaction

                trans_amt = '{:.2f}'.format(trans_amt)

                meta = data.new_metadata(f.name, index)

                txn = data.Transaction(
                    meta=meta,
                    date=trans_date,
                    flag=flags.FLAG_OKAY,
                    payee=trans_desc,
                    narration="",
                    tags=set(),
                    links=set(),
                    postings=[],
                )

                txn.postings.append(
                    data.Posting(self.account, amount.Amount(D(trans_amt),
                        'USD'), None, None, None, None)
                )

                entries.append(txn)

        return entries

USAGE.md

Note: everything here is pretty specific to my usage/accounts and not written for public use... You'll probably have to tweak a bunch of stuff.

$ bean-extract config.py ~/Downloads # the csvs should be in here