Convert QFX/OFX to CSV

ofx2csv.py

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)
    

Great!!!!
I was looking for exactly the same thing. Thanks for share!!!

I get the following error:
UnicodeDecodeError: 'ascii' codec can't decode byte 0xe2 in position 14: ordinal not in range(128)
Working on resolving it now.

Thanks for the script! I like many others have found it useful. I was looking to convert an investment statement qfx that I downloaded from Wealthfront. Thus, I ended up modifying the script to support both investment and bank statements. I'm adding it here like others have done as someone else may find it useful.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from csv import DictWriter
from glob import glob
import ofxparse
import csv


DATE_FORMAT = "%m/%d/%Y"

def write_csv(statement, out_file):
    if len(statement) == 0:
        print("Empty statement. Not outputting conversion")
        return
    print("Writing: " + out_file)
    header = statement[0].keys()
    with open(out_file, 'w', newline='') as f:
        writer = DictWriter(f, fieldnames=header)
        wr = csv.writer(f, dialect='excel')
        wr.writerow(header)
        writer.writerows(statement)


def get_bank_statement_from_qfx(qfx):
    balance = qfx.account.statement.balance
    statement = []
    credit_transactions = ['credit', 'dep', 'int', 'directdep']
    debit_transactions  = ['debit', 'atm', 'pos', 'xfer', 'check', 'fee', 'payment']
    for transaction in qfx.account.statement.transactions:
        credit = ""
        debit = ""
        balance = balance + transaction.amount
        if transaction.type in credit_transactions:
            credit = transaction.amount
        elif transaction.type in debit_transactions:
            debit = -transaction.amount
        else:
            raise ValueError("Unknown transaction type:" + transaction.type)
        line = {
            'date': transaction.date.strftime(DATE_FORMAT),
            'payee': transaction.payee,
            'memo' : transaction.memo,
            'debit': debit,
            'credit': credit,
            'balance': balance
        }
        statement.append(line)
    return statement


def get_investment_transactions_statement_from_qfx(qfx):
    statement = []
    for transaction in qfx.account.statement.transactions:
        line = {
            'type': getattr(transaction, 'type', ''),
            'tradeDate': getattr(transaction, 'tradeDate', ''),
            'settleDate': getattr(transaction, 'settleDate', ''),
            'memo': getattr(transaction, 'memo', ''),
            'income_type': getattr(transaction, 'income_type', ''),
            'units': getattr(transaction, 'units', ''),
            'unit_price': getattr(transaction, 'unit_price', ''),
            'comission': getattr(transaction, 'comission', ''),
            'fees': getattr(transaction, 'fees', ''),
            'total': getattr(transaction, 'total', ''),
            'tferaction': getattr(transaction, 'tferaction', '')
        }
        line.update(get_security_info(getattr(transaction, 'security', ''), qfx))
        statement.append(line)
    return statement


def get_security_info(sec_id, qfx):
    sec_dict = {
        'securityid':'',
        'securityname': '',
        'securityticker': '',
        'securitymemo': ''
    }
    if sec_id == '' or sec_id is None:
        return sec_dict
    for sec in qfx.security_list:
        if sec.uniqueid == sec_id :
            sec_dict['securityid'] = getattr(sec,'uniqueid','')
            sec_dict['securityname'] = getattr(sec,'name','')
            sec_dict['securityticker'] = getattr(sec,'ticker','')
            sec_dict['securitymemo'] = getattr(sec,'memo','')
            return sec_dict


def get_statement_from_qfx(qfx):
    if isinstance(qfx.account.statement, ofxparse.ofxparse.InvestmentStatement):
        return get_investment_transactions_statement_from_qfx(qfx)
    else:
        return get_bank_statement_from_qfx(qfx)


files = glob("*.qfx")
for qfx_file in files:
    qfx = ofxparse.OfxParser.parse(open(qfx_file, encoding="latin-1"))
    statement = get_statement_from_qfx(qfx)
    out_file = "converted_" + qfx_file.replace(".qfx",".csv")
    write_csv(statement, out_file)

Very useful. I made some revisions and added support for an optional parameter that allows the output to be a single combined CSV file or a JSON file. Also, repeating account FID and organization per row.

#!/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 = "singlecsv, json, or manycsv", default="singlecsv")
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']
    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
        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("*.qfx")
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 == 'manycsv':
        out_file = "converted_" + qfx_file.replace(".qfx", ".csv")
        write_csv(statement, out_file)
if outputtype == 'singlecsv':
    out_file = "qfx-transactions.csv"
    write_csv(allStatements, out_file)
else:
    with open('qfx-transactions.json', 'w') as outfile:
        json.dump(jsonBody, outfile)

Great find. I modified this script to pull out data for my 401K contributions from John Hancock's website.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Adapted from https://gist.github.com/whistler/e7c21c70d1cbb9c4b15d

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', 'memo', 'ticker', 'cusip', 'unit price', 'units', 'total price']
    with open(out_file, 'w', newline='') 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 = []
    ticker_list = []
    investment_transactions = ['buymf', 'sellmf']

    # Get all tickers from file
    for fund in qfx.security_list:
        line = {
            'name': fund.name,
            'ticker': fund.ticker,
            'cusip': fund.uniqueid
        }
        ticker_list.append(line)

    # Create the statement
    for transaction in qfx.account.statement.transactions:
        ticker = ""
        cusip = ""
        for fund in ticker_list:
            if fund['name'] == transaction.memo:
                ticker = fund['ticker']
                cusip = fund['cusip']
                break

        if transaction.type in investment_transactions:
            line = {
                'date': transaction.tradeDate.strftime(DATE_FORMAT),
                'memo': transaction.memo,
                'ticker': ticker,
                'cusip': cusip,
                'unit price': transaction.unit_price,
                'units': transaction.units,
                'total price': transaction.unit_price * transaction.units
            }
            statement.append(line)

    return statement

files = glob("quicken-statement-file.qfx")
for qfx_file in files:
    qfx = OfxParser.parse(open(qfx_file, encoding="latin-1"))
    fund_list = get_statement_from_qfx(qfx)
    out_file = "converted_" + qfx_file.replace(".qfx", ".csv")
    write_csv(fund_list, out_file)

can anyone give me a quick help on how to run it?

Thank you very much for this script.
It didn't run on a specific .OFX from one of my bank accounts (Banco do Brasil > Brazil) because they export the file with transaction type to 'OTHER' as default. So I added an extra if statement under get_statement_from_qfx(qfx) for this case. Not sure if the best solution, but now it's working as with other files.

Based on @basophil01

#!/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 = "singlecsv, json, or manycsv", default="singlecsv")
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("*.ofx")
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 == 'manycsv':
        out_file = "converted_" + qfx_file.replace(".ofx", ".csv")
        write_csv(statement, out_file)
if outputtype == 'singlecsv':
    out_file = "qfx-transactions.csv"
    write_csv(allStatements, out_file)
else:
    with open('qfx-transactions.json', 'w') as outfile:
        json.dump(jsonBody, outfile)

@Vinimuller

can anyone give me a quick help on how to run it?

Have python installed, save the code into a .py file, put it in the same folder as your .ofx files and run the .py file.

Have python installed, save the code into a .py file, put it in the same folder as your .ofx files and run the .py file.

@arthurbittar
Thanks!

Need to install some non-standard libraries -- any help there?

Need to install some non-standard libraries -- any help there?

I ended up using this code

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

[@oravecz] Need to install some non-standard libraries -- any help there?

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.

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