Spaces:
Configuration error
Configuration error
File size: 9,043 Bytes
83bcbe3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 | """
alert_engine.py — Core alert logic with state-machine deduplication,
alert cooldown, and in-memory alert history.
State machine per asset:
OUTSIDE — price is outside the user-defined range
INSIDE — price has entered the range (alert was already sent)
Transitions:
OUTSIDE → INSIDE → triggers an alert (subject to cooldown)
INSIDE → INSIDE → no-op (dedup)
INSIDE → OUTSIDE → resets state so next entry can trigger again
OUTSIDE → OUTSIDE → no-op
Cooldown:
After an alert fires for an asset, the same asset cannot fire again
for ALERT_COOLDOWN_SECONDS seconds, even if it leaves and re-enters
the range. This prevents spam when price oscillates at a boundary.
"""
import threading
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum, auto
from typing import Callable, Deque, Optional
import config
from logger import get_logger
log = get_logger("alert_engine")
class RangeState(Enum):
OUTSIDE = auto()
INSIDE = auto()
@dataclass
class AssetAlert:
"""Tracked state for a single asset."""
name: str
min_price: float
max_price: float
state: RangeState = RangeState.OUTSIDE
last_price: Optional[float] = None
last_alert_time: Optional[datetime] = None
# Whether alerts for this asset are paused by the user
paused: bool = False
@dataclass
class AlertEvent:
"""Data object passed to the alert callback and stored in history."""
asset_name: str
price: float
min_price: float
max_price: float
timestamp: datetime
class AlertEngine:
"""
Thread-safe alert engine that evaluates price updates against
user-defined ranges and fires callbacks on state transitions.
Features:
- Deduplication via INSIDE/OUTSIDE state machine
- Per-asset alert cooldown (configurable via config.ALERT_COOLDOWN_SECONDS)
- In-memory alert history (capped at config.ALERT_HISTORY_MAX)
- Per-asset pause/resume
- save/load via persistence module
"""
def __init__(self, on_alert: Callable[[AlertEvent], None]) -> None:
self._lock = threading.Lock()
self._assets: dict[str, AssetAlert] = {}
self._on_alert = on_alert
self._history: Deque[AlertEvent] = deque(maxlen=config.ALERT_HISTORY_MAX)
# Global pause flag — silences ALL alerts when True
self._globally_paused: bool = False
# ── Pause / Resume ───────────────────────────────────────────────────
def pause_all(self) -> None:
"""Pause all alert notifications globally."""
with self._lock:
self._globally_paused = True
log.info("All alerts paused globally")
def resume_all(self) -> None:
"""Resume all alert notifications globally."""
with self._lock:
self._globally_paused = False
log.info("All alerts resumed globally")
def is_paused(self) -> bool:
with self._lock:
return self._globally_paused
# ── Asset management ─────────────────────────────────────────────────
def add_asset(self, name: str, min_price: float, max_price: float) -> None:
"""Add or update a tracked asset with a price range."""
name = name.upper()
with self._lock:
if name in self._assets:
existing = self._assets[name]
existing.min_price = min_price
existing.max_price = max_price
existing.state = RangeState.OUTSIDE # reset on range change
log.info("Updated range for %s: $%.4f – $%.4f", name, min_price, max_price)
else:
self._assets[name] = AssetAlert(
name=name, min_price=min_price, max_price=max_price
)
log.info("Added asset %s: $%.4f – $%.4f", name, min_price, max_price)
def remove_asset(self, name: str) -> bool:
"""Remove a tracked asset. Returns True if it existed."""
name = name.upper()
with self._lock:
if name in self._assets:
del self._assets[name]
log.info("Removed asset %s", name)
return True
return False
def get_tracked_assets(self) -> dict[str, tuple[float, float]]:
"""Return {name: (min, max)} for all tracked assets."""
with self._lock:
return {
name: (a.min_price, a.max_price)
for name, a in self._assets.items()
}
def get_asset_details(self) -> list[dict]:
"""Return detailed info for all tracked assets (for /list command and persistence)."""
with self._lock:
result = []
for a in self._assets.values():
result.append({
"name": a.name,
"min": a.min_price,
"max": a.max_price,
"state": a.state.name,
"last_price": a.last_price,
"last_alert": a.last_alert_time.isoformat() if a.last_alert_time else None,
"paused": a.paused,
})
return result
def get_asset_price(self, name: str) -> Optional[float]:
"""Return the latest known price for an asset, or None if unavailable."""
name = name.upper()
with self._lock:
asset = self._assets.get(name)
return asset.last_price if asset else None
# ── Alert History ────────────────────────────────────────────────────
def get_alert_history(self, limit: int = 10) -> list[AlertEvent]:
"""Return up to `limit` most-recent alert events (newest first)."""
with self._lock:
events = list(self._history)
events.reverse()
return events[:limit]
# ── Price evaluation ─────────────────────────────────────────────────
def on_price_update(self, name: str, price: float) -> None:
"""
Called by fetchers whenever a new price is received.
Evaluates the state machine and fires the alert callback if needed.
"""
name = name.upper()
with self._lock:
asset = self._assets.get(name)
if asset is None:
return # not tracked
asset.last_price = price
in_range = asset.min_price <= price <= asset.max_price
if in_range and asset.state == RangeState.OUTSIDE:
# Check global pause
if self._globally_paused or asset.paused:
log.debug("%s would alert at $%.4f but is paused", name, price)
asset.state = RangeState.INSIDE # still update state
return
# Check cooldown
if asset.last_alert_time is not None:
now_utc = datetime.now(timezone.utc)
elapsed = (now_utc - asset.last_alert_time).total_seconds()
if elapsed < config.ALERT_COOLDOWN_SECONDS:
remaining = int(config.ALERT_COOLDOWN_SECONDS - elapsed)
log.debug(
"%s in range but cooldown active (%ds remaining)", name, remaining
)
asset.state = RangeState.INSIDE
return
# Price just entered the range → fire alert
asset.state = RangeState.INSIDE
now = datetime.now(timezone.utc)
asset.last_alert_time = now
event = AlertEvent(
asset_name=name,
price=price,
min_price=asset.min_price,
max_price=asset.max_price,
timestamp=now,
)
self._history.append(event)
log.info(
"🔔 ALERT: %s entered range [$%.4f – $%.4f] at $%.4f",
name, asset.min_price, asset.max_price, price,
)
# Release lock before callback to avoid deadlocks
self._lock.release()
try:
self._on_alert(event)
finally:
self._lock.acquire()
elif not in_range and asset.state == RangeState.INSIDE:
# Price exited the range → reset for re-alerting
asset.state = RangeState.OUTSIDE
log.debug(
"%s exited range at $%.4f — state reset to OUTSIDE", name, price
)
|