Created
April 19, 2015 23:32
-
-
Save whistler/e7c21c70d1cbb9c4b15d to your computer and use it in GitHub Desktop.
Convert QFX/OFX to CSV
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from csv import DictWriter | |
from glob import glob | |
from ofxparse import OfxParser | |
DATE_FORMAT = "%m/%d/%Y" | |
def write_csv(statement, out_file): | |
print "Writing: " + out_file | |
fields = ['date', 'payee', 'debit', 'credit', 'balance'] | |
with open(out_file, 'w') as f: | |
writer = DictWriter(f, fieldnames=fields) | |
for line in statement: | |
writer.writerow(line) | |
def get_statement_from_qfx(qfx): | |
balance = qfx.account.statement.balance | |
statement = [] | |
for transaction in qfx.account.statement.transactions: | |
credit = "" | |
debit = "" | |
balance = balance + transaction.amount | |
if transaction.type == 'credit': | |
credit = transaction.amount | |
elif transaction.type == 'debit': | |
debit = -transaction.amount | |
else: | |
raise Error("Unknown transaction type") | |
line = { | |
'date': transaction.date.strftime(DATE_FORMAT), | |
'payee': transaction.payee, | |
'debit': debit, | |
'credit': credit, | |
'balance': balance | |
} | |
statement.append(line) | |
return statement | |
files = glob("*.qfx") | |
for qfx_file in files: | |
qfx = OfxParser.parse(file(qfx_file)) | |
statement = get_statement_from_qfx(qfx) | |
out_file = "converted_" + qfx_file.replace(".qfx",".csv") | |
write_csv(statement, out_file) | |
Nice! I added a -i input argument to @schulzebittar 's script so I can select specific files to convert, if I want. Hope that helps.
#!/usr/bin/env` python3
# -*- coding: utf-8 -*-
from csv import DictWriter
from glob import glob
from ofxparse import OfxParser
import json
import urllib.request
import datetime
import argparse
argparser = argparse.ArgumentParser()
argparser.add_argument("-o", "--outputtype", help = "csv or json", default="csv")
argparser.add_argument("-i", "--input", nargs='+', help = "input file(s)", default="*.ofx")
args = argparser.parse_args()
DATE_FORMAT = "%m/%d/%Y"
jsonBody = {}
outputtype = args.outputtype
jsonBody["data"] = []
allStatements = []
def write_csv(statement, out_file):
print("Writing: " + out_file)
fields = ['Date', 'Description (payee)', 'Transaction Type (type)', 'UID', 'Amount',
'sic', 'mcc', 'Notes (memo)', 'Debit', 'Credit', 'Balance', 'FID', 'Organization']
with open(out_file, 'w', newline='') as f:
writer = DictWriter(f, fieldnames=fields)
writer.writeheader()
for line in statement:
writer.writerow(line)
def get_statement_from_qfx(qfx):
#print(qfx.account.account_id)
#print(qfx.account.institution.organization)
#print(qfx.account.institution.fid)
balance = qfx.account.statement.balance
statement = []
credit_transactions = ['credit', 'dep', 'int', 'directdep']
debit_transactions = ['debit', 'atm', 'pos',
'xfer', 'check', 'fee', 'payment']
other_transactions = ['other']
for transaction in qfx.account.statement.transactions:
#print(transaction.type)
credit = ""
debit = ""
balance = balance + transaction.amount
if transaction.type in credit_transactions:
credit = transaction.amount
elif transaction.type in debit_transactions:
debit = -transaction.amount
elif transaction.type in other_transactions:
if transaction.amount < 0:
debit = -transaction.amount
else:
credit = transaction.amount
else:
raise ValueError("Unknown transaction type:" + transaction.type)
line = {
'Date': transaction.date.strftime(DATE_FORMAT),
'Description (payee)': transaction.payee,
'Transaction Type (type)': transaction.type,
'Notes (memo)': transaction.memo,
'UID': transaction.id,
'Amount': str(transaction.amount),
'sic': transaction.sic,
'mcc': transaction.mcc,
'Debit': str(debit),
'Credit': str(credit),
'Balance': str(balance),
'FID': qfx.account.institution.fid,
'Organization': qfx.account.institution.organization}
statement.append(line)
jsonBody["data"].append(line)
return statement
files = glob(args.input)
for qfx_file in files:
qfx = OfxParser.parse(open(qfx_file, encoding="latin-1"))
statement = get_statement_from_qfx(qfx)
allStatements = allStatements + statement
#print(statement)
if outputtype == 'csv' and (len(args.input) > 1 or args.input == '*.ofx'):
out_file = "convert ed_" + qfx_file.replace(".ofx", ".csv")
write_csv(statement, out_file)
if outputtype == 'singlecsv':
out_file = "qfx-transactions.csv"
write_csv(allStatements, out_file)
elif outputtype == 'json' and (len(args.input) > 1 or args.input == '*.ofx'):
print('json output supports one file only')
else:
with open('qfx-transactions.json', 'w') as outfile:
json.dump(jsonBody, outfile)
I make in more generic to extract as much data from the QFX:
here the code: https://github.com/nhatkhai/ofx2csv/blob/main/ofx2csv.py
I make in more generic to extract as much data from the QFX: here the code: https://github.com/nhatkhai/ofx2csv/blob/main/ofx2csv.py
This looks good. I opened a PR to fix a few little bugs nhatkhai/ofx2csv#1
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Thanks for this.
Here's a fork that turns this into a CLI with various input and output formats: https://gist.github.com/abirchall/54b0f2eb47d02eb3e36e33b3077298a5
You can run:
pip3 install ofxparse
.You can find info about this library here: https://pypi.org/project/ofxparse/
The underlying code is simple enough, just a bunch of parsing logic specific to the OFX spec.