115 lines
4.1 KiB
Python
115 lines
4.1 KiB
Python
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.analysis.confluence import cluster_levels
|
|
from app.analysis.levels import Level, LevelKind, Side
|
|
from app.analysis.manual_lines import ManualLine
|
|
from app.api.ws import router as ws_router, snapshot
|
|
from app.bars.models import Bar, Timeframe
|
|
from app.config import Settings
|
|
from app.runtime import Runtime
|
|
|
|
|
|
def ma(id_: str, period: int, price: float) -> Level:
|
|
return Level(
|
|
id_, LevelKind.MA, Timeframe.M1, Side.RESISTANCE, 1, 1,
|
|
f"SMA{period}", 1000, price, 0, None, 0, 1000, 1000,
|
|
False, False, period,
|
|
)
|
|
|
|
|
|
def test_websocket_layer_preferences_are_isolated_per_connection(tmp_path):
|
|
app = FastAPI()
|
|
app.include_router(ws_router)
|
|
runtime = Runtime(Settings(manual_lines_path=tmp_path / "manual_lines.json"))
|
|
runtime.levels = [ma("ma20", 20, 100.0), ma("ma50", 50, 100.1)]
|
|
runtime.price = 99.9
|
|
runtime.atr15 = 1
|
|
runtime.stream.last_bar_t = 1000
|
|
runtime.clusters = cluster_levels(runtime.levels, 1000, runtime.price, runtime.atr15)
|
|
app.state.runtime = runtime
|
|
client = TestClient(app)
|
|
original_levels = list(runtime.levels)
|
|
original_clusters = list(runtime.clusters)
|
|
|
|
with client.websocket_connect("/ws") as filtered, client.websocket_connect("/ws") as default:
|
|
assert len(filtered.receive_json()["clusters"]) == 1
|
|
assert len(default.receive_json()["clusters"]) == 1
|
|
|
|
filtered.send_json(
|
|
{
|
|
"type": "prefs",
|
|
"hidden_levels_score": False,
|
|
"enabled": {"ma": {"1m": [20]}},
|
|
}
|
|
)
|
|
assert filtered.receive_json()["clusters"] == []
|
|
|
|
default.send_json({"type": "subscribe", "tf": "1m"})
|
|
assert len(default.receive_json()["clusters"]) == 1
|
|
|
|
assert runtime.levels == original_levels
|
|
assert runtime.clusters == original_clusters
|
|
|
|
|
|
def test_quote_change_always_uses_the_daily_session_open(tmp_path):
|
|
runtime = Runtime(Settings(manual_lines_path=tmp_path / "manual_lines.json"))
|
|
runtime.store.put(Bar(Timeframe.D1, 1000, 6123.25, 6140, 6100, 6130, 1, False, "/ES", "test"))
|
|
runtime.store.put(Bar(Timeframe.H1, 2000, 6130, 6150, 6120, 6145, 1, False, "/ES", "test"))
|
|
|
|
message = snapshot(runtime, Timeframe.H1)
|
|
|
|
assert message["session_open"] == 6123.25
|
|
|
|
|
|
def test_snapshot_carries_source_times_for_manual_trendlines(tmp_path):
|
|
runtime = Runtime(Settings(
|
|
manual_lines_path=tmp_path / "manual_lines.json",
|
|
trendline_source_geometry=True,
|
|
))
|
|
runtime.manual_lines.add(ManualLine(
|
|
"ml_30m", Timeframe.M30, Side.SUPPORT,
|
|
1000, 100, 1 / 1800, 2800, 2800,
|
|
))
|
|
runtime.store.put(Bar(Timeframe.M30, 1000, 100, 101, 99, 100, 1, True, "/ES", "test"))
|
|
runtime.store.put(Bar(Timeframe.M30, 2800, 100, 101, 99, 100, 1, True, "/ES", "test"))
|
|
runtime.rebuild_levels()
|
|
|
|
geometry = snapshot(runtime, Timeframe.M1)["trendline_geometry"]
|
|
|
|
assert geometry["mode"] == "source_tf"
|
|
source = geometry["series"]["30m"]
|
|
assert source["times"] == [1000, 2800]
|
|
assert source["duration"] == 1800
|
|
assert source["future_times"][:2] == [4600, 6400]
|
|
assert source["future_durations"][:2] == [1800, 1800]
|
|
|
|
|
|
def test_snapshot_rollback_gate_omits_source_geometry(tmp_path):
|
|
runtime = Runtime(Settings(
|
|
manual_lines_path=tmp_path / "manual_lines.json",
|
|
trendline_source_geometry=False,
|
|
))
|
|
|
|
assert snapshot(runtime, Timeframe.M1)["trendline_geometry"] == {
|
|
"mode": "legacy", "series": {},
|
|
}
|
|
|
|
|
|
def test_browser_can_request_full_geometry_after_a_source_delta_gap(tmp_path):
|
|
app = FastAPI()
|
|
app.include_router(ws_router)
|
|
runtime = Runtime(Settings(
|
|
manual_lines_path=tmp_path / "manual_lines.json",
|
|
trendline_source_geometry=True,
|
|
))
|
|
app.state.runtime = runtime
|
|
|
|
with TestClient(app).websocket_connect("/ws") as websocket:
|
|
websocket.receive_json()
|
|
websocket.send_json({"type": "trendline_geometry"})
|
|
|
|
assert websocket.receive_json() == {
|
|
"type": "trendline_geometry",
|
|
"geometry": {"mode": "source_tf", "series": {}},
|
|
}
|