70 lines
2.7 KiB
Python
70 lines
2.7 KiB
Python
"""Dead-man's switch in the qabot — alert on a stale event heartbeat during
|
|
market hours, at most every 30 min, silent when fresh or when the market is
|
|
closed. Clock, heartbeat read, and send are all injected: no Telegram, no
|
|
Postgres (uses the real holiday-aware market calendar with fixed dates)."""
|
|
|
|
from datetime import datetime
|
|
|
|
from telegram_bot import DeadmanCheck
|
|
|
|
TUE_10AM = datetime(2026, 7, 7, 10, 0) # regular trading day
|
|
SAT_10AM = datetime(2026, 7, 11, 10, 0) # weekend
|
|
TUE_EVENING = datetime(2026, 7, 7, 20, 35) # after the 20:30 ping gate
|
|
|
|
|
|
def make(now, max_ts, **kw):
|
|
"""DeadmanCheck with a settable clock/heartbeat and a capturing send."""
|
|
state = {"now": now, "max_ts": max_ts, "sent": []}
|
|
dm = DeadmanCheck(lambda: state["max_ts"], state["sent"].append,
|
|
stale_min=kw.pop("stale_min", 5),
|
|
daily_ping=kw.pop("daily_ping", "1"),
|
|
now_et=lambda: state["now"], **kw)
|
|
return dm, state
|
|
|
|
|
|
def test_stale_during_market_hours_alerts_once():
|
|
dm, st = make(TUE_10AM, TUE_10AM.replace(hour=9, minute=40)) # 20 min stale
|
|
msg = dm.check()
|
|
assert msg is not None and "DEAD-MAN" in msg and "20 min" in msg
|
|
assert st["sent"] == [msg]
|
|
|
|
|
|
def test_stale_realert_suppressed_within_30_min():
|
|
dm, st = make(TUE_10AM, TUE_10AM.replace(hour=9, minute=40))
|
|
assert dm.check() is not None
|
|
st["now"] = TUE_10AM.replace(minute=10) # +10 min, still stale
|
|
assert dm.check() is None
|
|
assert len(st["sent"]) == 1
|
|
st["now"] = TUE_10AM.replace(minute=31) # 31 min after first alert
|
|
assert dm.check() is not None # re-alert allowed
|
|
assert len(st["sent"]) == 2
|
|
|
|
|
|
def test_fresh_heartbeat_no_alert():
|
|
dm, st = make(TUE_10AM, datetime(2026, 7, 7, 9, 59)) # 1 min old
|
|
assert dm.check() is None
|
|
assert st["sent"] == []
|
|
|
|
|
|
def test_stale_outside_market_hours_no_alert():
|
|
dm, st = make(SAT_10AM, datetime(2026, 7, 6, 15, 59)) # very stale, Saturday
|
|
assert dm.check() is None
|
|
assert st["sent"] == []
|
|
|
|
|
|
def test_nightly_ping_once_after_2030():
|
|
dm, st = make(datetime(2026, 7, 7, 20, 0), datetime(2026, 7, 7, 15, 59))
|
|
assert dm.check() is None # before 20:30: no ping
|
|
st["now"] = TUE_EVENING
|
|
msg = dm.check()
|
|
assert msg is not None and "heartbeat" in msg and "✅" in msg
|
|
st["now"] = TUE_EVENING.replace(minute=50)
|
|
assert dm.check() is None # once per day only
|
|
assert len(st["sent"]) == 1
|
|
|
|
|
|
def test_nightly_ping_flags_a_day_with_no_events():
|
|
dm, _ = make(TUE_EVENING, datetime(2026, 7, 6, 15, 59)) # last event yesterday
|
|
msg = dm.check()
|
|
assert msg is not None and "NO events today" in msg
|