39 lines
1.3 KiB
Python
39 lines
1.3 KiB
Python
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, tf=Timeframe.H1):
|
|
return Level(id_, LevelKind.MA, tf, Side.RESISTANCE, weight, 1, id_, 100, price, 0, None, 0, 100, 100, False, False)
|
|
|
|
|
|
def test_single_linkage_cluster_has_known_score_and_effective_side():
|
|
clusters = cluster_levels(
|
|
[level("a", 99.8, 2), level("b", 100.1, 4), level("c", 105, 1)],
|
|
current_t=200,
|
|
current_price=99,
|
|
atr15=1,
|
|
)
|
|
|
|
assert len(clusters) == 1
|
|
assert clusters[0].low == 99.8
|
|
assert clusters[0].high == 100.1
|
|
assert clusters[0].score == 6
|
|
assert clusters[0].side is Side.RESISTANCE
|
|
|
|
|
|
def test_lone_daily_level_is_emitted():
|
|
clusters = cluster_levels([level("daily", 98, 12, Timeframe.D1)], 200, 100, 1)
|
|
assert len(clusters) == 1
|
|
assert clusters[0].side is Side.SUPPORT
|
|
|
|
|
|
def test_levels_on_opposite_sides_of_price_do_not_cluster():
|
|
clusters = cluster_levels([level("below", 99.9, 2), level("above", 100.1, 2)], 200, 100, 1)
|
|
assert clusters == []
|
|
|
|
|
|
def test_level_ended_before_current_time_is_excluded():
|
|
ended = level("ended", 98, 12, Timeframe.D1)
|
|
ended.cutoff_t = 150
|
|
assert cluster_levels([ended], 200, 100, 1) == []
|