164 lines
6.2 KiB
Python
164 lines
6.2 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.H1, 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_price_level_alert_offset_survives_json_round_trip(tmp_path):
|
|
path = tmp_path / "manual_lines.json"
|
|
line = sample_line()
|
|
line.slope = 0.0
|
|
line.alert_early_points = 1.25
|
|
ManualLineStore(path).add(line)
|
|
|
|
loaded = ManualLineStore(path).lines["ml_test"]
|
|
assert loaded.alert_early_points == 1.25
|
|
assert loaded.to_level().alert_early_points == 1.25
|
|
|
|
|
|
def test_hourly_line_uses_absolute_time_on_one_minute_chart():
|
|
level = sample_line().to_level()
|
|
instant = 160
|
|
assert level.tf is Timeframe.H1
|
|
assert level.price_at(instant) == 4999.4
|
|
assert level.weight == 4
|
|
|
|
|
|
def test_unnamed_trendlines_are_named_for_their_direction():
|
|
resistance = sample_line()
|
|
support = ManualLine(
|
|
"ml_support", Timeframe.M1, Side.SUPPORT, 100, 5000, 0.01, 200, 300
|
|
)
|
|
|
|
assert resistance.default_label() == "down1h"
|
|
assert support.default_label() == "up1m"
|
|
|
|
|
|
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 == 16
|
|
|
|
|
|
def test_an_annotation_is_never_a_level(tmp_path):
|
|
# Comments live with the lines so they share numbering, filtering and
|
|
# deletion — but a comment reaching levels() would join a confluence
|
|
# cluster and fire a push notification about a piece of text.
|
|
from app.analysis.manual_lines import ManualLine, ManualLineStore
|
|
from app.analysis.levels import Side
|
|
from app.bars.models import Timeframe
|
|
|
|
store = ManualLineStore(tmp_path / "lines.json")
|
|
common = dict(tf=Timeframe.M1, side=Side.SUPPORT, anchor_p=100.0,
|
|
anchor_t=1000, last_t=2000, created_at=1000)
|
|
store.add(ManualLine(id="ml_level", slope=0.0, **common))
|
|
store.add(ManualLine(id="ml_note", slope=0.0, kind="comment",
|
|
note="watch this", **common))
|
|
store.add(ManualLine(id="ml_symbol", slope=0.0, kind="symbol",
|
|
icon="skull", note="Skull", **common))
|
|
store.add(ManualLine(id="ml_fib", slope=0.01, kind="fibonacci",
|
|
note="fib", **common))
|
|
|
|
assert [level.id for level in store.levels()] == ["ml_level"]
|
|
assert [line.id for line in store.drawings()] == ["ml_level", "ml_note", "ml_symbol", "ml_fib"]
|
|
|
|
|
|
def test_drawing_kind_is_derived_for_lines_saved_before_comments(tmp_path):
|
|
from app.analysis.manual_lines import ManualLine, ManualLineStore
|
|
from app.analysis.levels import Side
|
|
from app.bars.models import Timeframe
|
|
|
|
store = ManualLineStore(tmp_path / "lines.json")
|
|
common = dict(tf=Timeframe.M1, side=Side.SUPPORT, anchor_p=100.0,
|
|
anchor_t=1000, last_t=2000, created_at=1000)
|
|
flat = store.add(ManualLine(id="ml_flat", slope=0.0, **common))
|
|
sloped = store.add(ManualLine(id="ml_sloped", slope=0.5, **common))
|
|
|
|
assert flat.drawing_kind == "level"
|
|
assert sloped.drawing_kind == "trendline"
|
|
|
|
|
|
def test_a_mark_scale_survives_json_and_does_not_move_its_anchor(tmp_path):
|
|
path = tmp_path / "lines.json"
|
|
store = ManualLineStore(path)
|
|
common = dict(tf=Timeframe.M1, side=Side.SUPPORT, anchor_p=100.0,
|
|
anchor_t=1000, last_t=2000, created_at=1000)
|
|
store.add(ManualLine(id="ml_mark", slope=0.0, kind="symbol",
|
|
icon="skull", note="Skull", scale=2.0, **common))
|
|
raw = ManualLine(
|
|
id="ml_old", slope=0.0, kind="symbol", icon="play", note="Go", **common,
|
|
).to_dict()
|
|
del raw["scale"]
|
|
store.add(ManualLine.from_dict(raw))
|
|
|
|
loaded = ManualLineStore(path)
|
|
mark = loaded.lines["ml_mark"]
|
|
old = loaded.lines["ml_old"]
|
|
assert mark.scale == 2.0
|
|
assert (mark.anchor_t, mark.anchor_p) == (1000, 100.0)
|
|
assert old.scale == 1.0
|
|
assert (old.anchor_t, old.anchor_p) == (1000, 100.0)
|
|
assert [level.id for level in loaded.levels()] == []
|
|
|
|
|
|
def test_every_drawing_gets_a_number_including_comments(tmp_path):
|
|
from app.analysis.manual_lines import ManualLine, ManualLineStore
|
|
from app.analysis.levels import Side
|
|
from app.bars.models import Timeframe
|
|
|
|
store = ManualLineStore(tmp_path / "lines.json")
|
|
common = dict(tf=Timeframe.M1, side=Side.SUPPORT, anchor_p=100.0,
|
|
anchor_t=1000, last_t=2000, created_at=1000)
|
|
first = store.add(ManualLine(id="ml_a", slope=0.0, **common))
|
|
note = store.add(ManualLine(id="ml_b", slope=0.0, kind="comment", **common))
|
|
third = store.add(ManualLine(id="ml_c", slope=1.0, **common))
|
|
|
|
assert [first.number, note.number, third.number] == [1, 2, 3]
|
|
|
|
|
|
def test_a_null_cutoff_clears_an_ended_line(tmp_path):
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.api.routes import router
|
|
from app.config import Settings
|
|
from app.runtime import Runtime
|
|
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
app.state.runtime = Runtime(Settings(manual_lines_path=tmp_path / "manual_lines.json"))
|
|
client = TestClient(app)
|
|
created = client.post("/api/lines", json={
|
|
"tf": "1m", "side": "support",
|
|
"anchor_t": 100, "anchor_p": 1.0,
|
|
"end_t": 200, "end_p": 2.0,
|
|
"cutoff_t": 150,
|
|
}).json()
|
|
assert created["cutoff_t"] == 150
|
|
|
|
cleared = client.patch(f"/api/lines/{created['id']}", json={"cutoff_t": None}).json()
|
|
assert cleared["cutoff_t"] is None
|