""" FX service for TripTalley. Fetches historical exchange rates from Frankfurter with business-day fallback. Rates are cached in the database by date, from_currency, to_currency. """ import asyncio import logging from datetime import date, timedelta from decimal import Decimal from typing import Dict, List, Optional, Tuple import httpx logger = logging.getLogger(__name__) class FXService: """ Historical FX rate service using Frankfurter API with caching. Frankfurter API: https://www.frankfurter.dev/ Supports: 34 currencies, historical rates, business day lookup Strategy: 1. Try to get rate from database cache 2. If not found, fetch from Frankfurter API 3. Cache the result for future use 4. Use business-day fallback: if rate not found for exact date, look backwards """ FRANKFURTER_API = "https://api.frankfurter.dev/v1" def __init__(self, db_session): """ Initialize FX service. Args: db_session: SQLAlchemy session for caching rates """ self.db = db_session async def get_rate( self, from_currency: str, to_currency: str, on_date: date, max_lookback_days: int = 7 ) -> Decimal: """ Get historical FX rate with business-day fallback. Args: from_currency: Source currency (e.g., "USD") to_currency: Target currency (e.g., "EUR") on_date: The date to get rate for max_lookback_days: Maximum days to look back for fallback Returns: FX rate as Decimal Raises: ValueError: If no rate found within lookback period """ # Normalize currencies to uppercase from_currency = from_currency.upper() to_currency = to_currency.upper() if from_currency == to_currency: return Decimal("1.0") # Try to get from cache first cached_rate = await self._get_cached_rate(from_currency, to_currency, on_date) if cached_rate is not None: return cached_rate # Not in cache, fetch from Frankfurter API rate = await self._fetch_rate(from_currency, to_currency, on_date) if rate is not None: await self._cache_rate(from_currency, to_currency, on_date, rate) return rate # Business-day fallback: look backwards for days_back in range(1, max_lookback_days + 1): fallback_date = on_date - timedelta(days=days_back) # Check cache first cached_rate = await self._get_cached_rate( from_currency, to_currency, fallback_date ) if cached_rate is not None: return cached_rate # Fetch from API rate = await self._fetch_rate(from_currency, to_currency, fallback_date) if rate is not None: await self._cache_rate( from_currency, to_currency, fallback_date, rate ) return rate raise ValueError( f"No FX rate found for {from_currency}→{to_currency} " f"on or near {on_date} (looked back {max_lookback_days} days)" ) async def get_rate_for_expense( self, from_currency: str, to_currency: str, spent_on: date ) -> Decimal: """ Get FX rate for an expense, using spent_on date. This is a convenience wrapper that doesn't require max_lookback_days. Args: from_currency: Expense currency to_currency: Target currency (settlement or display) spent_on: Date expense was spent Returns: FX rate as Decimal """ return await self.get_rate(from_currency, to_currency, spent_on) async def _get_cached_rate( self, from_currency: str, to_currency: str, on_date: date ) -> Optional[Decimal]: """ Get rate from database cache. Returns Decimal rate or None if not found. """ # TODO: Implement database query # This is a placeholder - actual implementation will query fx_rates table return None async def _cache_rate( self, from_currency: str, to_currency: str, on_date: date, rate: Decimal ) -> None: """ Cache a rate in the database. """ # TODO: Implement database insert/update # This is a placeholder - actual implementation will insert into fx_rates table logger.debug( f"Cached FX rate: {from_currency}→{to_currency} on {on_date} = {rate}" ) async def _fetch_rate( self, from_currency: str, to_currency: str, on_date: date ) -> Optional[Decimal]: """ Fetch rate from Frankfurter API. Returns Decimal rate or None if API returns 404 (no rate for date). Raises other HTTP errors. """ # Frankfurter API: https://api.frankfurter.dev/v1/{date}?from={base}&to={currencies} url = f"{self.FRANKFURTER_API}/{on_date.strftime('%Y-%m-%d')}" params = { "from": from_currency, "to": to_currency } async with httpx.AsyncClient() as client: try: response = await client.get(url, params=params, timeout=10.0) if response.status_code == 404: # No rate for this date return None response.raise_for_status() data = response.json() # Frankfurter returns {"amount": 1.0, "base": "USD", "date": "2024-01-15", "rates": {"EUR": 0.91366}} rates = data.get("rates", {}) if to_currency in rates: return Decimal(str(rates[to_currency])) logger.warning( f"Frankfurter API missing {to_currency} in response: {data}" ) return None except httpx.TimeoutException: logger.warning( f"Timeout fetching FX rate for {from_currency}→{to_currency} on {on_date}" ) return None except httpx.HTTPError as e: logger.error( f"HTTP error fetching FX rate: {e}" ) raise async def get_rates_for_date( self, on_date: date, base_currency: str = "USD" ) -> Dict[str, Decimal]: """ Get all rates for a specific date relative to a base currency. This is more efficient than fetching individual rates. Args: on_date: The date to get rates for base_currency: Base currency for rates Returns: Dict mapping currency codes to rates """ url = f"{self.FRANKFURTER_API}/{on_date.strftime('%Y-%m-%d')}" params = { "from": base_currency, "to": ",".join([ "EUR", "GBP", "JPY", "AUD", "CAD", "CHF", "CNY", "SEK", "NOK", "DKK", "NZD", "MXN", "SGD", "HKD", "KRW", "INR", "BRL", "ZAR", "RUB", "TRY", "BHD", "KWD", "OMR", "JOD" ]) } async with httpx.AsyncClient() as client: response = await client.get(url, params=params, timeout=10.0) response.raise_for_status() data = response.json() rates = data.get("rates", {}) return {k: Decimal(str(v)) for k, v in rates.items()} # Example usage async def main(): """Test the FX service.""" # In production, pass a real SQLAlchemy session service = FXService(db_session=None) # Get rate for a specific date rate = await service.get_rate("USD", "EUR", date(2024, 1, 15)) print(f"USD→EUR on 2024-01-15: {rate}") # Test business-day fallback try: rate_bhd = await service.get_rate("USD", "BHD", date(2024, 1, 1)) # New Year's Day print(f"USD→BHD on 2024-01-01 (New Year): {rate_bhd}") except ValueError as e: print(f"Fallback result: {e}") if __name__ == "__main__": asyncio.run(main())