update_trigger.py (6836B)
1 import datetime 2 import operator 3 from typing import Tuple, List, Callable 4 5 from token_bot.persistant_database import Alert, AlertType 6 from token_bot.token_database.flavor import Flavor 7 8 9 class UpdateTrigger: 10 def __init__(self, alert: Alert): 11 self._alert: Alert = alert 12 self._last_trigger: Tuple[datetime.datetime, int] | None = None 13 self._last_alerting: Tuple[datetime.datetime, int] | None = None 14 self._squelched: bool = False 15 16 @property 17 def alert(self) -> Alert: 18 return self._alert 19 20 @property 21 def last_trigger(self) -> Tuple[datetime.datetime, int] | None: 22 return self._last_trigger 23 24 @property 25 def last_alerting(self) -> Tuple[datetime.datetime, int] | None: 26 return self._last_alerting 27 28 @property 29 def squelched(self): 30 return self._squelched 31 32 @squelched.setter 33 def squelched(self, value): 34 self._squelched = value 35 36 def _find_next_trigger( 37 self, 38 comparison_operator: Callable, 39 starting_point: datetime.datetime, 40 history: List[Tuple[datetime.datetime, int]], 41 ): 42 candidate_datum: Tuple[datetime.datetime, int] | None = None 43 for datum in history: 44 if datum[0] > starting_point and datum != history[-1]: 45 if candidate_datum is None or comparison_operator( 46 datum[1], candidate_datum[1] 47 ): 48 candidate_datum = datum 49 self._last_trigger = candidate_datum 50 51 def check_and_update( 52 self, 53 new_datum: Tuple[datetime.datetime, int], 54 history: List[Tuple[datetime.datetime, int]], 55 ) -> bool: 56 match self.alert.flavor: 57 case Flavor.RETAIL: 58 start_time = datetime.datetime.fromisoformat( 59 "2020-11-15 00:00:01.000000000+00:00" 60 ) 61 case Flavor.CLASSIC: 62 start_time = datetime.datetime.fromisoformat( 63 "2023-05-23 00:00:01.000000000+00:00" 64 ) 65 case _: 66 raise NotImplementedError 67 68 now = datetime.datetime.now(tz=datetime.timezone.utc) 69 match self._alert.alert_type: 70 case AlertType.DAILY_LOW: 71 time_range = datetime.timedelta(days=1) 72 comparison_operator = operator.lt 73 case AlertType.DAILY_HIGH: 74 time_range = datetime.timedelta(days=1) 75 comparison_operator = operator.gt 76 case AlertType.WEEKLY_LOW: 77 time_range = datetime.timedelta(weeks=1) 78 comparison_operator = operator.lt 79 case AlertType.WEEKLY_HIGH: 80 time_range = datetime.timedelta(weeks=1) 81 comparison_operator = operator.gt 82 case AlertType.MONTHLY_LOW: 83 time_range = datetime.timedelta(days=31) 84 comparison_operator = operator.lt 85 case AlertType.MONTHLY_HIGH: 86 time_range = datetime.timedelta(days=31) 87 comparison_operator = operator.gt 88 case AlertType.YEARLY_LOW: 89 time_range = datetime.timedelta(days=365) 90 comparison_operator = operator.lt 91 case AlertType.YEARLY_HIGH: 92 time_range = datetime.timedelta(days=365) 93 comparison_operator = operator.gt 94 case AlertType.ALL_TIME_LOW: 95 time_range = now - start_time 96 comparison_operator = operator.lt 97 case AlertType.ALL_TIME_HIGH: 98 time_range = now - start_time 99 comparison_operator = operator.gt 100 case AlertType.SPECIFIC_PRICE: 101 # For custom price alerts, check if the price crosses the threshold 102 # We alert when price moves from below to above (or vice versa) 103 target_price = self._alert.price 104 if self._last_trigger is None: 105 # First time - initialize tracking 106 self._last_trigger = new_datum 107 if new_datum[1] >= target_price: 108 # Price already at/above target - alert and squelch 109 self._last_alerting = new_datum 110 self._squelched = True 111 return True 112 return False 113 else: 114 # Check if we crossed the threshold 115 old_price = self._last_trigger[1] 116 new_price = new_datum[1] 117 118 # Alert if we cross the threshold in either direction 119 crossed_up = old_price < target_price <= new_price 120 crossed_down = old_price >= target_price > new_price 121 122 # Always update last_trigger for tracking 123 self._last_trigger = new_datum 124 125 if crossed_up or crossed_down: 126 # We're crossing the threshold 127 if self._squelched: 128 # Currently squelched - this crossing unsquelches us 129 # but doesn't alert (prevents rapid-fire alerts on oscillation) 130 self._squelched = False 131 return False 132 else: 133 # Not squelched - send alert and squelch 134 self._last_alerting = new_datum 135 self._squelched = True 136 return True 137 return False 138 case _: 139 time_range = datetime.timedelta(days=int(365.25 * 6)) 140 comparison_operator = operator.eq 141 142 if new_datum[0] > now - time_range and self._alert.alert_type != AlertType.SPECIFIC_PRICE: 143 if self._last_trigger is None: 144 self._last_trigger = new_datum 145 self._last_alerting = new_datum 146 return True 147 148 # If the self._last_trigger falls out of scope of the alert, find the next thing that would have triggered 149 # the alert so the next time a high or low comes up it's correctly comparing against the range of the alert 150 # rather than since the alert triggered 151 if self._last_trigger[0] < now - time_range: 152 self._find_next_trigger(comparison_operator, now - time_range, history) 153 154 if comparison_operator(new_datum[1], self._last_trigger[1]): 155 self._last_trigger = new_datum 156 self._last_alerting = new_datum 157 was_squelched = self._squelched 158 self._squelched = True 159 return not was_squelched 160 elif self._squelched: 161 self._squelched = False 162 return False