161 lines
6.7 KiB
Python
161 lines
6.7 KiB
Python
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
|
|
|
|
@squelched.setter
|
|
def squelched(self, value):
|
|
self._squelched = value
|
|
|
|
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 AlertType.SPECIFIC_PRICE:
|
|
# For custom price alerts, check if the price crosses the threshold
|
|
# We alert when price moves from below to above (or vice versa)
|
|
target_price = self._alert.price
|
|
if self._last_trigger is None:
|
|
# First time - check if current price crossed the threshold
|
|
if new_datum[1] >= target_price:
|
|
self._last_trigger = new_datum
|
|
self._last_alerting = new_datum
|
|
return True
|
|
return False
|
|
else:
|
|
# Check if we crossed the threshold
|
|
old_price = self._last_trigger[1]
|
|
new_price = new_datum[1]
|
|
|
|
# Alert if we cross the threshold in either direction
|
|
crossed_up = old_price < target_price <= new_price
|
|
crossed_down = old_price >= target_price > new_price
|
|
|
|
if crossed_up or crossed_down:
|
|
self._last_trigger = new_datum
|
|
self._last_alerting = new_datum
|
|
was_squelched = self._squelched
|
|
self._squelched = True
|
|
return not was_squelched
|
|
elif self._squelched:
|
|
# Update last_trigger but don't alert (we're tracking current price)
|
|
self._last_trigger = new_datum
|
|
# If we moved away from threshold, allow next crossing to alert
|
|
if (crossed_up and new_price < target_price) or (crossed_down and new_price >= target_price):
|
|
self._squelched = False
|
|
else:
|
|
# Just update the last trigger for tracking
|
|
self._last_trigger = new_datum
|
|
return False
|
|
case _:
|
|
time_range = datetime.timedelta(days=int(365.25 * 6))
|
|
comparison_operator = operator.eq
|
|
|
|
if new_datum[0] > now - time_range and self._alert.alert_type != AlertType.SPECIFIC_PRICE:
|
|
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
|