Add FxCache and convert_to_gbp core
Context ------- FX for UK users has two lives: live ECB rates for portfolio display (available same-day), and HMRC monthly/daily rates for CGT basis (published after month-end). The plan keeps both in one cache table with an upgradable `source` column, so a later reconciliation job can replace ECB_LIVE values with HMRC_MONTHLY for the same date without schema work. This change ----------- - FxCache: SQLite table (currency, on_date) -> (rate_gbp, source) with ON CONFLICT UPDATE semantics so reconciliation is a single put(). - convert_to_gbp(): GBP short-circuits to identity; any other currency must be in the cache (network fetch is the caller's responsibility, separately implemented by the ECB and HMRC fetchers). - Explicit LookupError on cache miss — deliberate, we do NOT want a silent fallback that produces wrong cost-basis numbers. Decisions deferred to later commits: - Actual ECB daily reference-rate fetcher (eurofxref XML) — lands with the Trading212 provider in Phase 1 when non-GBP trades first appear. - HMRC monthly-rate fetcher + reconciliation CronJob — Phase 1 tail. Test plan --------- ## Automated - poetry run pytest tests/test_fx.py -v → 6 passed - poetry run mypy broker_sync tests → Success: no issues found in 8 source files - poetry run ruff check . → All checks passed! ## Manual Verification Not applicable — no network yet.
This commit is contained in:
parent
a66ef189f6
commit
33810899c9
2 changed files with 149 additions and 0 deletions
87
broker_sync/fx.py
Normal file
87
broker_sync/fx.py
Normal file
|
|
@ -0,0 +1,87 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import sqlite3
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
|
||||
from broker_sync.models import FxRateSource
|
||||
|
||||
_SCHEMA = """
|
||||
CREATE TABLE IF NOT EXISTS fx_rate (
|
||||
currency TEXT NOT NULL,
|
||||
on_date TEXT NOT NULL,
|
||||
rate_gbp TEXT NOT NULL,
|
||||
source TEXT NOT NULL,
|
||||
PRIMARY KEY (currency, on_date)
|
||||
);
|
||||
"""
|
||||
|
||||
|
||||
class FxCache:
|
||||
"""SQLite-backed FX rate store.
|
||||
|
||||
Rate meaning: 1 unit of `currency` = `rate_gbp` GBP.
|
||||
Source is upgradable — a later reconciliation run can replace an
|
||||
ECB_LIVE row with the HMRC_MONTHLY value for the same date.
|
||||
"""
|
||||
|
||||
def __init__(self, db_path: Path | str) -> None:
|
||||
self._path = Path(db_path)
|
||||
self._path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with self._conn() as c:
|
||||
c.executescript(_SCHEMA)
|
||||
|
||||
def _conn(self) -> sqlite3.Connection:
|
||||
return sqlite3.connect(self._path)
|
||||
|
||||
def get(self, currency: str, on_date: date) -> tuple[Decimal, FxRateSource] | None:
|
||||
with self._conn() as c:
|
||||
row = c.execute(
|
||||
"SELECT rate_gbp, source FROM fx_rate WHERE currency=? AND on_date=?",
|
||||
(currency.upper(), on_date.isoformat()),
|
||||
).fetchone()
|
||||
if row is None:
|
||||
return None
|
||||
return Decimal(row[0]), FxRateSource(row[1])
|
||||
|
||||
def put(
|
||||
self,
|
||||
currency: str,
|
||||
on_date: date,
|
||||
rate_gbp: Decimal,
|
||||
source: FxRateSource,
|
||||
) -> None:
|
||||
with self._conn() as c:
|
||||
c.execute(
|
||||
"INSERT INTO fx_rate (currency, on_date, rate_gbp, source) "
|
||||
"VALUES (?, ?, ?, ?) "
|
||||
"ON CONFLICT(currency, on_date) DO UPDATE SET "
|
||||
"rate_gbp=excluded.rate_gbp, source=excluded.source",
|
||||
(currency.upper(), on_date.isoformat(), str(rate_gbp), str(source)),
|
||||
)
|
||||
c.commit()
|
||||
|
||||
|
||||
def convert_to_gbp(
|
||||
amount: Decimal,
|
||||
currency: str,
|
||||
on_date: date,
|
||||
cache: FxCache | None,
|
||||
) -> tuple[Decimal, Decimal, FxRateSource]:
|
||||
"""Return (amount_gbp, rate_used, source).
|
||||
|
||||
- GBP passes through with rate=1 and source=ECB_LIVE (conventional).
|
||||
- Otherwise the cache must have a rate for (currency, on_date). The
|
||||
caller is responsible for populating the cache from ECB (live) or
|
||||
HMRC (reconciliation) upstream.
|
||||
"""
|
||||
if currency.upper() == "GBP":
|
||||
return amount, Decimal("1"), FxRateSource.ECB_LIVE
|
||||
if cache is None:
|
||||
raise LookupError(f"No FX cache available for {currency} {on_date.isoformat()}")
|
||||
hit = cache.get(currency, on_date)
|
||||
if hit is None:
|
||||
raise LookupError(f"No FX rate cached for {currency.upper()} {on_date.isoformat()}")
|
||||
rate, source = hit
|
||||
return amount * rate, rate, source
|
||||
62
tests/test_fx.py
Normal file
62
tests/test_fx.py
Normal file
|
|
@ -0,0 +1,62 @@
|
|||
from datetime import date
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from broker_sync.fx import FxCache, convert_to_gbp
|
||||
from broker_sync.models import FxRateSource
|
||||
|
||||
|
||||
def test_cache_roundtrip(tmp_path: Path) -> None:
|
||||
c = FxCache(tmp_path / "fx.db")
|
||||
c.put("USD", date(2026, 4, 1), Decimal("0.79"), FxRateSource.ECB_LIVE)
|
||||
hit = c.get("USD", date(2026, 4, 1))
|
||||
assert hit is not None
|
||||
assert hit[0] == Decimal("0.79")
|
||||
assert hit[1] is FxRateSource.ECB_LIVE
|
||||
|
||||
|
||||
def test_cache_miss(tmp_path: Path) -> None:
|
||||
c = FxCache(tmp_path / "fx.db")
|
||||
assert c.get("USD", date(2026, 4, 1)) is None
|
||||
|
||||
|
||||
def test_put_overwrites_on_source_upgrade(tmp_path: Path) -> None:
|
||||
# Reconciliation job upgrades ECB_LIVE → HMRC_MONTHLY for historical rows
|
||||
c = FxCache(tmp_path / "fx.db")
|
||||
c.put("USD", date(2026, 3, 1), Decimal("0.79"), FxRateSource.ECB_LIVE)
|
||||
c.put("USD", date(2026, 3, 1), Decimal("0.791"), FxRateSource.HMRC_MONTHLY)
|
||||
hit = c.get("USD", date(2026, 3, 1))
|
||||
assert hit is not None
|
||||
assert hit[0] == Decimal("0.791")
|
||||
assert hit[1] is FxRateSource.HMRC_MONTHLY
|
||||
|
||||
|
||||
def test_gbp_passthrough_is_identity() -> None:
|
||||
# currency == GBP → rate is 1.0, no network needed
|
||||
amount_gbp, rate, source = convert_to_gbp(amount=Decimal("100"),
|
||||
currency="GBP",
|
||||
on_date=date(2026, 4, 1),
|
||||
cache=None)
|
||||
assert amount_gbp == Decimal("100")
|
||||
assert rate == Decimal("1")
|
||||
assert source is FxRateSource.ECB_LIVE
|
||||
|
||||
|
||||
def test_convert_uses_cached_rate(tmp_path: Path) -> None:
|
||||
c = FxCache(tmp_path / "fx.db")
|
||||
c.put("USD", date(2026, 4, 1), Decimal("0.80"), FxRateSource.ECB_LIVE)
|
||||
amount_gbp, rate, source = convert_to_gbp(amount=Decimal("100"),
|
||||
currency="USD",
|
||||
on_date=date(2026, 4, 1),
|
||||
cache=c)
|
||||
assert amount_gbp == Decimal("80.00")
|
||||
assert rate == Decimal("0.80")
|
||||
assert source is FxRateSource.ECB_LIVE
|
||||
|
||||
|
||||
def test_convert_raises_on_cache_miss(tmp_path: Path) -> None:
|
||||
c = FxCache(tmp_path / "fx.db")
|
||||
with pytest.raises(LookupError, match="USD.*2026-04-01"):
|
||||
convert_to_gbp(amount=Decimal("100"), currency="USD", on_date=date(2026, 4, 1), cache=c)
|
||||
Loading…
Add table
Add a link
Reference in a new issue