chart/tests/test_alerts.py

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


def test_the_push_carries_the_number_and_time_but_the_screen_message_does_not(tmp_path):
    # ntfy is plain text, so the number and the local time have to live in the
    # body. The browser gets `number` as a field and draws a badge, so printing
    # them in the message too would show the same number twice in one row.
    from app.analysis.alerts import AlertEngine
    from app.analysis.confluence import Cluster
    from app.analysis.levels import Level, LevelKind, Side
    from app.bars.models import Timeframe

    level = Level("pd:high", LevelKind.HORIZONTAL, Timeframe.D1, Side.RESISTANCE, 16, 1,
                  "PDH", 7784, 5000, 0, None, 0, 7784, 7784, False, False)
    cluster = Cluster("cl_x", Side.RESISTANCE, 7784.0, 7783.0, 7785.5, 21, [level], 0.25)
    engine = AlertEngine(1.0, cooldown_seconds=0, timezone_name="America/Chicago")
    engine._next_number = 47

    alert = engine.evaluate([cluster], 7784.25, 4.0, 1786430800, "/ES")[0]

    assert alert.number == 47
    assert alert.push.startswith("#47 ")
    assert "CDT" in alert.push
    assert not alert.message.startswith("#")
    assert "CDT" not in alert.message