Skip to content

Instantly share code, notes, and snippets.

@mterwill
Last active February 16, 2024 09:23
Show Gist options
  • Save mterwill/7fdcc573dc1aa158648aacd4e33786e8 to your computer and use it in GitHub Desktop.
Save mterwill/7fdcc573dc1aa158648aacd4e33786e8 to your computer and use it in GitHub Desktop.
Beancount importers, scripts, etc.
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'),
]
#!/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))
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
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
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
#!/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

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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment