Convert an expense CSV using dated exchange rates in Python

import argparse
import csv
import io
import re
import sys
from pathlib import Path

from parseapi import ParseAPI, ParseAPIError

from datetime import date as CalendarDate, datetime, timezone
from decimal import Decimal, InvalidOperation, ROUND_HALF_EVEN, localcontext

DESCRIPTION = "Convert an expense CSV using dated reference exchange rates."
REQUIRED = ["amount", "currency", "date"]
COLUMNS = ["parse_base", "parse_quote", "parse_rate", "parse_rate_date", "parse_converted",
           "parse_digits", "parse_status", "parse_error", "parse_http_status"]
SAMPLE = """id,amount,currency,date,note
001,12.50,EUR,2026-08-28,Lunch
002,5.25,eur,2026-08-28,Same pair and day
003,1200,JPY,2026-08-29,Weekend expense
004,-8.00,EUR,2026-08-28,Refund
005,,EUR,2026-08-28,Missing amount
006,10.00,EUR,03/04/2026,Date needs interpretation
"""

# example:read
def read_rows(stream):
    reader = csv.reader(stream, strict=True)
    fields = next(reader, None)
    if (not fields or not set(REQUIRED).issubset(fields)
            or any(not field.strip() for field in fields)
            or len(fields) != len(set(fields))):
        raise ValueError("Use distinct, nonempty headers including " + ", ".join(REQUIRED) + ".")
    if set(fields) & set(COLUMNS):
        raise ValueError("The input already contains output columns. Use the original CSV.")
    rows = []
    for cells in reader:
        if len(cells) != len(fields):
            raise ValueError("Every row must have the same number of cells as the header.")
        rows.append(dict(zip(fields, cells)))
        if len(rows) > 200:
            raise ValueError("Use 1 to 200 data rows for this example.")
    if not rows:
        raise ValueError("The CSV needs at least one data row.")
    return fields, rows
# /example:read


def outcome(status, error="", http_status="", **values):
    return {**dict.fromkeys(COLUMNS, ""), "parse_status": status,
            "parse_error": error, "parse_http_status": http_status, **values}


def write_rows(stream, fields, rows):
    writer = csv.DictWriter(stream, fieldnames=fields + COLUMNS)
    writer.writeheader()
    writer.writerows(rows)

def iso_day(value):
    if not isinstance(value, str) or not re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}", value):
        raise ValueError("Use an ISO calendar date.")
    return CalendarDate.fromisoformat(value)


# example:amount
def expense(row):
    amount, base, day = row["amount"].strip(), row["currency"].strip().upper(), row["date"].strip()
    if not amount or not base or not day:
        return None, "missing_input"
    # Decimal point only, with up to 15 whole digits and 6 decimal places.
    if not re.fullmatch(r"[+-]?[0-9]{1,15}(?:\.[0-9]{1,6})?", amount):
        return None, "review_amount"
    if not re.fullmatch(r"[A-Z]{3}", base):
        return None, "review_currency"
    try:
        date = iso_day(day)
    except ValueError:
        return None, "review_date"
    if date > datetime.now(timezone.utc).date():
        return None, "future_date"
    return (Decimal(amount), base, day), None
# /example:amount


def quote_digits(data, quote):
    if (not isinstance(data, dict) or data.get("currency") != quote
            or type(data.get("digits")) is not int or not 0 <= data["digits"] <= 4):
        raise ValueError("unexpected_currency_response")
    return data["digits"]


def rate_answer(data, base, quote, day):
    if (not isinstance(data, dict) or data.get("base") != base or data.get("quote") != quote
            or type(data.get("rate")) not in (int, float)):
        raise ValueError("unexpected_rate_response")
    rate = Decimal(str(data["rate"]))
    if not rate.is_finite() or rate <= 0 or iso_day(data.get("date")) > iso_day(day):
        raise ValueError("unexpected_rate_response")
    return rate, data["date"]


# example:result
def converted(amount, base, quote, rate, rate_date, digits):
    # Keep decimal amounts exact, multiply once, then round the row once.
    try:
        with localcontext() as context:
            context.prec = 50
            rounded = (amount * rate).quantize(Decimal(1).scaleb(-digits), rounding=ROUND_HALF_EVEN)
    except InvalidOperation:
        return outcome("error", "unexpected_response")
    if rounded == 0:
        rounded = abs(rounded)
    return outcome("converted", http_status=200, parse_base=base, parse_quote=quote,
                   parse_rate=format(rate, "f"), parse_rate_date=rate_date,
                   parse_converted=format(rounded, "f"), parse_digits=digits)
# /example:result


def enrich(rows, parse, quote="USD"):
    quote = quote.strip().upper()
    if not re.fullmatch(r"[A-Z]{3}", quote):
        raise ValueError("Use a three-letter quote currency.")
    results, cache, stopped = [], {}, False
    digits, metadata_error = None, None
    for row in rows:
        values, issue = expense(row)
        if issue:
            results.append({**row, **outcome(issue)})
            continue
        amount, base, day = values
        identity = (base, quote, day)
        if identity in cache:
            answer = cache[identity]
        elif stopped:
            answer = outcome("not_attempted")
        elif metadata_error is not None:
            answer = metadata_error
        else:
            try:
                # example:request
                if digits is None:
                    digits = quote_digits(parse.currency(quote), quote)
                # The amount stays in Decimal locally. The API supplies a reusable rate.
                data = parse.currency.rate(base, quote, date=day)
                answer = rate_answer(data, base, quote, day)
                # /example:request
            except ParseAPIError as error:
                status = "rate_unavailable" if error.status == 404 else "error"
                answer = outcome(status, error.code, error.status)
                stopped = error.status in (401, 403, 429)
            except (ValueError, InvalidOperation):
                answer = outcome("error", "unexpected_response")
            except Exception:
                answer = outcome("error", "request_failed")
            if digits is None:
                metadata_error = answer
            cache[identity] = answer
        if isinstance(answer, tuple):
            rate, rate_date = answer
            result = converted(amount, base, quote, rate, rate_date, digits)
        else:
            result = answer
        results.append({**row, **result})
    return results

def main():
    parser = argparse.ArgumentParser(description=DESCRIPTION)
    parser.add_argument("input", nargs="?", help="UTF-8 CSV filename, or omit to run the sample")
    parser.add_argument("--output", help="New CSV filename, or omit for standard output")
    parser.add_argument("--quote", default="USD", help="Three-letter target currency, default USD")
    args = parser.parse_args()
    try:
        if not re.fullmatch(r"[A-Za-z]{3}", args.quote.strip()):
            raise ValueError("Use a three-letter quote currency.")
        if args.output and (Path(args.output).exists() or Path(args.output).is_symlink()):
            raise ValueError("The output path already exists. Choose a new filename.")
        if args.input:
            path = Path(args.input)
            if path.stat().st_size > 65536:
                raise ValueError("Use a CSV no larger than 64 KiB.")
            with path.open(encoding="utf-8-sig", newline="") as stream:
                fields, rows = read_rows(stream)
        else:
            fields, rows = read_rows(io.StringIO(SAMPLE))
        # Reads PARSEAPI_KEY. Reuse one client without automatic retries.
        with ParseAPI(timeout=10.0, retries=0) as parse:
            results = enrich(rows, parse, args.quote)
        if args.output:
            # Exclusive creation protects against a file appearing during the lookups.
            with Path(args.output).open("x", encoding="utf-8", newline="") as stream:
                write_rows(stream, fields, results)
        else:
            write_rows(sys.stdout, fields, results)
    except (ValueError, OSError, csv.Error) as error:
        print(f"Could not complete: {error}", file=sys.stderr)
        return 1
    review = sum(row["parse_status"] != "converted" for row in results)
    print(f"{len(results)} rows written, {review} need review.", file=sys.stderr)
    return 2 if review else 0


if __name__ == "__main__":
    sys.exit(main())