446 lines
12 KiB
Python
446 lines
12 KiB
Python
'''
|
|
Typed chart-local gap-overlay regressions.
|
|
|
|
'''
|
|
from collections.abc import Iterator
|
|
import os
|
|
from types import SimpleNamespace
|
|
|
|
os.environ['QT_QPA_PLATFORM'] = 'offscreen'
|
|
|
|
import msgspec
|
|
import numpy as np
|
|
import pyqtgraph as pg
|
|
import pytest
|
|
|
|
from piker.ui._annotate import GapAnnotations
|
|
from piker.ui._gaps import (
|
|
GapOverlay,
|
|
GapOverlayMngr,
|
|
GapOverlayPayload,
|
|
GapSpec,
|
|
SetGapOverlay,
|
|
gap_specs_from_ohlcv,
|
|
)
|
|
from piker.ui.qt import QApplication
|
|
|
|
|
|
FQME: str = 'gap.test'
|
|
|
|
|
|
class _ChartStub:
|
|
'''
|
|
Real Qt plot with the small chart API required by `gapman`.
|
|
|
|
'''
|
|
def __init__(
|
|
self,
|
|
fqme: str,
|
|
array: np.ndarray,
|
|
|
|
) -> None:
|
|
self.fqme: str = fqme
|
|
self.widget: pg.PlotWidget = pg.PlotWidget()
|
|
self.viz: SimpleNamespace = SimpleNamespace(
|
|
plot=self.widget.plotItem,
|
|
shm=SimpleNamespace(array=array),
|
|
)
|
|
|
|
def get_viz(
|
|
self,
|
|
fqme: str,
|
|
|
|
) -> SimpleNamespace:
|
|
'''
|
|
Return this chart's only viz.
|
|
|
|
'''
|
|
assert fqme == self.fqme
|
|
return self.viz
|
|
|
|
def close(self) -> None:
|
|
'''
|
|
Release this test-owned plot widget.
|
|
|
|
'''
|
|
self.widget.close()
|
|
self.widget.deleteLater()
|
|
|
|
|
|
def _ohlcv_array(
|
|
times: tuple[float, ...],
|
|
|
|
) -> np.ndarray:
|
|
'''
|
|
Build the minimal structured OHLCV used by gap rendering.
|
|
|
|
'''
|
|
dtype: np.dtype = np.dtype([
|
|
('index', 'i8'),
|
|
('time', 'f8'),
|
|
('open', 'f8'),
|
|
('close', 'f8'),
|
|
])
|
|
rows: list[tuple[int, float, float, float]] = []
|
|
i: int
|
|
time_s: float
|
|
for i, time_s in enumerate(times):
|
|
price: float = 100 + i
|
|
rows.append((i, time_s, price, price + 0.5))
|
|
|
|
return np.array(rows, dtype=dtype)
|
|
|
|
|
|
def _display_state(
|
|
fqme: str,
|
|
|
|
) -> tuple[
|
|
SimpleNamespace,
|
|
tuple[_ChartStub, _ChartStub],
|
|
]:
|
|
'''
|
|
Build independent realtime and history chart stubs.
|
|
|
|
'''
|
|
rt_chart: _ChartStub = _ChartStub(
|
|
fqme,
|
|
_ohlcv_array((1, 2, 4, 5)),
|
|
)
|
|
hist_chart: _ChartStub = _ChartStub(
|
|
fqme,
|
|
_ohlcv_array((60, 120, 300, 360)),
|
|
)
|
|
ds: SimpleNamespace = SimpleNamespace(
|
|
fqme=fqme,
|
|
chart=rt_chart,
|
|
viz=rt_chart.viz,
|
|
hist_chart=hist_chart,
|
|
hist_viz=hist_chart.viz,
|
|
)
|
|
return ds, (rt_chart, hist_chart)
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def qapp() -> Iterator[QApplication]:
|
|
'''
|
|
Keep one offscreen Qt application alive for graphics tests.
|
|
|
|
'''
|
|
app: QApplication|None = QApplication.instance()
|
|
if app is None:
|
|
app = QApplication(['piker-gap-tests'])
|
|
app.setQuitOnLastWindowClosed(False)
|
|
|
|
yield app
|
|
|
|
app.processEvents()
|
|
|
|
|
|
def test_gap_specs_and_wire_roundtrip() -> None:
|
|
'''
|
|
Local detection and remote IPC share one typed request model.
|
|
|
|
Build a small sorted OHLCV array with one positive timestamp gap,
|
|
derive its geometry without a per-row dataframe scan, and
|
|
roundtrip the request/reply through msgpack. This proves callers
|
|
invoke one chart renderer with one concrete schema.
|
|
|
|
'''
|
|
dtype: np.dtype = np.dtype([
|
|
('index', 'i8'),
|
|
('time', 'f8'),
|
|
('open', 'f8'),
|
|
('close', 'f8'),
|
|
])
|
|
array: np.ndarray = np.array([
|
|
(10, 60, 100, 101),
|
|
(11, 120, 101, 102),
|
|
(12, 300, 95, 96),
|
|
(13, 360, 96, 97),
|
|
], dtype=dtype)
|
|
|
|
specs: list[GapSpec] = gap_specs_from_ohlcv(
|
|
array,
|
|
period_s=60,
|
|
)
|
|
assert len(specs) == 1
|
|
spec: GapSpec = specs[0]
|
|
assert spec.start_time == 120
|
|
assert spec.end_time == 300
|
|
assert spec.start_pos == (11.9, 102)
|
|
assert spec.end_pos == (12.1, 95)
|
|
assert spec.pointing == 'down'
|
|
|
|
req: SetGapOverlay = SetGapOverlay(
|
|
fqme='mnq.cme.ib',
|
|
timeframe=60,
|
|
specs=specs,
|
|
request_id='req-616',
|
|
)
|
|
req_wire: bytes = msgspec.msgpack.encode(req)
|
|
assert msgspec.msgpack.decode(
|
|
req_wire,
|
|
type=SetGapOverlay,
|
|
) == req
|
|
|
|
resp: GapOverlay = GapOverlay(
|
|
fqme=req.fqme,
|
|
timeframe=req.timeframe,
|
|
visible=True,
|
|
gap_count=1,
|
|
aid=616,
|
|
request_id=req.request_id,
|
|
)
|
|
resp_wire: bytes = msgspec.msgpack.encode(resp)
|
|
assert msgspec.msgpack.decode(
|
|
resp_wire,
|
|
type=GapOverlay,
|
|
) == resp
|
|
|
|
|
|
def test_gap_overlay_payload_rejects_other_msgs() -> None:
|
|
'''
|
|
Reject payloads outside the strict gap-overlay dialog protocol.
|
|
|
|
The sibling Tractor context installs `GapOverlayPayload` so
|
|
legacy annotation dicts cannot enter its stream as partially
|
|
validated commands. Encode one such dict and prove the same
|
|
msgspec decoder rejects it before endpoint dispatch.
|
|
|
|
'''
|
|
invalid_wire: bytes = msgspec.msgpack.encode({
|
|
'cmd': 'SelectRect',
|
|
})
|
|
|
|
with pytest.raises(msgspec.ValidationError):
|
|
msgspec.msgpack.decode(
|
|
invalid_wire,
|
|
type=GapOverlayPayload,
|
|
)
|
|
|
|
req: SetGapOverlay = SetGapOverlay(
|
|
fqme='mnq.cme.ib',
|
|
timeframe=60,
|
|
specs=[],
|
|
)
|
|
resp: GapOverlay = GapOverlay(
|
|
fqme=req.fqme,
|
|
timeframe=req.timeframe,
|
|
visible=False,
|
|
gap_count=0,
|
|
)
|
|
with pytest.raises(msgspec.ValidationError):
|
|
msgspec.msgpack.decode(
|
|
msgspec.msgpack.encode(resp),
|
|
type=SetGapOverlay,
|
|
)
|
|
with pytest.raises(msgspec.ValidationError):
|
|
msgspec.msgpack.decode(
|
|
msgspec.msgpack.encode(req),
|
|
type=GapOverlay,
|
|
)
|
|
|
|
|
|
def test_gap_overlay_unknown_fqme_returns_typed_error() -> None:
|
|
'''
|
|
Keep a stale remote FQME request inside the typed dialog.
|
|
|
|
A chart can unload or replace display state after a client learns
|
|
its FQME set. Submit a valid request to an empty controller and
|
|
prove it returns `GapOverlay.error` instead of leaking a
|
|
`KeyError` which would terminate the whole Tractor context.
|
|
|
|
'''
|
|
gapman: GapOverlayMngr = GapOverlayMngr(
|
|
dss={},
|
|
annots={},
|
|
)
|
|
req: SetGapOverlay = SetGapOverlay(
|
|
fqme='gone.test',
|
|
timeframe=60,
|
|
specs=[],
|
|
request_id='req-stale',
|
|
)
|
|
|
|
resp: GapOverlay = gapman.apply(req)
|
|
|
|
assert resp.request_id == req.request_id
|
|
assert resp.error == 'No display state for fqme=gone.test'
|
|
|
|
|
|
def test_gap_manager_real_qt_lifecycle(
|
|
qapp: QApplication,
|
|
|
|
) -> None:
|
|
'''
|
|
Remove overlays from every PyQtGraph registry.
|
|
|
|
Gap layers are inserted through `PlotItem.addItem()`. Removing
|
|
the item directly from its scene leaves stale `PlotItem.items`
|
|
and `ViewBox.addedItems` refs which can leak memory and affect
|
|
range calculations. This test uses real offscreen Qt plots,
|
|
replaces one remotely owned layer, hides it, and proves each
|
|
registry releases the old graphics object.
|
|
|
|
'''
|
|
ds: SimpleNamespace
|
|
charts: tuple[_ChartStub, _ChartStub]
|
|
ds, charts = _display_state(FQME)
|
|
annots: dict[int, GapAnnotations] = {}
|
|
gapman: GapOverlayMngr = GapOverlayMngr(
|
|
dss={FQME: ds},
|
|
annots=annots,
|
|
)
|
|
specs: list[GapSpec] = gap_specs_from_ohlcv(
|
|
ds.hist_viz.shm.array,
|
|
period_s=60,
|
|
)
|
|
aids: set[int] = set()
|
|
req: SetGapOverlay = SetGapOverlay(
|
|
fqme=FQME,
|
|
timeframe=60,
|
|
specs=specs,
|
|
request_id='first',
|
|
)
|
|
|
|
try:
|
|
first: GapOverlay = gapman.apply(
|
|
req=req,
|
|
owner='remote-test',
|
|
aids=aids,
|
|
)
|
|
assert first.aid is not None
|
|
first_item: GapAnnotations = annots[first.aid]
|
|
assert first_item.scene() is charts[1].widget.scene()
|
|
assert first_item in ds.hist_viz.plot.items
|
|
assert first_item not in ds.hist_viz.plot.vb.addedItems
|
|
assert aids == {first.aid}
|
|
|
|
second: GapOverlay = gapman.apply(
|
|
req=SetGapOverlay(
|
|
fqme=FQME,
|
|
timeframe=60,
|
|
specs=specs,
|
|
request_id='second',
|
|
),
|
|
owner='remote-test',
|
|
aids=aids,
|
|
)
|
|
assert second.aid is not None
|
|
assert second.aid != first.aid
|
|
assert first.aid not in annots
|
|
assert first_item.scene() is None
|
|
assert first_item not in ds.hist_viz.plot.items
|
|
assert first_item not in ds.hist_viz.plot.vb.addedItems
|
|
assert aids == {second.aid}
|
|
|
|
hidden: GapOverlay = gapman.apply(
|
|
req=SetGapOverlay(
|
|
fqme=FQME,
|
|
timeframe=60,
|
|
specs=specs,
|
|
visible=False,
|
|
request_id='hidden',
|
|
),
|
|
owner='remote-test',
|
|
aids=aids,
|
|
)
|
|
assert hidden.visible is False
|
|
assert hidden.aid is None
|
|
assert aids == set()
|
|
assert annots == {}
|
|
assert gapman._layers == {}
|
|
assert gapman._plots == {}
|
|
item: object
|
|
for item in ds.hist_viz.plot.items:
|
|
assert not isinstance(item, GapAnnotations)
|
|
|
|
local: GapOverlay = gapman.apply(
|
|
req=SetGapOverlay(
|
|
fqme=FQME,
|
|
timeframe=60,
|
|
specs=specs,
|
|
request_id='local',
|
|
),
|
|
)
|
|
remote: GapOverlay = gapman.apply(
|
|
req=SetGapOverlay(
|
|
fqme=FQME,
|
|
timeframe=60,
|
|
specs=specs,
|
|
request_id='remote',
|
|
),
|
|
owner='remote-test',
|
|
)
|
|
assert local.aid is not None
|
|
assert remote.aid is not None
|
|
gapman.remove_owner('remote-test')
|
|
assert local.aid in annots
|
|
assert remote.aid not in annots
|
|
assert annots[local.aid].scene() is not None
|
|
gapman.remove_owner('chart-local')
|
|
assert annots == {}
|
|
finally:
|
|
chart: _ChartStub
|
|
for chart in charts:
|
|
chart.close()
|
|
qapp.processEvents()
|
|
|
|
|
|
def test_duplicate_fqme_layers_use_local_chart_identity(
|
|
qapp: QApplication,
|
|
|
|
) -> None:
|
|
'''
|
|
Keep cached displays for one FQME in separate local layers.
|
|
|
|
The actor-global display lookup is keyed by FQME and can
|
|
represent only its currently registered display. Local cached
|
|
charts still pass their exact `DisplayState` into
|
|
`GapOverlayMngr.refresh()`. Build two independent chart pairs
|
|
with the same FQME, refresh both, and prove their layers survive
|
|
independently.
|
|
|
|
'''
|
|
first_ds: SimpleNamespace
|
|
first_charts: tuple[_ChartStub, _ChartStub]
|
|
first_ds, first_charts = _display_state(FQME)
|
|
second_ds: SimpleNamespace
|
|
second_charts: tuple[_ChartStub, _ChartStub]
|
|
second_ds, second_charts = _display_state(FQME)
|
|
annots: dict[int, GapAnnotations] = {}
|
|
gapman: GapOverlayMngr = GapOverlayMngr(
|
|
dss={FQME: second_ds},
|
|
annots=annots,
|
|
)
|
|
|
|
try:
|
|
first: GapOverlay = gapman.refresh(
|
|
ds=first_ds,
|
|
timeframe=60,
|
|
)
|
|
second: GapOverlay = gapman.refresh(
|
|
ds=second_ds,
|
|
timeframe=60,
|
|
)
|
|
assert first.aid is not None
|
|
assert second.aid is not None
|
|
assert first.aid != second.aid
|
|
assert len(gapman._layers) == 2
|
|
chart_ids: set[int] = set()
|
|
key: tuple[str, int, str, float]
|
|
for key in gapman._layers:
|
|
chart_ids.add(key[1])
|
|
assert chart_ids == {
|
|
id(first_ds.hist_chart),
|
|
id(second_ds.hist_chart),
|
|
}
|
|
assert annots[first.aid].scene() is not None
|
|
assert annots[second.aid].scene() is not None
|
|
finally:
|
|
gapman.remove_owner('chart-local')
|
|
chart: _ChartStub
|
|
for chart in (*first_charts, *second_charts):
|
|
chart.close()
|
|
qapp.processEvents()
|