Currency Conversion

currency_conversion.py

from dataclasses import dataclass
from collections import defaultdict, deque


def parseRates(lines, converter):
    for l in lines:
        parts = l.strip().split(",")
        if len(parts) != 3:
            print(f"MALFORMED LINE: {l}")
            continue
        frm, to, rate = parts
        converter.rates[frm][to] = float(rate)
        # inverse edge, unless an explicit rate for it exists or comes later
        converter.rates[to].setdefault(frm, 1 / float(rate))


def parseRequests(lines):
    reqs = []
    for l in lines:
        parts = l.strip().split()
        if len(parts) != 3:
            print(f"MALFORMED LINE: {l}")
            continue
        [amt, frm, to] = parts
        reqs.append(Request(frm, to, float(amt)))
    return reqs


@dataclass
class Rate:
    frm: str
    to: str
    rate: float


@dataclass
class Request:
    frm: str
    to: str
    amount: float


MAX_HOPS = 3  # hops = conversions (edges)


class Converter:
    def __init__(self) -> None:
        self.rates = defaultdict(dict)  # frm -> {to: rate}, adjacency map

    def convert(self, req):
        # best rate over all paths of <= MAX_HOPS conversions
        # direct and inverse rates are 1-hop paths, so they are covered too
        rate = self.dfs_best(req.frm, req.to, 1.0, set(), MAX_HOPS)
        if rate is None:
            print(f"N/A {req.frm} -> {req.to}")
            return None
        res = round(rate * req.amount, 4)
        print(f"Conversion {req.frm} -> {req.to}: {req.amount} -> {res}")
        return res

    def bfs(self, frm, to):
        """rate along the path with the fewest hops, None if unreachable."""
        if frm == to:
            return 1.0
        q = deque([(frm, 1.0)])
        seen = {frm}
        while q:
            cur, rt = q.popleft()
            for nxt_cur, rate in self.rates.get(cur, {}).items():
                if nxt_cur == to:
                    return rt * rate
                if nxt_cur not in seen:
                    seen.add(nxt_cur)
                    q.append((nxt_cur, rt * rate))
        return None

    def dfs_best(self, frm, to, rt, seen, hops):
        """Best rate from frm to `to` using at most `hops` conversions; None if unreachable."""
        if frm == to:
            return rt
        if hops == 0:
            return None
        seen.add(frm)
        best = None
        for nxt_cur, rate in self.rates.get(frm, {}).items():
            if nxt_cur in seen:
                continue
            res = self.dfs_best(nxt_cur, to, rt * rate, seen, hops - 1)
            if res is not None and (best is None or res > best):
                best = res
        seen.remove(frm)  # backtrack: frm is free for other paths
        return best


def run(rates, requests):
    converter = Converter()
    parseRates(rates, converter)
    reqs = parseRequests(requests)
    return [converter.convert(req) for req in reqs]

test_currency_conversion.py

from currency_conversion import Converter, parseRates, run


def check(res, expected):
    assert len(res) == len(expected), res
    for got, exp in zip(res, expected):
        assert (got is None and exp is None) or round(got, 4) == exp, (got, exp)


def test_stage1_direct():
    check(run(["USD,EUR,0.5", "bad line"], ["100 USD EUR", "100 USD USD", "100 USD XXX", "100 USD"]),
          [50.0, 100.0, None])


def test_stage2_inverse():
    check(run(["USD,EUR,0.5"], ["100 EUR USD"]), [200.0])
    check(run(["USD,EUR,0.5", "EUR,USD,3"], ["100 EUR USD"]), [300.0])  # explicit rate wins


def test_stage3_multi_hop():
    rates = ["USD,EUR,0.5", "EUR,GBP,0.8", "JPY,CHF,0.01"]
    check(run(rates, ["100 USD GBP", "100 GBP USD", "100 USD CHF"]), [40.0, 250.0, None])
    c = Converter()
    parseRates(["USD,EUR,0.5", "EUR,GBP,0.8", "USD,GBP,0.39"], c)
    assert round(c.bfs("USD", "GBP"), 4) == 0.39  # fewest hops, even with a worse rate


def test_stage4_best_rate():
    assert run(["USD,EUR,0.5", "EUR,GBP,0.8", "USD,GBP,0.39"], ["100 USD GBP", "100 GBP USD"]) == [40.0, 256.4103]
    # hop limit: A->E through B, C, D would be 4 conversions
    rates = ["A,B,1", "B,C,1", "C,D,1", "D,E,2", "A,D,0.5", "C,X,3"]
    assert run(rates, ["1 A E", "1 A D", "1 A X"]) == [1.0, 1.0, 3.0]
    assert run(["USD,EUR,0.123456789"], ["1 USD EUR"]) == [0.1235]


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