Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Convert QFX/OFX to CSV
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)
@rverrips

This comment has been minimized.

Copy link

rverrips commented May 30, 2017

Thanks - Great little script ...
Added the MEMO column

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', 'memo', '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,
            'memo' : transaction.memo,
            '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)  
@jonshell

This comment has been minimized.

Copy link

jonshell commented Jul 13, 2017

Hi there. Awesomely useful utility script, much appreciated. I've updated it for python3, with minor optimizations:

  • added lists to enumerate multiple types of credit and debit transactions.
  • Updated to use python3-style open() instead of file()
  • Forced utf8 on open() (banks especially like to add cp 1252 junk)
  • changed "raise Error" to "raise ValueError" and print the unknown transaction type

Appreciate the work folks. Hopefully these updates will be useful to other folks too.

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

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', 'memo', '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 = []
    credit_transactions = ['credit', 'dep', 'int']
    debit_transactions  = ['debit', 'atm', 'pos', 'xfer', 'check']
    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
    

files = glob("*.qfx")
for qfx_file in files:
    qfx = 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)  

@neheller

This comment has been minimized.

Copy link

neheller commented Apr 13, 2019

Thanks for the script! I had to add some transaction types to get it to run on my file

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

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', 'memo', '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 = []
    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
    

files = glob("*.qfx")
for qfx_file in files:
    qfx = 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)  
@im-alexandre

This comment has been minimized.

Copy link

im-alexandre commented Jan 2, 2020

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

@Haeven

This comment has been minimized.

Copy link

Haeven commented Apr 20, 2020

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.

@egill512

This comment has been minimized.

Copy link

egill512 commented May 8, 2020

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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.