28 lines
1.2 KiB
Python
28 lines
1.2 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 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") == []
|