62 lines
2.4 KiB
Python
62 lines
2.4 KiB
Python
import asyncio
|
|
import time
|
|
|
|
from app.analysis.event_log import EventLog
|
|
from app.analysis.confluence import cluster_levels
|
|
from app.analysis.levels import Level, LevelKind, Side
|
|
from app.api.ws import snapshot
|
|
from app.bars.models import Timeframe
|
|
from app.config import Settings
|
|
from app.runtime import Runtime
|
|
|
|
|
|
def test_log_keeps_old_entries_and_recent_is_the_last_day(tmp_path):
|
|
log = EventLog(tmp_path / "events.json")
|
|
now = int(time.time())
|
|
log.add("alert", "old", number=1, at=now - 3 * 86400)
|
|
log.add("alert", "yesterday", number=2, at=now - 12 * 3600)
|
|
log.add("stream", "dropped", at=now)
|
|
|
|
day, more = log.recent()
|
|
assert [entry["message"] for entry in day] == ["dropped", "yesterday"]
|
|
assert more is True
|
|
assert EventLog(tmp_path / "events.json")._entries[0]["message"] == "old"
|
|
assert EventLog(tmp_path / "events.json")._entries[0]["symbol"] == "/ES"
|
|
|
|
|
|
def test_more_pages_older_than_the_cutoff(tmp_path):
|
|
log = EventLog(tmp_path / "events.json")
|
|
now = int(time.time())
|
|
log.add("alert", "old", number=1, at=now - 3 * 86400)
|
|
log.add("alert", "new", number=2, at=now)
|
|
page, more = log.page(before=now - 86400, limit=50)
|
|
assert [entry["message"] for entry in page] == ["old"]
|
|
assert more is False
|
|
|
|
|
|
def test_dispatched_alerts_land_in_the_event_log(tmp_path):
|
|
runtime = Runtime(Settings(
|
|
manual_lines_path=tmp_path / "lines.json",
|
|
events_path=tmp_path / "events.json",
|
|
alert_state_path=tmp_path / "alerts.json",
|
|
))
|
|
line = Level(
|
|
"ml_1", LevelKind.MANUAL, Timeframe.M5, Side.RESISTANCE, 1, 1,
|
|
"up1h", 100, 100, 0, None, 0, 100, 100, False, False, number=27,
|
|
)
|
|
now = int(time.time())
|
|
alerts = runtime.alert_engine.evaluate(cluster_levels([line], 100, 100, 1), 100, 1, now, "/ES")
|
|
|
|
async def send():
|
|
runtime.dispatch_alerts(alerts)
|
|
await asyncio.sleep(0)
|
|
|
|
asyncio.run(send())
|
|
events, _ = runtime.events.recent(since=0)
|
|
assert events[0]["kind"] == "alert"
|
|
assert "#27" in events[0]["message"]
|
|
assert "confluence" not in events[0]["message"]
|
|
assert snapshot(runtime, Timeframe.M1)["events"][0]["number"] == alerts[0].number
|
|
assert snapshot(runtime, Timeframe.M1)["events"][0]["symbol"] == "/ES"
|
|
assert snapshot(runtime, Timeframe.M1)["instrument"]["id"] == "es"
|
|
assert snapshot(runtime, Timeframe.M1)["instrument"]["tick"] == 0.25
|