Chris Amow
9da7d43434
A hand-drawn line could never produce an alert, for two independent reasons: - cluster_levels dropped any single-member group scoring under 8, and a manual line weighs 1 on 5m rising to 4 on 1h. A lone drawn line was discarded before it ever reached the alert engine. - Even had it survived, the engine gates on confluence score, and the threshold is 28. Both are wrong for a drawn line specifically. Weight exists to rank levels nobody asked for; a line you drew by hand is an explicit statement that this price matters, so it survives clustering on its own and bypasses the score threshold. Everything else still has to earn its place. Alerts naming a drawn line say so — "BEARISH LINE" with the line's label rather than "BEARISH ZONE ... confluence 1", since knowing which drawing to go look at is the actionable part. A line that happens to coincide with other levels still reports as a zone, with the line named alongside. Verified against the running app: a line placed at the current price now forms a cluster, where before it was discarded outright. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
85 lines
3.7 KiB
Python
85 lines
3.7 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_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
|