csv2ofx

To handle CSV files with non-standard field names, you must create a mapping object. A mapping is a dictionary where keys are OFX/QIF attributes and values are functions (or callables) that extract the value from a CSV record (a dictionary representing a row).

Required Attributes:

• account: Transaction account
• date: Transaction date
• amount: Transaction amount

Optional Attributes:

• desc: Transaction description
• payee: Transaction payee
• notes: Transaction notes
• check_num: Check or transaction number
• id: Transaction ID
• bank: Bank name
• account: Transaction account type (e.g., checking, savings)
• account_id: Transaction account ID
• type: Transaction type (CREDIT or DEBIT)
• balance: Account balance
• class: Transaction class

Example Mapping (using itemgetter):

from operator import itemgetter

mapping = {
    'bank': 'BetterBank',
    'account': itemgetter('Account'),
    'date': itemgetter('Date'),
    'amount': itemgetter('Amount')
}

Example Mapping (using complex logic):

mapping = {
    'account': lambda r: r['Details'].split(':')[0],
    'date': lambda r: '%s/%s/%s' % (r['Month'], r['Day'], r['Year']),
    'amount': lambda r: r['Amount'] * 2,
    'first_row': 1,
    'last_row': 10,
    'filter': lambda r: float(r['Amount']) > 10,
}

New mappings should be placed in the csv2ofx/mappings folder.

from operator import itemgetter

mapping = {
    'bank': 'BetterBank',
    'account': itemgetter('Account'),
    'date': itemgetter('Date'),
    'amount': itemgetter('Amount')}

To use csv2ofx within a Python script to generate OFX files, follow this workflow:

1. Initialize an OFX object with a mapping.
2. Read the CSV records (e.g., using meza.io.read_csv).
3. Generate groups, transactions, and clean them using the OFX instance methods.
4. Use utils.gen_data to prepare the data.
5. Chain the header, body, and footer to create the final content stream.

import itertools as it

from meza.io import read_csv, IterStringIO
from csv2ofx import utils
from csv2ofx.ofx import OFX
from csv2ofx.mappings.default import mapping

ofx = OFX(mapping)
records = read_csv('path/to/file.csv', has_header=True)
groups = ofx.gen_groups(records)
trxns = ofx.gen_trxns(groups)
cleaned_trxns = ofx.clean_trxns(trxns)
data = utils.gen_data(cleaned_trxns)
content = it.chain([ofx.header(), ofx.gen_body(data), ofx.footer()])

for line in IterStringIO(content):
    print(line)

When defining mappings or using the library, you can control how the CSV data is interpreted using these attributes:

These methods manage the lifecycle of an account statement in the OFX file.

account_start() Generates the <STMTRS> and <BANKACCTFROM> blocks.

• Required Kwargs: currency, bank_id, account_id, account_type (one of CHECKING, SAVINGS, MONEYMRKT, CREDITLINE).
• Note: If ms_money is enabled, dates are automatically adjusted to Noon GMT (e.g., YYYYMMDD120000) to ensure compatibility.

account_end() Generates the <LEDGERBAL> and closing tags.

• Required Kwargs: date (datetime), balance (number).
• Automatic Balance Detection: If balance is not provided in kwargs, the class attempts to infer the ending balance from the transaction history by analyzing date and balance order (ascending vs descending).

import datetime
from csv2ofx.ofx import OFX

ofx = OFX()

# Account Start
start_block = ofx.account_start(
    currency='USD', 
    bank_id='1', 
    account_id='1', 
    account_type='CHECKING', 
    start=datetime.datetime(2012, 1, 1), 
    end=datetime.datetime(2012, 2, 1)
)

# Account End
end_block = ofx.account_end(
    date=datetime.datetime(2012, 1, 15), 
    balance=150.00
)

The get(name, trxn=None, default=None) method is used to retrieve transaction attributes. It is highly flexible: it can return a static attribute of the Content instance, a value from a mapping function applied to the transaction, or a value from the mapping dictionary itself.

• If name refers to a normal attribute on Content, it returns that attribute.
• If name refers to a mapping function, it calls that function with trxn as an argument.
• If the attribute is not found, it returns the default value.

import datetime
from csv2ofx.mappings.mint import mapping

trxn = {'Transaction Type': 'DEBIT', 'Amount': 1000.00}
content = Content(mapping)

# Get a normal attribute
start_date = content.get('start')

# Get a mapping function value (requires the transaction dict)
amount = content.get('amount', trxn)

# Get a mapping attribute
has_header = content.get('has_header')

For transactions involving transfers between accounts, use the following methods:

• transfer(**kwargs): Generates the <INTRARS> block. Requires account_type, currency, id, amount, bank_id, and account_id.
• split_content(**kwargs): Generates the <BANKACCTTO> block. If split_account is provided in kwargs, it uses the split account details; otherwise, it uses the primary account details.
• transfer_end(**kwargs): Generates the closing <XFERINFO> and <INTRARS> tags. Requires date (datetime).

Note: OFX does not support more than 2 splits per transaction. gen_body will raise a TypeError if a group contains more than 2 splits.

import datetime
from csv2ofx.ofx import OFX

ofx = OFX()
kwargs = {
    'currency': 'USD', 
    'date': datetime.datetime(2012, 1, 15), 
    'bank_id': 1, 
    'account_id': 1, 
    'account_type': 'CHECKING', 
    'amount': 100, 
    'id': 'jbaevf',
    'split_account': 'Checking',
    'split_account_id': 2,
    'split_account_type': 'CHECKING'
}

transfer_start = ofx.transfer(**kwargs)
split_part = ofx.split_content(**kwargs)
transfer_end = ofx.transfer_end(date=kwargs['date'])