45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
"""Currency exchange rates."""
|
|
|
|
import configparser
|
|
import json
|
|
import os
|
|
import typing
|
|
from datetime import datetime, timedelta
|
|
from decimal import Decimal
|
|
|
|
import httpx
|
|
|
|
|
|
async def get_gbpusd(config: configparser.ConfigParser) -> Decimal:
|
|
"""Get the current value for GBPUSD, with caching."""
|
|
access_key = config.get("exchangerate", "access_key")
|
|
data_dir = config.get("data", "dir")
|
|
|
|
now = datetime.now()
|
|
now_str = now.strftime("%Y-%m-%d_%H:%M")
|
|
fx_dir = os.path.join(data_dir, "fx")
|
|
existing_data = os.listdir(fx_dir)
|
|
existing = [f for f in existing_data if f.endswith("_GBPUSD.json")]
|
|
if existing:
|
|
recent_filename = max(existing)
|
|
recent = datetime.strptime(recent_filename, "%Y-%m-%d_%H:%M_GBPUSD.json")
|
|
delta = now - recent
|
|
|
|
if existing and delta < timedelta(hours=6):
|
|
full = os.path.join(fx_dir, recent_filename)
|
|
data = json.load(open(full), parse_float=Decimal)
|
|
if "quotes" in data and "USDGBP" in data["quotes"]:
|
|
return typing.cast(Decimal, 1 / data["quotes"]["USDGBP"])
|
|
|
|
url = "http://api.exchangerate.host/live"
|
|
params = {"currencies": "GBP,USD", "access_key": access_key}
|
|
|
|
filename = f"{fx_dir}/{now_str}_GBPUSD.json"
|
|
async with httpx.AsyncClient() as client:
|
|
r = await client.get(url, params=params)
|
|
|
|
open(filename, "w").write(r.text)
|
|
data = json.loads(r.text, parse_float=Decimal)
|
|
|
|
return typing.cast(Decimal, 1 / data["quotes"]["USDGBP"])
|