202 lines
7.6 KiB
Python
202 lines
7.6 KiB
Python
import asyncio
|
|
|
|
import pytest
|
|
|
|
from app.analysis.alerts import Alert
|
|
from app.analysis.confluence import Cluster, cluster_levels
|
|
from app.analysis.levels import Level, LevelKind, Side
|
|
from app.analysis.manual_lines import ManualLine
|
|
from app.bars.models import Timeframe
|
|
from app.config import Settings
|
|
from app.runtime import Runtime
|
|
|
|
|
|
def runtime(tmp_path, **overrides) -> Runtime:
|
|
settings = Settings(
|
|
manual_lines_path=tmp_path / "manual_lines.json",
|
|
# Isolated per test: the default is relative to the working directory,
|
|
# so without this every test shares one alert-suppression file and they
|
|
# silence each other.
|
|
alert_state_path=tmp_path / "alert_state.json",
|
|
user_prefs_path=tmp_path / "user_prefs.json",
|
|
ntfy_topic=overrides.pop("ntfy_topic", ""),
|
|
**overrides,
|
|
)
|
|
return Runtime(settings)
|
|
|
|
|
|
def alert() -> Alert:
|
|
level = Level(
|
|
"pd:high", LevelKind.HORIZONTAL, Timeframe.D1, Side.RESISTANCE, 16, 1, "PDH",
|
|
100, 5000, 0, None, 0, 100, 100, False, False,
|
|
)
|
|
cluster = Cluster("cl_x", Side.RESISTANCE, 5000, 5000, 5000, 16, [level], 1.0)
|
|
return Alert(cluster, "BEARISH ZONE /ES")
|
|
|
|
|
|
def test_one_engine_serves_every_connection(tmp_path):
|
|
# Previously each WebSocket built its own engine, so reloading the page
|
|
# cleared the cooldown and the same zone alerted again immediately.
|
|
instance = runtime(tmp_path)
|
|
assert instance.alert_engine is not None
|
|
assert instance.alert_engine.min_score == instance.settings.confluence_min_score
|
|
|
|
|
|
def test_alerts_reach_subscribers(tmp_path):
|
|
instance = runtime(tmp_path)
|
|
queue: asyncio.Queue = asyncio.Queue(maxsize=10)
|
|
instance.subscribers.add(queue)
|
|
|
|
async def scenario():
|
|
instance.dispatch_alerts([alert()])
|
|
return queue.get_nowait()
|
|
|
|
event = asyncio.run(scenario())
|
|
assert event["type"] == "alert"
|
|
assert event["message"] == "BEARISH ZONE /ES"
|
|
|
|
|
|
def test_ntfy_failure_does_not_propagate(tmp_path, monkeypatch):
|
|
# send_ntfy used to be awaited inside the WebSocket loop, whose except
|
|
# clause only caught disconnects — so a push outage killed the connection.
|
|
instance = runtime(tmp_path, ntfy_topic="chart-test")
|
|
|
|
async def explode(*args, **kwargs):
|
|
raise RuntimeError("ntfy is down")
|
|
|
|
monkeypatch.setattr("app.runtime.send_ntfy", explode)
|
|
asyncio.run(instance.notify("anything"))
|
|
|
|
|
|
def test_blank_topic_sends_nothing(tmp_path, monkeypatch):
|
|
instance = runtime(tmp_path)
|
|
calls = []
|
|
|
|
async def record(server, topic, message):
|
|
calls.append(topic)
|
|
|
|
monkeypatch.setattr("app.runtime.send_ntfy", record)
|
|
asyncio.run(instance.notify("anything"))
|
|
assert calls == [""] # send_ntfy itself is the one that short-circuits
|
|
|
|
|
|
def test_tick_bars_update_higher_timeframes_without_doubling_volume(tmp_path):
|
|
# The aggregator accumulates with `current.v += incoming.v`, so a forming
|
|
# minute re-sent on every tick would add its volume to each higher
|
|
# timeframe again and again. The provisional path must combine, not
|
|
# accumulate: five ticks in one minute leave the hour's volume equal to the
|
|
# closed minutes plus the live one, exactly once.
|
|
from app.bars.models import Bar
|
|
|
|
instance = runtime(tmp_path)
|
|
|
|
def minute(t, close, volume, closed=True):
|
|
return Bar(tf=Timeframe.M1, t=t, o=close, h=close, l=close, c=close,
|
|
v=volume, closed=closed, symbol="/ES", source="test")
|
|
|
|
base = 1786356000 # top of an hour
|
|
asyncio.run(instance.on_bar(minute(base, 100.0, 10)))
|
|
asyncio.run(instance.on_bar(minute(base + 60, 101.0, 20)))
|
|
settled = [b for b in instance.store.get(Timeframe.H1) if b.t == base][-1].v
|
|
assert settled == 30
|
|
|
|
for _ in range(5):
|
|
asyncio.run(instance.on_bar(minute(base + 120, 102.0, 7, closed=False)))
|
|
|
|
hour = [b for b in instance.store.get(Timeframe.H1) if b.t == base][-1]
|
|
assert hour.v == 37, "the live minute's volume must be added once, not per tick"
|
|
assert hour.c == 102.0
|
|
assert hour.closed is False
|
|
|
|
|
|
def test_a_tripped_manual_alert_stays_disarmed_after_rebuild_and_restart(tmp_path):
|
|
instance = runtime(tmp_path)
|
|
instance.manual_lines.add(
|
|
ManualLine(
|
|
"ml_once", Timeframe.D1, Side.RESISTANCE, 1000, 5000, 0.0,
|
|
1000, 1000, note="one shot",
|
|
)
|
|
)
|
|
instance.rebuild_levels()
|
|
clusters = cluster_levels(instance.levels, 1000, 5000, 1)
|
|
alerts = instance.alert_engine.evaluate(clusters, 5000, 1, 1000, "/ES")
|
|
|
|
async def dispatch():
|
|
instance.dispatch_alerts(alerts)
|
|
await asyncio.gather(*instance._notify_tasks)
|
|
|
|
asyncio.run(dispatch())
|
|
|
|
assert len(alerts) == 1
|
|
assert alerts[0].tripped == ("ml_once",)
|
|
assert instance.manual_lines.lines["ml_once"].armed is False
|
|
assert next(level for level in instance.levels if level.id == "ml_once").armed is False
|
|
|
|
restarted = runtime(tmp_path)
|
|
assert restarted.manual_lines.lines["ml_once"].armed is False
|
|
assert next(level for level in restarted.levels if level.id == "ml_once").armed is False
|
|
|
|
|
|
def test_a_broadcast_from_a_worker_thread_reaches_subscribers(tmp_path):
|
|
# The mutating routes are sync `def`, so FastAPI runs them in a threadpool,
|
|
# and they reach broadcast through rebuild_levels. asyncio.Queue is not
|
|
# thread-safe — it wakes a consumer by resolving a Future, which only the
|
|
# loop thread may do — so writing it from there can drop the wakeup and
|
|
# leave one browser's drawing invisible to another until the next tick.
|
|
instance = runtime(tmp_path)
|
|
|
|
async def exercise():
|
|
instance._loop = asyncio.get_running_loop()
|
|
queue: asyncio.Queue = asyncio.Queue(maxsize=10)
|
|
instance.subscribers.add(queue)
|
|
waiting = asyncio.ensure_future(queue.get())
|
|
await asyncio.sleep(0) # park the consumer on the Future
|
|
await asyncio.to_thread(instance.broadcast, {"type": "bar", "bar": "sentinel"})
|
|
return await asyncio.wait_for(waiting, timeout=2)
|
|
|
|
assert asyncio.run(exercise())["bar"] == "sentinel"
|
|
|
|
|
|
def test_a_cross_thread_broadcast_is_posted_through_the_loop(tmp_path):
|
|
"""The contract, asserted directly.
|
|
|
|
The race itself is timing-dependent and usually masked — a foreign-thread
|
|
put_nowait often lands in the loop's ready queue before it sleeps, and the
|
|
stream ticking once a second papers over the times it does not. So this
|
|
asserts the rule rather than trying to provoke the failure: a broadcast from
|
|
off the loop must go through call_soon_threadsafe, never touch the queue.
|
|
"""
|
|
instance = runtime(tmp_path)
|
|
|
|
async def exercise():
|
|
loop = asyncio.get_running_loop()
|
|
instance._loop = loop
|
|
posted = []
|
|
original = loop.call_soon_threadsafe
|
|
|
|
def spy(callback, *args):
|
|
posted.append(callback)
|
|
return original(callback, *args)
|
|
|
|
loop.call_soon_threadsafe = spy
|
|
try:
|
|
await asyncio.to_thread(instance.broadcast, {"type": "bar", "bar": "x"})
|
|
finally:
|
|
loop.call_soon_threadsafe = original
|
|
return posted
|
|
|
|
assert asyncio.run(exercise()), "a worker-thread broadcast bypassed the loop"
|
|
|
|
|
|
def test_broadcasting_on_the_loop_still_delivers_synchronously(tmp_path):
|
|
# The stream's own path must not pay for a hop it does not need.
|
|
instance = runtime(tmp_path)
|
|
|
|
async def exercise():
|
|
instance._loop = asyncio.get_running_loop()
|
|
queue: asyncio.Queue = asyncio.Queue(maxsize=10)
|
|
instance.subscribers.add(queue)
|
|
instance.broadcast({"type": "bar", "bar": "direct"})
|
|
return queue.get_nowait() # already there, no await needed
|
|
|
|
assert asyncio.run(exercise())["bar"] == "direct"
|