diff --git a/app/analysis/alerts.py b/app/analysis/alerts.py index 8b2bb52..dc4dc4c 100644 --- a/app/analysis/alerts.py +++ b/app/analysis/alerts.py @@ -1,6 +1,7 @@ from dataclasses import dataclass from app.analysis.confluence import Cluster +from app.analysis.levels import LevelKind @dataclass(slots=True) @@ -61,7 +62,13 @@ class AlertEngine: # Strongest first, so when several overlapping zones qualify at once the # one that survives suppression is the most significant. for cluster in sorted(clusters, key=lambda item: item.score, reverse=True): - if cluster.score < self.min_score or abs(cluster.center - current_price) > tolerance: + drawn = [member for member in cluster.members if member.kind is LevelKind.MANUAL] + # A drawn line bypasses the score threshold entirely. Weights run + # from 1 (5m) to 4 (1h) against a threshold of 28, so gating on + # score would mean a line you deliberately drew could never alert. + if not drawn and cluster.score < self.min_score: + continue + if abs(cluster.center - current_price) > tolerance: continue if any( entry.side == cluster.side.value @@ -72,10 +79,17 @@ class AlertEngine: self._fired.append(_Fired(cluster.side.value, cluster.center, now)) direction = "BEARISH" if cluster.side.value == "resistance" else "BULLISH" timeframes = ", ".join(dict.fromkeys(member.tf.value for member in cluster.members)) - message = ( - f"{direction} ZONE {symbol} {current_price:.2f}\n" + # Naming the line matters: "your line" is actionable in a way that + # "confluence 4" is not, and it says which drawing to go look at. + headline = "LINE" if drawn and len(cluster.members) == len(drawn) else "ZONE" + detail = ( f"{cluster.side.value.title()} confluence {cluster.score:g} " - f"@ {cluster.low:.2f}-{cluster.high:.2f}\n{timeframes}" + f"@ {cluster.low:.2f}-{cluster.high:.2f}" + ) + if drawn: + detail += "\n" + ", ".join(member.label for member in drawn) + message = ( + f"{direction} {headline} {symbol} {current_price:.2f}\n{detail}\n{timeframes}" ) alerts.append(Alert(cluster, message)) return alerts diff --git a/app/analysis/confluence.py b/app/analysis/confluence.py index b7a31f6..b4c7f3b 100644 --- a/app/analysis/confluence.py +++ b/app/analysis/confluence.py @@ -2,7 +2,7 @@ from dataclasses import dataclass from hashlib import sha1 from typing import Any -from app.analysis.levels import Level, Side +from app.analysis.levels import Level, LevelKind, Side @dataclass(slots=True) @@ -64,7 +64,11 @@ def cluster_levels( clusters: list[Cluster] = [] for group in groups: score = sum(level.weight for _, level in group) - if len(group) < 2 and score < 8: + # A hand-drawn line survives on its own however little it weighs: it is + # an explicit statement that this price matters. Everything else has to + # earn its place by clustering or by being a heavyweight daily level. + drawn = any(level.kind is LevelKind.MANUAL for _, level in group) + if not drawn and len(group) < 2 and score < 8: continue low, high = group[0][0], group[-1][0] center = (low + high) / 2 diff --git a/tests/test_alerts.py b/tests/test_alerts.py index b6f180b..8c74640 100644 --- a/tests/test_alerts.py +++ b/tests/test_alerts.py @@ -8,6 +8,10 @@ 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)] @@ -41,6 +45,36 @@ def test_a_third_level_joining_the_zone_does_not_re_alert(): 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)