1
0

exchange_rate: clear up confusion re type of ExchangeBase._history

follow-up 41c0558595
This commit is contained in:
SomberNight
2025-05-05 15:01:11 +00:00
parent f25ddbc8f9
commit c50aa7e777

View File

@@ -46,7 +46,7 @@ class ExchangeBase(Logger):
def __init__(self, on_quotes, on_history):
Logger.__init__(self)
self._history = {} # type: Dict[str, Dict[str, float]]
self._history = {} # type: Dict[str, Dict[str, str | float]]
self._quotes = {} # type: Dict[str, Optional[Decimal]]
self._quotes_timestamp = 0 # type: Union[int, float]
self.on_quotes = on_quotes
@@ -101,7 +101,7 @@ class ExchangeBase(Logger):
@staticmethod
def _read_historical_rates_from_file(
*, exchange_name: str, ccy: str, cache_dir: str,
) -> Tuple[Optional[dict], Optional[float]]:
) -> Tuple[Optional[Dict[str, str]], Optional[float]]:
filename = os.path.join(cache_dir, f"{exchange_name}_{ccy}")
if not os.path.exists(filename):
return None, None
@@ -125,6 +125,7 @@ class ExchangeBase(Logger):
)
if not h:
return None
assert timestamp is not None
h['timestamp'] = timestamp
self._history[ccy] = h
self.on_history()
@@ -132,11 +133,17 @@ class ExchangeBase(Logger):
@staticmethod
def _write_historical_rates_to_file(
*, exchange_name: str, ccy: str, cache_dir: str, history: Dict[str, float],
*, exchange_name: str, ccy: str, cache_dir: str, history: Dict[str, str],
) -> None:
# sanity check types of history dict
assert 'timestamp' not in history
for key, rate in history.items():
assert isinstance(key, str), f"{exchange_name=}. {ccy=}. {key=!r}. {rate=!r}"
assert isinstance(rate, str), f"{exchange_name=}. {ccy=}. {key=!r}. {rate=!r}"
# write to file
filename = os.path.join(cache_dir, f"{exchange_name}_{ccy}")
with open(filename, 'w', encoding='utf-8') as f:
f.write(json.dumps(history))
f.write(json.dumps(history, sort_keys=True))
@log_exceptions
async def get_historical_rates_safe(self, ccy: str, cache_dir: str) -> None:
@@ -151,7 +158,7 @@ class ExchangeBase(Logger):
self.logger.exception(f"failed fx history: {repr(e)}")
return
# cast rates to str
h_new = {date_str: str(rate) for (date_str, rate) in h_new.items()}
h_new = {date_str: str(rate) for (date_str, rate) in h_new.items()} # type: Dict[str, str]
# merge old history and new history. resolve duplicate dates using new data.
h_old, _timestamp = self._read_historical_rates_from_file(
exchange_name=self.name(), ccy=ccy, cache_dir=cache_dir,
@@ -162,7 +169,7 @@ class ExchangeBase(Logger):
self._write_historical_rates_to_file(
exchange_name=self.name(), ccy=ccy, cache_dir=cache_dir, history=h,
)
h['timestamp'] = time.time()
h['timestamp'] = time.time() # note: this is the only item in h that has a float value
self._history[ccy] = h
self.on_history()