- Validate and enforce positive price for SPECIFIC_PRICE alerts
- improve error handling and alert squelching logic.
This commit is contained in:
2026-01-28 17:49:28 -08:00
parent 62b20cf4ab
commit 35ad75cd7c
4 changed files with 48 additions and 39 deletions

View File

@@ -20,7 +20,7 @@ class HistoryManager:
self._history[flavor][Region(region)] = History(flavor, Region(region))
async def _retrieve_data(
self, flavor: Flavor, region: Region
self, flavor: Flavor, region: Region
) -> List[Tuple[datetime.datetime, int]]:
high_fidelity_time = datetime.datetime.now(
tz=datetime.UTC
@@ -32,7 +32,7 @@ class HistoryManager:
final_response = []
def _convert_to_datetime(
data: Tuple[str, int]
data: Tuple[str, int]
) -> Tuple[datetime.datetime, int]:
return datetime.datetime.fromisoformat(data[0]), data[1]

View File

@@ -34,24 +34,24 @@ class UpdateTrigger:
self._squelched = value
def _find_next_trigger(
self,
comparison_operator: Callable,
starting_point: datetime.datetime,
history: List[Tuple[datetime.datetime, int]],
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]
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]],
self,
new_datum: Tuple[datetime.datetime, int],
history: List[Tuple[datetime.datetime, int]],
) -> bool:
match self.alert.flavor:
case Flavor.RETAIL:
@@ -102,10 +102,12 @@ class UpdateTrigger:
# 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
# First time - initialize tracking
self._last_trigger = new_datum
if new_datum[1] >= target_price:
self._last_trigger = new_datum
# Price already at/above target - alert and squelch
self._last_alerting = new_datum
self._squelched = True
return True
return False
else:
@@ -117,21 +119,21 @@ class UpdateTrigger:
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:
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):
# 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
else:
# Just update the last trigger for tracking
self._last_trigger = new_datum
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))