Ledger

Input: lines of the form timestamp,from,to,amount_cents,currency, processed in order. An empty from is a deposit: money enters the system and no account is debited.

Stages

  1. Balances. Apply every transfer and print each account's final balance in USD, sorted by account name.
  2. Malformed lines. Skip lines with the wrong number of fields, a non-integer amount or a negative amount. Count them and print the count at the end.
  3. Fees. A transfer above 100000 cents pays a 1% fee, rounded down. The receiver gets the amount minus the fee, and the fee goes to the account FEES.
  4. No overdraft. An account may never go below 0. Reject a transfer that would overdraw the sender, apply nothing for it, and print the timestamps of the rejected transfers.

Files

  • ledger.py: the solution for all four stages.
  • test_ledger.py: tests grouped by stage. Run with python3 test_ledger.py or pytest.

Solutions

ledger.py

from collections import defaultdict
from dataclasses import dataclass


@dataclass
class Txn:
    ts: str
    frm: str
    to: str
    amount: int
    currency: str


def parse(line):  # validation lives here
    parts = line.strip().split(",")
    if len(parts) != 5:
        return None
    ts, frm, to, amt, cur = parts
    if not amt.isdigit():  # rejects non-ints and negatives
        return None
    return Txn(ts, frm, to, int(amt), cur)


FEE_THRESHOLD = 100000


class Ledger:
    def __init__(self):
        self.bal = defaultdict(lambda: defaultdict(int))  # bal[acct][cur]
        self.rejected = []

    # empty `from` = deposit: always allowed, no sender to debit
    def transfer(self, t):
        fee = t.amount // 100 if t.amount > FEE_THRESHOLD else 0
        if t.frm != "":
            if self.bal[t.frm][t.currency] < t.amount:
                self.rejected.append(t)
                return
            self.bal[t.frm][t.currency] -= t.amount
        self.bal[t.to][t.currency] += t.amount - fee
        if fee:
            self.bal["FEES"][t.currency] += fee

    def report(self, cur="USD"):
        for acct in sorted(self.bal):
            print(f"{acct}: {self.bal[acct][cur]}")

        for r in self.rejected:  # already in input order; Txn isn't sortable
            print(f"{r.ts} error : insufficient balance")


def run(lines):
    ledger, bad = Ledger(), 0
    for l in lines:
        t = parse(l)
        if t is None:
            bad += 1
            continue
        ledger.transfer(t)
    ledger.report()
    print(f"malformed lines: {bad}")
    return ledger, bad

test_ledger.py

from ledger import Ledger, parse, run


def test_stage1_balances():
    l, _ = run(["1,,A,500,USD", "2,A,B,200,USD", "3,,C,100,EUR"])
    assert l.bal["A"]["USD"] == 300
    assert l.bal["B"]["USD"] == 200
    assert l.bal["C"]["USD"] == 0  # EUR is tracked but not reported


def test_stage2_malformed():
    assert parse("x,y") is None  # wrong field count
    assert parse("1,A,B,1.5,USD") is None  # not an integer
    assert parse("1,A,B,-5,USD") is None  # negative
    _, bad = run(["x,y", "1,,A,10,USD", "1,A,B,abc,USD"])
    assert bad == 2


def test_stage3_fees():
    l, _ = run(["1,,A,300000,USD", "2,A,B,200000,USD"])
    assert l.bal["A"]["USD"] == 300000 - 3000 - 200000  # deposit fee, then send
    assert l.bal["B"]["USD"] == 200000 - 2000
    assert l.bal["FEES"]["USD"] == 3000 + 2000
    assert sum(b["USD"] for b in l.bal.values()) == 300000  # money conserved
    l, _ = run(["1,,A,100000,USD"])  # exactly the threshold: no fee
    assert l.bal["A"]["USD"] == 100000 and l.bal["FEES"]["USD"] == 0


def test_stage4_no_overdraft():
    l, bad = run(["1,,A,300000,USD", "2,A,B,200000,USD", "3,B,C,999999,USD", "x,y"])
    assert [t.ts for t in l.rejected] == ["3"] and bad == 1
    assert l.bal["C"]["USD"] == 0  # nothing applied for the rejected transfer


if __name__ == "__main__":
    for name, f in list(globals().items()):
        if name.startswith("test_"):
            f()
    print("all tests passed")