74 lines
2.7 KiB
Python
74 lines
2.7 KiB
Python
import datetime
|
|
from typing import List, Dict, Tuple
|
|
|
|
from token_bot.history_manager.history import History
|
|
from token_bot.persistant_database import Alert
|
|
from token_bot.token_database import database as tdb
|
|
from token_bot.token_database.flavor import Flavor
|
|
from token_bot.token_database.region import Region
|
|
|
|
|
|
class HistoryManager:
|
|
HIGH_FIDELITY_PERIOD = "72h"
|
|
|
|
def __init__(self, token_db: tdb.Database):
|
|
self._history: Dict[Flavor, Dict[Region, History]] = {}
|
|
self._tdb: tdb.Database = token_db
|
|
for flavor in Flavor:
|
|
self._history[flavor] = {}
|
|
for region in Region:
|
|
self._history[flavor][Region(region)] = History(flavor, Region(region))
|
|
|
|
async def _retrieve_data(
|
|
self, flavor: Flavor, region: Region
|
|
) -> List[Tuple[datetime.datetime, int]]:
|
|
high_fidelity_time = datetime.datetime.now(
|
|
tz=datetime.UTC
|
|
) - datetime.timedelta(hours=72)
|
|
all_history = await self._tdb.history(flavor, region)
|
|
high_fidelity_history = await self._tdb.history(
|
|
flavor, region, self.HIGH_FIDELITY_PERIOD
|
|
)
|
|
final_response = []
|
|
|
|
def _convert_to_datetime(
|
|
data: Tuple[str, int]
|
|
) -> Tuple[datetime.datetime, int]:
|
|
return datetime.datetime.fromisoformat(data[0]), data[1]
|
|
|
|
for data_point in all_history:
|
|
datetime_tuple = _convert_to_datetime(data_point)
|
|
if datetime_tuple[0] < high_fidelity_time:
|
|
final_response.append(datetime_tuple)
|
|
|
|
final_response.extend(
|
|
_convert_to_datetime(data_point) for data_point in high_fidelity_history
|
|
)
|
|
|
|
return final_response
|
|
|
|
async def load_data(self):
|
|
for flavor in Flavor:
|
|
for r in Region:
|
|
region = Region(r)
|
|
history = History(flavor, region)
|
|
history_response = await self._retrieve_data(flavor, region)
|
|
for item in history_response:
|
|
await history.add_price(item)
|
|
self._history[flavor][region] = history
|
|
|
|
async def update_data(self, flavor: Flavor, region: Region) -> List[Alert]:
|
|
history = self._history[flavor][region]
|
|
current_price_data = await self._tdb.current(flavor)
|
|
current_region_data = current_price_data[region.value.lower()]
|
|
datum = (
|
|
datetime.datetime.fromisoformat(current_region_data[0]),
|
|
current_region_data[1],
|
|
)
|
|
if datum != history.last_price_datum:
|
|
return await history.add_price(datum)
|
|
return []
|
|
|
|
def get_history(self, flavor, region) -> History:
|
|
return self._history[flavor][region]
|