Comment stream drain idea

Likely it's not going to be that useful but keep it for reference for
the moment.

piker/data/_sharedmem.py

@@ -505,3 +505,35 @@ def maybe_open_shm_array(
         # to fail if a block has been allocated
         # on the OS by someone else.
         return open_shm_array(key=key, dtype=dtype, **kwargs), True
+
+
+def try_read(
+    array: np.ndarray
+
+) -> Optional[np.ndarray]:
+    '''
+    Try to read the last row from a shared mem array or ``None``
+    if the array read returns a zero-length array result.
+
+    Can be used to check for backfilling race conditions where an array
+    is currently being (re-)written by a writer actor but the reader is
+    unaware and reads during the window where the first and last indexes
+    are being updated.
+
+    '''
+    try:
+        return array[-1]
+    except IndexError:
+        # XXX: race condition with backfilling shm.
+        #
+        # the underlying issue is that a backfill (aka prepend) and subsequent
+        # shm array first/last index update could result in an empty array
+        # read here since the indices may be updated in such a way that
+        # a read delivers an empty array (though it seems like we
+        # *should* be able to prevent that?). also, as and alt and
+        # something we need anyway, maybe there should be some kind of
+        # signal that a prepend is taking place and this consumer can
+        # respond (eg. redrawing graphics) accordingly.
+
+        # the array read was emtpy
+        return None

piker/data/_source.py

@@ -106,6 +106,18 @@ class Symbol(BaseModel):
         mult = 1 / self.tick_size
         return round(value * mult) / mult
 
+    def front_feed(self) -> tuple[str, str]:
+        '''
+        Return the "current" feed key for this symbol.
+
+        (i.e. the broker + symbol key in a tuple).
+
+        '''
+        return (
+            list(self.broker_info.keys())[0],
+            self.key,
+        )
+
 
 @validate_arguments
 def mk_symbol(

piker/fsp/_engine.py

@@ -186,15 +186,17 @@ async def fsp_compute(
 async def cascade(
 
     ctx: tractor.Context,
+
+    # data feed key
     brokername: str,
+    symbol: str,
 
     src_shm_token: dict,
     dst_shm_token: tuple[str, np.dtype],
 
-    symbol: str,
     func_name: str,
-    zero_on_step: bool = False,
 
+    zero_on_step: bool = False,
     loglevel: Optional[str] = None,
 
 ) -> None:

piker/fsp/_momo.py

@@ -1,5 +1,5 @@
 # piker: trading gear for hackers
-# Copyright (C) 2018-present  Tyler Goodlet (in stewardship of piker0)
+# Copyright (C) Tyler Goodlet (in stewardship of pikers)
 
 # 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

piker/ui/_anchors.py

@@ -18,21 +18,26 @@
 Anchor funtions for UI placement of annotions.
 
 '''
-from typing import Callable
+from __future__ import annotations
+from typing import Callable, TYPE_CHECKING
 
 from PyQt5.QtCore import QPointF
 from PyQt5.QtWidgets import QGraphicsPathItem
 
+if TYPE_CHECKING:
+    from ._axes import PriceAxis
+    from ._chart import ChartPlotWidget
     from ._label import Label
 
 
 def marker_right_points(
-
+    chart: ChartPlotWidget,  # noqa
-    chart: 'ChartPlotWidget',  # noqa
     marker_size: int = 20,
 
 ) -> (float, float, float):
-    '''Return x-dimension, y-axis-aware, level-line marker oriented scene values.
+    '''
+    Return x-dimension, y-axis-aware, level-line marker oriented scene
+    values.
 
     X values correspond to set the end of a level line, end of
     a paried level line marker, and the right most side of the "right"
@@ -57,16 +62,17 @@ def vbr_left(
     label: Label,
 
 ) -> Callable[..., float]:
-    """Return a closure which gives the scene x-coordinate for the
+    '''
-    leftmost point of the containing view box.
+    Return a closure which gives the scene x-coordinate for the leftmost
+    point of the containing view box.
 
-    """
+    '''
     return label.vbr().left
 
 
 def right_axis(
 
-    chart: 'ChartPlotWidget',  # noqa
+    chart: ChartPlotWidget,  # noqa
     label: Label,
 
     side: str = 'left',
@@ -141,13 +147,13 @@ def gpath_pin(
         return path_br.bottomRight() - QPointF(label.w, label.h / 6)
 
 
-
 def pp_tight_and_right(
     label: Label
 
 ) -> QPointF:
-    '''Place *just* right of the pp label.
+    '''
+    Place *just* right of the pp label.
 
     '''
-    txt = label.txt
+    # txt = label.txt
     return label.txt.pos() + QPointF(label.w - label.h/3, 0)

piker/ui/_axes.py

@@ -18,8 +18,8 @@
 Chart axes graphics and behavior.
 
 """
-import functools
+from functools import lru_cache
-from typing import List, Tuple, Optional
+from typing import List, Tuple, Optional, Callable
 from math import floor
 
 import pandas as pd
@@ -27,8 +27,10 @@ import pyqtgraph as pg
 from PyQt5 import QtCore, QtGui, QtWidgets
 from PyQt5.QtCore import QPointF
 
-from ._style import DpiAwareFont, hcolor, _font
 from ..data._source import float_digits
+from ._label import Label
+from ._style import DpiAwareFont, hcolor, _font
+from ._interaction import ChartView
 
 _axis_pen = pg.mkPen(hcolor('bracket'))
 
@@ -42,7 +44,6 @@ class Axis(pg.AxisItem):
         self,
         linkedsplits,
         typical_max_str: str = '100 000.000',
-        min_tick: int = 2,
         **kwargs
 
     ) -> None:
@@ -52,7 +53,6 @@ class Axis(pg.AxisItem):
         # self.setCacheMode(QtWidgets.QGraphicsItem.DeviceCoordinateCache)
 
         self.linkedsplits = linkedsplits
-        self._min_tick = min_tick
         self._dpi_font = _font
 
         self.setTickFont(_font.font)
@@ -74,7 +74,10 @@ class Axis(pg.AxisItem):
         })
 
         self.setTickFont(_font.font)
+        # NOTE: this is for surrounding "border"
         self.setPen(_axis_pen)
+        # this is the text color
+        self.setTextPen(_axis_pen)
         self.typical_br = _font._qfm.boundingRect(typical_max_str)
 
         # size the pertinent axis dimension to a "typical value"
@@ -83,40 +86,102 @@ class Axis(pg.AxisItem):
     def size_to_values(self) -> None:
         pass
 
-    def set_min_tick(self, size: int) -> None:
-        self._min_tick = size
-
     def txt_offsets(self) -> Tuple[int, int]:
         return tuple(self.style['tickTextOffset'])
 
 
 class PriceAxis(Axis):
 
+    def __init__(
+        self,
+        *args,
+        min_tick: int = 2,
+        title: str = '',
+        formatter: Optional[Callable[[float], str]] = None,
+        **kwargs
+
+    ) -> None:
+        super().__init__(*args, **kwargs)
+        self.formatter = formatter
+        self._min_tick: int = min_tick
+        self.title = None
+
+    def set_title(
+        self,
+        title: str,
+        view: Optional[ChartView] = None
+
+    ) -> Label:
+        '''
+        Set a sane UX label using our built-in ``Label``.
+
+        '''
+        # XXX: built-in labels but they're huge, and placed weird..
+        # self.setLabel(title)
+        # self.showLabel()
+
+        label = self.title = Label(
+            view=view or self.linkedView(),
+            fmt_str=title,
+            color='bracket',
+            parent=self,
+            # update_on_range_change=False,
+        )
+
+        def below_axis() -> QPointF:
+            return QPointF(
+                0,
+                self.size().height(),
+            )
+
+        # XXX: doesn't work? have to pass it above
+        # label.txt.setParent(self)
+        label.scene_anchor = below_axis
+        label.render()
+        label.show()
+        label.update()
+        return label
+
+    def set_min_tick(
+        self,
+        size: int
+    ) -> None:
+        self._min_tick = size
+
     def size_to_values(self) -> None:
+        # self.typical_br = _font._qfm.boundingRect(typical_max_str)
         self.setWidth(self.typical_br.width())
 
     # XXX: drop for now since it just eats up h space
 
     def tickStrings(
         self,
-        vals,
+        vals: tuple[float],
-        scale,
+        scale: float,
-        spacing,
+        spacing: float,
-    ):
 
-        # TODO: figure out how to enforce min tick spacing by passing
+    ) -> list[str]:
-        # it into the parent type
+        # TODO: figure out how to enforce min tick spacing by passing it
-        digits = max(float_digits(spacing * scale), self._min_tick)
+        # into the parent type
+        digits = max(
+            float_digits(spacing * scale),
+            self._min_tick,
+        )
+        if self.title:
+            self.title.update()
 
         # print(f'vals: {vals}\nscale: {scale}\nspacing: {spacing}')
         # print(f'digits: {digits}')
 
+        if not self.formatter:
             return [
                 ('{value:,.{digits}f}').format(
                     digits=digits,
                     value=v,
                 ).replace(',', ' ') for v in vals
             ]
+        else:
+            return list(map(self.formatter, vals))
 
 
 class DynamicDateAxis(Axis):
@@ -136,6 +201,7 @@ class DynamicDateAxis(Axis):
     def _indexes_to_timestrs(
         self,
         indexes: List[int],
+
     ) -> List[str]:
 
         chart = self.linkedsplits.chart
@@ -165,9 +231,10 @@ class DynamicDateAxis(Axis):
     def tickStrings(
         self,
         values: tuple[float],
-        scale,
+        scale: float,
-        spacing,
+        spacing: float,
-    ):
+
+    ) -> list[str]:
         # info = self.tickStrings.cache_info()
         # print(info)
         return self._indexes_to_timestrs(values)
@@ -220,6 +287,8 @@ class AxisLabel(pg.GraphicsObject):
         self.path = None
         self.rect = None
 
+        self._pw = self.pixelWidth()
+
     def paint(
         self,
         p: QtGui.QPainter,
@@ -269,9 +338,10 @@ class AxisLabel(pg.GraphicsObject):
 
 
     def boundingRect(self):  # noqa
-        """Size the graphics space from the text contents.
+        '''
+        Size the graphics space from the text contents.
 
-        """
+        '''
         if self.label_str:
             self._size_br_from_str(self.label_str)
 
@@ -287,23 +357,32 @@ class AxisLabel(pg.GraphicsObject):
 
         return QtCore.QRectF()
 
-        # return self.rect or QtCore.QRectF()
+    # TODO: but the input probably needs to be the "len" of
+    # the current text value:
+    @lru_cache
+    def _size_br_from_str(
+        self,
+        value: str
 
-    def _size_br_from_str(self, value: str) -> None:
+    ) -> tuple[float, float]:
-        """Do our best to render the bounding rect to a set margin
+        '''
+        Do our best to render the bounding rect to a set margin
         around provided string contents.
 
-        """
+        '''
         # size the filled rect to text and/or parent axis
         # if not self._txt_br:
-        #     # XXX: this can't be c
+        #     # XXX: this can't be called until stuff is rendered?
         #     self._txt_br = self._dpifont.boundingRect(value)
 
         txt_br = self._txt_br = self._dpifont.boundingRect(value)
         txt_h, txt_w = txt_br.height(), txt_br.width()
+        # print(f'wsw: {self._dpifont.boundingRect(" ")}')
 
         # allow subtypes to specify a static width and height
         h, w = self.size_hint()
+        # print(f'axis size: {self._parent.size()}')
+        # print(f'axis geo: {self._parent.geometry()}')
 
         self.rect = QtCore.QRectF(
             0, 0,
@@ -314,7 +393,7 @@ class AxisLabel(pg.GraphicsObject):
         # hb = self.path.controlPointRect()
         # hb_size = hb.size()
 
-        return self.rect
+        return (self.rect.width(), self.rect.height())
 
 # _common_text_flags = (
 #     QtCore.Qt.TextDontClip |
@@ -342,6 +421,7 @@ class XAxisLabel(AxisLabel):
         abs_pos: QPointF,  # scene coords
         value: float,  # data for text
         offset: int = 0  # if have margins, k?
+
     ) -> None:
 
         timestrs = self._parent._indexes_to_timestrs([int(value)])
@@ -356,17 +436,19 @@ class XAxisLabel(AxisLabel):
 
         w = self.boundingRect().width()
 
-        self.setPos(QPointF(
+        self.setPos(
-            abs_pos.x() - w/2,
+            QPointF(
+                abs_pos.x() - w/2 - self._pw,
                 y_offset/2,
-        ))
+            )
+        )
         self.update()
 
     def _draw_arrow_path(self):
         y_offset = self._parent.style['tickTextOffset'][1]
         path = QtGui.QPainterPath()
         h, w = self.rect.height(), self.rect.width()
-        middle = w/2 - 0.5
+        middle = w/2 - self._pw * 0.5
         aw = h/2
         left = middle - aw
         right = middle + aw
@@ -410,8 +492,12 @@ class YAxisLabel(AxisLabel):
             self.x_offset, y_offset = self._parent.txt_offsets()
 
     def size_hint(self) -> Tuple[float, float]:
-        # size to parent axis width
+        # size to parent axis width(-ish)
-        return None, self._parent.width()
+        wsh = self._dpifont.boundingRect(' ').height() / 2
+        return (
+            None,
+            self._parent.size().width() - wsh,
+        )
 
     def update_label(
         self,
@@ -432,16 +518,19 @@ class YAxisLabel(AxisLabel):
         br = self.boundingRect()
         h = br.height()
 
-        self.setPos(QPointF(
+        self.setPos(
+            QPointF(
                 x_offset,
-            abs_pos.y() - h / 2 - self._y_margin / 2
+                abs_pos.y() - h / 2 - self._pw,
-        ))
+            )
+        )
         self.update()
 
     def update_on_resize(self, vr, r):
-        """Tiis is a ``.sigRangeChanged()`` handler.
+        '''
+        This is a ``.sigRangeChanged()`` handler.
 
-        """
+        '''
         index, last = self._last_datum
         if index is not None:
             self.update_from_data(index, last)
@@ -451,11 +540,13 @@ class YAxisLabel(AxisLabel):
         index: int,
         value: float,
         _save_last: bool = True,
+
     ) -> None:
-        """Update the label's text contents **and** position from
+        '''
+        Update the label's text contents **and** position from
         a view box coordinate datum.
 
-        """
+        '''
         if _save_last:
             self._last_datum = (index, value)
 
@@ -469,7 +560,7 @@ class YAxisLabel(AxisLabel):
         path = QtGui.QPainterPath()
         h = self.rect.height()
         path.moveTo(0, 0)
-        path.lineTo(-x_offset - h/4, h/2.)
+        path.lineTo(-x_offset - h/4, h/2. - self._pw/2)
         path.lineTo(0, h)
         path.closeSubpath()
         self.path = path

piker/ui/_chart.py

@@ -19,8 +19,6 @@ High level chart-widget apis.
 
 '''
 from __future__ import annotations
-from functools import partial
-from dataclasses import dataclass
 from typing import Optional
 
 from PyQt5 import QtCore, QtWidgets
@@ -383,8 +381,9 @@ class LinkedSplits(QWidget):
 
         style: str = 'bar',
 
-    ) -> 'ChartPlotWidget':
+    ) -> ChartPlotWidget:
-        '''Start up and show main (price) chart and all linked subcharts.
+        '''
+        Start up and show main (price) chart and all linked subcharts.
 
         The data input struct array must include OHLC fields.
 
@@ -480,14 +479,20 @@ class LinkedSplits(QWidget):
             axisItems=axes,
             **cpw_kwargs,
         )
+        cpw.hideAxis('left')
+        cpw.hideAxis('bottom')
 
         if self.xaxis_chart:
+            self.xaxis_chart.hideAxis('bottom')
+
             # presuming we only want it at the true bottom of all charts.
             # XXX: uses new api from our ``pyqtgraph`` fork.
             # https://github.com/pikers/pyqtgraph/tree/plotitemoverlay_onto_pg_master
-            _ = self.xaxis_chart.removeAxis('bottom', unlink=False)
+            # _ = self.xaxis_chart.removeAxis('bottom', unlink=False)
-            assert 'bottom' not in self.xaxis_chart.plotItem.axes
+            # assert 'bottom' not in self.xaxis_chart.plotItem.axes
+
             self.xaxis_chart = cpw
+            cpw.showAxis('bottom')
 
         if self.xaxis_chart is None:
             self.xaxis_chart = cpw
@@ -537,7 +542,7 @@ class LinkedSplits(QWidget):
             )
             self.cursor.contents_labels.add_label(
                 cpw,
-                'ohlc',
+                name,
                 anchor_at=('top', 'left'),
                 update_func=ContentsLabel.update_from_ohlc,
             )
@@ -611,11 +616,11 @@ class LinkedSplits(QWidget):
 
 # import pydantic
 
-# class ArrayScene(pydantic.BaseModel):
+# class Graphics(pydantic.BaseModel):
 #     '''
 #     Data-AGGRegate: high level API onto multiple (categorized)
-#     ``ShmArray``s with high level processing routines mostly for
+#     ``ShmArray``s with high level processing routines for
-#     graphics summary and display.
+#     graphics computations and display.
 
 #     '''
 #     arrays: dict[str, np.ndarray] = {}
@@ -727,18 +732,13 @@ class ChartPlotWidget(pg.PlotWidget):
         self._static_yrange = static_yrange  # for "known y-range style"
         self._view_mode: str = 'follow'
 
-        # show only right side axes
-        self.hideAxis('left')
-        self.showAxis('right')
-        # self.showAxis('left')
-
         # show background grid
         self.showGrid(x=False, y=True, alpha=0.3)
 
         self.default_view()
         self.cv.enable_auto_yrange()
 
-        self.overlay: PlotItemOverlay = PlotItemOverlay(self.plotItem)
+        self.pi_overlay: PlotItemOverlay = PlotItemOverlay(self.plotItem)
 
     def resume_all_feeds(self):
         for feed in self._feeds.values():
@@ -849,7 +849,7 @@ class ChartPlotWidget(pg.PlotWidget):
 
         # adds all bar/candle graphics objects for each data point in
         # the np array buffer to be drawn on next render cycle
-        self.addItem(graphics)
+        self.plotItem.addItem(graphics)
 
         # draw after to allow self.scene() to work...
         graphics.draw_from_data(data)
@@ -860,6 +860,62 @@ class ChartPlotWidget(pg.PlotWidget):
 
         return graphics, data_key
 
+    def overlay_plotitem(
+        self,
+        name: str,
+        index: Optional[int] = None,
+        axis_title: Optional[str] = None,
+        axis_side: str = 'right',
+        axis_kwargs: dict = {},
+
+    ) -> pg.PlotItem:
+
+        # Custom viewbox impl
+        cv = self.mk_vb(name)
+        cv.chart = self
+
+        allowed_sides = {'left', 'right'}
+        if axis_side not in allowed_sides:
+            raise ValueError(f'``axis_side``` must be in {allowed_sides}')
+        yaxis = PriceAxis(
+            orientation=axis_side,
+            linkedsplits=self.linked,
+            **axis_kwargs,
+        )
+
+        pi = pg.PlotItem(
+            parent=self.plotItem,
+            name=name,
+            enableMenu=False,
+            viewBox=cv,
+            axisItems={
+                # 'bottom': xaxis,
+                axis_side: yaxis,
+            },
+            default_axes=[],
+        )
+        pi.hideButtons()
+
+        cv.enable_auto_yrange()
+
+        # compose this new plot's graphics with the current chart's
+        # existing one but with separate axes as neede and specified.
+        self.pi_overlay.add_plotitem(
+            pi,
+            index=index,
+
+            # only link x-axes,
+            link_axes=(0,),
+        )
+
+        # add axis title
+        # TODO: do we want this API to still work?
+        # raxis = pi.getAxis('right')
+        axis = self.pi_overlay.get_axis(pi, axis_side)
+        axis.set_title(axis_title or name, view=pi.getViewBox())
+
+        return pi
+
     def draw_curve(
         self,
 
@@ -868,7 +924,6 @@ class ChartPlotWidget(pg.PlotWidget):
 
         array_key: Optional[str] = None,
         overlay: bool = False,
-        separate_axes: bool = False,
         color: Optional[str] = None,
         add_label: bool = True,
 
@@ -907,11 +962,11 @@ class ChartPlotWidget(pg.PlotWidget):
             **pdi_kwargs,
         )
 
-        # XXX: see explanation for differenct caching modes:
+        # XXX: see explanation for different caching modes:
         # https://stackoverflow.com/a/39410081
         # seems to only be useful if we don't re-generate the entire
         # QPainterPath every time
-        # curve.curve.setCacheMode(QtWidgets.QGraphicsItem.DeviceCoordinateCache)
+        # curve.setCacheMode(QtWidgets.QGraphicsItem.DeviceCoordinateCache)
 
         # don't ever use this - it's a colossal nightmare of artefacts
         # and is disastrous for performance.
@@ -921,83 +976,28 @@ class ChartPlotWidget(pg.PlotWidget):
         self._graphics[name] = curve
         self._arrays[data_key] = data
 
+        pi = self.plotItem
+
+        # TODO: this probably needs its own method?
         if overlay:
             # anchor_at = ('bottom', 'left')
             self._overlays[name] = None
 
-            if separate_axes:
+            if isinstance(overlay, pg.PlotItem):
-
+                if overlay not in self.pi_overlay.overlays:
-                # Custom viewbox impl
+                    raise RuntimeError(
-                cv = self.mk_vb(name)
+                            f'{overlay} must be from `.plotitem_overlay()`'
-
-                def maxmin():
-                    return self.maxmin(name=data_key)
-
-                # ensure view maxmin is computed from correct array
-                # cv._maxmin = partial(self.maxmin, name=data_key)
-
-                cv._maxmin = maxmin
-
-                cv.chart = self
-
-                # xaxis = DynamicDateAxis(
-                #     orientation='bottom',
-                #     linkedsplits=self.linked,
-                # )
-                yaxis = PriceAxis(
-                    orientation='right',
-                    linkedsplits=self.linked,
-                )
-
-                plotitem = pg.PlotItem(
-                    parent=self.plotItem,
-                    name=name,
-                    enableMenu=False,
-                    viewBox=cv,
-                    axisItems={
-                        # 'bottom': xaxis,
-                        'right': yaxis,
-                    },
-                    default_axes=[],
-                )
-                # plotitem.setAxisItems(
-                #     add_to_layout=False,
-                #     axisItems={
-                #         'bottom': xaxis,
-                #         'right': yaxis,
-                #     },
-                # )
-                # plotite.hideAxis('right')
-                # plotite.hideAxis('bottom')
-                plotitem.addItem(curve)
-                cv.enable_auto_yrange()
-
-                # config
-                # plotitem.setAutoVisible(y=True)
-                # plotitem.enableAutoRange(axis='y')
-                plotitem.hideButtons()
-
-                self.overlay.add_plotitem(
-                    plotitem,
-                    # only link x-axes,
-                    link_axes=(0,),
                         )
+                pi = overlay
 
         else:
-                # this intnernally calls `PlotItem.addItem()` on the
-                # graphics object
-                self.addItem(curve)
-        else:
-            # this intnernally calls `PlotItem.addItem()` on the
-            # graphics object
-            self.addItem(curve)
-
             # anchor_at = ('top', 'left')
 
             # TODO: something instead of stickies for overlays
             # (we need something that avoids clutter on x-axis).
             self._add_sticky(name, bg_color=color)
 
+        pi.addItem(curve)
         return curve, data_key
 
     # TODO: make this a ctx mngr
@@ -1020,7 +1020,8 @@ class ChartPlotWidget(pg.PlotWidget):
         # add y-axis "last" value label
         last = self._ysticks[name] = YAxisLabel(
             chart=self,
-            parent=self.getAxis('right'),
+            # parent=self.getAxis('right'),
+            parent=self.pi_overlay.get_axis(self.plotItem, 'right'),
             # TODO: pass this from symbol data
             digits=digits,
             opacity=1,
@@ -1036,7 +1037,8 @@ class ChartPlotWidget(pg.PlotWidget):
         **kwargs,
 
     ) -> pg.GraphicsObject:
-        '''Update the named internal graphics from ``array``.
+        '''
+        Update the named internal graphics from ``array``.
 
         '''
         self._arrays[self.name] = array
@@ -1053,7 +1055,8 @@ class ChartPlotWidget(pg.PlotWidget):
         **kwargs,
 
     ) -> pg.GraphicsObject:
-        '''Update the named internal graphics from ``array``.
+        '''
+        Update the named internal graphics from ``array``.
 
         '''
         assert len(array)
@@ -1123,7 +1126,7 @@ class ChartPlotWidget(pg.PlotWidget):
     def get_index(self, time: float) -> int:
 
         # TODO: this should go onto some sort of
-        # data-view strimg thinger..right?
+        # data-view thinger..right?
         ohlc = self._shm.array
 
         # XXX: not sure why the time is so off here
@@ -1178,15 +1181,9 @@ class ChartPlotWidget(pg.PlotWidget):
             yhigh = np.nanmax(bars['high'])
 
         else:
-            try:
             view = bars[name or self.data_key]
-            except:
-                breakpoint()
-            # if self.data_key != 'volume':
-            # else:
-            #     view = bars
             ylow = np.nanmin(view)
             yhigh = np.nanmax(view)
-            # print(f'{(ylow, yhigh)}')
 
+        # print(f'{(ylow, yhigh)}')
         return ylow, yhigh

piker/ui/_cursor.py

@@ -24,7 +24,7 @@ from typing import Optional, Callable
 import inspect
 import numpy as np
 import pyqtgraph as pg
-from PyQt5 import QtCore, QtGui, QtWidgets
+from PyQt5 import QtCore, QtWidgets
 from PyQt5.QtCore import QPointF, QRectF
 
 from ._style import (
@@ -246,12 +246,16 @@ class ContentsLabels:
         # for name, (label, update) in self._labels.items():
         for chart, name, label, update in self._labels:
 
-            if not (index >= 0 and index < chart._arrays['ohlc'][-1]['index']):
+            array = chart._arrays[name]
+            if not (
+                index >= 0
+                and index < array[-1]['index']
+            ):
                 # out of range
                 print('out of range?')
                 continue
 
-            array = chart._arrays[name]
+            # array = chart._arrays[name]
 
             # call provided update func with data point
             try:
@@ -365,7 +369,13 @@ class Cursor(pg.GraphicsObject):
         self,
         plot: 'ChartPlotWidget',  # noqa
         digits: int = 0,
+
     ) -> None:
+        '''
+        Add chart to tracked set such that a cross-hair and possibly
+        curve tracking cursor can be drawn on the plot.
+
+        '''
         # add ``pg.graphicsItems.InfiniteLine``s
         # vertical and horizonal lines and a y-axis label
 
@@ -378,7 +388,8 @@ class Cursor(pg.GraphicsObject):
 
         yl = YAxisLabel(
             chart=plot,
-            parent=plot.getAxis('right'),
+            # parent=plot.getAxis('right'),
+            parent=plot.pi_overlay.get_axis(plot.plotItem, 'right'),
             digits=digits or self.digits,
             opacity=_ch_label_opac,
             bg_color=self.label_color,
@@ -420,19 +431,25 @@ class Cursor(pg.GraphicsObject):
 
         # ONLY create an x-axis label for the cursor
         # if this plot owns the 'bottom' axis.
-        if 'bottom' in plot.plotItem.axes:
+        # if 'bottom' in plot.plotItem.axes:
-            self.xaxis_label = XAxisLabel(
+        if plot.linked.xaxis_chart is plot:
+            xlabel = self.xaxis_label = XAxisLabel(
                 parent=self.plots[plot_index].getAxis('bottom'),
+                # parent=self.plots[plot_index].pi_overlay.get_axis(plot.plotItem, 'bottom'),
                 opacity=_ch_label_opac,
                 bg_color=self.label_color,
             )
             # place label off-screen during startup
-            self.xaxis_label.setPos(self.plots[0].mapFromView(QPointF(0, 0)))
+            xlabel.setPos(
+                self.plots[0].mapFromView(QPointF(0, 0))
+            )
+            xlabel.show()
 
     def add_curve_cursor(
         self,
         plot: 'ChartPlotWidget',  # noqa
         curve: 'PlotCurveItem',  # noqa
+
     ) -> LineDot:
         # if this plot contains curves add line dot "cursors" to denote
         # the current sample under the mouse
@@ -462,12 +479,15 @@ class Cursor(pg.GraphicsObject):
 
     def mouseMoved(
         self,
-        evt: 'tuple[QMouseEvent]',  # noqa
+        coords: tuple[QPointF],  # noqa
-    ) -> None:  # noqa
+
-        """Update horizonal and vertical lines when mouse moves inside
+    ) -> None:
+        '''
+        Update horizonal and vertical lines when mouse moves inside
         either the main chart or any indicator subplot.
-        """
+
-        pos = evt[0]
+        '''
+        pos = coords[0]
 
         # find position inside active plot
         try:
@@ -486,24 +506,27 @@ class Cursor(pg.GraphicsObject):
 
         ix = round(x)  # since bars are centered around index
 
+        # px perfect...
+        line_offset = self._lw / 2
+
         # round y value to nearest tick step
         m = self._y_incr_mult
         iy = round(y * m) / m
-
+        vl_y = iy - line_offset
-        # px perfect...
-        line_offset = self._lw / 2
 
         # update y-range items
         if iy != last_iy:
 
             if self._y_label_update:
                 self.graphics[self.active_plot]['yl'].update_label(
-                    abs_pos=plot.mapFromView(QPointF(ix, iy)),
+                    # abs_pos=plot.mapFromView(QPointF(ix, iy)),
+                    abs_pos=plot.mapFromView(QPointF(ix, vl_y)),
                     value=iy
                 )
 
                 # only update horizontal xhair line if label is enabled
-                self.graphics[plot]['hl'].setY(iy)
+                # self.graphics[plot]['hl'].setY(iy)
+                self.graphics[plot]['hl'].setY(vl_y)
 
             # update all trackers
             for item in self._trackers:
@@ -516,28 +539,34 @@ class Cursor(pg.GraphicsObject):
                 # with cursor movement
                 self.contents_labels.update_labels(ix)
 
+            vl_x = ix + line_offset
             for plot, opts in self.graphics.items():
 
-                # update the chart's "contents" label
-                # plot.update_contents_labels(ix)
-
                 # move the vertical line to the current "center of bar"
-                opts['vl'].setX(ix + line_offset)
+                opts['vl'].setX(vl_x)
 
                 # update all subscribed curve dots
                 for cursor in opts.get('cursors', ()):
                     cursor.setIndex(ix)
 
                 # update the label on the bottom of the crosshair
-                if 'bottom' in plot.plotItem.axes:
+                axes = plot.plotItem.axes
-                    self.xaxis_label.update_label(
 
-                        # XXX: requires:
+                # TODO: make this an up-front calc that we update
-                        # https://github.com/pyqtgraph/pyqtgraph/pull/1418
+                # on axis-widget resize events.
-                        # otherwise gobbles tons of CPU..
+                # left axis offset width for calcuating
+                # absolute x-axis label placement.
+                left_axis_width = 0
+                left = axes.get('left')
+                if left:
+                    left_axis_width = left['item'].width()
 
                 # map back to abs (label-local) coordinates
-                        abs_pos=plot.mapFromView(QPointF(ix + line_offset, iy)),
+                self.xaxis_label.update_label(
+                    abs_pos=(
+                        plot.mapFromView(QPointF(vl_x, iy)) -
+                        QPointF(left_axis_width, 0)
+                    ),
                     value=ix,
                 )
 

piker/ui/_display.py

@@ -21,36 +21,32 @@ this module ties together quote and computational (fsp) streams with
 graphics update methods via our custom ``pyqtgraph`` charting api.
 
 '''
-from contextlib import asynccontextmanager as acm
 from functools import partial
-from itertools import cycle
 import time
-from types import ModuleType
+from typing import Optional
-from typing import Optional, AsyncGenerator
 
 import numpy as np
-from pydantic import create_model
-import pyqtgraph as pg
 import tractor
 import trio
 
 from .. import brokers
-from .._cacheables import maybe_open_context
-from tractor.trionics import gather_contexts
 from ..data.feed import open_feed, Feed
 from ._chart import (
     ChartPlotWidget,
     LinkedSplits,
     GodWidget,
 )
-from .. import fsp
 from ._l1 import L1Labels
-from ..data._sharedmem import ShmArray, maybe_open_shm_array
+from ._fsp import (
+    update_fsp_chart,
+    start_fsp_displays,
+    has_vlm,
+    open_vlm_displays,
+)
+from ..data._sharedmem import ShmArray, try_read
 from ._forms import (
     FieldsForm,
-    mk_form,
     mk_order_pane_layout,
-    open_form_input_handling,
 )
 from .order_mode import open_order_mode
 from ..log import get_logger
@@ -61,78 +57,6 @@ log = get_logger(__name__)
 _quote_throttle_rate: int = 58  # Hz
 
 
-def try_read(
-    array: np.ndarray
-
-) -> Optional[np.ndarray]:
-    '''
-    Try to read the last row from a shared mem array or ``None``
-    if the array read returns a zero-length array result.
-
-    Can be used to check for backfilling race conditions where an array
-    is currently being (re-)written by a writer actor but the reader is
-    unaware and reads during the window where the first and last indexes
-    are being updated.
-
-    '''
-    try:
-        return array[-1]
-    except IndexError:
-        # XXX: race condition with backfilling shm.
-        #
-        # the underlying issue is that a backfill (aka prepend) and subsequent
-        # shm array first/last index update could result in an empty array
-        # read here since the indices may be updated in such a way that
-        # a read delivers an empty array (though it seems like we
-        # *should* be able to prevent that?). also, as and alt and
-        # something we need anyway, maybe there should be some kind of
-        # signal that a prepend is taking place and this consumer can
-        # respond (eg. redrawing graphics) accordingly.
-
-        # the array read was emtpy
-        return None
-
-
-def update_fsp_chart(
-    chart: ChartPlotWidget,
-    shm: ShmArray,
-    graphics_name: str,
-    array_key: Optional[str],
-
-) -> None:
-
-    array = shm.array
-    last_row = try_read(array)
-
-    # guard against unreadable case
-    if not last_row:
-        log.warning(f'Read-race on shm array: {graphics_name}@{shm.token}')
-        return
-
-    # update graphics
-    # NOTE: this does a length check internally which allows it
-    # staying above the last row check below..
-    chart.update_curve_from_array(
-        graphics_name,
-        array,
-        array_key=array_key or graphics_name,
-    )
-    chart.cv._set_yrange()
-
-    # XXX: re: ``array_key``: fsp func names must be unique meaning we
-    # can't have duplicates of the underlying data even if multiple
-    # sub-charts reference it under different 'named charts'.
-
-    # read from last calculated value and update any label
-    last_val_sticky = chart._ysticks.get(graphics_name)
-    if last_val_sticky:
-        # array = shm.array[array_key]
-        # if len(array):
-        #     value = array[-1]
-        last = last_row[array_key]
-        last_val_sticky.update_from_data(-1, last)
-
-
 # a working tick-type-classes template
 _tick_groups = {
     'clears': {'trade', 'utrade', 'last'},
@@ -151,6 +75,10 @@ def chart_maxmin(
     float,
     float,
 ]:
+    '''
+    Compute max and min datums "in view" for range limits.
+
+    '''
     # TODO: implement this
     # https://arxiv.org/abs/cs/0610046
     # https://github.com/lemire/pythonmaxmin
@@ -178,7 +106,7 @@ def chart_maxmin(
     return last_bars_range, mx, max(mn, 0), mx_vlm_in_view
 
 
-async def update_chart_from_quotes(
+async def update_linked_charts_graphics(
     linked: LinkedSplits,
     stream: tractor.MsgStream,
     ohlcv: np.ndarray,
@@ -187,12 +115,14 @@ async def update_chart_from_quotes(
     vlm_chart: Optional[ChartPlotWidget] = None,
 
 ) -> None:
-    '''The 'main' (price) chart real-time update loop.
+    '''
+    The 'main' (price) chart real-time update loop.
 
-    Receive from the quote stream and update the OHLC chart.
+    Receive from the primary instrument quote stream and update the OHLC
+    chart.
 
     '''
-    # TODO: bunch of stuff:
+    # TODO: bunch of stuff (some might be done already, can't member):
     # - I'm starting to think all this logic should be
     #   done in one place and "graphics update routines"
     #   should not be doing any length checking and array diffing.
@@ -252,19 +182,44 @@ async def update_chart_from_quotes(
     view = chart.view
     last_quote = time.time()
 
+    # async def iter_drain_quotes():
+    #     # NOTE: all code below this loop is expected to be synchronous
+    #     # and thus draw instructions are not picked up jntil the next
+    #     # wait / iteration.
+    #     async for quotes in stream:
+    #         while True:
+    #             try:
+    #                 moar = stream.receive_nowait()
+    #             except trio.WouldBlock:
+    #                 yield quotes
+    #                 break
+    #             else:
+    #                 for sym, quote in moar.items():
+    #                     ticks_frame = quote.get('ticks')
+    #                     if ticks_frame:
+    #                         quotes[sym].setdefault(
+    #                             'ticks', []).extend(ticks_frame)
+    #                     print('pulled extra')
+
+    #                 yield quotes
+
+    # async for quotes in iter_drain_quotes():
+
     async for quotes in stream:
 
-        now = time.time()
         quote_period = time.time() - last_quote
         quote_rate = round(
             1/quote_period, 1) if quote_period > 0 else float('inf')
-
         if (
             quote_period <= 1/_quote_throttle_rate
-            and quote_rate > _quote_throttle_rate * 1.5
+
+            # in the absolute worst case we shouldn't see more then
+            # twice the expected throttle rate right!?
+            and quote_rate >= _quote_throttle_rate * 1.5
         ):
             log.warning(f'High quote rate {symbol.key}: {quote_rate}')
-        last_quote = now
+
+        last_quote = time.time()
 
         # chart isn't active/shown so skip render cycle and pause feed(s)
         if chart.linked.isHidden():
@@ -297,12 +252,20 @@ async def update_chart_from_quotes(
                     mx_vlm_in_view != last_mx_vlm or
                     mx_vlm_in_view > last_mx_vlm
                 ):
-                    print(f'mx vlm: {last_mx_vlm} -> {mx_vlm_in_view}')
+                    # print(f'mx vlm: {last_mx_vlm} -> {mx_vlm_in_view}')
                     vlm_view._set_yrange(
                         yrange=(0, mx_vlm_in_view * 1.375)
                     )
                     last_mx_vlm = mx_vlm_in_view
 
+                for curve_name, shm in vlm_chart._overlays.items():
+                    update_fsp_chart(
+                        vlm_chart,
+                        shm,
+                        curve_name,
+                        array_key=curve_name,
+                    )
+
             ticks_frame = quote.get('ticks', ())
 
             frames_by_type: dict[str, dict] = {}
@@ -420,7 +383,7 @@ async def update_chart_from_quotes(
                 (mx > last_mx) or (mn < last_mn)
                 and not chart._static_yrange == 'axis'
             ):
-                print(f'new y range: {(mn, mx)}')
+                # print(f'new y range: {(mn, mx)}')
                 view._set_yrange(
                     yrange=(mn, mx),
                     # TODO: we should probably scale
@@ -458,382 +421,6 @@ async def update_chart_from_quotes(
                 # chart._set_yrange()
 
 
-def maybe_mk_fsp_shm(
-    sym: str,
-    field_name: str,
-    display_name: Optional[str] = None,
-    readonly: bool = True,
-
-) -> (ShmArray, bool):
-    '''
-    Allocate a single row shm array for an symbol-fsp pair if none
-    exists, otherwise load the shm already existing for that token.
-
-    '''
-    uid = tractor.current_actor().uid
-    if not display_name:
-        display_name = field_name
-
-    # TODO: load function here and introspect
-    # return stream type(s)
-
-    # TODO: should `index` be a required internal field?
-    fsp_dtype = np.dtype([('index', int), (field_name, float)])
-
-    key = f'{sym}.fsp.{display_name}.{".".join(uid)}'
-
-    shm, opened = maybe_open_shm_array(
-        key,
-        # TODO: create entry for each time frame
-        dtype=fsp_dtype,
-        readonly=True,
-    )
-    return shm, opened
-
-
-@acm
-async def open_fsp_sidepane(
-    linked: LinkedSplits,
-    conf: dict[str, dict[str, str]],
-
-) -> FieldsForm:
-
-    schema = {}
-
-    assert len(conf) == 1  # for now
-
-    # add (single) selection widget
-    for display_name, config in conf.items():
-        schema[display_name] = {
-                'label': '**fsp**:',
-                'type': 'select',
-                'default_value': [display_name],
-            }
-
-        # add parameters for selection "options"
-        params = config.get('params', {})
-        for name, config in params.items():
-
-            default = config['default_value']
-            kwargs = config.get('widget_kwargs', {})
-
-            # add to ORM schema
-            schema.update({
-                name: {
-                    'label': f'**{name}**:',
-                    'type': 'edit',
-                    'default_value': default,
-                    'kwargs': kwargs,
-                },
-            })
-
-    sidepane: FieldsForm = mk_form(
-        parent=linked.godwidget,
-        fields_schema=schema,
-    )
-
-    # https://pydantic-docs.helpmanual.io/usage/models/#dynamic-model-creation
-    FspConfig = create_model(
-        'FspConfig',
-        name=display_name,
-        **params,
-    )
-    sidepane.model = FspConfig()
-
-    # just a logger for now until we get fsp configs up and running.
-    async def settings_change(key: str, value: str) -> bool:
-        print(f'{key}: {value}')
-        return True
-
-    # TODO:
-    async with (
-        open_form_input_handling(
-            sidepane,
-            focus_next=linked.godwidget,
-            on_value_change=settings_change,
-        )
-    ):
-        yield sidepane
-
-
-@acm
-async def open_fsp_cluster(
-    workers: int = 2
-
-) -> AsyncGenerator[int, dict[str, tractor.Portal]]:
-
-    from tractor._clustering import open_actor_cluster
-
-    profiler = pg.debug.Profiler(
-        delayed=False,
-        disabled=False
-    )
-    async with open_actor_cluster(
-        count=2,
-        names=['fsp_0', 'fsp_1'],
-        modules=['piker.fsp._engine'],
-    ) as cluster_map:
-        profiler('started fsp cluster')
-        yield cluster_map
-
-
-@acm
-async def maybe_open_fsp_cluster(
-    workers: int = 2,
-    **kwargs,
-
-) -> AsyncGenerator[int, dict[str, tractor.Portal]]:
-
-    kwargs.update(
-        {'workers': workers}
-    )
-
-    async with maybe_open_context(
-        # for now make a cluster per client?
-        acm_func=open_fsp_cluster,
-        kwargs=kwargs,
-    ) as (cache_hit, cluster_map):
-        if cache_hit:
-            log.info('re-using existing fsp cluster')
-            yield cluster_map
-        else:
-            yield cluster_map
-
-
-async def start_fsp_displays(
-    cluster_map: dict[str, tractor.Portal],
-    linkedsplits: LinkedSplits,
-    fsps: dict[str, str],
-    sym: str,
-    src_shm: list,
-    brokermod: ModuleType,
-    group_status_key: str,
-    loglevel: str,
-
-) -> None:
-    '''
-    Create sub-actors (under flat tree)
-    for each entry in config and attach to local graphics update tasks.
-
-    Pass target entrypoint and historical data.
-
-    '''
-    linkedsplits.focus()
-
-    profiler = pg.debug.Profiler(
-        delayed=False,
-        disabled=False
-    )
-
-    async with trio.open_nursery() as n:
-        # Currently we spawn an actor per fsp chain but
-        # likely we'll want to pool them eventually to
-        # scale horizonatlly once cores are used up.
-        for (display_name, conf), (name, portal) in zip(
-            fsps.items(),
-
-            # round robin to cluster for now..
-            cycle(cluster_map.items()),
-        ):
-            func_name = conf['func_name']
-            shm, opened = maybe_mk_fsp_shm(
-                sym,
-                field_name=func_name,
-                display_name=display_name,
-                readonly=True,
-            )
-
-            profiler(f'created shm for fsp actor: {display_name}')
-
-            # XXX: fsp may have been opened by a duplicate chart.
-            # Error for now until we figure out how to wrap fsps as
-            # "feeds".  assert opened, f"A chart for {key} likely
-            # already exists?"
-
-            profiler(f'attached to fsp portal: {display_name}')
-
-            # init async
-            n.start_soon(
-                partial(
-                    update_chart_from_fsp,
-
-                    portal,
-                    linkedsplits,
-                    brokermod,
-                    sym,
-                    src_shm,
-                    func_name,
-                    display_name,
-                    conf=conf,
-                    shm=shm,
-                    is_overlay=conf.get('overlay', False),
-                    group_status_key=group_status_key,
-                    loglevel=loglevel,
-                    profiler=profiler,
-                )
-            )
-
-    # blocks here until all fsp actors complete
-
-
-async def update_chart_from_fsp(
-    portal: tractor.Portal,
-    linkedsplits: LinkedSplits,
-    brokermod: ModuleType,
-    sym: str,
-    src_shm: ShmArray,
-    func_name: str,
-    display_name: str,
-    conf: dict[str, dict],
-
-    shm: ShmArray,
-    is_overlay: bool,
-
-    group_status_key: str,
-    loglevel: str,
-    profiler: pg.debug.Profiler,
-
-) -> None:
-    '''
-    FSP stream chart update loop.
-
-    This is called once for each entry in the fsp
-    config map.
-
-    '''
-
-    profiler(f'started chart task for fsp: {func_name}')
-
-    done = linkedsplits.window().status_bar.open_status(
-        f'loading fsp, {display_name}..',
-        group_key=group_status_key,
-    )
-
-    async with (
-        portal.open_context(
-
-            # chaining entrypoint
-            fsp.cascade,
-
-            # name as title of sub-chart
-            brokername=brokermod.name,
-            src_shm_token=src_shm.token,
-            dst_shm_token=shm.token,
-            symbol=sym,
-            func_name=func_name,
-            loglevel=loglevel,
-            zero_on_step=conf.get('zero_on_step', False),
-
-        ) as (ctx, last_index),
-        ctx.open_stream() as stream,
-
-        open_fsp_sidepane(linkedsplits, {display_name: conf},) as sidepane,
-    ):
-        profiler(f'fsp:{func_name} attached to fsp ctx-stream')
-
-        if is_overlay:
-            chart = linkedsplits.chart
-            chart.draw_curve(
-                name=display_name,
-                data=shm.array,
-                overlay=True,
-                color='default_light',
-            )
-            # specially store ref to shm for lookup in display loop
-            chart._overlays[display_name] = shm
-
-        else:
-            chart = linkedsplits.add_plot(
-                name=display_name,
-                array=shm.array,
-
-                array_key=conf['func_name'],
-                sidepane=sidepane,
-
-                # curve by default
-                ohlc=False,
-
-                # settings passed down to ``ChartPlotWidget``
-                **conf.get('chart_kwargs', {})
-                # static_yrange=(0, 100),
-            )
-
-            # XXX: ONLY for sub-chart fsps, overlays have their
-            # data looked up from the chart's internal array set.
-            # TODO: we must get a data view api going STAT!!
-            chart._shm = shm
-
-            # should **not** be the same sub-chart widget
-            assert chart.name != linkedsplits.chart.name
-
-        array_key = func_name
-
-        profiler(f'fsp:{func_name} chart created')
-
-        # first UI update, usually from shm pushed history
-        update_fsp_chart(
-            chart,
-            shm,
-            display_name,
-            array_key=array_key,
-        )
-
-        chart.linked.focus()
-
-        # TODO: figure out if we can roll our own `FillToThreshold` to
-        # get brush filled polygons for OS/OB conditions.
-        # ``pg.FillBetweenItems`` seems to be one technique using
-        # generic fills between curve types while ``PlotCurveItem`` has
-        # logic inside ``.paint()`` for ``self.opts['fillLevel']`` which
-        # might be the best solution?
-        # graphics = chart.update_from_array(chart.name, array[func_name])
-        # graphics.curve.setBrush(50, 50, 200, 100)
-        # graphics.curve.setFillLevel(50)
-
-        if func_name == 'rsi':
-            from ._lines import level_line
-            # add moveable over-[sold/bought] lines
-            # and labels only for the 70/30 lines
-            level_line(chart, 20)
-            level_line(chart, 30, orient_v='top')
-            level_line(chart, 70, orient_v='bottom')
-            level_line(chart, 80, orient_v='top')
-
-        chart.cv._set_yrange()
-        done()  # status updates
-
-        profiler(f'fsp:{func_name} starting update loop')
-        profiler.finish()
-
-        # update chart graphics
-        last = time.time()
-        async for value in stream:
-
-            # chart isn't actively shown so just skip render cycle
-            if chart.linked.isHidden():
-                continue
-
-            else:
-                now = time.time()
-                period = now - last
-
-                if period <= 1/_quote_throttle_rate:
-                    # faster then display refresh rate
-                    print(f'fsp too fast: {1/period}')
-                    continue
-
-                # run synchronous update
-                update_fsp_chart(
-                    chart,
-                    shm,
-                    display_name,
-                    array_key=func_name,
-                )
-
-                # set time of last graphics update
-                last = time.time()
-
-
 async def check_for_new_bars(
     feed: Feed,
     ohlcv: np.ndarray,
@@ -908,93 +495,6 @@ async def check_for_new_bars(
             price_chart.increment_view()
 
 
-def has_vlm(ohlcv: ShmArray) -> bool:
-    # make sure that the instrument supports volume history
-    # (sometimes this is not the case for some commodities and
-    # derivatives)
-    volm = ohlcv.array['volume']
-    return not bool(np.all(np.isin(volm, -1)) or np.all(np.isnan(volm)))
-
-
-@acm
-async def maybe_open_vlm_display(
-    linked: LinkedSplits,
-    ohlcv: ShmArray,
-
-) -> ChartPlotWidget:
-
-    if not has_vlm(ohlcv):
-        log.warning(f"{linked.symbol.key} does not seem to have volume info")
-        yield
-        return
-    else:
-
-        shm, opened = maybe_mk_fsp_shm(
-            linked.symbol.key,
-            'vlm',
-            readonly=True,
-        )
-
-        async with open_fsp_sidepane(
-            linked, {
-                'vlm': {
-                    'params': {
-                        'price_func': {
-                            'default_value': 'chl3',
-                            # tell target ``Edit`` widget to not allow
-                            # edits for now.
-                            'widget_kwargs': {'readonly': True},
-                        },
-                    },
-                }
-            },
-        ) as sidepane:
-
-            # built-in $vlm
-            shm = ohlcv
-            chart = linked.add_plot(
-                name='volume',
-                array=shm.array,
-
-                array_key='volume',
-                sidepane=sidepane,
-
-                # curve by default
-                ohlc=False,
-
-                # Draw vertical bars from zero.
-                # we do this internally ourselves since
-                # the curve item internals are pretty convoluted.
-                style='step',
-            )
-
-            # XXX: ONLY for sub-chart fsps, overlays have their
-            # data looked up from the chart's internal array set.
-            # TODO: we must get a data view api going STAT!!
-            chart._shm = shm
-
-            # should **not** be the same sub-chart widget
-            assert chart.name != linked.chart.name
-
-            # sticky only on sub-charts atm
-            last_val_sticky = chart._ysticks[chart.name]
-
-            # read from last calculated value
-            value = shm.array['volume'][-1]
-
-            last_val_sticky.update_from_data(-1, value)
-
-            chart.update_curve_from_array(
-                'volume',
-                shm.array,
-            )
-
-            # size view to data once at outset
-            chart.cv._set_yrange()
-
-            yield chart
-
-
 async def display_symbol_data(
     godwidget: GodWidget,
     provider: str,
@@ -1084,57 +584,6 @@ async def display_symbol_data(
         # TODO: a data view api that makes this less shit
         chart._shm = ohlcv
 
-        # TODO: eventually we'll support some kind of n-compose syntax
-        fsp_conf = {
-
-            # 'dolla_vlm': {
-            #     'func_name': 'dolla_vlm',
-            #     'zero_on_step': True,
-            #     'params': {
-            #         'price_func': {
-            #             'default_value': 'chl3',
-            #             # tell target ``Edit`` widget to not allow
-            #             # edits for now.
-            #             'widget_kwargs': {'readonly': True},
-            #         },
-            #     },
-            #     'chart_kwargs': {'style': 'step'}
-            # },
-
-            # 'rsi': {
-            #     'func_name': 'rsi',  # literal python func ref lookup name
-
-            #     # map of parameters to place on the fsp sidepane widget
-            #     # which should map to dynamic inputs available to the
-            #     # fsp function at runtime.
-            #     'params': {
-            #         'period': {
-            #             'default_value': 14,
-            #             'widget_kwargs': {'readonly': True},
-            #         },
-            #     },
-
-            #     # ``ChartPlotWidget`` options passthrough
-            #     'chart_kwargs': {
-            #         'static_yrange': (0, 100),
-            #     },
-            # },
-        }
-
-        if has_vlm(ohlcv):  # and provider != 'binance':
-            # binance is too fast atm for FSPs until we wrap
-            # the fsp streams as throttled ``Feeds``, see
-            #
-
-            # add VWAP to fsp config for downstream loading
-            fsp_conf.update({
-                'vwap': {
-                    'func_name': 'vwap',
-                    'overlay': True,
-                    'anchor': 'session',
-                },
-            })
-
         # NOTE: we must immediately tell Qt to show the OHLC chart
         # to avoid a race where the subplots get added/shown to
         # the linked set *before* the main price chart!
@@ -1142,32 +591,30 @@ async def display_symbol_data(
         linkedsplits.focus()
         await trio.sleep(0)
 
-        vlm_chart = None
+        vlm_chart: Optional[ChartPlotWidget] = None
+        async with trio.open_nursery() as ln:
 
-        async with gather_contexts(
+            # if available load volume related built-in display(s)
-            (
+            if has_vlm(ohlcv):
-                trio.open_nursery(),
+                vlm_chart = await ln.start(
-                maybe_open_vlm_display(linkedsplits, ohlcv),
+                    open_vlm_displays,
-                maybe_open_fsp_cluster(),
+                    linkedsplits,
+                    ohlcv,
                 )
-        ) as (ln, vlm_chart, cluster_map):
 
-            # load initial fsp chain (otherwise known as "indicators")
+            # load (user's) FSP set (otherwise known as "indicators")
+            # from an input config.
             ln.start_soon(
                 start_fsp_displays,
-                cluster_map,
                 linkedsplits,
-                fsp_conf,
-                sym,
                 ohlcv,
-                brokermod,
                 loading_sym_key,
                 loglevel,
             )
 
             # start graphics update loop after receiving first live quote
             ln.start_soon(
-                update_chart_from_quotes,
+                update_linked_charts_graphics,
                 linkedsplits,
                 feed.stream,
                 ohlcv,
@@ -1175,6 +622,7 @@ async def display_symbol_data(
                 vlm_chart,
             )
 
+            # start sample step incrementer
             ln.start_soon(
                 check_for_new_bars,
                 feed,
@@ -1196,5 +644,15 @@ async def display_symbol_data(
                 await trio.sleep(0)
                 linkedsplits.resize_sidepanes()
 
-                # let the app run.
+                # NOTE: we pop the volume chart from the subplots set so
+                # that it isn't double rendered in the display loop
+                # above since we do a maxmin calc on the volume data to
+                # determine if auto-range adjustements should be made.
+                linkedsplits.subplots.pop('volume', None)
+
+                # TODO: make this not so shit XD
+                # close group status
+                sbar._status_groups[loading_sym_key][1]()
+
+                # let the app run.. bby
                 await trio.sleep_forever()

piker/ui/_fsp.py

@@ -0,0 +1,841 @@
+# piker: trading gear for hackers
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
+
+# 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/>.
+
+'''
+FSP UI and graphics components.
+
+Financial signal processing cluster and real-time graphics management.
+
+'''
+from contextlib import asynccontextmanager as acm
+from functools import partial
+from itertools import cycle
+from typing import Optional, AsyncGenerator, Any
+
+import numpy as np
+from pydantic import create_model
+import tractor
+# from tractor.trionics import gather_contexts
+import pyqtgraph as pg
+import trio
+from trio_typing import TaskStatus
+
+from ._axes import PriceAxis
+from .._cacheables import maybe_open_context
+from ..calc import humanize
+from ..data._sharedmem import (
+    ShmArray,
+    maybe_open_shm_array,
+    try_read,
+)
+from ._chart import (
+    ChartPlotWidget,
+    LinkedSplits,
+)
+from .. import fsp
+from ._forms import (
+    FieldsForm,
+    mk_form,
+    open_form_input_handling,
+)
+from ..log import get_logger
+
+log = get_logger(__name__)
+
+
+def maybe_mk_fsp_shm(
+    sym: str,
+    field_name: str,
+    display_name: Optional[str] = None,
+    readonly: bool = True,
+
+) -> (ShmArray, bool):
+    '''
+    Allocate a single row shm array for an symbol-fsp pair if none
+    exists, otherwise load the shm already existing for that token.
+
+    '''
+    uid = tractor.current_actor().uid
+    if not display_name:
+        display_name = field_name
+
+    # TODO: load function here and introspect
+    # return stream type(s)
+
+    # TODO: should `index` be a required internal field?
+    fsp_dtype = np.dtype([('index', int), (field_name, float)])
+
+    key = f'{sym}.fsp.{display_name}.{".".join(uid)}'
+
+    shm, opened = maybe_open_shm_array(
+        key,
+        # TODO: create entry for each time frame
+        dtype=fsp_dtype,
+        readonly=True,
+    )
+    return shm, opened
+
+
+def has_vlm(ohlcv: ShmArray) -> bool:
+    # make sure that the instrument supports volume history
+    # (sometimes this is not the case for some commodities and
+    # derivatives)
+    vlm = ohlcv.array['volume']
+    return not bool(np.all(np.isin(vlm, -1)) or np.all(np.isnan(vlm)))
+
+
+def update_fsp_chart(
+    chart: ChartPlotWidget,
+    shm: ShmArray,
+    graphics_name: str,
+    array_key: Optional[str],
+
+) -> None:
+
+    array = shm.array
+    last_row = try_read(array)
+
+    # guard against unreadable case
+    if not last_row:
+        log.warning(f'Read-race on shm array: {graphics_name}@{shm.token}')
+        return
+
+    # update graphics
+    # NOTE: this does a length check internally which allows it
+    # staying above the last row check below..
+    chart.update_curve_from_array(
+        graphics_name,
+        array,
+        array_key=array_key or graphics_name,
+    )
+
+    # XXX: re: ``array_key``: fsp func names must be unique meaning we
+    # can't have duplicates of the underlying data even if multiple
+    # sub-charts reference it under different 'named charts'.
+
+    # read from last calculated value and update any label
+    last_val_sticky = chart._ysticks.get(graphics_name)
+    if last_val_sticky:
+        # array = shm.array[array_key]
+        # if len(array):
+        #     value = array[-1]
+        last = last_row[array_key]
+        last_val_sticky.update_from_data(-1, last)
+
+
+@acm
+async def open_fsp_sidepane(
+    linked: LinkedSplits,
+    conf: dict[str, dict[str, str]],
+
+) -> FieldsForm:
+
+    schema = {}
+
+    assert len(conf) == 1  # for now
+
+    # add (single) selection widget
+    for display_name, config in conf.items():
+        schema[display_name] = {
+                'label': '**fsp**:',
+                'type': 'select',
+                'default_value': [display_name],
+            }
+
+        # add parameters for selection "options"
+        params = config.get('params', {})
+        for name, config in params.items():
+
+            default = config['default_value']
+            kwargs = config.get('widget_kwargs', {})
+
+            # add to ORM schema
+            schema.update({
+                name: {
+                    'label': f'**{name}**:',
+                    'type': 'edit',
+                    'default_value': default,
+                    'kwargs': kwargs,
+                },
+            })
+
+    sidepane: FieldsForm = mk_form(
+        parent=linked.godwidget,
+        fields_schema=schema,
+    )
+
+    # https://pydantic-docs.helpmanual.io/usage/models/#dynamic-model-creation
+    FspConfig = create_model(
+        'FspConfig',
+        name=display_name,
+        **params,
+    )
+    sidepane.model = FspConfig()
+
+    # just a logger for now until we get fsp configs up and running.
+    async def settings_change(key: str, value: str) -> bool:
+        print(f'{key}: {value}')
+        return True
+
+    # TODO:
+    async with (
+        open_form_input_handling(
+            sidepane,
+            focus_next=linked.godwidget,
+            on_value_change=settings_change,
+        )
+    ):
+        yield sidepane
+
+
+@acm
+async def open_fsp_actor_cluster(
+    names: list[str] = ['fsp_0', 'fsp_1'],
+
+) -> AsyncGenerator[int, dict[str, tractor.Portal]]:
+
+    from tractor._clustering import open_actor_cluster
+
+    # profiler = pg.debug.Profiler(
+    #     delayed=False,
+    #     disabled=False
+    # )
+    async with open_actor_cluster(
+        count=2,
+        names=names,
+        modules=['piker.fsp._engine'],
+
+    ) as cluster_map:
+        # profiler('started fsp cluster')
+        yield cluster_map
+
+
+async def run_fsp_ui(
+
+    linkedsplits: LinkedSplits,
+    shm: ShmArray,
+    started: trio.Event,
+    func_name: str,
+    display_name: str,
+    conf: dict[str, dict],
+    loglevel: str,
+    # profiler: pg.debug.Profiler,
+    # _quote_throttle_rate: int = 58,
+
+) -> None:
+    '''
+    Taskf for UI spawning around a ``LinkedSplits`` chart for fsp
+    related graphics/UX management.
+
+    This is normally spawned/called once for each entry in the fsp
+    config.
+
+    '''
+    # profiler(f'started UI task for fsp: {func_name}')
+
+    async with (
+        # side UI for parameters/controls
+        open_fsp_sidepane(
+            linkedsplits,
+            {display_name: conf},
+        ) as sidepane,
+    ):
+        await started.wait()
+        # profiler(f'fsp:{func_name} attached to fsp ctx-stream')
+
+        overlay_with = conf.get('overlay', False)
+        if overlay_with:
+            if overlay_with == 'ohlc':
+                chart = linkedsplits.chart
+            else:
+                chart = linkedsplits.subplots[overlay_with]
+
+            chart.draw_curve(
+                name=display_name,
+                data=shm.array,
+                overlay=True,
+                color='default_light',
+                array_key=func_name,
+                separate_axes=conf.get('separate_axes', False),
+                **conf.get('chart_kwargs', {})
+            )
+            # specially store ref to shm for lookup in display loop
+            chart._overlays[display_name] = shm
+
+        else:
+            # create a new sub-chart widget for this fsp
+            chart = linkedsplits.add_plot(
+                name=display_name,
+                array=shm.array,
+
+                array_key=func_name,
+                sidepane=sidepane,
+
+                # curve by default
+                ohlc=False,
+
+                # settings passed down to ``ChartPlotWidget``
+                **conf.get('chart_kwargs', {})
+            )
+
+            # XXX: ONLY for sub-chart fsps, overlays have their
+            # data looked up from the chart's internal array set.
+            # TODO: we must get a data view api going STAT!!
+            chart._shm = shm
+
+            # should **not** be the same sub-chart widget
+            assert chart.name != linkedsplits.chart.name
+
+        array_key = func_name
+
+        # profiler(f'fsp:{func_name} chart created')
+
+        # first UI update, usually from shm pushed history
+        update_fsp_chart(
+            chart,
+            shm,
+            display_name,
+            array_key=array_key,
+        )
+
+        chart.linked.focus()
+
+        # TODO: figure out if we can roll our own `FillToThreshold` to
+        # get brush filled polygons for OS/OB conditions.
+        # ``pg.FillBetweenItems`` seems to be one technique using
+        # generic fills between curve types while ``PlotCurveItem`` has
+        # logic inside ``.paint()`` for ``self.opts['fillLevel']`` which
+        # might be the best solution?
+
+        # graphics = chart.update_from_array(chart.name, array[func_name])
+        # graphics.curve.setBrush(50, 50, 200, 100)
+        # graphics.curve.setFillLevel(50)
+
+        # if func_name == 'rsi':
+        #     from ._lines import level_line
+        #     # add moveable over-[sold/bought] lines
+        #     # and labels only for the 70/30 lines
+        #     level_line(chart, 20)
+        #     level_line(chart, 30, orient_v='top')
+        #     level_line(chart, 70, orient_v='bottom')
+        #     level_line(chart, 80, orient_v='top')
+
+        chart.view._set_yrange()
+        # done()  # status updates
+
+        # profiler(f'fsp:{func_name} starting update loop')
+        # profiler.finish()
+
+        # update chart graphics
+        # last = time.time()
+
+        # XXX: this currently doesn't loop since
+        # the FSP engine does **not** push updates atm
+        # since we do graphics update in the main loop
+        # in ``._display.
+        # async for value in stream:
+        #     print(value)
+
+        #     # chart isn't actively shown so just skip render cycle
+        #     if chart.linked.isHidden():
+        #         continue
+
+        #     else:
+        #         now = time.time()
+        #         period = now - last
+
+        #         if period <= 1/_quote_throttle_rate:
+        #             # faster then display refresh rate
+        #             print(f'fsp too fast: {1/period}')
+        #             continue
+
+        #         # run synchronous update
+        #         update_fsp_chart(
+        #             chart,
+        #             shm,
+        #             display_name,
+        #             array_key=func_name,
+        #         )
+
+        #         # set time of last graphics update
+        #         last = time.time()
+
+
+class FspAdmin:
+    '''
+    Client API for orchestrating FSP actors and displaying
+    real-time graphics output.
+
+    '''
+    def __init__(
+        self,
+        tn: trio.Nursery,
+        cluster: dict[str, tractor.Portal],
+        linked: LinkedSplits,
+        src_shm: ShmArray,
+
+    ) -> None:
+        self.tn = tn
+        self.cluster = cluster
+        self.linked = linked
+        self._rr_next_actor = cycle(cluster.items())
+        self._registry: dict[
+            tuple,
+            tuple[tractor.MsgStream, ShmArray]
+        ] = {}
+        self.src_shm = src_shm
+
+    def rr_next_portal(self) -> tractor.Portal:
+        name, portal = next(self._rr_next_actor)
+        return portal
+
+    async def open_chain(
+        self,
+
+        portal: tractor.Portal,
+        complete: trio.Event,
+        started: trio.Event,
+        dst_shm: ShmArray,
+        conf: dict,
+        func_name: str,
+        loglevel: str,
+
+    ) -> None:
+        '''
+        Task which opens a remote FSP endpoint in the managed
+        cluster and sleeps until signalled to exit.
+
+        '''
+        brokername, sym = self.linked.symbol.front_feed()
+        async with (
+            portal.open_context(
+
+                # chaining entrypoint
+                fsp.cascade,
+
+                # data feed key
+                brokername=brokername,
+                symbol=sym,
+
+                # mems
+                src_shm_token=self.src_shm.token,
+                dst_shm_token=dst_shm.token,
+
+                # target
+                func_name=func_name,
+
+                loglevel=loglevel,
+                zero_on_step=conf.get('zero_on_step', False),
+
+            ) as (ctx, last_index),
+            ctx.open_stream() as stream,
+        ):
+            # register output data
+            self._registry[(brokername, sym, func_name)] = (
+                stream, dst_shm, complete)
+
+            started.set()
+
+            # wait for graceful shutdown signal
+            await complete.wait()
+
+    async def start_engine_task(
+        self,
+
+        display_name: str,
+        conf: dict[str, dict[str, Any]],
+
+        worker_name: Optional[str] = None,
+        loglevel: str = 'error',
+
+    ) -> (ShmArray, trio.Event):
+
+        # unpack FSP details from config dict
+        func_name = conf['func_name']
+
+        # allocate an output shm array
+        dst_shm, opened = maybe_mk_fsp_shm(
+            self.linked.symbol.front_feed(),
+            field_name=func_name,
+            display_name=display_name,
+            readonly=True,
+        )
+        if not opened:
+            raise RuntimeError(f'Already started FSP {func_name}')
+
+        portal = self.cluster.get(worker_name) or self.rr_next_portal()
+        complete = trio.Event()
+        started = trio.Event()
+        self.tn.start_soon(
+            self.open_chain,
+
+            portal,
+            complete,
+            started,
+            dst_shm,
+            conf,
+            func_name,
+            loglevel,
+        )
+
+        return dst_shm, started
+
+    async def open_fsp_chart(
+        self,
+        display_name: str,
+        conf: dict,  # yeah probably dumb..
+        loglevel: str = 'error',
+
+    ) -> (trio.Event, ChartPlotWidget):
+
+        func_name = conf['func_name']
+
+        shm, started = await self.start_engine_task(
+            display_name,
+            conf,
+            loglevel,
+        )
+
+        # init async
+        self.tn.start_soon(
+            partial(
+                run_fsp_ui,
+
+                self.linked,
+                shm,
+                started,
+                func_name,
+                display_name,
+
+                conf=conf,
+                loglevel=loglevel,
+            )
+        )
+        return started
+
+
+@acm
+async def open_fsp_admin(
+    linked: LinkedSplits,
+    src_shm: ShmArray,
+    **kwargs,
+
+) -> AsyncGenerator[dict, dict[str, tractor.Portal]]:
+
+    async with (
+        maybe_open_context(
+            # for now make a cluster per client?
+            acm_func=open_fsp_actor_cluster,
+            kwargs=kwargs,
+        ) as (cache_hit, cluster_map),
+
+        trio.open_nursery() as tn,
+    ):
+        if cache_hit:
+            log.info('re-using existing fsp cluster')
+
+        admin = FspAdmin(
+            tn,
+            cluster_map,
+            linked,
+            src_shm,
+        )
+        try:
+            yield admin
+        finally:
+            # terminate all tasks via signals
+            for key, entry in admin._registry.items():
+                _, _, event = entry
+                event.set()
+
+
+async def open_vlm_displays(
+
+    linked: LinkedSplits,
+    ohlcv: ShmArray,
+    dvlm: bool = True,
+
+    task_status: TaskStatus[ChartPlotWidget] = trio.TASK_STATUS_IGNORED,
+
+) -> ChartPlotWidget:
+    '''
+    Volume subchart displays.
+
+    Since "volume" is often included directly alongside OHLCV price
+    data, we don't really need a separate FSP-actor + shm array for it
+    since it's likely already directly adjacent to OHLC samples from the
+    data provider.
+
+    Further only if volume data is detected (it sometimes isn't provided
+    eg. forex, certain commodities markets) will volume dependent FSPs
+    be spawned here.
+
+    '''
+    async with (
+        open_fsp_sidepane(
+            linked, {
+                'vlm': {
+                    'params': {
+                        'price_func': {
+                            'default_value': 'chl3',
+                            # tell target ``Edit`` widget to not allow
+                            # edits for now.
+                            'widget_kwargs': {'readonly': True},
+                        },
+                    },
+                }
+            },
+        ) as sidepane,
+        open_fsp_admin(linked, ohlcv) as admin,
+    ):
+        # built-in vlm which we plot ASAP since it's
+        # usually data provided directly with OHLC history.
+        shm = ohlcv
+        chart = linked.add_plot(
+            name='volume',
+            array=shm.array,
+
+            array_key='volume',
+            sidepane=sidepane,
+
+            # curve by default
+            ohlc=False,
+
+            # Draw vertical bars from zero.
+            # we do this internally ourselves since
+            # the curve item internals are pretty convoluted.
+            style='step',
+        )
+
+        # force 0 to always be in view
+        def maxmin(name) -> tuple[float, float]:
+            mxmn = chart.maxmin(name=name)
+            if mxmn:
+                return 0, mxmn[1]
+
+            return 0, 0
+
+        chart.view._maxmin = partial(maxmin, name='volume')
+
+        # TODO: fix the x-axis label issue where if you put
+        # the axis on the left it's totally not lined up...
+        # show volume units value on LHS (for dinkus)
+        # chart.hideAxis('right')
+        # chart.showAxis('left')
+
+        # XXX: ONLY for sub-chart fsps, overlays have their
+        # data looked up from the chart's internal array set.
+        # TODO: we must get a data view api going STAT!!
+        chart._shm = shm
+
+        # send back new chart to caller
+        task_status.started(chart)
+
+        # should **not** be the same sub-chart widget
+        assert chart.name != linked.chart.name
+
+        # sticky only on sub-charts atm
+        last_val_sticky = chart._ysticks[chart.name]
+
+        # read from last calculated value
+        value = shm.array['volume'][-1]
+
+        last_val_sticky.update_from_data(-1, value)
+
+        vlm_curve = chart.update_curve_from_array(
+            'volume',
+            shm.array,
+        )
+
+        # size view to data once at outset
+        chart.view._set_yrange()
+
+        # add axis title
+        axis = chart.getAxis('right')
+        axis.set_title(' vlm')
+
+        if dvlm:
+
+            # spawn and overlay $ vlm on the same subchart
+            shm, started = await admin.start_engine_task(
+                'dolla_vlm',
+                # linked.symbol.front_feed(),  # data-feed symbol key
+                {  # fsp engine conf
+                    'func_name': 'dolla_vlm',
+                    'zero_on_step': True,
+                    'params': {
+                        'price_func': {
+                            'default_value': 'chl3',
+                        },
+                    },
+                },
+                # loglevel,
+            )
+            # profiler(f'created shm for fsp actor: {display_name}')
+
+            await started.wait()
+
+            pi = chart.overlay_plotitem(
+                'dolla_vlm',
+                index=0,  # place axis on inside (nearest to chart)
+                axis_title=' $vlm',
+                axis_side='right',
+                axis_kwargs={
+                    'typical_max_str': ' 100.0 M ',
+                    'formatter': partial(
+                        humanize,
+                        digits=2,
+                    ),
+                },
+
+            )
+
+            # add custom auto range handler
+            pi.vb._maxmin = partial(maxmin, name='dolla_vlm')
+
+            curve, _ = chart.draw_curve(
+
+                name='dolla_vlm',
+                data=shm.array,
+
+                array_key='dolla_vlm',
+                overlay=pi,
+                # color='bracket',
+                # TODO: this color or dark volume
+                # color='charcoal',
+                step_mode=True,
+                # **conf.get('chart_kwargs', {})
+            )
+            # TODO: is there a way to "sync" the dual axes such that only
+            # one curve is needed?
+            # hide the original vlm curve since the $vlm one is now
+            # displayed and the curves are effectively the same minus
+            # liquidity events (well at least on low OHLC periods - 1s).
+            vlm_curve.hide()
+
+            # TODO: we need a better API to do this..
+            # specially store ref to shm for lookup in display loop
+            # since only a placeholder of `None` is entered in
+            # ``.draw_curve()``.
+            chart._overlays['dolla_vlm'] = shm
+
+            # XXX: old dict-style config before it was moved into the
+            # helper task
+            #     'dolla_vlm': {
+            #         'func_name': 'dolla_vlm',
+            #         'zero_on_step': True,
+            #         'overlay': 'volume',
+            #         'separate_axes': True,
+            #         'params': {
+            #             'price_func': {
+            #                 'default_value': 'chl3',
+            #                 # tell target ``Edit`` widget to not allow
+            #                 # edits for now.
+            #                 'widget_kwargs': {'readonly': True},
+            #             },
+            #         },
+            #         'chart_kwargs': {'step_mode': True}
+            #     },
+
+            # }
+
+            for name, axis_info in pi.axes.items():
+                # lol this sux XD
+                axis = axis_info['item']
+                if isinstance(axis, PriceAxis):
+                    axis.size_to_values()
+
+        # built-in vlm fsps
+        for display_name, conf in {
+            'vwap': {
+                'func_name': 'vwap',
+                'overlay': 'ohlc',  # overlays with OHLCV (main) chart
+                'anchor': 'session',
+            },
+        }.items():
+            started = await admin.open_fsp_chart(
+                display_name,
+                conf,
+            )
+
+
+async def start_fsp_displays(
+
+    linked: LinkedSplits,
+    ohlcv: ShmArray,
+    group_status_key: str,
+    loglevel: str,
+
+) -> None:
+    '''
+    Create fsp charts from a config input attached to a local actor
+    compute cluster.
+
+    Pass target entrypoint and historical data via ``ShmArray``.
+
+    '''
+    linked.focus()
+
+    # TODO: eventually we'll support some kind of n-compose syntax
+    fsp_conf = {
+        # 'rsi': {
+        #     'func_name': 'rsi',  # literal python func ref lookup name
+
+        #     # map of parameters to place on the fsp sidepane widget
+        #     # which should map to dynamic inputs available to the
+        #     # fsp function at runtime.
+        #     'params': {
+        #         'period': {
+        #             'default_value': 14,
+        #             'widget_kwargs': {'readonly': True},
+        #         },
+        #     },
+
+        #     # ``ChartPlotWidget`` options passthrough
+        #     'chart_kwargs': {
+        #         'static_yrange': (0, 100),
+        #     },
+        # },
+    }
+    profiler = pg.debug.Profiler(
+        delayed=False,
+        disabled=False
+    )
+
+    # async with gather_contexts((
+    async with (
+
+        # NOTE: this admin internally opens an actor cluster
+        open_fsp_admin(linked, ohlcv) as admin,
+    ):
+        statuses = []
+        for display_name, conf in fsp_conf.items():
+            started = await admin.open_fsp_chart(
+                display_name,
+                conf,
+            )
+            done = linked.window().status_bar.open_status(
+                f'loading fsp, {display_name}..',
+                group_key=group_status_key,
+            )
+            statuses.append((started, done))
+
+        for fsp_loaded, status_cb in statuses:
+            await fsp_loaded.wait()
+            profiler(f'attached to fsp portal: {display_name}')
+            status_cb()
+
+    # blocks on nursery until all fsp actors complete

piker/ui/_label.py

@@ -34,7 +34,7 @@ from ._style import (
 
 
 class Label:
-    """
+    '''
     A plain ol' "scene label" using an underlying ``QGraphicsTextItem``.
 
     After hacking for many days on multiple "label" systems inside
@@ -50,10 +50,8 @@ class Label:
     small, re-usable label components that can actually be used to build
     production grade UIs...
 
-    """
+    '''
-
     def __init__(
-
         self,
         view: pg.ViewBox,
         fmt_str: str,
@@ -63,6 +61,7 @@ class Label:
         font_size: str = 'small',
         opacity: float = 1,
         fields: dict = {},
+        parent: pg.GraphicsObject = None,
         update_on_range_change: bool = True,
 
     ) -> None:
@@ -71,11 +70,13 @@ class Label:
         self._fmt_str = fmt_str
         self._view_xy = QPointF(0, 0)
 
-        self.scene_anchor: Optional[Callable[..., QPointF]] = None
+        self.scene_anchor: Optional[
+            Callable[..., QPointF]
+        ] = None
 
         self._x_offset = x_offset
 
-        txt = self.txt = QtWidgets.QGraphicsTextItem()
+        txt = self.txt = QtWidgets.QGraphicsTextItem(parent=parent)
         txt.setCacheMode(QtWidgets.QGraphicsItem.DeviceCoordinateCache)
 
         vb.scene().addItem(txt)
@@ -86,7 +87,6 @@ class Label:
         )
         dpi_font.configure_to_dpi()
         txt.setFont(dpi_font.font)
-
         txt.setOpacity(opacity)
 
         # register viewbox callbacks
@@ -109,7 +109,7 @@ class Label:
         # self.setTextInteractionFlags(QtGui.Qt.TextEditorInteraction)
 
     @property
-    def color(self):
+    def color(self) -> str:
         return self._hcolor
 
     @color.setter
@@ -118,9 +118,10 @@ class Label:
         self._hcolor = color
 
     def update(self) -> None:
-        '''Update this label either by invoking its
+        '''
-        user defined anchoring function, or by positioning
+        Update this label either by invoking its user defined anchoring
-        to the last recorded data view coordinates.
+        function, or by positioning to the last recorded data view
+        coordinates.
 
         '''
         # move label in scene coords to desired position
@@ -234,7 +235,8 @@ class Label:
 
 
 class FormatLabel(QLabel):
-    '''Kinda similar to above but using the widget apis.
+    '''
+    Kinda similar to above but using the widget apis.
 
     '''
     def __init__(
@@ -273,8 +275,8 @@ class FormatLabel(QLabel):
             QSizePolicy.Expanding,
             QSizePolicy.Expanding,
         )
-        self.setAlignment(Qt.AlignVCenter
+        self.setAlignment(
-            | Qt.AlignLeft
+            Qt.AlignVCenter | Qt.AlignLeft
         )
         self.setText(self.fmt_str)
 

piker/ui/_lines.py

@@ -334,10 +334,11 @@ class LevelLine(pg.InfiniteLine):
         w: QtWidgets.QWidget
 
     ) -> None:
-        """Core paint which we override (yet again)
+        '''
+        Core paint which we override (yet again)
         from pg..
 
-        """
+        '''
         p.setRenderHint(p.Antialiasing)
 
         # these are in viewbox coords

piker/ui/_style.py

@@ -16,6 +16,7 @@
 
 """
 Qt UI styling.
+
 """
 from typing import Optional, Dict
 import math
@@ -141,7 +142,7 @@ class DpiAwareFont:
         self._font_inches = inches
         font_size = math.floor(inches * dpi)
 
-        log.info(
+        log.debug(
             f"screen:{screen.name()}]\n"
             f"pDPI: {pdpi}, lDPI: {ldpi}, scale: {scale}\n"
             f"\nOur best guess font size is {font_size}\n"

piker/ui/_window.py

@@ -1,5 +1,5 @@
 # piker: trading gear for hackers
-# Copyright (C) Tyler Goodlet (in stewardship for piker0)
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
 
 # 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
@@ -25,7 +25,7 @@ from typing import Callable, Optional, Union
 import uuid
 
 from pyqtgraph import QtGui
-from PyQt5 import QtCore, QtWidgets
+from PyQt5 import QtCore
 from PyQt5.QtWidgets import QLabel, QStatusBar
 
 from ..log import get_logger
@@ -55,7 +55,8 @@ class MultiStatus:
         group_key: Optional[Union[bool, str]] = False,
 
     ) -> Union[Callable[..., None], str]:
-        '''Add a status to the status bar and return a close callback which
+        '''
+        Add a status to the status bar and return a close callback which
         when called will remove the status ``msg``.
 
         '''
@@ -137,7 +138,8 @@ class MultiStatus:
         return ret
 
     def render(self) -> None:
-        '''Display all open statuses to bar.
+        '''
+        Display all open statuses to bar.
 
         '''
         if self.statuses: