add revolut

nlb-export/nlb.py

@@ -2,6 +2,7 @@ import csv
 import re
 import subprocess
 import sys
+from decimal import Decimal
 from pathlib import Path
 # extract transaction from izpisek pdf
 TRANSACTION_START = re.compile(r"^\d{2}\.\d{2}\.\d{2}\s")
@@ -24,6 +25,14 @@ def pdf_to_text(pdf_path):
     return result.stdout
 
 
+def to_number(text):
+    """Statements write 1.234,56 and mark a negative balance with a trailing minus."""
+    text = text.strip()
+    if text.endswith("-"):
+        text = "-" + text[:-1]
+    return Decimal(text.replace(".", "").replace(",", "."))
+
+
 def parse_transaction(line):
     m = TRANSACTION_LINE.match(line.rstrip())
     account = ""
@@ -33,15 +42,12 @@ def parse_transaction(line):
             account = f
         else:
             desc_parts.append(f)
-    balance = m.group("balance")
-    if balance.endswith("-"):
-        balance = "-" + balance[:-1]
     return {
         "date": m.group("date"),
         "desc": " ".join(desc_parts),
         "account": account,
-        "amount": m.group("amount"),
+        "amount": f"{to_number(m.group('amount')):+.2f}",
-        "balance": balance,
+        "balance": f"{to_number(m.group('balance')):.2f}",
         "currency": "EUR",
     }
 
@@ -70,13 +76,25 @@ def extract_transactions(pdf_path):
     return rows
 
 
+def check_balances(rows, path):
+    """Every balance must be the previous one plus the transaction amount."""
+    for prev, row in zip(rows, rows[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():
     pdf_dir = Path(sys.argv[1])
     out_path = pdf_dir.with_suffix(".csv")
 
     all_rows = []
     for pdf_path in sorted(pdf_dir.glob("*.pdf")):
-        all_rows.extend(extract_transactions(pdf_path))
+        rows = extract_transactions(pdf_path)
+        # consecutive izpiski of one account, so the chain carries over the file boundary
+        check_balances(all_rows[-1:] + rows, pdf_path)
+        all_rows.extend(rows)
 
     with open(out_path, "w", newline="") as f:
         writer = csv.DictWriter(f, fieldnames=["date", "desc", "account", "amount", "balance", "currency"])

nlb-export/revolut.py

@@ -0,0 +1,95 @@
+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()

nlb-export/trade_republic.py

@@ -0,0 +1,155 @@
+import csv
+import re
+import subprocess
+import sys
+from datetime import datetime
+from decimal import Decimal
+from pathlib import Path
+# extract transactions from trade republic account statement pdf
+HEADER = re.compile(r"^\s*DATE\b.*\bMONEY IN\b.*\bMONEY OUT\b.*\bBALANCE\b")
+TEXT_COLUMNS = ("DATE", "TYPE", "DESCRIPTION")
+MONEY_COLUMNS = ("MONEY IN", "MONEY OUT", "BALANCE")
+AMOUNT = re.compile(r"-?€\s?-?[\d,]+\.\d{2}")
+TOKEN = re.compile(r"\S+")
+FULL_DATE = re.compile(r"^\d{2} [A-Z][a-z]{2} \d{4}$")
+IBAN = re.compile(r"\b[A-Z]{2}\d{2}[A-Z0-9]{11,30}\b")
+AMOUNT_SLACK = 5
+
+
+def pdf_to_text(pdf_path):
+    result = subprocess.run(
+        ["pdftotext", "-layout", str(pdf_path), "-"],
+        capture_output=True, text=True, check=True,
+    )
+    return result.stdout
+
+
+def column_starts(header_line):
+    return {c: header_line.index(c) for c in TEXT_COLUMNS + MONEY_COLUMNS}
+
+
+def nearest(pos, cols):
+    return min(cols, key=lambda c: abs(pos - cols[c]))
+
+
+def to_number(text):
+    return Decimal(text.replace("€", "").replace(",", "").replace(" ", ""))
+
+
+def join_wrapped(chunks):
+    """Glue description fragments split across lines, honouring hyphen wraps."""
+    out = ""
+    for chunk in chunks:
+        if not out:
+            out = chunk
+        elif len(out) > 1 and out.endswith("-") and not out.endswith(" -"):
+            out += chunk
+        else:
+            out += " " + chunk
+    return out
+
+
+def parse_block(lines, cols):
+    """A block is the consecutive lines making up one transaction.
+
+    Layout varies between statements: everything on one line, or the date,
+    type and description stacked across three lines. Both are handled by
+    assigning every token to the column its start position is closest to.
+    """
+    text_cols = {c: cols[c] for c in TEXT_COLUMNS}
+    money_cols = {c: cols[c] for c in MONEY_COLUMNS}
+    words = {c: [] for c in TEXT_COLUMNS}
+    desc_chunks = []
+    amounts = {c: None for c in MONEY_COLUMNS}
+
+    for line in lines:
+        cut = len(line)
+        for m in AMOUNT.finditer(line):
+            if m.start() < cols["MONEY IN"] - AMOUNT_SLACK:
+                continue
+            cut = min(cut, m.start())
+            amounts[nearest(m.start(), money_cols)] = to_number(m.group())
+        words["DESCRIPTION"] = []
+        for m in TOKEN.finditer(line[:cut]):
+            words[nearest(m.start(), text_cols)].append(m.group())
+        if words["DESCRIPTION"]:
+            desc_chunks.append(" ".join(words["DESCRIPTION"]))
+
+    date = " ".join(words["DATE"])
+    if not FULL_DATE.match(date) or amounts["BALANCE"] is None:
+        return None
+
+    desc = join_wrapped(desc_chunks)
+    account = IBAN.search(desc)
+    amount = (amounts["MONEY IN"] or 0) - (amounts["MONEY OUT"] or 0)
+    return {
+        "date": datetime.strptime(date, "%d %b %Y").strftime("%d.%m.%y"),
+        "desc": desc,
+        "account": account.group() if account else "",
+        "amount": f"{amount:+.2f}",
+        "balance": f"{amounts['BALANCE']:.2f}",
+        "currency": "EUR",
+        "type": " ".join(words["TYPE"]),
+    }
+
+
+def blocks(lines):
+    block = []
+    for line in lines:
+        if line.strip():
+            block.append(line)
+        elif block:
+            yield block
+            block = []
+    if block:
+        yield block
+
+
+def extract_transactions(path):
+    print(path)
+    rows = []
+    for page_text in pdf_to_text(path).split("\f"):
+        page_lines = page_text.split("\n")
+        header = next((i for i, l in enumerate(page_lines) if HEADER.match(l)), None)
+        if header is None:
+            continue
+        cols = column_starts(page_lines[header])
+        for block in blocks(page_lines[header + 1:]):
+            row = parse_block(block, cols)
+            if row:
+                rows.append(row)
+    return rows
+
+
+def check_balances(rows, path):
+    """Every balance must be the previous one plus the transaction amount."""
+    for prev, row in zip(rows, rows[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("*.pdf")) if target.is_dir() else [target]
+    out_path = target.with_suffix(".csv")
+
+    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=["date", "desc", "account", "amount", "balance", "currency", "type"]
+        )
+        writer.writeheader()
+        writer.writerows(all_rows)
+
+    print(f"Wrote {len(all_rows)} transactions to {out_path}")
+
+
+if __name__ == "__main__":
+    main()