piker/tests/test_trend_graphics.py

# piker: trading gear for hackers
# Copyright (C) 2018-present Tyler Goodlet (in stewardship of pikers)
# SPDX-License-Identifier: AGPL-3.0-or-later

# This program is free software: you can redistribute it and/or
# modify it under the terms of the GNU Affero General Public License
# as published by the Free Software Foundation, either version 3
# of the License, or (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.

# You should have received a copy of the GNU Affero General Public
# License along with this program. If not, see
# <https://www.gnu.org/licenses/>.

'''
Trend overlay causality, coordinate and bounded Qt lifetime checks.

'''
from collections.abc import AsyncIterator, Iterator
from contextlib import asynccontextmanager
import os
from types import SimpleNamespace
from typing import Literal

os.environ['QT_QPA_PLATFORM'] = 'offscreen'

import numpy as np
import pyqtgraph as pg
import pytest
import trio
from trio.testing import wait_all_tasks_blocked

from piker.ui._trend import (
    RegimeInterval,
    ReversalPoint,
    TrendOverlay,
    classification_interval,
    regime_rectangles,
    reversal_points,
    times_to_x,
    trend_intervals,
)
from piker.ui.qt import QApplication, QtGui, QPointF
from piker.fsp._jev import JevClassification, RegimeAnswer
from piker.fsp import Fsp
from piker.ui._fsp import FspAdmin, update_trend_overlays
from piker.ui import _fsp as fsp_ui


def _chart(
    times: tuple[float, ...],
    first: int = 1000,

) -> np.ndarray:
    '''
    Make chart bars with absolute indices independent of timestamp.

    '''
    rows: np.ndarray = np.zeros(len(times), dtype=[
        ('index', 'i8'),
        ('time', 'f8'),
        ('low', 'f8'),
        ('high', 'f8'),
        ('close', 'f8'),
    ])
    rows['index'] = np.arange(first, first + len(times))
    rows['time'] = times
    rows['low'] = 99
    rows['high'] = 101
    rows['close'] = 100
    return rows


def _source(
    times: tuple[float, ...],
    trends: tuple[float, ...],

) -> np.ndarray:
    '''
    Make sparse FSP fields, including a deliberately mutable tail.

    '''
    fields: list[tuple[str, str]] = [
        ('index', 'i8'),
        ('time', 'f8'),
        ('ready', 'f8'),
    ]
    order: int
    for order in range(1, 5):
        field: str = 'trend' if order == 1 else f'trend_{order}'
        fields.append((field, 'f8'))
        kind: str
        for kind in ('top', 'bottom'):
            fields.extend([
                (f'{kind}_{order}', 'f8'),
                (f'{kind}_time_{order}', 'f8'),
            ])

    rows: np.ndarray = np.empty(len(times), dtype=fields)
    name: str
    for name in rows.dtype.names:
        if name == 'index':
            rows[name] = np.arange(5000, 5000 + len(rows))
        else:
            rows[name] = np.nan
    rows['time'] = times
    rows['ready'] = 4
    rows['trend'] = trends
    return rows


@pytest.fixture(scope='session')
def qapp() -> Iterator[QApplication]:
    '''
    Keep one offscreen application alive without launching a GUI.

    '''
    app: QApplication|None = QApplication.instance()
    if app is None:
        app = QApplication(['piker-trend-tests'])
    app.setQuitOnLastWindowClosed(False)
    yield app
    app.processEvents()


@pytest.mark.parametrize('index_field', ['index', 'time'])
def test_timestamp_mapping_preserves_gaps(
    index_field: Literal['index', 'time'],

) -> None:
    '''
    A timestamp inside a missing bucket must not snap to a future
    bar or confuse slice offsets with absolute SHM coordinates.
    Irregular 10-second bars and off-grid queries exercise both
    chart axes; NaNs prove out-of-range observations stay hidden.

    '''
    rows: np.ndarray = _chart((100, 110, 140))
    queries: np.ndarray = np.array((99, 100, 115, 125, 145, 150))
    actual: np.ndarray = times_to_x(rows, queries, 10, index_field)
    expected: list[float] = (
        [np.nan, 1000, 1001.5, np.nan, 1002.5, np.nan]
        if index_field == 'index'
        else [np.nan, 100, 115, np.nan, 145, np.nan]
    )
    np.testing.assert_allclose(actual, expected, equal_nan=True)


def test_pivots_confirm_at_later_bar_close() -> None:
    '''
    Plotting a provisional reversal makes an intrabar repaint look
    like a confirmed signal. A confirmed top and mutable-tail bottom
    share an earlier pivot time here. Only the top may render, with
    its 30-second source detection time and 10-second chart position
    retained separately; ``source['index']`` is unrelated on purpose.

    '''
    source: np.ndarray = _source((100, 130, 160, 190), (0, 1, -1, 1))
    source['top_1'][2] = 110
    source['top_time_1'][2] = 130
    source['bottom_1'][3] = 90
    source['bottom_time_1'][3] = 160
    chart: np.ndarray = _chart(tuple(range(100, 201, 10)))
    points: list[ReversalPoint] = reversal_points(
        source,
        chart,
        source_period_s=30,
        chart_period_s=10,
    )
    assert len(points) == 1
    assert points[0].x == 1003
    assert points[0].pivot_time == 130
    assert points[0].detected_at == 190
    assert points[0].period_s == 30
    assert points[0].kind == 'top'


def test_reversal_order_is_independent_of_timeframe() -> None:
    '''
    A second reversal order must use its own pivot fields without
    implicitly changing the source sampling period. Distinct order
    one and order two prices at the same source time identify which
    columns were consumed and prevent order/timeframe conflation.

    '''
    source: np.ndarray = _source((10, 20, 30), (1, -1, 1))
    source['top_1'][1] = 105
    source['top_time_1'][1] = 10
    source['bottom_2'][1] = 95
    source['bottom_time_2'][1] = 10
    points: list[ReversalPoint] = reversal_points(
        source,
        _chart((10, 20, 30)),
        source_period_s=10,
        chart_period_s=10,
        order=2,
    )
    assert len(points) == 1
    assert points[0].price == 95
    assert points[0].order == 2
    assert points[0].period_s == 10


def test_regime_runs_start_at_confirmation() -> None:
    '''
    Coloring from a pivot or detection-bar open falsely implies an
    earlier available regime. A warm-up row, direction change, and
    source gap arrange all run boundaries. Every interval starts at
    finalized bar close and the mutable tail's opposite direction
    cannot create a new regime.

    '''
    source: np.ndarray = _source(
        (100, 110, 120, 130, 170, 180),
        (1, 1, 1, -1, -1, 1),
    )
    source['ready'][0] = 0
    spans: list[RegimeInterval] = trend_intervals(source, 10)
    span: RegimeInterval  # noqa: F842
    assert [(span.start_time, span.end_time) for span in spans] == [
        (120, 140),
        (140, 150),
        (180, 190),
    ]
    assert [span.label for span in spans] == [
        'uptrend', 'downtrend', 'downtrend',
    ]


def test_decimal_subsecond_epochs_do_not_invent_gaps() -> None:
    '''
    Float64 Unix timestamps cannot represent every 100 ms boundary
    exactly; successive differences alternate around 0.1 seconds.
    Comparing each difference strictly to the requested period splits
    one regime into spurious holes. Realistic epochs prove rounding
    preserves one run, while deleting one actual bucket still splits
    both regime evidence and chart rectangle geometry.

    '''
    times: np.ndarray = 1_700_000_000 + np.arange(8) * 0.1
    source: np.ndarray = _source(tuple(times), (1,) * 8)
    spans: list[RegimeInterval] = trend_intervals(source, 0.1)
    assert len(spans) == 1
    rectangles: np.ndarray
    _: list[RegimeInterval]
    rectangles, _ = regime_rectangles(
        spans, _chart(tuple(times)), 0.1,
    )
    assert len(rectangles) == 1

    source['time'][4:] += 0.1
    spans = trend_intervals(source, 0.1)
    assert len(spans) == 2
    whole: RegimeInterval = RegimeInterval(
        float(times[0]), float(times[-1] + 0.3), 'uptrend',
    )
    rectangles, _ = regime_rectangles(
        [whole], _chart(tuple(source['time'])), 0.1,
    )
    assert len(rectangles) == 2


@pytest.mark.parametrize('index_field', ['index', 'time'])
def test_delayed_regime_and_chart_gaps(
    index_field: Literal['index', 'time'],

) -> None:
    '''
    A classifier may report an earlier onset after the chart has
    advanced. Its rectangle must begin only at response arrival and
    must not fill an absent chart bucket. Fractional arrival within
    a bar, plus a 20-second hole, proves causal clipping and split
    geometry on both time and absolute-index chart axes.

    '''
    chart: np.ndarray = _chart((100, 110, 140, 150))
    span: RegimeInterval = RegimeInterval(
        start_time=100,
        end_time=155,
        label='ranging',
        probability=0.8,
        source='jev',
        observed_at=115,
    )
    rectangles: np.ndarray
    metadata: list[RegimeInterval]
    rectangles, metadata = regime_rectangles(
        [span], chart, 10, index_field=index_field,
    )
    expected: list[list[float]] = (
        [[1001.5, 99, 0.5, 2], [1002, 99, 1.5, 2]]
        if index_field == 'index'
        else [[115, 99, 5, 2], [140, 99, 15, 2]]
    )
    np.testing.assert_allclose(rectangles, expected)
    assert metadata == [span, span]
    assert metadata[0].probability == 0.8
    assert metadata[0].source == 'jev'


def test_empty_and_unavailable_geometry() -> None:
    '''
    A layer can attach before history arrives, and late classifier
    responses can lie beyond retained history. Empty chart arrays
    and already-expired intervals must produce empty geometry rather
    than indexing the final row or drawing a retrospective rectangle.

    '''
    span: RegimeInterval = RegimeInterval(10, 20, 'ranging')
    empty: np.ndarray
    metadata: list[RegimeInterval]
    empty, metadata = regime_rectangles([span], _chart(()), 1)
    assert empty.shape == (0, 4)
    assert metadata == []
    span = RegimeInterval(10, 20, 'ranging', observed_at=21)
    empty, metadata = regime_rectangles([span], _chart((10, 20)), 1)
    assert empty.shape == (0, 4)
    assert metadata == []


@pytest.mark.parametrize('period', [0, -1, np.nan, np.inf])
def test_invalid_period_rejected(period: float) -> None:
    '''
    Invalid sampling periods break binary-search interval semantics
    and can create infinite widths. All non-positive/non-finite
    period classes are rejected before constructing chart geometry.

    '''
    with pytest.raises(ValueError, match='period'):
        times_to_x(_chart((10, 20)), np.array([10]), period)


@pytest.mark.parametrize(
    'market,revision,period,end_time,accepted',
    [
        ('test.market', 3, 5, 130, True),
        ('other.market', 3, 5, 130, False),
        ('test.market', 4, 5, 130, False),
        ('test.market', 3, 15, 130, False),
        ('test.market', 3, 5, 119, False),
    ],
    ids=['matching', 'market', 'revision', 'period', 'future'],
)
def test_jev_adapter_retains_causal_evidence(
    market: str,
    revision: int,
    period: float,
    end_time: float,
    accepted: bool,

) -> None:
    '''
    Late or misrouted classifications must not color an unrelated
    market, source revision, or timeframe. A typed Jev response
    claims source evidence at 100 but arrives at 120; matching
    evidence retains its original probability/model and starts at
    receipt time. The rejected cases vary one ownership bound each.

    '''
    result: JevClassification = JevClassification(
        market='test.market',
        revision=3,
        sequence=40,
        segment=2,
        source_as_of=100,
        requested_at=118,
        received_at=120,
        latency=2,
        model='jev-test',
        prompt_version='test',
        answers=(RegimeAnswer(
            question='regime_5',
            period=5,
            label='range',
            probability=0.82,
            confidence=0.7,
            probabilities=(('range', 0.82), ('unknown', 0.18)),
        ),),
        input_tokens=100,
        output_tokens=20,
    )
    interval: RegimeInterval|None = classification_interval(
        result,
        fqme=market,
        revision=revision,
        period_s=period,
        end_time=end_time,
    )
    if not accepted:
        assert interval is None
        return
    assert interval is not None
    assert interval.start_time == 100
    assert interval.observed_at == 120
    assert interval.label == 'range'
    assert interval.probability == 0.82
    assert interval.source == 'typesafe.ai/jev-test'


def test_overlay_rebuild_and_owned_teardown(
    qapp: QApplication,

) -> None:
    '''
    Equal-bound backfills rewrite earlier pivot fields without
    changing the live tail. An ordinary paint gate intentionally
    skips that change; a forced revision must replace stale markers
    and rectangles, preserve a sibling layer, and release exactly
    the owning layer's two items on repeated cancellation cleanup.
    Real Qt items prove scene ownership rather than a mocked API.

    '''
    source: np.ndarray = _source(
        (100, 110, 120, 130, 140), (1, 1, -1, -1, 1),
    )
    source['top_1'][1] = 105
    source['top_time_1'][1] = 100
    widget: pg.PlotWidget = pg.PlotWidget()
    viz: SimpleNamespace = SimpleNamespace(
        plot=widget.plotItem,
        shm=SimpleNamespace(array=_chart((100, 110, 120, 130, 140))),
        index_field='index',
    )
    first: TrendOverlay = TrendOverlay(
        viz,
        SimpleNamespace(array=source),
        fqme='first.test',
        source_period_s=10,
        chart_period_s=10,
    )
    sibling: TrendOverlay = TrendOverlay(
        viz,
        SimpleNamespace(array=source),
        fqme='second.test',
        source_period_s=10,
        chart_period_s=10,
    )
    try:
        assert first.update()
        assert sibling.update()
        assert len(first.points) == 1
        assert not first.update()
        source['top_1'][1] = np.nan
        source['bottom_1'][1] = 95
        source['bottom_time_1'][1] = 100
        assert not first.update()
        assert first.update(force=True)
        assert first.points[0].kind == 'bottom'
        assert sibling.points[0].kind == 'top'
        assert len(widget.plotItem.items) == 4

        # A synthetic full-scene paint verifies Qt primitive types
        # and brush batching without a GUI or screenshot dependency.
        image: QtGui.QImage = QtGui.QImage(
            160, 100, QtGui.QImage.Format.Format_ARGB32,
        )
        image.fill(0)
        painter: QtGui.QPainter = QtGui.QPainter(image)
        try:
            widget.scene().render(painter)
        finally:
            painter.end()

        first.remove()
        first.remove()
        assert len(widget.plotItem.items) == 2
        assert sibling.markers in widget.plotItem.items
        assert not first.update(force=True)
    finally:
        first.remove()
        sibling.remove()
        widget.close()
        widget.deleteLater()
        qapp.processEvents()


def test_provisional_updates_reuse_confirmed_markers(
    qapp: QApplication,

) -> None:
    '''
    Quotes may repaint a provisional reversal repeatedly during one
    bar. Rebuilding historical markers each time wastes UI time and
    can present a temporary reversal as confirmed. Mutate only the
    FSP tail, then the chart candle envelope: the first update is
    skipped and the second updates rectangles while preserving the
    same ``TrendOverlay.points`` list. Changing ``chart['index']``
    requires remapping those retained points.

    '''
    source: np.ndarray = _source((10, 20, 30), (1, 1, -1))
    source['top_1'][1] = 105
    source['top_time_1'][1] = 10
    widget: pg.PlotWidget = pg.PlotWidget()
    chart: np.ndarray = _chart((10, 20, 30))
    viz: SimpleNamespace = SimpleNamespace(
        plot=widget.plotItem,
        shm=SimpleNamespace(array=chart),
        index_field='index',
    )
    overlay: TrendOverlay = TrendOverlay(
        viz,
        SimpleNamespace(array=source),
        fqme='live.test',
        source_period_s=10,
        chart_period_s=10,
    )
    try:
        assert overlay.update()
        original: list[ReversalPoint] = overlay.points
        source['trend'][-1] = 1
        source['bottom_1'][-1] = 91
        source['bottom_time_1'][-1] = 20
        assert not overlay.update()
        assert overlay.points is original

        chart['high'][-1] = 109
        assert overlay.update()
        assert overlay.points is original
        assert overlay.rectangles.rectangles[-1, 3] == 10

        chart['high'][-1] = 100
        assert overlay.update()
        assert overlay.points is original
        assert overlay.rectangles.rectangles[-1, 3] == 2

        chart['index'] += 500
        assert overlay.update()
        assert overlay.points is not original
        assert overlay.points[0].x == 1500
    finally:
        overlay.remove()
        widget.close()
        widget.deleteLater()
        qapp.processEvents()


def test_overlay_limits_and_classifier_metadata(
    qapp: QApplication,

) -> None:
    '''
    Unbounded event layers can allocate thousands of Qt objects and
    delayed classifier results can overwrite deterministic meaning.
    Many synthetic pivots exercise the configured retention limits;
    replacing rectangles with a Jev result must retain probability,
    source, and markers while keeping exactly two scene items.

    '''
    source: np.ndarray = _source(tuple(range(20)), (1, -1) * 10)
    source['top_1'] = 105
    source['top_time_1'] = source['time']
    widget: pg.PlotWidget = pg.PlotWidget()
    viz: SimpleNamespace = SimpleNamespace(
        plot=widget.plotItem,
        shm=SimpleNamespace(array=_chart(tuple(range(20)))),
        index_field='index',
    )
    overlay: TrendOverlay = TrendOverlay(
        viz,
        SimpleNamespace(array=source),
        fqme='bounded.test',
        source_period_s=1,
        chart_period_s=1,
        max_points=3,
        max_rectangles=2,
    )
    try:
        overlay.update()
        assert len(overlay.points) == 3
        assert overlay.points[0].pivot_time == 16
        assert len(overlay.rectangles.rectangles) == 2
        assert len(widget.plotItem.items) == 2

        classified: RegimeInterval = RegimeInterval(
            start_time=15,
            end_time=20,
            label='ranging',
            probability=0.73,
            source='jev',
            observed_at=18,
        )
        overlay.set_regimes([classified])
        assert len(overlay.points) == 3
        assert overlay.rectangles.intervals == [classified]
        assert overlay.rectangles.rectangles[0, 0] == 1018
        tip: str = overlay.rectangles.tooltip_at(QPointF(1019, 100))
        assert 'ranging: 73.0%' in tip
        assert '(1s)' in tip
        assert 'jev' in tip
        assert '00:00:18.000' in tip
        overlay.set_regimes([])
        assert len(overlay.rectangles.rectangles) == 0
        overlay.set_regimes(None)
        assert len(overlay.rectangles.rectangles) == 2
        tip = overlay.rectangles.tooltip_at(QPointF(1019, 100))
        assert 'trend-calculus' in tip
        assert '%' not in tip
    finally:
        overlay.remove()
        widget.close()
        widget.deleteLater()
        qapp.processEvents()


class _PriceChart:
    '''
    Expose the price-chart API while retaining actual Qt ownership.

    '''
    _trend_overlays: dict[tuple[str, str], list[TrendOverlay]]

    def __init__(self) -> None:
        '''
        Build a chart whose SHM token identifies its base timeframe.

        '''
        self.widget: pg.PlotWidget = pg.PlotWidget()
        self.viz: SimpleNamespace = SimpleNamespace(
            plot=self.widget.plotItem,
            shm=SimpleNamespace(
                array=_chart((10, 20, 30, 40, 50)),
                _token='ohlcv',
            ),
            index_field='index',
        )

    def get_viz(self, fqme: str) -> SimpleNamespace|None:
        '''
        Resolve exactly the test market rather than any chart key.

        '''
        return self.viz if fqme == 'admin.test' else None

    def close(self) -> None:
        '''
        Release the disposable offscreen widget after its layers.

        '''
        self.widget.close()
        self.widget.deleteLater()


async def _admin_bootstrap_scenario(
    qapp: QApplication,
    monkeypatch: pytest.MonkeyPatch,

) -> None:
    '''
    Attaching before FSP bootstrap exposes empty history, and opening
    several reversal orders must not start multiple SHM writers.
    The ``attach_first()`` task calls ``FspAdmin.open_trend_overlay``
    while the mocked engine's ``started`` event stays unset.
    ``wait_all_tasks_blocked()`` reaches that wait deterministically;
    one engine call and an empty Qt plot rule out early attachment.
    Only setting ``started`` lets ``attached`` confirm publication.
    Reopening another order retains one writer, and repeated
    ``FspAdmin.close_trend_overlays()`` preserves a sibling's items.
    This exercises local lifecycle ordering, not actor IPC.

    '''
    chart: _PriceChart = _PriceChart()
    source: np.ndarray = _source((10, 20, 30, 40, 50), (1,) * 5)
    source['trend_2'] = 1
    destination: SimpleNamespace = SimpleNamespace(
        rt_shm=SimpleNamespace(array=source, _token='trend'),
    )
    flume: SimpleNamespace = SimpleNamespace(
        mkt=SimpleNamespace(fqme='admin.test'),
        rt_shm=chart.viz.shm,
        rt_sample_period_s=10,
        _hist_shm_token=None,
    )
    started: trio.Event = trio.Event()
    attached: trio.Event = trio.Event()
    calls: list[tuple[str, dict, str]] = []
    attached_layers: list[TrendOverlay] = []

    async def start_engine(
        target: Fsp,
        conf: dict,
        loglevel: str,

    ) -> tuple[SimpleNamespace, trio.Event]:
        '''
        Return an allocated destination before publishing bootstrap.

        '''
        calls.append((target.name, conf, loglevel))
        return destination, started

    try:
        nursery: trio.Nursery
        async with trio.open_nursery() as nursery:
            admin: FspAdmin = FspAdmin(
                nursery, {}, SimpleNamespace(chart=chart), flume,
            )
            monkeypatch.setattr(
                admin, 'start_engine_task', start_engine,
            )
            with pytest.raises(ValueError, match='positive'):
                await admin.open_trend_overlay(chart_period_s=0)
            assert calls == []

            async def attach_first() -> None:
                '''
                Signal only after startup has produced visible data.

                '''
                layer: TrendOverlay = (
                    await admin.open_trend_overlay()
                )
                attached_layers.append(layer)
                attached.set()

            nursery.start_soon(attach_first)
            await wait_all_tasks_blocked()
            assert len(calls) == 1
            assert not attached.is_set()
            assert chart.widget.plotItem.items == []

            started.set()
            await attached.wait()
            first: TrendOverlay = attached_layers[0]
            second: TrendOverlay = await admin.open_trend_overlay(
                order=2,
            )
            assert await admin.open_trend_overlay() is first
            with pytest.raises(ValueError, match='period differs'):
                await admin.open_trend_overlay(chart_period_s=60)
            assert second is not first
            assert len(calls) == 1
            assert calls[0][0] == 'trend_calculus'
            assert chart._trend_overlays == {
                ('admin.test', 'trend'): [first, second],
            }
            assert len(chart.widget.plotItem.items) == 4

            sibling: TrendOverlay = TrendOverlay(
                chart.viz,
                SimpleNamespace(array=source, _token='other'),
                fqme='other.test',
                source_period_s=10,
                chart_period_s=10,
            )
            foreign_key: tuple[str, str] = ('other.test', 'other')
            chart._trend_overlays[foreign_key] = [sibling]
            try:
                admin.close_trend_overlays()
                admin.close_trend_overlays()
                assert len(chart.widget.plotItem.items) == 2
                assert chart._trend_overlays == {
                    ('other.test', 'other'): [sibling],
                }
            finally:
                sibling.remove()
    finally:
        chart.close()
        qapp.processEvents()


def test_admin_waits_for_bootstrap_and_reuses_writer(
    qapp: QApplication,
    monkeypatch: pytest.MonkeyPatch,

) -> None:
    '''
    Exercise bootstrap publication and owner cleanup in local Trio.

    ``_admin_bootstrap_scenario()`` controls its ``started`` event
    and waits for ``attach_first()`` to block in
    ``FspAdmin.open_trend_overlay()`` before inspecting the plot.
    The unset ``attached`` event and empty plot prove attachment
    waits for bootstrap rather than depending on timer luck. A
    second order shares the writer; cleanup retains a sibling's
    ``_PriceChart._trend_overlays`` entry and its two Qt items.

    '''
    trio.run(_admin_bootstrap_scenario, qapp, monkeypatch)


def test_registry_filters_and_defers_hidden_revisions(
    qapp: QApplication,

) -> None:
    '''
    A destination rewrite while hidden must survive an unchanged
    source-tail paint signature when the chart becomes visible.
    An equal-bound earlier pivot edit arranges that case. Wrong
    market/token/timeframe invalidations leave the layer alone;
    a matching hidden notification invalidates without touching Qt
    data, and the next ordinary refresh consumes the corrected pivot.

    '''
    chart: _PriceChart = _PriceChart()
    source: np.ndarray = _source((10, 20, 30, 40, 50), (1,) * 5)
    source['top_1'][1] = 105
    source['top_time_1'][1] = 10
    overlay: TrendOverlay = TrendOverlay(
        chart.viz,
        SimpleNamespace(array=source, _token='trend'),
        fqme='admin.test',
        source_period_s=10,
        chart_period_s=10,
    )
    chart._trend_overlays = {('admin.test', 'trend'): [overlay]}
    try:
        assert update_trend_overlays(chart) == 1
        original: list[ReversalPoint] = overlay.points
        source['top_1'][1] = 109
        assert update_trend_overlays(
            chart, fqme='other.test', force=True,
        ) == 0
        assert update_trend_overlays(
            chart, shm_token='other', force=True,
        ) == 0
        assert update_trend_overlays(
            chart, chart_period_s=60, force=True,
        ) == 0
        assert update_trend_overlays(chart) == 0
        assert overlay.points is original

        admin: FspAdmin = FspAdmin(
            None, {}, SimpleNamespace(chart=chart), None,
        )
        admin._trend_overlays.append((chart, overlay))
        admin.refresh_trend_overlays('trend', render=False)
        assert overlay.points is original
        assert overlay.points[0].price == 105
        assert update_trend_overlays(chart) == 1
        assert overlay.points[0].price == 109
    finally:
        overlay.remove()
        chart.close()
        qapp.processEvents()


def test_admin_context_exit_releases_layers(
    qapp: QApplication,
    monkeypatch: pytest.MonkeyPatch,

) -> None:
    '''
    A finished FSP-admin context must not leave its graphics attached
    to a reused chart widget. Replace only actor-cluster startup and
    engine allocation; ``open_fsp_admin()`` and its Trio nursery run
    normal exit with actual Qt items. Empty
    ``FspAdmin._trend_overlays`` and ``_PriceChart._trend_overlays``
    plus the signalled ``complete`` event from
    ``FspAdmin._registry`` prove graphics teardown and existing
    engine shutdown notification both run. This local stub does not
    prove remote actor shutdown or shared-memory visibility.

    '''
    chart: _PriceChart = _PriceChart()

    @asynccontextmanager
    async def cluster_context(
        **kwargs: object,

    ) -> AsyncIterator[tuple[bool, dict]]:
        '''
        Supply a local cluster stub without spawning actor processes.

        '''
        yield False, {}

    monkeypatch.setattr(
        fsp_ui, 'maybe_open_context', cluster_context,
    )

    async def scenario() -> None:
        '''
        Observe cleanup after the actual admin context has exited.

        '''
        flume: SimpleNamespace = SimpleNamespace(
            mkt=SimpleNamespace(fqme='admin.test'),
            rt_shm=chart.viz.shm,
            rt_sample_period_s=10,
            _hist_shm_token=None,
        )
        source: np.ndarray = _source(
            (10, 20, 30, 40, 50), (1,) * 5,
        )
        started: trio.Event = trio.Event()
        started.set()
        complete: trio.Event = trio.Event()

        async def start_engine(
            target: Fsp,
            conf: dict,
            loglevel: str,

        ) -> tuple[SimpleNamespace, trio.Event]:
            '''
            Return an already published, test-owned FSP destination.

            '''
            return SimpleNamespace(rt_shm=SimpleNamespace(
                array=source,
                _token='trend',
            )), started

        admin: FspAdmin
        async with fsp_ui.open_fsp_admin(
            SimpleNamespace(chart=chart), flume,
        ) as admin:
            monkeypatch.setattr(
                admin, 'start_engine_task', start_engine,
            )
            admin._registry[('test',)] = (None, None, complete)
            await admin.open_trend_overlay()
            assert len(chart.widget.plotItem.items) == 2

        assert chart.widget.plotItem.items == []
        assert chart._trend_overlays == {}
        assert admin._trend_overlays == []
        assert complete.is_set()

    try:
        trio.run(scenario)
    finally:
        chart.close()
        qapp.processEvents()