Chris Amow
51a9636ea7
Suppression matched on cluster side as well as position, but side is positional: a level sitting at price is resistance when price is a tick below it and support a tick later. Every crossing failed the side match and fired as a brand-new zone — which is exactly when a level is least newsworthy, not most. Found the honest way. A flat line placed at the live price produced four phone pushes in two minutes. Zones are now matched on position alone. Side still determines whether the message reads BULLISH or BEARISH; it just no longer decides whether you are told twice. Over the same six replayed sessions at threshold 28 that is 40 alerts down to 26, and 10 at the configured four-hour cooldown with a worst session of 5. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
100 lines
4.5 KiB
Python
100 lines
4.5 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
|