68 lines
2.5 KiB
Python
68 lines
2.5 KiB
Python
from app.analysis.alerts import AlertEngine
|
|
from app.analysis.confluence import cluster_levels
|
|
from app.analysis.levels import Side
|
|
from app.analysis.manual_lines import ManualLine
|
|
from app.bars.models import Timeframe
|
|
|
|
|
|
def price_alert(price: float, note: str = "", **changes) -> ManualLine:
|
|
return ManualLine(
|
|
"ml_price", Timeframe.D1, Side.RESISTANCE, 1000, price, 0.0, 4600, 1000,
|
|
note=note, **changes,
|
|
)
|
|
|
|
|
|
def test_a_typed_level_holds_its_price_at_any_time():
|
|
level = price_alert(7800).to_level()
|
|
assert level.price_at(0) == 7800
|
|
assert level.price_at(10**9) == 7800
|
|
|
|
|
|
def test_unlabelled_alert_is_named_by_its_price():
|
|
# "1d resistance" tells you nothing about which alert fired.
|
|
assert price_alert(7800).to_level().label == "@ 7800.00"
|
|
assert price_alert(7800, note="gap fill").to_level().label == "gap fill"
|
|
|
|
|
|
def test_typed_level_alerts_regardless_of_confluence_score():
|
|
engine = AlertEngine(min_score=28, cooldown_seconds=900)
|
|
clusters = cluster_levels([price_alert(7800, "gap fill").to_level()], 5000, 7800, 1)
|
|
|
|
assert len(clusters) == 1
|
|
alerts = engine.evaluate(clusters, 7800, 1, 5000, "/ES")
|
|
assert len(alerts) == 1
|
|
assert "gap fill" in alerts[0].message
|
|
|
|
|
|
def test_resistance_level_can_alert_a_fixed_distance_early():
|
|
engine = AlertEngine(min_score=28)
|
|
level = price_alert(100, alert_early_points=2).to_level()
|
|
|
|
too_early = cluster_levels([level], 5000, 97.75, 1)
|
|
in_window = cluster_levels([level], 5000, 98, 1)
|
|
assert engine.evaluate(too_early, 97.75, 1, 5000, "/ES") == []
|
|
assert len(engine.evaluate(in_window, 98, 1, 5001, "/ES")) == 1
|
|
|
|
|
|
def test_support_level_can_alert_a_fixed_distance_early():
|
|
engine = AlertEngine(min_score=28)
|
|
line = price_alert(100, alert_early_points=2)
|
|
line.side = Side.SUPPORT
|
|
level = line.to_level()
|
|
|
|
too_early = cluster_levels([level], 5000, 102.25, 1)
|
|
in_window = cluster_levels([level], 5000, 102, 1)
|
|
assert engine.evaluate(too_early, 102.25, 1, 5000, "/ES") == []
|
|
assert len(engine.evaluate(in_window, 102, 1, 5001, "/ES")) == 1
|
|
|
|
|
|
def test_price_alerts_survive_the_json_round_trip(tmp_path):
|
|
from app.analysis.manual_lines import ManualLineStore
|
|
|
|
store = ManualLineStore(tmp_path / "manual_lines.json")
|
|
store.add(price_alert(7800, "gap fill"))
|
|
|
|
reloaded = ManualLineStore(tmp_path / "manual_lines.json").lines["ml_price"]
|
|
assert reloaded.slope == 0.0
|
|
assert reloaded.horizontal
|
|
assert reloaded.anchor_p == 7800
|