import datetime import operator from typing import Tuple, List, Callable from token_bot.persistant_database import Alert, AlertType from token_bot.token_database.flavor import Flavor class UpdateTrigger: def __init__(self, alert: Alert): self._alert: Alert = alert self._last_trigger: Tuple[datetime.datetime, int] | None = None self._last_alerting: Tuple[datetime.datetime, int] | None = None self._squelched: bool = False @property def alert(self) -> Alert: return self._alert @property def last_trigger(self) -> Tuple[datetime.datetime, int] | None: return self._last_trigger @property def last_alerting(self) -> Tuple[datetime.datetime, int] | None: return self._last_alerting @property def squelched(self): return self._squelched def _find_next_trigger( self, comparison_operator: Callable, starting_point: datetime.datetime, history: List[Tuple[datetime.datetime, int]], ): candidate_datum: Tuple[datetime.datetime, int] | None = None for datum in history: if datum[0] > starting_point and datum != history[-1]: if candidate_datum is None or comparison_operator( datum[1], candidate_datum[1] ): candidate_datum = datum self._last_trigger = candidate_datum def check_and_update( self, new_datum: Tuple[datetime.datetime, int], history: List[Tuple[datetime.datetime, int]], ) -> bool: match self.alert.flavor: case Flavor.RETAIL: start_time = datetime.datetime.fromisoformat( "2020-11-15 00:00:01.000000000+00:00" ) case Flavor.CLASSIC: start_time = datetime.datetime.fromisoformat( "2023-05-23 00:00:01.000000000+00:00" ) case _: raise NotImplementedError now = datetime.datetime.now(tz=datetime.timezone.utc) match self._alert.alert_type: case AlertType.DAILY_LOW: time_range = datetime.timedelta(days=1) comparison_operator = operator.lt case AlertType.DAILY_HIGH: time_range = datetime.timedelta(days=1) comparison_operator = operator.gt case AlertType.WEEKLY_LOW: time_range = datetime.timedelta(weeks=1) comparison_operator = operator.lt case AlertType.WEEKLY_HIGH: time_range = datetime.timedelta(weeks=1) comparison_operator = operator.gt case AlertType.MONTHLY_LOW: time_range = datetime.timedelta(days=31) comparison_operator = operator.lt case AlertType.MONTHLY_HIGH: time_range = datetime.timedelta(days=31) comparison_operator = operator.gt case AlertType.YEARLY_LOW: time_range = datetime.timedelta(days=365) comparison_operator = operator.lt case AlertType.YEARLY_HIGH: time_range = datetime.timedelta(days=365) comparison_operator = operator.gt case AlertType.ALL_TIME_LOW: time_range = now - start_time comparison_operator = operator.lt case AlertType.ALL_TIME_HIGH: time_range = now - start_time comparison_operator = operator.gt case _: # TODO: The logic here is certainly wrong for Custom time_range = datetime.timedelta(days=int(365.25 * 6)) comparison_operator = operator.eq if new_datum[0] > now - time_range: if self._last_trigger is None: self._last_trigger = new_datum self._last_alerting = new_datum return True # If the self._last_trigger falls out of scope of the alert, find the next thing that would have triggered # the alert so the next time a high or low comes up it's correctly comparing against the range of the alert # rather than since the alert triggered if self._last_trigger[0] < now - time_range: self._find_next_trigger(comparison_operator, now - time_range, history) if comparison_operator(new_datum[1], self._last_trigger[1]): self._last_trigger = new_datum self._last_alerting = new_datum was_squelched = self._squelched self._squelched = True return not was_squelched elif self._squelched: self._squelched = False return False