history.py (3003B)
1 import datetime 2 from typing import List, Tuple 3 4 from token_bot.history_manager.update_trigger import UpdateTrigger 5 from token_bot.persistant_database import AlertType, Alert 6 from token_bot.token_database.flavor import Flavor 7 from token_bot.token_database.region import Region 8 9 10 class History: 11 def __init__(self, flavor: Flavor, region: Region): 12 self._flavor: Flavor = flavor 13 self._region: Region = region 14 self._history: List[Tuple[datetime, int]] = [] 15 self._last_price_movement: int = 0 16 self._latest_price_datum: Tuple[datetime.datetime, int] | None = None 17 self._update_triggers: List[UpdateTrigger] = [] 18 # Create triggers for all non-custom alert types 19 for alert_type in AlertType: 20 if alert_type != AlertType.SPECIFIC_PRICE: 21 self._update_triggers.append( 22 UpdateTrigger(Alert(alert_type, flavor, self._region)) 23 ) 24 # SPECIFIC_PRICE triggers are created on-demand as they have unique prices 25 26 @property 27 def flavor(self) -> Flavor: 28 return self._flavor 29 30 @property 31 def region(self) -> Region: 32 return self._region 33 34 @property 35 def last_price_datum(self) -> Tuple[datetime.datetime, int] | None: 36 return self._latest_price_datum 37 38 @property 39 def last_price_movement(self) -> int: 40 return self._last_price_movement 41 42 @property 43 def history(self) -> List[Tuple[datetime.datetime, int]]: 44 return self._history 45 46 async def _process_update_triggers(self) -> List[Alert]: 47 alerts = [] 48 for trigger in self._update_triggers: 49 if trigger.check_and_update(self._latest_price_datum, self._history): 50 alerts.append(trigger.alert) 51 return alerts 52 53 async def add_price(self, datum: Tuple[datetime.datetime, int]) -> List[Alert]: 54 if self._latest_price_datum is None: 55 self._latest_price_datum = datum 56 self._last_price_movement = datum[1] - self._latest_price_datum[1] 57 self._latest_price_datum = datum 58 self._history.append(datum) 59 return await self._process_update_triggers() 60 61 async def find_update_trigger_from_alert(self, alert: Alert, initial_import: bool = False) -> UpdateTrigger: 62 for trigger in self._update_triggers: 63 if trigger.alert == alert: 64 return trigger 65 66 # If not found and it's a SPECIFIC_PRICE alert, create it on-demand 67 if alert.alert_type == AlertType.SPECIFIC_PRICE: 68 new_trigger = UpdateTrigger(alert) 69 if initial_import: 70 new_trigger.squelched = True 71 self._update_triggers.append(new_trigger) 72 # Initialize the trigger with current history 73 if self._latest_price_datum is not None: 74 new_trigger.check_and_update(self._latest_price_datum, self._history) 75 new_trigger.squelched = False 76 return new_trigger 77 78 raise ValueError