Convert QFX/OFX to CSV

ofx2csv.py

#!/usr/bin/env python3

"""
Parse .qfx or .ofx files and output as .json or .csv format.
"""

from typing import Dict, List, Optional

import argparse
import math
import os
import sys

from csv import DictWriter
from datetime import datetime
from decimal import Decimal
from glob import glob

# Must `pip3 install ofxparse` prior to script execution.
# See https://pypi.org/project/ofxparse/ for library details.
import ofxparse


DEFAULT_DATE_FORMAT = "%Y-%m-%d"
DEFAULT_EXTENSION = "qfx"


def get_arg_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser()

    # require one of the --file or --directory arg
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument(
        "-f", "--file", action="store", type=str,
        help="Convert this file.")
    group.add_argument(
        "-d", "--directory", action="store", type=str,
        help="Convert all matching files in this directory.")

    parser.add_argument(
        "-e", "--extension", action="store", choices=["qfx", "ofx"],
        nargs="?", default=DEFAULT_EXTENSION, type=str,
        help=("Search for this file extension when using --directory. Default: '"
              + DEFAULT_EXTENSION
              + "'"))

    parser.add_argument(
        "--date-format", action="store",
        nargs="?", default=DEFAULT_DATE_FORMAT, type=str,
        help=("Format for dates. Default: '"
            + DEFAULT_DATE_FORMAT.replace("%", "%%")
            + "'"))

    parser.add_argument(
        "--output-format", action="store",
        choices=["json-print", "csv-print", "csv-file"],
        nargs="?", default="json-print", type=str,
        help="Output Type. Default: 'json-print'")

    # TODO add an arg to control overwrite vs adding a new numbered file

    return parser


def write_csv(
    account_info: Dict[str, any],
    transactions: List[Dict[str, any]],
    handle
) -> None:

    # gather all the transaction columns which are present
    tr_fields_present = set()
    for tr in transactions:
        tr_fields_present.update(tr.keys())

    # fix order of transaction columns
    tr_fields = list(filter(
        lambda field: field in tr_fields_present,
        [
            "Transaction ID", "Activity Type",
            "Activity Date", "Transaction Date", "Settlement Date",
            "Amount", "Quantity", "Unit Price",
            "Security ID", "Security Ticker", "Security Name",
            "Fees", "Commission", "SIC", "MCC", "Payee", "Check Number", "Memo",
        ]
    ))

    # write account info
    account_info_writer = DictWriter(handle, fieldnames=account_info.keys())
    account_info_writer.writeheader()
    account_info_writer.writerow(account_info)

    # write an empty line between the account info and transactions
    account_info_writer.writerow({})

    # write transactions
    tr_writer = DictWriter(handle, fieldnames=tr_fields)
    tr_writer.writeheader()
    tr_writer.writerows(transactions)


def get_transactions(
    args: argparse.Namespace,
    ofx: ofxparse.ofxparse.Ofx,
    security_list: Dict[str, Dict[str, any]]
) -> List[Dict[str, any]]:

    transactions = []
    def add_transaction(tr: Dict[str, any]) -> None:
        transactions.append({k:v for k,v in tr.items() if v})

    # Financial Institution Transaction ID
    def parse_transaction_id(tr: Dict[str, any]) -> any:
        # this member is not initialized in the InvestmentTransaction constructor
        transaction_id = tr.id if hasattr(tr, "id") else ""

        # truncate ".00000" from transaction_id
        if not math.isnan(float(transaction_id)):
            transaction_id = float(transaction_id)
            if transaction_id.is_integer():
                transaction_id = int(transaction_id)
        return transaction_id

    def parse_date(d: Optional[datetime]) -> Optional[str]:
        return d.strftime(args.date_format) if d else None

    def parse_float(v: Optional[Decimal]) -> Optional[float]:
        return float(v) if v is not None else None

    for tr in ofx.account.statement.transactions:
        if type(tr) == ofxparse.ofxparse.InvestmentTransaction:
            activity_type = tr.type
            if tr.type == "income":
                activity_type = tr.income_type
            if tr.type == "transfer":
                activity_type += " " + tr.tferaction

            sec = security_list[tr.security] if tr.security in security_list else {}
            add_transaction({
                "Transaction ID": parse_transaction_id(tr),
                "Activity Type": activity_type.lower(),
                "Activity Date": parse_date(tr.tradeDate),
                "Settlement Date": parse_date(tr.settleDate),
                "Security ID": tr.security,
                "Security Ticker": sec["Ticker"] if "Ticker" in sec else None,
                "Amount": parse_float(tr.total),
                "Quantity": parse_float(tr.units),
                "Unit Price": parse_float(tr.unit_price),
                "Memo": tr.memo,
                "Security Name": sec["Name"] if "Name" in sec else None,
                "Fees": parse_float(tr.fees),
                "Commission": parse_float(tr.commission),
            })

        elif type(tr) == ofxparse.ofxparse.Transaction:
            add_transaction({
                "Transaction ID": parse_transaction_id(tr),
                "Activity Type": tr.type,
                "Activity Date": parse_date(tr.date),
                "Transaction Date": parse_date(tr.user_date),
                "Amount": parse_float(tr.amount),
                "Memo": tr.memo,
                "SIC": tr.sic,
                "MCC": tr.mcc,
                "Payee": tr.payee,
                "Check Number": tr.checknum,
            })

        else:
            print("Unknown transaction type!!!")
    return transactions


def get_account_info(ofx: ofxparse.ofxparse.Ofx) -> Dict[str, any]:
    account = ofx.account

    info = {
        "curdef": account.curdef,
        "account_id": account.account_id,
        "routing_number": account.routing_number,
        "branch_id": account.branch_id,
        "account_type": account.account_type,
    }

    # parse institution info
    inst = account.institution
    if inst:
        info["institution_organization"] = inst.organization
        info["institution_fid"] = inst.fid

    if account.type == ofxparse.AccountType.Investment:
        info["brokerid"] = account.brokerid

    # filter down to set values
    return {k:v for k,v in info.items() if v}


def get_security_list(
    args: argparse.Namespace,
    ofx: ofxparse.ofxparse.Ofx
) -> Dict[str, Dict[str, any]]:
    """
    Return a dictionry indexed by Security ID.
    """
    if not hasattr(ofx, "security_list"):
        return {}

    securities = {}
    for sec in ofx.security_list:
        securities[sec.uniqueid] = { k:v for k,v in {
            "ID": sec.uniqueid,
            "Ticker": sec.ticker,
            "Name": sec.name,
            "Memo": sec.memo,
        }.items() if v }
    return securities


def parse_file(args: argparse.Namespace, file_name: str) -> None:
    if not os.path.exists(file_name):
        raise Exception("File not found: " + file_name)

    file = open(file_name, "r")
    ofx = ofxparse.OfxParser.parse(file)

    account_info = get_account_info(ofx)
    security_list = get_security_list(args, ofx)
    transactions = get_transactions(args, ofx, security_list)

    if args.output_format == "json-print":
        print(account_info)
        for tx in transactions:
            print(tx)

    if args.output_format == "csv-file":
        (file_root, file_ext) = os.path.splitext(file_name)
        out_file_name = file_root + ".csv"

        print("Writing to: " + out_file_name)
        with open(out_file_name, "w") as handle:
            write_csv(account_info, transactions, handle)

    if args.output_format == "csv-print":
        with sys.stdout as handle:
            write_csv(account_info, transactions, handle)


def main() -> None:
    parser = get_arg_parser()
    args = parser.parse_args()

    # attempt to parse a single file
    file_name = args.file
    if file_name:
        parse_file(args, file_name)

    # attempt to parse all files in the directory
    directory_name = args.directory
    if directory_name:
        files = glob("*." + args.extension)
        for file_name in files:
            parse_file(args, file_name)


main()