Changeset - 31146b884361
Parent rev.
Child rev.
[Not reviewed]
0 1 0
Ben Sturmfels (bsturmfels) - 2 years ago 2022-02-09 01:29:44
ben@sturm.com.au
reconcile: Add initial FR support to statement reconciler.
1 file changed with 47 insertions and 13 deletions:
conservancy_beancount/reconcile/prototype_amex_reconciler.py
47
13
0 comments (0 inline, 0 general)
conservancy_beancount/reconcile/prototype_amex_reconciler.py
➞
Show inline comments
...
 
@@ -2,42 +2,54 @@
 

			




	
	
	
		
 
Run like this:


			




	
	
	
		
 

			




	
	
	
		
 
$ python3 -m pip install thefuzz


			




	
	
	
		
 
$ python3 conservancy_beancount/reconcile/prototype_amex_reconciler.py \


			




	
	
	
		
 
  --beancount-file=$HOME/conservancy/beancount/books/2021.beancount \


			




	
	
	
		
 
  --amex-csv=$HOME/conservancy/confidential/2021-09-10_AMEX_activity.csv


			




	
	
	
		
 
  --csv-statement=$HOME/conservancy/confidential/2021-09-10_AMEX_activity.csv \


			




	
	
	
		
 
  --account=amex


			




	
	
	
		
 

			




	
	
	
		
 
Conservancy currently enter data by hand rather than using Beancount importers.


			




	
	
	
		
 
This tool is still somewhat like an importer in that it needs to extract


			




	
	
	
		
 
transaction details from a third-party statement. Instead of creating


			




	
	
	
		
 
directives, it just checks to see that similar directives are already present.


			




	
	
	
		
 

			




	
	
	
		
 
Problem this attempts to address:


			




	
	
	
		
 
 - errors in the books take hours to find during reconciliation ("you're entering a world of pain"


			




	
	
	
		
 
 - balance checks are manually updated in svn/Financial/Ledger/sanity-check-balances.yaml


			




	
	
	
		
 
 - paper checks are entered in the books when written, but may not be cashed until months later (reconcile errors)


			




	
	
	
		
 
 - adding statement/reconciliation metadata to books is manual and prone to mistakes


			




	
	
	
		
 
 - creating reconciliation reports


			




	
	
	
		
 
 - normally transactions are entered manually, but potentially could create transaction directives (a.k.a. importing)


			




	
	
	
		
 
 - jumping to an individual transaction in the books isn't trivial - Emacs grep mode is helpful


			




	
	
	
		
 

			




	
	
	
		
 
Q. How are reconciliation reports created currently? How do you read them?


			




	
	
	
		
 

			




	
	
	
		
 
TODO/ISSUES:


			




	
	
	
		
 
 - AMEX statement doesn't provide bank balance or running total


			




	
	
	
		
 

			




	
	
	
		
 
"""


			




	
	
	
		
 
import argparse


			




	
	
	
		
 
import csv


			




	
	
	
		
 
import collections


			




	
	
	
		
 
import datetime


			




	
	
	
		
 
import decimal


			




	
	
	
		
 
from typing import Dict, List, Tuple


			




	
	
	
		
 

			




	
	
	
		
 
from beancount import loader


			




	
	
	
		
 
from beancount.query.query import run_query


			




	
	
	
		
 
from thefuzz import fuzz  # type: ignore


			




	
	
	
		
 

			




	
	
	
		
 
# NOTE: Statement doesn't seem to give us a running balance or a final total.


			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def standardize_amex_record(row: Dict) -> Dict:


			




	
	
	
		
 
    """Turn an AMEX CSV row into a standard dict format representing a transaction."""


			




	
	
	
		
 
    return {


			




	
	
	
		
 
        'date': datetime.datetime.strptime(row['Date'], '%m/%d/%Y').date(),


			




	
	
	
		
 
        'amount': -1 * decimal.Decimal(row['Amount']),


			




	
	
	
		
 
        'payee': row['Description'],


			




	
	
	
		
 
        'payee': row['Description'] or '',


			




	
	
	
		
 
    }


			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def standardize_beancount_record(row) -> Dict:  # type: ignore[no-untyped-def]


			




	
	
	
		
 
    """Turn a Beancount query result row into a standard dict representing a transaction."""


			




	
	
	
		
 
    return {


			




	...
	
		
 
@@ -46,15 +58,22 @@ def standardize_beancount_record(row) -> Dict:  # type: ignore[no-untyped-def]


			




	
	
	
		
 
        'payee': row.payee if row.payee else row.narration,


			




	
	
	
		
 
        'filename': row.filename,


			




	
	
	
		
 
        'line': row.posting_line,


			




	
	
	
		
 
        'statement': row.posting_statement,


			




	
	
	
		
 
    }


			




	
	
	
		
 

			




	
	
	
		
 
def standardize_fr_record(row: Dict) -> Dict:


			




	
	
	
		
 
    return {


			




	
	
	
		
 
        'date': datetime.datetime.strptime(row['Date'], '%m/%d/%Y').date(),


			




	
	
	
		
 
        'amount': decimal.Decimal(row['Amount']),


			




	
	
	
		
 
        'payee': row['Detail'] or '',


			




	
	
	
		
 
    }


			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def format_record(record: Dict) -> str:


			




	
	
	
		
 
    return f"{record['date'].isoformat()}: {record['amount']:>8} {record['payee'][:20]:<20}"


			




	
	
	
		
 
    return f"{record['date'].isoformat()}: {record['amount']:>11} {record['payee'][:20]:<20}"


			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def sort_records(records: List) -> List:


			




	
	
	
		
 
    return sorted(records, key=lambda x: (x['date'], x['amount']))


			




	
	
	
		
 

			




	
	
	
		
 

			




	...
	
		
 
@@ -71,21 +90,29 @@ def records_match(r1: Dict, r2: Dict) -> Tuple[bool, str]:


			




	
	
	
		
 
    else:


			




	
	
	
		
 
        return False, ''


			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
parser = argparse.ArgumentParser(description='Reconciliation helper')


			




	
	
	
		
 
parser.add_argument('--beancount-file', required=True)


			




	
	
	
		
 
parser.add_argument('--amex-csv', required=True)


			




	
	
	
		
 
parser.add_argument('--csv-statement', required=True)


			




	
	
	
		
 
parser.add_argument('--account', required=True, help='eg. Liabilities:CreditCard:AMEX')


			




	
	
	
		
 
parser.add_argument('--grep-output-filename')


			




	
	
	
		
 
# parser.add_argument('--report-group-regex')


			




	
	
	
		
 
parser.add_argument('--show-reconciled-matches', action='store_true')


			




	
	
	
		
 
args = parser.parse_args()


			




	
	
	
		
 

			




	
	
	
		
 
with open(args.amex_csv) as f:


			




	
	
	
		
 
# TODO: Should put in a sanity check to make sure the statement you're feeding


			




	
	
	
		
 
# in matches the account you've provided.


			




	
	
	
		
 
if 'AMEX' in args.account:


			




	
	
	
		
 
    standardize_statement_record = standardize_amex_record


			




	
	
	
		
 
else:


			




	
	
	
		
 
    standardize_statement_record = standardize_fr_record


			




	
	
	
		
 

			




	
	
	
		
 
with open(args.csv_statement) as f:


			




	
	
	
		
 
    reader = csv.DictReader(f)


			




	
	
	
		
 
    statement_trans = sort_records([standardize_amex_record(row) for row in reader])


			




	
	
	
		
 
    statement_trans = sort_records([standardize_statement_record(row) for row in reader])


			




	
	
	
		
 

			




	
	
	
		
 
begin_date = statement_trans[0]['date']


			




	
	
	
		
 
end_date = statement_trans[-1]['date']


			




	
	
	
		
 

			




	
	
	
		
 
# Do we traverse and filter the in-memory entries list and filter that, or do we


			




	
	
	
		
 
# use Beancount Query Language (BQL) to get a list of transactions? Currently


			




	...
	
		
 
@@ -95,18 +122,20 @@ end_date = statement_trans[-1]['date']


			




	
	
	
		
 
# beancount.query.query_execute.filter_entries() look useful in this respect,


			




	
	
	
		
 
# but I'm not clear on how to use compile(). An example would help.


			




	
	
	
		
 
entries, _, options = loader.load_file(args.beancount_file)


			




	
	
	
		
 

			




	
	
	
		
 
# String concatenation looks bad, but there's no SQL injection possible here


			




	
	
	
		
 
# because BQL can't write back to the Beancount files. I hope!


			




	
	
	
		
 
query = f"SELECT filename, META('lineno') AS posting_line, META('bank-statement') AS posting_statement, date, number(cost(position)), payee, narration where account = 'Liabilities:CreditCard:AMEX' and date >= {begin_date} and date <= {end_date}"


			




	
	
	
		
 
query = f"SELECT filename, META('lineno') AS posting_line, META('bank-statement') AS posting_statement, date, number(cost(position)), payee, narration where account = '{args.account}' and date >= {begin_date} and date <= {end_date}"


			




	
	
	
		
 
result_types, result_rows = run_query(


			




	
	
	
		
 
    entries,


			




	
	
	
		
 
    options,


			




	
	
	
		
 
    query,


			




	
	
	
		
 
)


			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
books_trans = sort_records([standardize_beancount_record(row) for row in result_rows])


			




	
	
	
		
 

			




	
	
	
		
 
num_statement_records = len(statement_trans)


			




	
	
	
		
 
num_books_trans = len(books_trans)


			




	
	
	
		
 
statement_index = 0


			




	
	
	
		
 
books_index = 0


			




	...
	
		
 
@@ -121,37 +150,42 @@ for r1 in statement_trans:


			




	
	
	
		
 
    for r2 in books_trans:


			




	
	
	
		
 
        match, note = records_match(r1, r2)


			




	
	
	
		
 
        if match:


			




	
	
	
		
 
            if not r2['statement'] or args.show_reconciled_matches:


			




	
	
	
		
 
                    matches.append([r2['date'], f'{format_record(r1)}    -->    {format_record(r2)}  ✓ {note}'])


			




	
	
	
		
 
            if not r2['statement']:


			




	
	
	
		
 
                metadata_to_apply.append((r2['filename'], r2['line'], f'    bank-statement: "{args.amex_csv}"\n'))


			




	
	
	
		
 
                metadata_to_apply.append((r2['filename'], r2['line'], f'    bank-statement: "{args.csv_statement}"\n'))


			




	
	
	
		
 
            books_trans.remove(r2)


			




	
	
	
		
 
            break


			




	
	
	
		
 
    else:


			




	
	
	
		
 
        matches.append([r1['date'], f'{format_record(r1)}    -->    {" ":^41}  ✗ Not in books'])


			




	
	
	
		
 
        matches.append([r1['date'], f'{format_record(r1)}    -->    {" ":^44}  ✗ Not in books'])


			




	
	
	
		
 
for r2 in books_trans:


			




	
	
	
		
 
    matches.append([r2['date'], f'{" ":^41}    -->    {format_record(r2)}  ✗ Not on statement'])


			




	
	
	
		
 
    matches.append([r2['date'], f'{" ":^44}    -->    {format_record(r2)}  ✗ Not on statement'])


			




	
	
	
		
 

			




	
	
	
		
 
print(f'-----------------------------------------------------------------------------------------------------------------')


			




	
	
	
		
 
print(f'{"STATEMENT":<40}            {"BOOKS":<40}   NOTES')


			




	
	
	
		
 
print(f'-----------------------------------------------------------------------------------------------------------------')


			




	
	
	
		
 
for _, output in sorted(matches):


			




	
	
	
		
 
    print(output)


			




	
	
	
		
 
print(f'-----------------------------------------------------------------------------------------------------------------')


			




	
	
	
		
 

			




	
	
	
		
 
# Write statement metadata back to books


			




	
	
	
		
 
if metadata_to_apply:


			




	
	
	
		
 
    print('Mark matched transactions as reconciled in the books? (y/N) ', end='')


			




	
	
	
		
 
    if input().lower() == 'y':


			




	
	
	
		
 
        files = {}


			




	
	
	
		
 
        for filename, line, metadata in metadata_to_apply:


			




	
	
	
		
 
        # Query results aren't necessarily sequential in a file, so need to sort


			




	
	
	
		
 
        # so that our line number offsets work.


			




	
	
	
		
 
        for filename, line, metadata in sorted(metadata_to_apply):


			




	
	
	
		
 
            if filename not in files:


			




	
	
	
		
 
                with open(filename, 'r') as f:


			




	
	
	
		
 
                    files[filename] = [0, f.readlines()]


			




	
	
	
		
 
                    print(f'Opening {filename}.')


			




	
	
	
		
 
                    files[filename] = [0, f.readlines()]  # Offset and contents


			




	
	
	
		
 
            files[filename][1].insert(line + files[filename][0], metadata)


			




	
	
	
		
 
            files[filename][0] += 1


			




	
	
	
		
 
            print(f'File {filename} offset {files[filename][0]}')


			




	
	
	
		
 
        for filename in files:


			




	
	
	
		
 
            with open(filename, 'w') as f:


			




	
	
	
		
 
                f.writelines(files[filename][1])


			




	
	
	
		
 
                print(f'Wrote {filename}.')


			




	
	
	
		
 

			




	
	
	
		
 
# Local Variables:


			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







    


    

  


    








    
        Server instance: kapok.sfconservancy.org-2166
    
    
        This site is powered by
            Kallithea 0.6.3,
        which is
        © 2010–2020 by various authors & licensed under GPLv3.
            – Support
        – Privacy Policy