96 lines
3.2 KiB
Python
96 lines
3.2 KiB
Python
import csv
|
|
import re
|
|
import sys
|
|
from collections import Counter
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
# convert revolut account statement csv exports into the same shape as nlb.py / trade_republic.py
|
|
STATEMENTS = "account-statement*.csv"
|
|
IBAN = re.compile(r"\b[A-Z]{2}\d{2}[A-Z0-9]{11,30}\b")
|
|
FIELDS = ["date", "desc", "account", "amount", "balance", "currency", "type"]
|
|
|
|
|
|
def decimals(text):
|
|
return len(text.partition(".")[2])
|
|
|
|
|
|
def scale(source):
|
|
"""Revolut writes JPY without decimals and EUR with two, so follow the file."""
|
|
places = {}
|
|
for r in source:
|
|
seen = [decimals(r[c]) for c in ("Amount", "Fee", "Balance") if r[c]]
|
|
places[r["Currency"]] = max(seen + [places.get(r["Currency"], 0)])
|
|
return places
|
|
|
|
|
|
def extract_transactions(path):
|
|
print(path)
|
|
with open(path, newline="") as f:
|
|
source = list(csv.DictReader(f))
|
|
source.sort(key=lambda r: r["Completed Date"])
|
|
places = scale(source)
|
|
|
|
rows = []
|
|
skipped = Counter()
|
|
for r in source:
|
|
if r["State"] != "COMPLETED" or not r["Balance"]:
|
|
skipped[r["State"]] += 1
|
|
continue
|
|
digits = places[r["Currency"]]
|
|
fee = Decimal(r["Fee"] or 0)
|
|
# the fee is deducted alongside the amount rather than booked separately
|
|
amount = Decimal(r["Amount"]) - fee
|
|
desc = r["Description"]
|
|
if fee:
|
|
desc += f" (fee {fee:.{digits}f})"
|
|
account = IBAN.search(desc)
|
|
rows.append({
|
|
"date": datetime.strptime(r["Completed Date"][:10], "%Y-%m-%d").strftime("%d.%m.%y"),
|
|
"desc": desc,
|
|
"account": account.group() if account else "",
|
|
"amount": f"{amount:+.{digits}f}",
|
|
"balance": f"{Decimal(r['Balance']):.{digits}f}",
|
|
"currency": r["Currency"],
|
|
"type": r["Type"],
|
|
})
|
|
for state, count in sorted(skipped.items()):
|
|
print(f" skipped {count} {state or 'unfinished'} transactions")
|
|
return rows
|
|
|
|
|
|
def check_balances(rows, path):
|
|
"""Every balance must be the previous one plus the transaction amount."""
|
|
for currency in dict.fromkeys(row["currency"] for row in rows):
|
|
chain = [row for row in rows if row["currency"] == currency]
|
|
for prev, row in zip(chain, chain[1:]):
|
|
expected = Decimal(prev["balance"]) + Decimal(row["amount"])
|
|
if expected != Decimal(row["balance"]):
|
|
print(f" warning: {path.name} {row['date']} {row['desc'][:40]!r} "
|
|
f"balance {row['balance']} != {expected}")
|
|
|
|
|
|
def main():
|
|
target = Path(sys.argv[1])
|
|
paths = sorted(target.glob(STATEMENTS)) if target.is_dir() else [target]
|
|
out_path = target.with_suffix(".csv")
|
|
if out_path in paths:
|
|
sys.exit(f"{out_path} would overwrite the statement it reads; pass the directory instead")
|
|
|
|
all_rows = []
|
|
for path in paths:
|
|
rows = extract_transactions(path)
|
|
check_balances(rows, path)
|
|
all_rows.extend(rows)
|
|
|
|
with open(out_path, "w", newline="") as f:
|
|
writer = csv.DictWriter(f, fieldnames=FIELDS)
|
|
writer.writeheader()
|
|
writer.writerows(all_rows)
|
|
|
|
print(f"Wrote {len(all_rows)} transactions to {out_path}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|