history_manager.py (3357B)
1 import datetime 2 from typing import List, Dict, Tuple 3 4 from token_bot.history_manager.history import History 5 from token_bot.persistant_database import Alert 6 from token_bot.token_database import database as tdb 7 from token_bot.token_database.flavor import Flavor 8 from token_bot.token_database.region import Region 9 10 11 class HistoryManager: 12 HIGH_FIDELITY_PERIOD = "72h" 13 14 def __init__(self, token_db: tdb.Database): 15 self._history: Dict[Flavor, Dict[Region, History]] = {} 16 self._tdb: tdb.Database = token_db 17 for flavor in Flavor: 18 self._history[flavor] = {} 19 for region in Region: 20 self._history[flavor][Region(region)] = History(flavor, Region(region)) 21 22 async def _retrieve_data( 23 self, flavor: Flavor, region: Region 24 ) -> List[Tuple[datetime.datetime, int]]: 25 high_fidelity_time = datetime.datetime.now( 26 tz=datetime.UTC 27 ) - datetime.timedelta(hours=72) 28 all_history = await self._tdb.history(flavor, region) 29 high_fidelity_history = await self._tdb.history( 30 flavor, region, self.HIGH_FIDELITY_PERIOD 31 ) 32 final_response = [] 33 34 def _convert_to_datetime( 35 data: Tuple[str, int] 36 ) -> Tuple[datetime.datetime, int]: 37 return datetime.datetime.fromisoformat(data[0]), data[1] 38 39 for data_point in all_history: 40 datetime_tuple = _convert_to_datetime(data_point) 41 if datetime_tuple[0] < high_fidelity_time: 42 final_response.append(datetime_tuple) 43 44 final_response.extend( 45 _convert_to_datetime(data_point) for data_point in high_fidelity_history 46 ) 47 48 return final_response 49 50 async def load_data(self): 51 for flavor in Flavor: 52 for r in Region: 53 region = Region(r) 54 history = History(flavor, region) 55 history_response = await self._retrieve_data(flavor, region) 56 for item in history_response: 57 await history.add_price(item) 58 self._history[flavor][region] = history 59 60 async def load_custom_alerts(self, custom_alerts: List[Alert]): 61 """Load custom price alerts and initialize their triggers with historical data.""" 62 for alert in custom_alerts: 63 history = self._history[alert.flavor][alert.region] 64 # This will create the trigger on-demand via find_update_trigger_from_alert 65 trigger = await history.find_update_trigger_from_alert(alert) 66 # Process all historical data through this trigger to initialize its state 67 for datum in history.history: 68 trigger.check_and_update(datum, history.history) 69 70 async def update_data(self, flavor: Flavor, region: Region) -> List[Alert]: 71 history = self._history[flavor][region] 72 current_price_data = await self._tdb.current(flavor) 73 current_region_data = current_price_data[region.value.lower()] 74 datum = ( 75 datetime.datetime.fromisoformat(current_region_data[0]), 76 current_region_data[1], 77 ) 78 if datum != history.last_price_datum: 79 return await history.add_price(datum) 80 return [] 81 82 def get_history(self, flavor, region) -> History: 83 return self._history[flavor][region]