42 lines
1.4 KiB
Python
42 lines
1.4 KiB
Python
from app.analysis.levels import Side
|
|
from app.analysis.manual_lines import ManualLine, ManualLineStore
|
|
from app.analysis.confluence import cluster_levels
|
|
from app.analysis.levels import Level, LevelKind
|
|
from app.bars.models import Timeframe
|
|
|
|
|
|
def sample_line():
|
|
return ManualLine("ml_test", Timeframe.H4, Side.RESISTANCE, 100, 5000, -0.01, 200, 300, number=1)
|
|
|
|
|
|
def test_json_persistence_round_trip(tmp_path):
|
|
path = tmp_path / "manual_lines.json"
|
|
store = ManualLineStore(path)
|
|
store.add(sample_line())
|
|
|
|
loaded = ManualLineStore(path)
|
|
assert list(loaded.lines.values()) == [sample_line()]
|
|
|
|
loaded.update("ml_test", {"note": "major swing"})
|
|
assert ManualLineStore(path).lines["ml_test"].note == "major swing"
|
|
loaded.delete("ml_test")
|
|
assert ManualLineStore(path).lines == {}
|
|
|
|
|
|
def test_four_hour_line_uses_absolute_time_on_one_minute_chart():
|
|
level = sample_line().to_level()
|
|
instant = 160
|
|
assert level.tf is Timeframe.H4
|
|
assert level.price_at(instant) == 4999.4
|
|
assert level.weight == 8
|
|
|
|
|
|
def test_manual_line_raises_existing_ma_cluster_score():
|
|
ma = Level(
|
|
"ma", LevelKind.MA, Timeframe.D1, Side.RESISTANCE, 12, 1, "1d SMA20",
|
|
100, 5000, 0, None, 0, 100, 100, False, False, 20,
|
|
)
|
|
before = cluster_levels([ma], 160, 4998, 2)[0]
|
|
after = cluster_levels([ma, sample_line().to_level()], 160, 4998, 2)[0]
|
|
assert before.score == 12
|
|
assert after.score == 20
|