106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
"""Currency exchange rates."""
|
|
|
|
import json
|
|
import os
|
|
import typing
|
|
from datetime import datetime, timedelta
|
|
from decimal import Decimal
|
|
|
|
import flask
|
|
import httpx
|
|
|
|
|
|
async def get_gbpusd(config: flask.config.Config) -> Decimal:
|
|
"""Get the current value for GBPUSD, with caching."""
|
|
access_key = config["EXCHANGERATE_ACCESS_KEY"]
|
|
data_dir = config["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"])
|
|
|
|
|
|
def read_cached_rates(filename: str, currencies: list[str]) -> dict[str, Decimal]:
|
|
"""Read FX rates from cache."""
|
|
with open(filename) as file:
|
|
data = json.load(file, parse_float=Decimal)
|
|
return {
|
|
cur: Decimal(data["quotes"][f"GBP{cur}"])
|
|
for cur in currencies
|
|
if f"GBP{cur}" in data["quotes"]
|
|
}
|
|
|
|
|
|
def get_rates(config: flask.config.Config) -> dict[str, Decimal]:
|
|
"""Get current values of exchange rates for a list of currencies against GBP."""
|
|
currencies = config["CURRENCIES"]
|
|
access_key = config["EXCHANGERATE_ACCESS_KEY"]
|
|
data_dir = config["DATA_DIR"]
|
|
|
|
now = datetime.now()
|
|
now_str = now.strftime("%Y-%m-%d_%H:%M")
|
|
fx_dir = os.path.join(data_dir, "fx")
|
|
os.makedirs(fx_dir, exist_ok=True) # Ensure the directory exists
|
|
|
|
currency_string = ",".join(sorted(currencies))
|
|
file_suffix = f"{currency_string}_to_GBP.json"
|
|
existing_data = os.listdir(fx_dir)
|
|
existing_files = [f for f in existing_data if f.endswith(file_suffix)]
|
|
|
|
if existing_files:
|
|
recent_filename = max(existing_files)
|
|
recent = datetime.strptime(recent_filename[:16], "%Y-%m-%d_%H:%M")
|
|
delta = now - recent
|
|
|
|
full_path = os.path.join(fx_dir, recent_filename)
|
|
if delta < timedelta(hours=12):
|
|
return read_cached_rates(full_path, currencies)
|
|
|
|
url = "http://api.exchangerate.host/live"
|
|
params = {"currencies": currency_string, "source": "GBP", "access_key": access_key}
|
|
|
|
filename = f"{now_str}_{file_suffix}"
|
|
try:
|
|
with httpx.Client() as client:
|
|
response = client.get(url, params=params)
|
|
except httpx.ConnectError:
|
|
return read_cached_rates(full_path, currencies)
|
|
|
|
try:
|
|
data = json.loads(response.text, parse_float=Decimal)
|
|
except json.decoder.JSONDecodeError:
|
|
return read_cached_rates(full_path, currencies)
|
|
|
|
with open(os.path.join(fx_dir, filename), "w") as file:
|
|
file.write(response.text)
|
|
|
|
return {
|
|
cur: Decimal(data["quotes"][f"GBP{cur}"])
|
|
for cur in currencies
|
|
if f"GBP{cur}" in data["quotes"]
|
|
}
|