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
...
 
@@ -5,20 +5,31 @@ 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


			




	...
	
		
 
@@ -29,12 +40,13 @@ 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 '',


			




	
	
	
		
 
    }


			




	
	
	
		
 

			




	
	
	
		
 

			




	...
	
		
 
@@ -49,9 +61,16 @@ def standardize_beancount_record(row) -> Dict:  # type: ignore[no-untyped-def]


			




	
	
	
		
 
        '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:


			




	...
	
		
 
@@ -74,15 +93,23 @@ def records_match(r1: Dict, r2: Dict) -> Tuple[bool, str]:


			




	
	
	
		
 

			




	
	
	
		
 
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']


			




	...
	
		
 
@@ -98,12 +125,14 @@ 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)


			




	...
	
		
 
@@ -124,13 +153,13 @@ for r1 in statement_trans:


			




	
	
	
		
 
            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')


			




	...
	
		
 
@@ -139,16 +168,21 @@ 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])


			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







    


    

  


    








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