Set a `PlotItem.viz` for interaction lookup

Inside `._interaction` routines we need access to `Viz` instances.
Instead of doing `CharPlotWidget._vizs: dict` lookups this ensures each
plot can lookup it's (parent) viz without error.

Also, adjusts `Viz.maxmin()` output parsing to new signature.

piker/_daemon.py

@@ -257,7 +257,7 @@ async def open_piker_runtime(
     # and spawn the service tree distributed per that.
     start_method: str = 'trio',
 
-    tractor_kwargs: dict = {},
+    **tractor_kwargs,
 
 ) -> tuple[
     tractor.Actor,

piker/data/feed.py

@@ -207,7 +207,7 @@ def get_feed_bus(
 
 ) -> _FeedsBus:
     '''
-    Retreive broker-daemon-local data feeds bus from process global
+    Retrieve broker-daemon-local data feeds bus from process global
     scope. Serialize task access to lock.
 
     '''
@@ -250,6 +250,7 @@ async def start_backfill(
     shm: ShmArray,
     timeframe: float,
     sampler_stream: tractor.MsgStream,
+    feed_is_live: trio.Event,
 
     last_tsdb_dt: Optional[datetime] = None,
     storage: Optional[Storage] = None,
@@ -281,9 +282,30 @@ async def start_backfill(
             - pendulum.from_timestamp(times[-2])
         ).seconds
 
-        if step_size_s == 60:
+        # if the market is open (aka we have a live feed) but the
+        # history sample step index seems off we report the surrounding
+        # data and drop into a bp. this case shouldn't really ever
+        # happen if we're doing history retrieval correctly.
+        if (
+            step_size_s == 60
+            and feed_is_live.is_set()
+        ):
             inow = round(time.time())
-            if (inow - times[-1]) > 60:
+            diff = inow - times[-1]
+            if abs(diff) > 60:
+                surr = array[-6:]
+                diff_in_mins = round(diff/60., ndigits=2)
+                log.warning(
+                    f'STEP ERROR `{bfqsn}` for period {step_size_s}s:\n'
+                    f'Off by `{diff}` seconds (or `{diff_in_mins}` mins)\n'
+                    'Surrounding 6 time stamps:\n'
+                    f'{list(surr["time"])}\n'
+                    'Here is surrounding 6 samples:\n'
+                    f'{surr}\nn'
+                )
+
+                # for now we expect a hacker to investigate this case
+                # manually..
                 await tractor.breakpoint()
 
         # frame's worth of sample-period-steps, in seconds
@@ -485,6 +507,7 @@ async def basic_backfill(
     bfqsn: str,
     shms: dict[int, ShmArray],
     sampler_stream: tractor.MsgStream,
+    feed_is_live: trio.Event,
 
 ) -> None:
 
@@ -504,6 +527,7 @@ async def basic_backfill(
                     shm,
                     timeframe,
                     sampler_stream,
+                    feed_is_live,
                 )
             )
         except DataUnavailable:
@@ -520,6 +544,7 @@ async def tsdb_backfill(
     bfqsn: str,
     shms: dict[int, ShmArray],
     sampler_stream: tractor.MsgStream,
+    feed_is_live: trio.Event,
 
     task_status: TaskStatus[
         tuple[ShmArray, ShmArray]
@@ -554,6 +579,8 @@ async def tsdb_backfill(
                     shm,
                     timeframe,
                     sampler_stream,
+                    feed_is_live,
+
                     last_tsdb_dt=last_tsdb_dt,
                     tsdb_is_up=True,
                     storage=storage,
@@ -856,6 +883,7 @@ async def manage_history(
                         60: hist_shm,
                     },
                     sample_stream,
+                    feed_is_live,
                 )
 
                 # yield back after client connect with filled shm
@@ -890,6 +918,7 @@ async def manage_history(
                     60: hist_shm,
                 },
                 sample_stream,
+                feed_is_live,
             )
             task_status.started((
                 hist_zero_index,
@@ -1051,7 +1080,10 @@ async def allocate_persistent_feed(
     # seed the buffer with a history datum - this is most handy
     # for many backends which don't sample @ 1s OHLC but do have
     # slower data such as 1m OHLC.
-    if not len(rt_shm.array):
+    if (
+        not len(rt_shm.array)
+        and hist_shm.array.size
+    ):
         rt_shm.push(hist_shm.array[-3:-1])
         ohlckeys = ['open', 'high', 'low', 'close']
         rt_shm.array[ohlckeys][-2:] = hist_shm.array['close'][-1]
@@ -1062,6 +1094,9 @@ async def allocate_persistent_feed(
         rt_shm.array['time'][0] = ts
         rt_shm.array['time'][1] = ts + 1
 
+    elif hist_shm.array.size == 0:
+        await tractor.breakpoint()
+
     # wait the spawning parent task to register its subscriber
     # send-stream entry before we start the sample loop.
     await sub_registered.wait()

piker/pp.py

@@ -54,7 +54,7 @@ def open_trade_ledger(
     broker: str,
     account: str,
 
-) -> str:
+) -> dict:
     '''
     Indempotently create and read in a trade log file from the
     ``<configuration_dir>/ledgers/`` directory.

piker/ui/_chart.py

@@ -50,7 +50,6 @@ from ._cursor import (
     ContentsLabel,
 )
 from ..data._sharedmem import ShmArray
-from ._l1 import L1Labels
 from ._ohlc import BarItems
 from ._curve import (
     Curve,
@@ -70,12 +69,10 @@ from ..data._source import Symbol
 from ..log import get_logger
 from ._interaction import ChartView
 from ._forms import FieldsForm
-from .._profile import pg_profile_enabled, ms_slower_then
 from ._overlay import PlotItemOverlay
 from ._dataviz import Viz
 from ._search import SearchWidget
 from . import _pg_overrides as pgo
-from .._profile import Profiler
 
 if TYPE_CHECKING:
     from ._display import DisplayState
@@ -632,11 +629,13 @@ class LinkedSplits(QWidget):
         for axis in axes.values():
             axis.pi = cpw.plotItem
 
+
         cpw.hideAxis('left')
         cpw.hideAxis('bottom')
 
         if (
-            _xaxis_at == 'bottom' and (
+            _xaxis_at == 'bottom'
+            and (
                 self.xaxis_chart
                 or (
                     not self.subplots
@@ -644,6 +643,8 @@ class LinkedSplits(QWidget):
                 )
             )
         ):
+            # hide the previous x-axis chart's bottom axis since we're
+            # presumably being appended to the bottom subplot.
             if self.xaxis_chart:
                 self.xaxis_chart.hideAxis('bottom')
 
@@ -688,7 +689,12 @@ class LinkedSplits(QWidget):
         # link chart x-axis to main chart
         # this is 1/2 of where the `Link` in ``LinkedSplit``
         # comes from ;)
-        cpw.setXLink(self.chart)
+        cpw.cv.setXLink(self.chart)
+
+        # NOTE: above is the same as the following,
+        # link this subchart's axes to the main top level chart.
+        # if self.chart:
+        #     cpw.cv.linkView(0, self.chart.cv)
 
         add_label = False
         anchor_at = ('top', 'left')
@@ -800,7 +806,9 @@ class LinkedSplits(QWidget):
 # write our own wrapper around `PlotItem`..
 class ChartPlotWidget(pg.PlotWidget):
     '''
-    ``GraphicsView`` subtype containing a single ``PlotItem``.
+    ``GraphicsView`` subtype containing a ``.plotItem: PlotItem`` as well
+    as a `.pi_overlay: PlotItemOverlay`` which helps manage and overlay flow
+    graphics view multiple compose view boxes.
 
     - The added methods allow for plotting OHLC sequences from
       ``np.ndarray``s with appropriate field names.
@@ -857,17 +865,17 @@ class ChartPlotWidget(pg.PlotWidget):
         self.sidepane: Optional[FieldsForm] = None
 
         # source of our custom interactions
-        self.cv = cv = self.mk_vb(name)
+        self.cv = self.mk_vb(name)
 
         pi = pgo.PlotItem(
-            viewBox=cv,
+            viewBox=self.cv,
             name=name,
             **kwargs,
         )
         pi.chart_widget = self
         super().__init__(
             background=hcolor(view_color),
-            viewBox=cv,
+            viewBox=self.cv,
             # parent=None,
             # plotItem=None,
             # antialias=True,
@@ -878,7 +886,9 @@ class ChartPlotWidget(pg.PlotWidget):
         # give viewbox as reference to chart
         # allowing for kb controls and interactions on **this** widget
         # (see our custom view mode in `._interactions.py`)
-        cv.chart = self
+        self.cv.chart = self
+
+        self.pi_overlay: PlotItemOverlay = PlotItemOverlay(self.plotItem)
 
         # ensure internal pi matches
         assert self.cv is self.plotItem.vb
@@ -907,8 +917,6 @@ class ChartPlotWidget(pg.PlotWidget):
         # show background grid
         self.showGrid(x=False, y=True, alpha=0.3)
 
-        self.pi_overlay: PlotItemOverlay = PlotItemOverlay(self.plotItem)
-
         # indempotent startup flag for auto-yrange subsys
         # to detect the "first time" y-domain graphics begin
         # to be shown in the (main) graphics view.
@@ -1101,6 +1109,7 @@ class ChartPlotWidget(pg.PlotWidget):
         # view **to** this parent and likewise *from* the
         # main/parent chart back *to* the created overlay.
         cv.enable_auto_yrange(src_vb=self.view)
+
         # makes it so that interaction on the new overlay will reflect
         # back on the main chart (which overlay was added to).
         self.view.enable_auto_yrange(src_vb=cv)
@@ -1175,6 +1184,7 @@ class ChartPlotWidget(pg.PlotWidget):
             # register curve graphics with this viz
             graphics=graphics,
         )
+        pi.viz = viz
         assert isinstance(viz.shm, ShmArray)
 
         # TODO: this probably needs its own method?
@@ -1302,13 +1312,6 @@ class ChartPlotWidget(pg.PlotWidget):
         If ``bars_range`` is provided use that range.
 
         '''
-        profiler = Profiler(
-            msg=f'`{str(self)}.maxmin(name={name})`: `{self.name}`',
-            disabled=not pg_profile_enabled(),
-            ms_threshold=ms_slower_then,
-            delayed=True,
-        )
-
         # TODO: here we should instead look up the ``Viz.shm.array``
         # and read directly from shm to avoid copying to memory first
         # and then reading it again here.
@@ -1316,36 +1319,21 @@ class ChartPlotWidget(pg.PlotWidget):
         viz = self._vizs.get(viz_key)
         if viz is None:
             log.error(f"viz {viz_key} doesn't exist in chart {self.name} !?")
-            key = res = 0, 0
+            return 0, 0
 
+        res = viz.maxmin()
+
+        if (
+            res is None
+        ):
+            mxmn = 0, 0
+            if not self._on_screen:
+                self.default_view(do_ds=False)
+                self._on_screen = True
         else:
-            (
+            x_range, read_slc, mxmn = res
-                l,
-                _,
-                lbar,
-                rbar,
-                _,
-                r,
-            ) = bars_range or viz.datums_range()
 
-            profiler(f'{self.name} got bars range')
+        return mxmn
-            key = lbar, rbar
-            res = viz.maxmin(*key)
-
-            if (
-                res is None
-            ):
-                log.warning(
-                    f"{viz_key} no mxmn for bars_range => {key} !?"
-                )
-                res = 0, 0
-                if not self._on_screen:
-                    self.default_view(do_ds=False)
-                    self._on_screen = True
-
-        profiler(f'yrange mxmn: {key} -> {res}')
-        # print(f'{viz_key} yrange mxmn: {key} -> {res}')
-        return res
 
     def get_viz(
         self,

piker/ui/_dataviz.py

@@ -60,6 +60,7 @@ from ..log import get_logger
 from .._profile import (
     Profiler,
     pg_profile_enabled,
+    ms_slower_then,
 )
 
 
@@ -276,7 +277,10 @@ class Viz(msgspec.Struct):  # , frozen=True):
     ] = (None, None)
 
     # cache of y-range values per x-range input.
-    _mxmns: dict[tuple[int, int], tuple[float, float]] = {}
+    _mxmns: dict[
+        tuple[int, int],
+        tuple[float, float],
+    ] = {}
 
     @property
     def shm(self) -> ShmArray:
@@ -324,32 +328,29 @@ class Viz(msgspec.Struct):  # , frozen=True):
 
     def maxmin(
         self,
-        lbar: int,
-        rbar: int,
 
+        # TODO: drop this right?
+        bars_range: Optional[tuple[
+            int, int, int, int, int, int
+        ]] = None,
+
+        x_range: slice | tuple[int, int] | None = None,
         use_caching: bool = True,
 
-    ) -> Optional[tuple[float, float]]:
+    ) -> tuple[float, float] | None:
         '''
         Compute the cached max and min y-range values for a given
         x-range determined by ``lbar`` and ``rbar`` or ``None``
         if no range can be determined (yet).
 
         '''
-        # TODO: hash the slice instead maybe?
+        name = self.name
-        # https://stackoverflow.com/a/29980872
+        profiler = Profiler(
-        rkey = (round(lbar), round(rbar))
+            msg=f'{name} -> `{str(self)}.maxmin()`',
-
+            disabled=not pg_profile_enabled(),
-        do_print: bool = False
+            ms_threshold=ms_slower_then,
-        if use_caching:
+            delayed=True,
-            cached_result = self._mxmns.get(rkey)
+        )
-            if cached_result:
-                if do_print:
-                    print(
-                        f'{self.name} CACHED maxmin\n'
-                        f'{rkey} -> {cached_result}'
-                    )
-                return cached_result
 
         shm = self.shm
         if shm is None:
@@ -357,6 +358,43 @@ class Viz(msgspec.Struct):  # , frozen=True):
 
         arr = shm.array
 
+        if x_range is None:
+            (
+                l,
+                _,
+                lbar,
+                rbar,
+                _,
+                r,
+            ) = (
+                # TODO: drop this yah?
+                bars_range
+                or self.datums_range()
+            )
+
+            profiler(f'{self.name} got bars range')
+            x_range = lbar, rbar
+
+        # TODO: hash the slice instead maybe?
+        # https://stackoverflow.com/a/29980872
+        ixrng = (round(lbar), round(rbar))
+
+        do_print: bool = False
+        if use_caching:
+            cached_result = self._mxmns.get(ixrng)
+            if cached_result:
+                if do_print:
+                    print(
+                        f'{self.name} CACHED maxmin\n'
+                        f'{ixrng} -> {cached_result}'
+                    )
+                read_slc, mxmn = cached_result
+                return (
+                    ixrng,
+                    read_slc,
+                    mxmn,
+                )
+
         # get relative slice indexes into array
         if self.index_field == 'time':
             read_slc = slice_from_time(
@@ -376,7 +414,10 @@ class Viz(msgspec.Struct):  # , frozen=True):
         slice_view = arr[read_slc]
 
         if not slice_view.size:
-            log.warning(f'{self.name} no maxmin in view?')
+            log.warning(
+                f'{self.name} no maxmin in view?\n'
+                f"{name} no mxmn for bars_range => {ixrng} !?"
+            )
             return None
 
         elif self.yrange:
@@ -384,9 +425,8 @@ class Viz(msgspec.Struct):  # , frozen=True):
             if do_print:
                 print(
                     f'{self.name} M4 maxmin:\n'
-                    f'{rkey} -> {mxmn}'
+                    f'{ixrng} -> {mxmn}'
                 )
-
         else:
             if self.is_ohlc:
                 ylow = np.min(slice_view['low'])
@@ -404,7 +444,7 @@ class Viz(msgspec.Struct):  # , frozen=True):
                 s = 3
                 print(
                     f'{self.name} MANUAL ohlc={self.is_ohlc} maxmin:\n'
-                    f'{rkey} -> {mxmn}\n'
+                    f'{ixrng} -> {mxmn}\n'
                     f'read_slc: {read_slc}\n'
                     # f'abs_slc: {slice_view["index"]}\n'
                     f'first {s}:\n{slice_view[:s]}\n'
@@ -413,9 +453,13 @@ class Viz(msgspec.Struct):  # , frozen=True):
 
         # cache result for input range
         assert mxmn
-        self._mxmns[rkey] = mxmn
+        self._mxmns[ixrng] = (read_slc, mxmn)
-
+        profiler(f'yrange mxmn cacheing: {x_range} -> {mxmn}')
-        return mxmn
+        return (
+            ixrng,
+            read_slc,
+            mxmn,
+        )
 
     def view_range(self) -> tuple[int, int]:
         '''
@@ -985,12 +1029,10 @@ class Viz(msgspec.Struct):  # , frozen=True):
         )
 
         if do_ds:
+            # view.interaction_graphics_update_cycle()
             view.maybe_downsample_graphics()
             view._set_yrange()
 
-            # caller should do this!
-            # self.linked.graphics_cycle()
-
     def incr_info(
         self,
         ds: DisplayState,

piker/ui/_display.py

@@ -986,32 +986,6 @@ async def link_views_with_region(
     # region.sigRegionChangeFinished.connect(update_pi_from_region)
 
 
-# force 0 to always be in view
-def multi_maxmin(
-    chart: ChartPlotWidget,
-    names: list[str],
-
-) -> tuple[float, float]:
-    '''
-    Viz "group" maxmin loop; assumes all named vizs
-    are in the same co-domain and thus can be sorted
-    as one set.
-
-    Iterates all the named vizs and calls the chart
-    api to find their range values and return.
-
-    TODO: really we should probably have a more built-in API
-    for this?
-
-    '''
-    mx = 0
-    for name in names:
-        ymn, ymx = chart.maxmin(name=name)
-        mx = max(mx, ymx)
-
-        return 0, mx
-
-
 _quote_throttle_rate: int = 60 - 6
 
 

piker/ui/_interaction.py

@@ -21,7 +21,11 @@ Chart view box primitives
 from __future__ import annotations
 from contextlib import asynccontextmanager
 import time
-from typing import Optional, Callable
+from typing import (
+    Optional,
+    Callable,
+    TYPE_CHECKING,
+)
 
 import pyqtgraph as pg
 # from pyqtgraph.GraphicsScene import mouseEvents
@@ -39,6 +43,9 @@ from .._profile import pg_profile_enabled, ms_slower_then
 from ._editors import SelectRect
 from . import _event
 
+if TYPE_CHECKING:
+    from ._chart import ChartPlotWidget
+
 
 log = get_logger(__name__)
 
@@ -374,7 +381,7 @@ class ChartView(ViewBox):
         )
 
         self.linked = None
-        self._chart: 'ChartPlotWidget' = None  # noqa
+        self._chart: ChartPlotWidget | None = None  # noqa
 
         # add our selection box annotator
         self.select_box = SelectRect(self)
@@ -445,11 +452,11 @@ class ChartView(ViewBox):
             yield self
 
     @property
-    def chart(self) -> 'ChartPlotWidget':  # type: ignore # noqa
+    def chart(self) -> ChartPlotWidget:  # type: ignore # noqa
         return self._chart
 
     @chart.setter
-    def chart(self, chart: 'ChartPlotWidget') -> None:  # type: ignore # noqa
+    def chart(self, chart: ChartPlotWidget) -> None:  # type: ignore # noqa
         self._chart = chart
         self.select_box.chart = chart
         if self._maxmin is None:
@@ -783,11 +790,9 @@ class ChartView(ViewBox):
 
                 if yrange is None:
                     log.warning(f'No yrange provided for {name}!?')
-                    print(f"WTF NO YRANGE {name}")
                     return
 
             ylow, yhigh = yrange
-
             profiler(f'callback ._maxmin(): {yrange}')
 
             # view margins: stay within a % of the "true range"

piker/ui/_ohlc.py

@@ -121,7 +121,7 @@ class BarItems(FlowGraphic):
 
         # XXX: causes this weird jitter bug when click-drag panning
         # where the path curve will awkwardly flicker back and forth?
-        # self.setCacheMode(QtWidgets.QGraphicsItem.DeviceCoordinateCache)
+        self.setCacheMode(QtWidgets.QGraphicsItem.DeviceCoordinateCache)
 
         self.path = QPainterPath()
         self._last_bar_lines: tuple[QLineF, ...] | None = None

piker/ui/_overlay.py

@@ -22,7 +22,6 @@ from collections import defaultdict
 from functools import partial
 from typing import (
     Callable,
-    Optional,
 )
 
 from pyqtgraph.graphicsItems.AxisItem import AxisItem
@@ -246,7 +245,7 @@ class ComposedGridLayout:
         plot: PlotItem,
         name: str,
 
-    ) -> Optional[AxisItem]:
+    ) -> AxisItem | None:
         '''
         Retrieve the named axis for overlayed ``plot`` or ``None``
         if axis for that name is not shown.
@@ -321,7 +320,7 @@ class PlotItemOverlay:
     def add_plotitem(
         self,
         plotitem: PlotItem,
-        index: Optional[int] = None,
+        index: int | None = None,
 
         # event/signal names which will be broadcasted to all added
         # (relayee) ``PlotItem``s (eg. ``ViewBox.mouseDragEvent``).
@@ -376,7 +375,7 @@ class PlotItemOverlay:
 
                         # TODO: drop this viewbox specific input and
                         # allow a predicate to be passed in by user.
-                        axis: 'Optional[int]' = None,
+                        axis: int | None = None,
 
                         *,
 
@@ -578,3 +577,93 @@ class PlotItemOverlay:
 
     #     '''
     #     ...
+
+    def group_maxmin(
+        self,
+        focus_around: str | None = None,
+        force_min: float | None = None,
+
+    ) -> tuple[
+        float,  # mn
+        float,  # mx
+        float,  # max range in % terms of highest sigma plot's y-range
+        PlotItem,  # front/selected plot
+    ]:
+        '''
+        Overlay "group" maxmin sorting.
+
+        Assumes all named flows are in the same co-domain and thus can
+        be sorted as one set.
+
+        Iterates all the named flows and calls the chart api to find
+        their range values and return.
+
+        TODO: really we should probably have a more built-in API for
+        this?
+
+        '''
+        # TODO:
+        # - use this in the ``.ui._fsp`` mutli-maxmin stuff
+        # -
+
+        # force 0 to always be in view
+        group_mx: float = 0
+        group_mn: float = 0
+        mx_up_rng: float = 0
+        mn_down_rng: float = 0
+        pis2ranges: dict[
+            PlotItem,
+            tuple[float, float],
+        ] = {}
+
+        for pi in self.overlays:
+
+            # TODO: can we remove this from the widget
+            # and place somewhere more related to UX/Viz?
+            # name = pi.name
+            # chartw = pi.chart_widget
+            viz = pi.viz
+            # viz = chartw._vizs[name]
+
+            out = viz.maxmin()
+            if out is None:
+                return None
+
+            (
+                (x_start, x_stop),
+                read_slc,
+                (ymn, ymx),
+            ) = out
+
+            arr = viz.shm.array
+
+            y_start = arr[read_slc.start - 1]
+            y_stop = arr[read_slc.stop - 1]
+            if viz.is_ohlc:
+                y_start = y_start['open']
+                y_stop = y_stop['close']
+            else:
+                y_start = y_start[viz.name]
+                y_stop = y_stop[viz.name]
+
+            # update max for group
+            up_rng = (ymx - y_start) / y_start
+            down_rng = (y_stop - ymn) / y_stop
+
+            # compute directional (up/down) y-range % swing/dispersion
+            mx_up_rng = max(mx_up_rng, up_rng)
+            mn_down_rng = min(mn_down_rng, down_rng)
+
+            pis2ranges[pi] = (ymn, ymx)
+
+            group_mx = max(group_mx, ymx)
+            if force_min is None:
+                group_mn = min(group_mn, ymn)
+
+        return (
+           group_mn if force_min is None else force_min,
+           group_mx,
+           mn_down_rng,
+           mx_up_rng,
+           pis2ranges,
+        )