65 lines
2.4 KiB
Python
65 lines
2.4 KiB
Python
"""Suppression has to survive a restart, or every deploy re-alerts."""
|
|
import json
|
|
|
|
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 zone(price: float = 100.0):
|
|
return cluster_levels([level("a", price, 3), level("b", price + 0.1, 4)], 100, price, 1)
|
|
|
|
|
|
def engine(tmp_path, cooldown=14400):
|
|
return AlertEngine(6, cooldown, tmp_path / "alert_state.json")
|
|
|
|
|
|
def test_fires_once_then_suppresses_within_the_process(tmp_path):
|
|
one = engine(tmp_path)
|
|
assert len(one.evaluate(zone(), 100, 1, 0, "/ES")) == 1
|
|
assert one.evaluate(zone(), 100, 1, 60, "/ES") == []
|
|
|
|
|
|
def test_suppression_survives_a_restart(tmp_path):
|
|
one = engine(tmp_path)
|
|
assert len(one.evaluate(zone(), 100, 1, 0, "/ES")) == 1
|
|
|
|
# A second engine over the same state file stands in for a redeploy.
|
|
two = engine(tmp_path)
|
|
assert two.evaluate(zone(), 100, 1, 60, "/ES") == []
|
|
|
|
|
|
def test_without_a_state_path_a_restart_still_refires(tmp_path):
|
|
"""Unchanged behaviour for local runs, which should not write files."""
|
|
assert len(AlertEngine(6, 14400).evaluate(zone(), 100, 1, 0, "/ES")) == 1
|
|
assert len(AlertEngine(6, 14400).evaluate(zone(), 100, 1, 60, "/ES")) == 1
|
|
|
|
|
|
def test_rearms_across_a_restart_after_cooldown_and_separation(tmp_path):
|
|
one = engine(tmp_path, cooldown=900)
|
|
assert len(one.evaluate(zone(), 100, 1, 0, "/ES")) == 1
|
|
|
|
two = engine(tmp_path, cooldown=900)
|
|
# Price genuinely left the zone, and the cooldown has elapsed.
|
|
assert two.evaluate(cluster_levels([level("a", 100, 3)], 100, 103, 1), 103, 1, 902, "/ES") == []
|
|
assert len(two.evaluate(zone(), 100, 1, 903, "/ES")) == 1
|
|
|
|
|
|
def test_corrupt_state_does_not_prevent_alerting(tmp_path):
|
|
(tmp_path / "alert_state.json").write_text("{not json", encoding="utf-8")
|
|
assert len(engine(tmp_path).evaluate(zone(), 100, 1, 0, "/ES")) == 1
|
|
|
|
|
|
def test_state_file_records_centre_and_time(tmp_path):
|
|
engine(tmp_path).evaluate(zone(), 100, 1, 42, "/ES")
|
|
payload = json.loads((tmp_path / "alert_state.json").read_text(encoding="utf-8"))
|
|
assert len(payload) == 1
|
|
assert payload[0]["at"] == 42
|