wow-token-app-bot/token_bot/history_manager/update_trigger.py

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 - initialize tracking
                    self._last_trigger = new_datum
                    if new_datum[1] >= target_price:
                        # Price already at/above target - alert and squelch
                        self._last_alerting = new_datum
                        self._squelched = True
                        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

                    # Always update last_trigger for tracking
                    self._last_trigger = new_datum

                    if crossed_up or crossed_down:
                        # We're crossing the threshold
                        if self._squelched:
                            # Currently squelched - this crossing unsquelches us
                            # but doesn't alert (prevents rapid-fire alerts on oscillation)
                            self._squelched = False
                            return False
                        else:
                            # Not squelched - send alert and squelch
                            self._last_alerting = new_datum
                            self._squelched = True
                            return True
                    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