Chris Amow
bc9fed764a
Forming daily H/L, not a full-width line. Config default on. A forming tick only moves the marks if the range actually expanded.
199 lines
7.6 KiB
Python
199 lines
7.6 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_drawings_off_hides_manual_lines_from_confluence():
|
|
from app.api.ws import level_enabled
|
|
|
|
line = Level(
|
|
"ml1", LevelKind.MANUAL, Timeframe.D1, Side.RESISTANCE, 12, 1,
|
|
"line", 1000, 100.0, 0, None, 0, 1000, 1000, False, False,
|
|
)
|
|
assert level_enabled(line, {"manual": True, "drawings": True})
|
|
assert not level_enabled(line, {"manual": True, "drawings": False})
|
|
assert not level_enabled(line, {"manual": False, "drawings": True})
|
|
# Prefs written before the drawings key still show manuals.
|
|
assert level_enabled(line, {"manual": True})
|
|
|
|
|
|
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_a_1m_snapshot_is_not_capped_shorter_than_the_store(tmp_path):
|
|
runtime = Runtime(Settings(manual_lines_path=tmp_path / "manual_lines.json"))
|
|
for minute in range(1001):
|
|
runtime.store.put(Bar(Timeframe.M1, 60 * minute, 1, 1, 1, 1, 1, True, "/ES", "test"))
|
|
|
|
assert len(snapshot(runtime, Timeframe.M1)["bars"]) == 1001
|
|
|
|
|
|
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_session_range_uses_the_forming_daily_bar(tmp_path):
|
|
runtime = Runtime(Settings(manual_lines_path=tmp_path / "manual_lines.json"))
|
|
runtime.store.put(Bar(Timeframe.D1, 1000, 100, 110, 90, 105, 1, True, "/ES", "test"))
|
|
runtime.store.put(Bar(Timeframe.D1, 2000, 105, 125, 102, 120, 1, False, "/ES", "test"))
|
|
|
|
message = snapshot(runtime, Timeframe.M1)
|
|
|
|
assert message["session_t"] == 2000
|
|
assert message["session_high"] == 125
|
|
assert message["session_low"] == 102
|
|
|
|
|
|
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]
|
|
assert len(source["future_times"]) == 180
|
|
|
|
|
|
def test_daily_snapshot_carries_session_aware_display_future_times(tmp_path):
|
|
runtime = Runtime(Settings(manual_lines_path=tmp_path / "manual_lines.json"))
|
|
thursday = 1786658400 # 2026-08-13 18:00 America/New_York
|
|
runtime.store.put(Bar(Timeframe.D1, thursday, 1, 1, 1, 1, 1, True, "/ES", "test"))
|
|
|
|
message = snapshot(runtime, Timeframe.D1)
|
|
|
|
assert len(message["future_times"]) == 180
|
|
assert message["future_times"][0] == 1786917600 # Sunday 18:00 Eastern
|
|
|
|
|
|
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_a_forming_tick_does_not_resend_future_times(tmp_path):
|
|
app = FastAPI()
|
|
app.include_router(ws_router)
|
|
runtime = Runtime(Settings(
|
|
manual_lines_path=tmp_path / "manual_lines.json",
|
|
trendline_source_geometry=True,
|
|
))
|
|
runtime.store.put(Bar(Timeframe.M1, 1000, 1, 1, 1, 1, 1, True, "/ES", "test"))
|
|
runtime.store.put(Bar(Timeframe.M30, 0, 1, 1, 1, 1, 1, True, "/ES", "test"))
|
|
runtime.manual_lines.add(ManualLine(
|
|
"ml_30m", Timeframe.M30, Side.SUPPORT, 0, 100, 1 / 1800, 1800, 1800,
|
|
))
|
|
runtime.rebuild_levels()
|
|
app.state.runtime = runtime
|
|
|
|
with TestClient(app).websocket_connect("/ws") as websocket:
|
|
websocket.receive_json()
|
|
first = Bar(Timeframe.M1, 1060, 1, 2, 1, 1.5, 1, False, "/ES", "test")
|
|
runtime.broadcast({"type": "bar", "bar": first})
|
|
opened = websocket.receive_json()
|
|
assert opened["type"] == "bar"
|
|
assert opened["future_times"]
|
|
|
|
runtime.broadcast({"type": "bar", "bar": first})
|
|
tick = websocket.receive_json()
|
|
assert tick["type"] == "bar"
|
|
assert "future_times" not in tick
|
|
assert "trendline_future_times" not in tick
|
|
assert "session_high" not in tick
|
|
|
|
higher = Bar(Timeframe.M30, 1800, 1, 2, 1, 1.5, 1, False, "/ES", "test")
|
|
runtime.broadcast({"type": "bar", "bar": higher})
|
|
assert websocket.receive_json()["type"] == "trendline_bar"
|
|
runtime.broadcast({"type": "bar", "bar": higher})
|
|
runtime.broadcast({"type": "bar", "bar": first})
|
|
assert websocket.receive_json()["type"] == "bar"
|
|
|
|
|
|
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": {}},
|
|
}
|