https://fanqiuwen.com/ledger-ofx/

ofx2ledger.py

#!/home/fanqiuwen/ledger/venv/bin/python

import os
import re
import subprocess
import tempfile

from ofxclient import Institution, CreditCardAccount

DOWNLOAD_DAYS = 5
LEDGER_FILENAME = '/home/fanqiuwen/ledger/journal.ledger'

def get_pass(pass_name):
    return subprocess.check_output(['pass', pass_name]).splitlines()[0].decode()

def ofx2csv(account, account_name, days, fout):
    try:
        trans = account.transactions(days)
    except Exception as e:
        print("Failed to download for %s: %s" % (account_name, e))
        return False

    if not trans:
        print("No transactions for %s" % account_name)
        return False

    for tran in sorted(trans, key=lambda x: x.date):
        line = ("%s,%s,\"%s\"" % (tran.date.strftime('%Y/%m/%d'),
                                  tran.amount, tran.payee))
        if tran.amount == 0:
            print("Skipping " + line)
        else:
            fout.write(line + "\n")
    fout.flush()
    return True

def csv2ledger(account_name, csv_filename, fout):
    with tempfile.NamedTemporaryFile('r') as tmp_ledger:
        subprocess.run(['reckon', '-f', csv_filename, '-a', account_name,
                        '-l', LEDGER_FILENAME, '-o', tmp_ledger.name])

        for line in tmp_ledger.readlines():
            line = re.sub(r'^\t(\S+)\s+(\$.*)', r'  \1\t\2', line)
            line = re.sub(r'^\t(\S+)\s+-\$.*', r'  \1', line)
            fout.write(line)

chase_args = {
    'id': '10898',
    'org': 'B1',
    'url': 'https://ofx.chase.com',
    'username': 'REDACTED',
    'password': get_pass('sites/chase.com'),
    'client_args': {
        'id': 'REDACTED',  # 32-char UUID that identifies this client
        'app_id': 'QWIN',
        'app_version': '2200',
        'ofx_version': '103'
    }
}

discover_args = {
    'id': '7101',
    'org': 'Discover Financial Services',
    'url': 'https://ofx.discovercard.com',
    'username': 'REDACTED',
    'password': get_pass('sites/discover.com'),
    'client_args': {
        'app_id': 'QWIN',
        'app_version': '2200',
        'ofx_version': '102'
    }
}

chase_sr = CreditCardAccount(institution=Institution(**chase_args),
                             number='REDACTED')
discover = CreditCardAccount(institution=Institution(**discover_args),
                             number='REDACTED')

if __name__ == '__main__':
    accounts = [
        (chase_sr, 'Liability:ChaseSR'),
        (discover, 'Liability:Discover')
    ]

    with open(LEDGER_FILENAME, 'a') as file_ledger:
        for account, account_name in accounts:
            with tempfile.NamedTemporaryFile('w') as file_csv:
                if ofx2csv(account, account_name, DOWNLOAD_DAYS, file_csv):
                    csv2ledger(account_name, file_csv.name, file_ledger)