Chris Amow
32e25b84aa
Alerts get a number, assigned server-side and shown in both the push and the Events list, so a notification on a phone can be matched to a row on a screen when several fire together. It could not come from the browser: that counter restarts on reload and differs between tabs. It is persisted next to the cooldown state, because numbering restarting after a deploy would collide with a phone's existing notification history — which changed that file from a list to an object, with the loader still reading the old shape. Pushes now carry a timestamp in the configured zone rather than the server's. ALERT_TIMEZONE defaults to America/Chicago; containers run UTC, and a push reading 02:14 to someone seeing 21:14 costs a translation every time. The browser already formats its own times locally and is unchanged. Confluence zones are one line each, ordered by price rather than by proximity, so the list reads top to bottom the way the chart does and all of them fit on screen — sixteen zones in 394px, about 25px each, where each previously took a four-line block. Ordering is a display concern only: the server still returns them nearest-first, which is what the alert path wants. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
142 lines
6 KiB
Python
142 lines
6 KiB
Python
from app.analysis.alerts import AlertEngine
|
|
from app.analysis.confluence import cluster_levels
|
|
from app.analysis.levels import Level, LevelKind, Side
|
|
from app.bars.models import Timeframe
|
|
|
|
|
|
def level(id_: str, price: float, weight: float):
|
|
return Level(id_, LevelKind.MA, Timeframe.D1, Side.RESISTANCE, weight, 1, id_, 100, price, 0, None, 0, 100, 100, False, False)
|
|
|
|
|
|
def drawn_line(id_: str, price: float, label: str = "swing high", weight: float = 1):
|
|
return Level(id_, LevelKind.MANUAL, Timeframe.M5, Side.RESISTANCE, weight, 1, label, 100, price, 0, None, 0, 100, 100, False, False)
|
|
|
|
|
|
def test_oscillation_fires_once_until_separation_and_cooldown():
|
|
engine = AlertEngine(min_score=6, cooldown_seconds=900)
|
|
levels = [level("a", 100, 3), level("b", 100.1, 4)]
|
|
cluster = cluster_levels(levels, 100, 100, 1)
|
|
|
|
assert len(engine.evaluate(cluster, 100, 1, 0, "/ES")) == 1
|
|
assert engine.evaluate(cluster, 100.2, 1, 60, "/ES") == []
|
|
assert engine.evaluate(cluster, 100, 1, 901, "/ES") == []
|
|
|
|
far_cluster = cluster_levels(levels, 100, 103, 1)
|
|
assert engine.evaluate(far_cluster, 103, 1, 902, "/ES") == []
|
|
assert len(engine.evaluate(cluster, 100, 1, 903, "/ES")) == 1
|
|
|
|
|
|
def test_score_threshold_blocks_two_daily_mas_at_default_calibration():
|
|
engine = AlertEngine(min_score=28)
|
|
cluster = cluster_levels([level("a", 100, 12), level("b", 100.1, 12)], 100, 100, 1)
|
|
assert engine.evaluate(cluster, 100, 1, 0, "/ES") == []
|
|
|
|
|
|
def test_a_third_level_joining_the_zone_does_not_re_alert():
|
|
# Membership churns constantly as levels drift in and out of tolerance.
|
|
# Suppression is by proximity precisely so this reads as one zone.
|
|
engine = AlertEngine(min_score=6, cooldown_seconds=900)
|
|
two = cluster_levels([level("a", 100, 3), level("b", 100.1, 4)], 100, 100, 1)
|
|
assert len(engine.evaluate(two, 100, 1, 0, "/ES")) == 1
|
|
|
|
three = cluster_levels(
|
|
[level("a", 100, 3), level("b", 100.1, 4), level("c", 100.2, 5)], 100, 100, 1
|
|
)
|
|
assert engine.evaluate(three, 100, 1, 60, "/ES") == []
|
|
|
|
|
|
def test_a_lone_drawn_line_alerts_despite_the_score_threshold():
|
|
# A 5m line weighs 1 against a threshold of 28. Gating drawn lines on score
|
|
# would mean a line you deliberately drew could never alert.
|
|
engine = AlertEngine(min_score=28, cooldown_seconds=900)
|
|
clusters = cluster_levels([drawn_line("ml_1", 100)], 100, 100, 1)
|
|
|
|
assert len(clusters) == 1, "a lone drawn line must survive clustering"
|
|
alerts = engine.evaluate(clusters, 100, 1, 0, "/ES")
|
|
assert len(alerts) == 1
|
|
assert "LINE" in alerts[0].message
|
|
assert "swing high" in alerts[0].message
|
|
|
|
|
|
def test_a_lone_weak_non_drawn_level_still_does_not_alert():
|
|
# The bypass is for drawn lines only; a lone 5m average stays quiet.
|
|
weak = level("ma", 100, 1)
|
|
weak.tf = Timeframe.M5
|
|
assert cluster_levels([weak], 100, 100, 1) == []
|
|
|
|
|
|
def test_drawn_line_clustering_with_levels_reports_as_a_zone():
|
|
engine = AlertEngine(min_score=28, cooldown_seconds=900)
|
|
clusters = cluster_levels([drawn_line("ml_1", 100), level("pd", 100.1, 16)], 100, 100, 1)
|
|
|
|
alerts = engine.evaluate(clusters, 100, 1, 0, "/ES")
|
|
assert len(alerts) == 1
|
|
assert "ZONE" in alerts[0].message
|
|
assert "swing high" in alerts[0].message # the line is still named
|
|
|
|
|
|
def test_price_crossing_a_level_does_not_re_alert_on_the_side_flip():
|
|
# Side is positional, so a level sitting at price flips between support and
|
|
# resistance on every tick across it. This produced a fresh alert per
|
|
# crossing — four in two minutes when first tried against a live line.
|
|
engine = AlertEngine(min_score=6, cooldown_seconds=900)
|
|
line = [drawn_line("ml_1", 100, weight=8)]
|
|
|
|
below = cluster_levels(line, 100, 99.9, 1) # level above price -> resistance
|
|
assert len(engine.evaluate(below, 99.9, 1, 0, "/ES")) == 1
|
|
|
|
above = cluster_levels(line, 100, 100.1, 1) # price crossed -> now support
|
|
assert above and above[0].side is not below[0].side, "the flip must actually occur"
|
|
assert engine.evaluate(above, 100.1, 1, 30, "/ES") == []
|
|
|
|
|
|
def test_a_genuinely_separate_zone_still_alerts_during_cooldown():
|
|
# The cooldown is per zone, not global: a level far away is new information.
|
|
engine = AlertEngine(min_score=6, cooldown_seconds=900)
|
|
near = cluster_levels([level("a", 100, 3), level("b", 100.1, 4)], 100, 100, 1)
|
|
assert len(engine.evaluate(near, 100, 1, 0, "/ES")) == 1
|
|
|
|
far = cluster_levels([level("c", 120, 3), level("d", 120.1, 4)], 100, 120, 1)
|
|
assert len(engine.evaluate(far, 120, 1, 60, "/ES")) == 1
|
|
|
|
|
|
def test_alerts_are_numbered_and_stamped_in_local_time(tmp_path):
|
|
# A push and a screen have to agree on which alert is which. The browser
|
|
# cannot supply that: its counter restarts on reload and differs per tab.
|
|
from app.analysis.alerts import AlertEngine
|
|
|
|
engine = AlertEngine(
|
|
1.0, cooldown_seconds=0, state_path=tmp_path / "alerts.json",
|
|
timezone_name="America/Chicago",
|
|
)
|
|
engine._next_number = 41
|
|
stamp = engine._stamp(1786360000) # Mon 2026-08-10 11:06:40 UTC
|
|
|
|
assert "06:06:40" in stamp and "CDT" in stamp, stamp
|
|
assert "11:06" not in stamp, "stamped in UTC rather than the configured zone"
|
|
|
|
|
|
def test_alert_numbers_survive_a_restart(tmp_path):
|
|
# Numbers restarting from 1 after a deploy would collide with the ones
|
|
# already sitting in a phone's notification history.
|
|
from app.analysis.alerts import AlertEngine
|
|
|
|
path = tmp_path / "alerts.json"
|
|
first = AlertEngine(1.0, cooldown_seconds=0, state_path=path)
|
|
first._next_number = 87
|
|
first._save()
|
|
|
|
assert AlertEngine(1.0, cooldown_seconds=0, state_path=path)._next_number == 87
|
|
|
|
|
|
def test_an_old_bare_list_state_file_still_loads(tmp_path):
|
|
# The file predates numbering and was a plain list of fired zones.
|
|
import json
|
|
from app.analysis.alerts import AlertEngine
|
|
|
|
path = tmp_path / "alerts.json"
|
|
path.write_text(json.dumps([{"center": 5000.0, "at": 1786360000}]), encoding="utf-8")
|
|
|
|
engine = AlertEngine(1.0, cooldown_seconds=0, state_path=path)
|
|
assert len(engine._fired) == 1
|
|
assert engine._next_number == 1
|