Just warn on `ib` symbol search lags

Initial test that starts a `binance` feed and reads the quote messages
alongside shm buffers for 1s and 1m OHLC; just prints to console for
now.

Template out parametrization for multi-symbol quote-multiplexed feeds
which coming soon B)

dockering/ib/docker-compose.yml

@@ -62,39 +62,39 @@ services:
       # - "127.0.0.1:4002:4002"
       # - "127.0.0.1:5900:5900"
 
-  ib_gw_live:
-    image: waytrade/ib-gateway:1012.2i
-    restart: always
-    network_mode: 'host'
+  # ib_gw_live:
+  #   image: waytrade/ib-gateway:1012.2i
+  #   restart: always
+  #   network_mode: 'host'
 
-    volumes:
-      - type: bind
-        source: ./jts_live.ini
-        target: /root/jts/jts.ini
-        # don't let ibc clobber this file for
-        # the main reason of not having a stupid
-        # timezone set..
-        read_only: true
+  #   volumes:
+  #     - type: bind
+  #       source: ./jts_live.ini
+  #       target: /root/jts/jts.ini
+  #       # don't let ibc clobber this file for
+  #       # the main reason of not having a stupid
+  #       # timezone set..
+  #       read_only: true
 
-      # force our own ibc config
-      - type: bind
-        source: ./ibc.ini
-        target: /root/ibc/config.ini
+  #     # force our own ibc config
+  #     - type: bind
+  #       source: ./ibc.ini
+  #       target: /root/ibc/config.ini
 
-      # force our noop script - socat isn't needed in host mode.
-      - type: bind
-        source: ./fork_ports_delayed.sh
-        target: /root/scripts/fork_ports_delayed.sh
+  #     # force our noop script - socat isn't needed in host mode.
+  #     - type: bind
+  #       source: ./fork_ports_delayed.sh
+  #       target: /root/scripts/fork_ports_delayed.sh
 
-      # force our noop script - socat isn't needed in host mode.
-      - type: bind
-        source: ./run_x11_vnc.sh
-        target: /root/scripts/run_x11_vnc.sh
-        read_only: true
+  #     # force our noop script - socat isn't needed in host mode.
+  #     - type: bind
+  #       source: ./run_x11_vnc.sh
+  #       target: /root/scripts/run_x11_vnc.sh
+  #       read_only: true
 
-    # NOTE: to fill these out, define an `.env` file in the same dir as
-    # this compose file which looks something like:
-    environment:
-      TRADING_MODE: 'live'
-      VNC_SERVER_PASSWORD: 'doggy'
-      VNC_SERVER_PORT: '3004'
+  #   # NOTE: to fill these out, define an `.env` file in the same dir as
+  #   # this compose file which looks something like:
+  #   environment:
+  #     TRADING_MODE: 'live'
+  #     VNC_SERVER_PASSWORD: 'doggy'
+  #     VNC_SERVER_PORT: '3004'

piker/__init__.py

@@ -18,3 +18,10 @@
 piker: trading gear for hackers.
 
 """
+from ._daemon import open_piker_runtime
+from .data.feed import open_feed
+
+__all__ = [
+    'open_piker_runtime',
+    'open_feed',
+]

piker/_daemon.py

@@ -35,7 +35,12 @@ log = get_logger(__name__)
 
 _root_dname = 'pikerd'
 
-_registry_addr = ('127.0.0.1', 6116)
+_registry_host: str = '127.0.0.1'
+_registry_port: int = 6116
+_registry_addr = (
+    _registry_host,
+    _registry_port,
+)
 _tractor_kwargs: dict[str, Any] = {
     # use a different registry addr then tractor's default
     'arbiter_addr': _registry_addr
@@ -135,6 +140,7 @@ async def open_pikerd(
     # XXX: you should pretty much never want debug mode
     # for data daemons when running in production.
     debug_mode: bool = False,
+    registry_addr: None | tuple[str, int] = None,
 
 ) -> Optional[tractor._portal.Portal]:
     '''
@@ -153,7 +159,7 @@ async def open_pikerd(
         tractor.open_root_actor(
 
             # passed through to ``open_root_actor``
-            arbiter_addr=_registry_addr,
+            arbiter_addr=registry_addr or _registry_addr,
             name=_root_dname,
             loglevel=loglevel,
             debug_mode=debug_mode,
@@ -193,7 +199,7 @@ async def open_piker_runtime(
     # for data daemons when running in production.
     debug_mode: bool = False,
 
-) -> Optional[tractor._portal.Portal]:
+) -> tractor.Actor:
     '''
     Start a piker actor who's runtime will automatically sync with
     existing piker actors on the local link based on configuration.
@@ -248,6 +254,7 @@ async def maybe_open_runtime(
 @acm
 async def maybe_open_pikerd(
     loglevel: Optional[str] = None,
+    registry_addr: None | tuple = None,
     **kwargs,
 
 ) -> Union[tractor._portal.Portal, Services]:
@@ -260,13 +267,21 @@ async def maybe_open_pikerd(
         get_console_log(loglevel)
 
     # subtle, we must have the runtime up here or portal lookup will fail
-    async with maybe_open_runtime(loglevel, **kwargs):
-
-        async with tractor.find_actor(_root_dname) as portal:
-            # assert portal is not None
-            if portal is not None:
-                yield portal
-                return
+    async with (
+        maybe_open_runtime(loglevel, **kwargs),
+        tractor.find_actor(_root_dname) as portal
+    ):
+        # connect to any existing daemon presuming
+        # its registry socket was selected.
+        if (
+            portal is not None
+            and (
+                registry_addr is None
+                or portal.channel.raddr == registry_addr
+            )
+        ):
+            yield portal
+            return
 
     # presume pikerd role since no daemon could be found at
     # configured address
@@ -274,6 +289,7 @@ async def maybe_open_pikerd(
 
         loglevel=loglevel,
         debug_mode=kwargs.get('debug_mode', False),
+        registry_addr=registry_addr,
 
     ) as _:
         # in the case where we're starting up the

piker/brokers/ib/broker.py

@@ -371,8 +371,8 @@ async def update_and_audit_msgs(
                     else:
                         entry = f'split_ratio = 1/{int(reverse_split_ratio)}'
 
-                    raise ValueError(
-                    # log.error(
+                    # raise ValueError(
+                    log.error(
                         f'POSITION MISMATCH ib <-> piker ledger:\n'
                         f'ib: {ibppmsg}\n'
                         f'piker: {msg}\n'
@@ -883,7 +883,7 @@ async def deliver_trade_events(
                     # execdict.pop('acctNumber')
 
                     fill_msg = BrokerdFill(
-                        # should match the value returned from
+                        # NOTE: should match the value returned from
                         # `.submit_limit()`
                         reqid=execu.orderId,
                         time_ns=time.time_ns(),  # cuz why not

piker/brokers/ib/feed.py

@@ -1038,7 +1038,13 @@ async def open_symbol_search(
                 stock_results = []
 
                 async def stash_results(target: Awaitable[list]):
-                    stock_results.extend(await target)
+                    try:
+                        results = await target
+                    except tractor.trionics.Lagged:
+                        print("IB SYM-SEARCH OVERRUN?!?")
+                        return
+
+                    stock_results.extend(results)
 
                 for i in range(10):
                     with trio.move_on_after(3) as cs:

piker/brokers/kraken/feed.py

@@ -85,6 +85,7 @@ class Pair(Struct):
     margin_call: str  # margin call level
     margin_stop: str  # stop-out/liquidation margin level
     ordermin: float  # minimum order volume for pair
+    tick_size: float # min price step size
 
 
 class OHLC(Struct):

piker/clearing/_ems.py

@@ -1239,8 +1239,7 @@ async def process_client_order_cmds(
                 pred = mk_check(trigger_price, last, action)
 
                 spread_slap: float = 5
-                sym = fqsn.replace(f'.{brokers[0]}', '')
-                min_tick = feed.symbols[sym].tick_size
+                min_tick = feed.symbols[fqsn].tick_size
 
                 if action == 'buy':
                     tickfilter = ('ask', 'last', 'trade')

piker/cli/__init__.py

@@ -27,7 +27,11 @@ import tractor
 
 from ..log import get_console_log, get_logger, colorize_json
 from ..brokers import get_brokermod
-from .._daemon import _tractor_kwargs
+from .._daemon import (
+    _tractor_kwargs,
+    _registry_host,
+    _registry_port,
+)
 from .. import config
 
 
@@ -39,13 +43,21 @@ DEFAULT_BROKER = 'questrade'
 @click.option('--loglevel', '-l', default='warning', help='Logging level')
 @click.option('--tl', is_flag=True, help='Enable tractor logging')
 @click.option('--pdb', is_flag=True, help='Enable tractor debug mode')
-@click.option('--host', '-h', default='127.0.0.1', help='Host address to bind')
+@click.option('--host', '-h', default=None, help='Host addr to bind')
+@click.option('--port', '-p', default=None, help='Port number to bind')
 @click.option(
     '--tsdb',
     is_flag=True,
     help='Enable local ``marketstore`` instance'
 )
-def pikerd(loglevel, host, tl, pdb, tsdb):
+def pikerd(
+    loglevel: str,
+    host: str,
+    port: int,
+    tl: bool,
+    pdb: bool,
+    tsdb: bool,
+):
     '''
     Spawn the piker broker-daemon.
 
@@ -62,12 +74,21 @@ def pikerd(loglevel, host, tl, pdb, tsdb):
             "\n"
         ))
 
+    reg_addr: None | tuple[str, int] = None
+    if host or port:
+        reg_addr = (
+            host or _registry_host,
+            int(port) or _registry_port,
+        )
+
     async def main():
 
         async with (
             open_pikerd(
                 loglevel=loglevel,
                 debug_mode=pdb,
+                registry_addr=reg_addr,
+
             ),  # normally delivers a ``Services`` handle
             trio.open_nursery() as n,
         ):
@@ -104,8 +125,19 @@ def pikerd(loglevel, host, tl, pdb, tsdb):
 @click.option('--loglevel', '-l', default='warning', help='Logging level')
 @click.option('--tl', is_flag=True, help='Enable tractor logging')
 @click.option('--configdir', '-c', help='Configuration directory')
+@click.option('--host', '-h', default=None, help='Host addr to bind')
+@click.option('--port', '-p', default=None, help='Port number to bind')
 @click.pass_context
-def cli(ctx, brokers, loglevel, tl, configdir):
+def cli(
+    ctx: click.Context,
+    brokers: list[str],
+    loglevel: str,
+    tl: bool,
+    configdir: str,
+    host: str,
+    port: int,
+
+) -> None:
     if configdir is not None:
         assert os.path.isdir(configdir), f"`{configdir}` is not a valid path"
         config._override_config_dir(configdir)
@@ -117,6 +149,13 @@ def cli(ctx, brokers, loglevel, tl, configdir):
     else:
         brokermods = [get_brokermod(broker) for broker in brokers]
 
+    reg_addr: None | tuple[str, int] = None
+    if host or port:
+        reg_addr = (
+            host or _registry_host,
+            int(port) or _registry_port,
+        )
+
     ctx.obj.update({
         'brokers': brokers,
         'brokermods': brokermods,
@@ -125,6 +164,7 @@ def cli(ctx, brokers, loglevel, tl, configdir):
         'log': get_console_log(loglevel),
         'confdir': config._config_dir,
         'wl_path': config._watchlists_data_path,
+        'registry_addr': reg_addr,
     })
 
     # allow enabling same loglevel in ``tractor`` machinery

piker/data/feed.py

@@ -746,12 +746,12 @@ async def manage_history(
 
         # we expect the sub-actor to write
         readonly=False,
-        size=4*_secs_in_day,
+        size=3*_secs_in_day,
     )
 
     # (for now) set the rt (hft) shm array with space to prepend
     # only a few days worth of 1s history.
-    days = 3
+    days = 2
     start_index = days*_secs_in_day
     rt_shm._first.value = start_index
     rt_shm._last.value = start_index
@@ -1410,7 +1410,7 @@ async def open_feed(
 
             # symbol.broker_info[brokername] = si
             feed.symbols[fqsn] = symbol
-            feed.symbols[sym] = symbol
+            feed.symbols[f'{sym}.{brokername}'] = symbol
 
             # cast shm dtype to list... can't member why we need this
             for shm_key, shm in [

piker/ui/_app.py

@@ -66,7 +66,7 @@ async def _async_main(
     # implicit required argument provided by ``qtractor_run()``
     main_widget: GodWidget,
 
-    sym: str,
+    syms: list[str],
     brokernames: str,
     loglevel: str,
 
@@ -113,12 +113,16 @@ async def _async_main(
         # godwidget.hbox.addWidget(search)
         godwidget.search = search
 
-        symbol, _, provider = sym.rpartition('.')
+        symbols: list[str] = []
+
+        for sym in syms:
+            symbol, _, provider = sym.rpartition('.')
+            symbols.append(symbol)
 
         # this internally starts a ``display_symbol_data()`` task above
-        order_mode_ready = await godwidget.load_symbol(
+        order_mode_ready = await godwidget.load_symbols(
             provider,
-            symbol,
+            symbols,
             loglevel
         )
 
@@ -166,7 +170,7 @@ async def _async_main(
 
 
 def _main(
-    sym: str,
+    syms: list[str],
     brokernames: [str],
     piker_loglevel: str,
     tractor_kwargs,
@@ -178,7 +182,7 @@ def _main(
     '''
     run_qtractor(
         func=_async_main,
-        args=(sym, brokernames, piker_loglevel),
+        args=(syms, brokernames, piker_loglevel),
         main_widget_type=GodWidget,
         tractor_kwargs=tractor_kwargs,
     )

piker/ui/_chart.py

@@ -186,10 +186,10 @@ class GodWidget(QWidget):
     ) -> tuple[LinkedSplits, LinkedSplits]:  # type: ignore
         return self._chart_cache.get(symbol_key)
 
-    async def load_symbol(
+    async def load_symbols(
         self,
         providername: str,
-        symbol_key: str,
+        symbol_keys: list[str],
         loglevel: str,
         reset: bool = False,
 
@@ -200,12 +200,20 @@ class GodWidget(QWidget):
         Expects a ``numpy`` structured array containing all the ohlcv fields.
 
         '''
-        # our symbol key style is always lower case
-        symbol_key = symbol_key.lower()
+        fqsns: list[str] = []
 
-        # fully qualified symbol name (SNS i guess is what we're making?)
-        fqsn = '.'.join([symbol_key, providername])
-        all_linked = self.get_chart_symbol(fqsn)
+        # our symbol key style is always lower case
+        for key in list(map(str.lower, symbol_keys)):
+
+            # fully qualified symbol name (SNS i guess is what we're making?)
+            fqsn = '.'.join([key, providername])
+            fqsns.append(fqsn)
+
+        # NOTE: for now we use the first symbol in the set as the "key"
+        # for the overlay of feeds on the chart.
+        group_key = fqsns[0]
+
+        all_linked = self.get_chart_symbol(group_key)
         order_mode_started = trio.Event()
 
         if not self.vbox.isEmpty():
@@ -238,7 +246,7 @@ class GodWidget(QWidget):
                 display_symbol_data,
                 self,
                 providername,
-                symbol_key,
+                fqsns,
                 loglevel,
                 order_mode_started,
             )
@@ -814,7 +822,8 @@ class ChartPlotWidget(pg.PlotWidget):
     # a better one?
     def mk_vb(self, name: str) -> ChartView:
         cv = ChartView(name)
-        cv.linkedsplits = self.linked
+        # link new view to chart's view set
+        cv.linked = self.linked
         return cv
 
     def __init__(
@@ -1179,19 +1188,27 @@ class ChartPlotWidget(pg.PlotWidget):
         )
         pi.hideButtons()
 
-        # cv.enable_auto_yrange(self.view)
-        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,
+            # only link x-axes and
+            # don't relay any ``ViewBox`` derived event
+            # handlers since we only care about keeping charts
+            # x-synced on interaction (at least for now).
             link_axes=(0,),
         )
 
+        # connect auto-yrange callbacks *from* this new
+        # 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)
+
         # add axis title
         # TODO: do we want this API to still work?
         # raxis = pi.getAxis('right')

piker/ui/_display.py

@@ -947,7 +947,7 @@ async def link_views_with_region(
 async def display_symbol_data(
     godwidget: GodWidget,
     provider: str,
-    sym: str,
+    fqsns: list[str],
     loglevel: str,
     order_mode_started: trio.Event,
 
@@ -961,11 +961,6 @@ async def display_symbol_data(
 
     '''
     sbar = godwidget.window.status_bar
-    loading_sym_key = sbar.open_status(
-        f'loading {sym}.{provider} ->',
-        group_key=True
-    )
-
     # historical data fetch
     # brokermod = brokers.get_brokermod(provider)
 
@@ -974,10 +969,18 @@ async def display_symbol_data(
     #     clear_on_next=True,
     #     group_key=loading_sym_key,
     # )
-    fqsn = '.'.join((sym, provider))
+
+    for fqsn in fqsns:
+
+        loading_sym_key = sbar.open_status(
+            f'loading {fqsn} ->',
+            group_key=True
+        )
+
+    first_fqsn = fqsns[0]
 
     async with open_feed(
-        [fqsn],
+        fqsns,
         loglevel=loglevel,
 
         # limit to at least display's FPS
@@ -988,7 +991,7 @@ async def display_symbol_data(
         ohlcv: ShmArray = feed.rt_shm
         hist_ohlcv: ShmArray = feed.hist_shm
 
-        symbol = feed.symbols[sym]
+        symbol = feed.symbols[first_fqsn]
         fqsn = symbol.front_fqsn()
 
         step_size_s = 1
@@ -1025,7 +1028,7 @@ async def display_symbol_data(
         godwidget.pp_pane = pp_pane
 
         # create main OHLC chart
-        chart = rt_linked.plot_ohlc_main(
+        ohlc_chart = rt_linked.plot_ohlc_main(
             symbol,
             ohlcv,
             # in the case of history chart we explicitly set `False`
@@ -1033,8 +1036,8 @@ async def display_symbol_data(
             sidepane=pp_pane,
         )
 
-        chart._feeds[symbol.key] = feed
-        chart.setFocus()
+        ohlc_chart._feeds[symbol.key] = feed
+        ohlc_chart.setFocus()
 
         # XXX: FOR SOME REASON THIS IS CAUSING HANGZ!?!
         # plot historical vwap if available
@@ -1044,7 +1047,7 @@ async def display_symbol_data(
         #     and 'bar_wap' in bars.dtype.fields
         # ):
         #     wap_in_history = True
-        #     chart.draw_curve(
+        #     ohlc_chart.draw_curve(
         #         name='bar_wap',
         #         shm=ohlcv,
         #         color='default_light',
@@ -1105,7 +1108,7 @@ async def display_symbol_data(
             await trio.sleep(0)
 
             # size view to data prior to order mode init
-            chart.default_view()
+            ohlc_chart.default_view()
             rt_linked.graphics_cycle()
             await trio.sleep(0)
 
@@ -1119,7 +1122,7 @@ async def display_symbol_data(
             godwidget.resize_all()
 
             await link_views_with_region(
-                chart,
+                ohlc_chart,
                 hist_chart,
                 feed,
             )
@@ -1135,7 +1138,7 @@ async def display_symbol_data(
             ):
                 if not vlm_chart:
                     # trigger another view reset if no sub-chart
-                    chart.default_view()
+                    ohlc_chart.default_view()
 
                 rt_linked.mode = mode
 

piker/ui/_fsp.py

@@ -624,6 +624,8 @@ async def open_vlm_displays(
         # built-in vlm which we plot ASAP since it's
         # usually data provided directly with OHLC history.
         shm = ohlcv
+        ohlc_chart = linked.chart
+
         chart = linked.add_plot(
             name='volume',
             shm=shm,
@@ -639,6 +641,9 @@ async def open_vlm_displays(
             # the curve item internals are pretty convoluted.
             style='step',
         )
+        ohlc_chart.view.enable_auto_yrange(
+            src_vb=chart.view,
+        )
 
         # force 0 to always be in view
         def multi_maxmin(

piker/ui/_interaction.py

@@ -329,7 +329,6 @@ async def handle_viewmode_mouse(
         ):
             # when in order mode, submit execution
             # msg.event.accept()
-            # breakpoint()
             view.order_mode.submit_order()
 
 
@@ -346,16 +345,6 @@ class ChartView(ViewBox):
     '''
     mode_name: str = 'view'
 
-    # "relay events" for making overlaid views work.
-    # NOTE: these MUST be defined here (and can't be monkey patched
-    # on later) due to signal construction requiring refs to be
-    # in place during the run of meta-class machinery.
-    mouseDragEventRelay = QtCore.Signal(object, object, object)
-    wheelEventRelay = QtCore.Signal(object, object, object)
-
-    event_relay_source: 'Optional[ViewBox]' = None
-    relays: dict[str, QtCore.Signal] = {}
-
     def __init__(
         self,
 
@@ -479,7 +468,7 @@ class ChartView(ViewBox):
         self,
         ev,
         axis=None,
-        relayed_from: ChartView = None,
+        # relayed_from: ChartView = None,
     ):
         '''
         Override "center-point" location for scrolling.
@@ -490,6 +479,13 @@ class ChartView(ViewBox):
         TODO: PR a method into ``pyqtgraph`` to make this configurable
 
         '''
+        linked = self.linked
+        if (
+            not linked
+        ):
+            # print(f'{self.name} not linked but relay from {relayed_from.name}')
+            return
+
         if axis in (0, 1):
             mask = [False, False]
             mask[axis] = self.state['mouseEnabled'][axis]
@@ -609,9 +605,20 @@ class ChartView(ViewBox):
         self,
         ev,
         axis: Optional[int] = None,
-        relayed_from: ChartView = None,
+        # relayed_from: ChartView = None,
 
     ) -> None:
+        # if relayed_from:
+        #     print(f'PAN: {self.name} -> RELAYED FROM: {relayed_from.name}')
+
+        # NOTE since in the overlay case axes are already
+        # "linked" any x-range change will already be mirrored
+        # in all overlaid ``PlotItems``, so we need to simply
+        # ignore the signal here since otherwise we get N-calls
+        # from N-overlays resulting in an "accelerated" feeling
+        # panning motion instead of the expect linear shift.
+        # if relayed_from:
+        #     return
 
         pos = ev.pos()
         lastPos = ev.lastPos()
@@ -849,33 +856,37 @@ class ChartView(ViewBox):
 
     ) -> None:
         '''
-        Assign callback for rescaling y-axis automatically
-        based on data contents and ``ViewBox`` state.
+        Assign callbacks for rescaling and resampling y-axis data
+        automatically based on data contents and ``ViewBox`` state.
 
         '''
         if src_vb is None:
             src_vb = self
 
-        # splitter(s) resizing
+        # widget-UIs/splitter(s) resizing
         src_vb.sigResized.connect(self._set_yrange)
 
+        # re-sampling trigger:
         # TODO: a smarter way to avoid calling this needlessly?
         # 2 things i can think of:
         # - register downsample-able graphics specially and only
         #   iterate those.
-        # - only register this when certain downsampleable graphics are
+        # - only register this when certain downsample-able graphics are
         #   "added to scene".
         src_vb.sigRangeChangedManually.connect(
             self.maybe_downsample_graphics
         )
-
         # mouse wheel doesn't emit XRangeChanged
         src_vb.sigRangeChangedManually.connect(self._set_yrange)
 
-        # src_vb.sigXRangeChanged.connect(self._set_yrange)
-        # src_vb.sigXRangeChanged.connect(
-        #     self.maybe_downsample_graphics
-        # )
+        # XXX: enabling these will cause "jittery"-ness
+        # on zoom where sharp diffs in the y-range will
+        # not re-size right away until a new sample update?
+        # if src_vb is not self:
+        #     src_vb.sigXRangeChanged.connect(self._set_yrange)
+        #     src_vb.sigXRangeChanged.connect(
+        #         self.maybe_downsample_graphics
+        #     )
 
     def disable_auto_yrange(self) -> None:
 
@@ -916,7 +927,6 @@ class ChartView(ViewBox):
         self,
         autoscale_overlays: bool = True,
     ):
-
         profiler = Profiler(
             msg=f'ChartView.maybe_downsample_graphics() for {self.name}',
             disabled=not pg_profile_enabled(),
@@ -931,8 +941,12 @@ class ChartView(ViewBox):
 
         # TODO: a faster single-loop-iterator way of doing this XD
         chart = self._chart
+        plots = {chart.name: chart}
+
         linked = self.linked
-        plots = linked.subplots | {chart.name: chart}
+        if linked:
+            plots |= linked.subplots
+
         for chart_name, chart in plots.items():
             for name, flow in chart._flows.items():
 

piker/ui/_overlay.py

@@ -18,23 +18,27 @@
 Charting overlay helpers.
 
 '''
-from typing import Callable, Optional
-
-from pyqtgraph.Qt.QtCore import (
-    # QObject,
-    # Signal,
-    Qt,
-    # QEvent,
+from collections import defaultdict
+from functools import partial
+from typing import (
+    Callable,
+    Optional,
 )
 
 from pyqtgraph.graphicsItems.AxisItem import AxisItem
 from pyqtgraph.graphicsItems.ViewBox import ViewBox
-from pyqtgraph.graphicsItems.GraphicsWidget import GraphicsWidget
+# from pyqtgraph.graphicsItems.GraphicsWidget import GraphicsWidget
 from pyqtgraph.graphicsItems.PlotItem.PlotItem import PlotItem
-from pyqtgraph.Qt.QtCore import QObject, Signal, QEvent
-from pyqtgraph.Qt.QtWidgets import QGraphicsGridLayout, QGraphicsLinearLayout
-
-from ._interaction import ChartView
+from pyqtgraph.Qt.QtCore import (
+    QObject,
+    Signal,
+    QEvent,
+    Qt,
+)
+from pyqtgraph.Qt.QtWidgets import (
+    # QGraphicsGridLayout,
+    QGraphicsLinearLayout,
+)
 
 __all__ = ["PlotItemOverlay"]
 
@@ -89,16 +93,11 @@ class ComposedGridLayout:
     def __init__(
         self,
         item: PlotItem,
-        grid: QGraphicsGridLayout,
-        reverse: bool = False,  # insert items to the "center"
 
     ) -> None:
-        self.items: list[PlotItem] = []
-        # self.grid = grid
-        self.reverse = reverse
 
-        # TODO: use a ``bidict`` here?
-        self._pi2axes: dict[
+        self.items: list[PlotItem] = []
+        self._pi2axes: dict[  # TODO: use a ``bidict`` here?
             int,
             dict[str, AxisItem],
         ] = {}
@@ -120,12 +119,13 @@ class ComposedGridLayout:
 
             if name in ('top', 'bottom'):
                 orient = Qt.Vertical
+
             elif name in ('left', 'right'):
                 orient = Qt.Horizontal
 
             layout.setOrientation(orient)
 
-        self.insert(0, item)
+        self.insert_plotitem(0, item)
 
         # insert surrounding linear layouts into the parent pi's layout
         # such that additional axes can be appended arbitrarily without
@@ -159,7 +159,7 @@ class ComposedGridLayout:
         # enter plot into list for index tracking
         self.items.insert(index, plotitem)
 
-    def insert(
+    def insert_plotitem(
         self,
         index: int,
         plotitem: PlotItem,
@@ -171,7 +171,9 @@ class ComposedGridLayout:
 
         '''
         if index < 0:
-            raise ValueError('`insert()` only supports an index >= 0')
+            raise ValueError(
+                '`.insert_plotitem()` only supports an index >= 0'
+            )
 
         # add plot's axes in sequence to the embedded linear layouts
         # for each "side" thus avoiding graphics collisions.
@@ -220,7 +222,7 @@ class ComposedGridLayout:
 
         return index
 
-    def append(
+    def append_plotitem(
         self,
         item: PlotItem,
 
@@ -232,7 +234,7 @@ class ComposedGridLayout:
         '''
         # for left and bottom axes we have to first remove
         # items and re-insert to maintain a list-order.
-        return self.insert(len(self.items), item)
+        return self.insert_plotitem(len(self.items), item)
 
     def get_axis(
         self,
@@ -249,16 +251,16 @@ class ComposedGridLayout:
         named = self._pi2axes[name]
         return named.get(index)
 
-    def pop(
-        self,
-        item: PlotItem,
+    # def pop(
+    #     self,
+    #     item: PlotItem,
 
-    ) -> PlotItem:
-        '''
-        Remove item and restack all axes in list-order.
+    # ) -> PlotItem:
+    #     '''
+    #     Remove item and restack all axes in list-order.
 
-        '''
-        raise NotImplementedError
+    #     '''
+    #     raise NotImplementedError
 
 
 # Unimplemented features TODO:
@@ -279,194 +281,6 @@ class ComposedGridLayout:
 #   axis?
 
 
-# TODO: we might want to enabled some kind of manual flag to disable
-# this method wrapping during type creation? As example a user could
-# definitively decide **not** to enable broadcasting support by
-# setting something like ``ViewBox.disable_relays = True``?
-def mk_relay_method(
-
-    signame: str,
-    slot: Callable[
-        [ViewBox,
-         'QEvent',
-         Optional[AxisItem]],
-        None,
-    ],
-
-) -> Callable[
-    [
-        ViewBox,
-        # lol, there isn't really a generic type thanks
-        # to the rewrite of Qt's event system XD
-        'QEvent',
-
-        'Optional[AxisItem]',
-        'Optional[ViewBox]',  # the ``relayed_from`` arg we provide
-    ],
-    None,
-]:
-
-    def maybe_broadcast(
-        vb: 'ViewBox',
-        ev: 'QEvent',
-        axis: 'Optional[int]' = None,
-        relayed_from: 'ViewBox' = None,
-
-    ) -> None:
-        '''
-        (soon to be) Decorator which makes an event handler
-        "broadcastable" to overlayed ``GraphicsWidget``s.
-
-        Adds relay signals based on the decorated handler's name
-        and conducts a signal broadcast of the relay signal if there
-        are consumers registered.
-
-        '''
-        # When no relay source has been set just bypass all
-        # the broadcast machinery.
-        if vb.event_relay_source is None:
-            ev.accept()
-            return slot(
-                vb,
-                ev,
-                axis=axis,
-            )
-
-        if relayed_from:
-            assert axis is None
-
-            # this is a relayed event and should be ignored (so it does not
-            # halt/short circuit the graphicscene loop). Further the
-            # surrounding handler for this signal must be allowed to execute
-            # and get processed by **this consumer**.
-            # print(f'{vb.name} rx relayed from {relayed_from.name}')
-            ev.ignore()
-
-            return slot(
-                vb,
-                ev,
-                axis=axis,
-            )
-
-        if axis is not None:
-            # print(f'{vb.name} handling axis event:\n{str(ev)}')
-            ev.accept()
-            return slot(
-                vb,
-                ev,
-                axis=axis,
-            )
-
-        elif (
-            relayed_from is None
-            and vb.event_relay_source is vb  # we are the broadcaster
-            and axis is None
-        ):
-            # Broadcast case: this is a source event which will be
-            # relayed to attached consumers and accepted after all
-            # consumers complete their own handling followed by this
-            # routine's processing. Sequence is,
-            # - pre-relay to all consumers *first* - ``.emit()`` blocks
-            #   until all downstream relay handlers have run.
-            # - run the source handler for **this** event and accept
-            #   the event
-
-            # Access the "bound signal" that is created
-            # on the widget type as part of instantiation.
-            signal = getattr(vb, signame)
-            # print(f'{vb.name} emitting {signame}')
-
-            # TODO/NOTE: we could also just bypass a "relay" signal
-            # entirely and instead call the handlers manually in
-            # a loop? This probably is a lot simpler and also doesn't
-            # have any downside, and allows not touching target widget
-            # internals.
-            signal.emit(
-                ev,
-                axis,
-                # passing this demarks a broadcasted/relayed event
-                vb,
-            )
-            # accept event so no more relays are fired.
-            ev.accept()
-
-            # call underlying wrapped method with an extra
-            # ``relayed_from`` value to denote that this is a relayed
-            # event handling case.
-            return slot(
-                vb,
-                ev,
-                axis=axis,
-            )
-
-    return maybe_broadcast
-
-
-# XXX: :( can't define signals **after** class compile time
-# so this is not really useful.
-# def mk_relay_signal(
-#     func,
-#     name: str = None,
-
-# ) -> Signal:
-#     (
-#         args,
-#         varargs,
-#         varkw,
-#         defaults,
-#         kwonlyargs,
-#         kwonlydefaults,
-#         annotations
-#     ) = inspect.getfullargspec(func)
-
-#     # XXX: generate a relay signal with 1 extra
-#     # argument for a ``relayed_from`` kwarg. Since
-#     # ``'self'`` is already ignored by signals we just need
-#     # to count the arguments since we're adding only 1 (and
-#     # ``args`` will capture that).
-#     numargs = len(args + list(defaults))
-#     signal = Signal(*tuple(numargs * [object]))
-#     signame = name or func.__name__ + 'Relay'
-#     return signame, signal
-
-
-def enable_relays(
-    widget: GraphicsWidget,
-    handler_names: list[str],
-
-) -> list[Signal]:
-    '''
-    Method override helper which enables relay of a particular
-    ``Signal`` from some chosen broadcaster widget to a set of
-    consumer widgets which should operate their event handlers normally
-    but instead of signals "relayed" from the broadcaster.
-
-    Mostly useful for overlaying widgets that handle user input
-    that you want to overlay graphically. The target ``widget`` type must
-    define ``QtCore.Signal``s each with a `'Relay'` suffix for each
-    name provided in ``handler_names: list[str]``.
-
-    '''
-    signals = []
-    for name in handler_names:
-        handler = getattr(widget, name)
-        signame = name + 'Relay'
-        # ensure the target widget defines a relay signal
-        relay = getattr(widget, signame)
-        widget.relays[signame] = name
-        signals.append(relay)
-        method = mk_relay_method(signame, handler)
-        setattr(widget, name, method)
-
-    return signals
-
-
-enable_relays(
-    ChartView,
-    ['wheelEvent', 'mouseDragEvent']
-)
-
-
 class PlotItemOverlay:
     '''
     A composite for managing overlaid ``PlotItem`` instances such that
@@ -482,16 +296,18 @@ class PlotItemOverlay:
     ) -> None:
 
         self.root_plotitem: PlotItem = root_plotitem
+        self.relay_handlers: defaultdict[
+            str,
+            list[Callable],
+        ] = defaultdict(list)
 
-        vb = root_plotitem.vb
-        vb.event_relay_source = vb  # TODO: maybe change name?
-        vb.setZValue(1000)  # XXX: critical for scene layering/relaying
+        # NOTE: required for scene layering/relaying; this guarantees
+        # the "root" plot receives priority for interaction
+        # events/signals.
+        root_plotitem.vb.setZValue(1000)
 
         self.overlays: list[PlotItem] = []
-        self.layout = ComposedGridLayout(
-            root_plotitem,
-            root_plotitem.layout,
-        )
+        self.layout = ComposedGridLayout(root_plotitem)
         self._relays: dict[str, Signal] = {}
 
     def add_plotitem(
@@ -499,8 +315,10 @@ class PlotItemOverlay:
         plotitem: PlotItem,
         index: Optional[int] = None,
 
-        # TODO: we could also put the ``ViewBox.XAxis``
-        # style enum here?
+        # event/signal names which will be broadcasted to all added
+        # (relayee) ``PlotItem``s (eg. ``ViewBox.mouseDragEvent``).
+        relay_events: list[str] = [],
+
         # (0,),  # link x
         # (1,),  # link y
         # (0, 1),  # link both
@@ -510,58 +328,155 @@ class PlotItemOverlay:
 
         index = index or len(self.overlays)
         root = self.root_plotitem
-        # layout: QGraphicsGridLayout = root.layout
         self.overlays.insert(index, plotitem)
         vb: ViewBox = plotitem.vb
 
-        # mark this consumer overlay as ready to expect relayed events
-        # from the root plotitem.
-        vb.event_relay_source = root.vb
-
         # TODO: some sane way to allow menu event broadcast XD
         # vb.setMenuEnabled(False)
 
-        # TODO: inside the `maybe_broadcast()` (soon to be) decorator
-        # we need have checks that consumers have been attached to
-        # these relay signals.
-        if link_axes != (0, 1):
+        # wire up any relay signal(s) from the source plot to added
+        # "overlays". We use a plain loop instead of mucking with
+        # re-connecting signal/slots which tends to be more invasive and
+        # harder to implement and provides no measurable performance
+        # gain.
+        if relay_events:
+            for ev_name in relay_events:
+                relayee_handler: Callable[
+                    [
+                        ViewBox,
+                        # lol, there isn't really a generic type thanks
+                        # to the rewrite of Qt's event system XD
+                        QEvent,
 
-            # wire up relay signals
-            for relay_signal_name, handler_name in vb.relays.items():
-                # print(handler_name)
-                # XXX: Signal class attrs are bound after instantiation
-                # of the defining type, so we need to access that bound
-                # version here.
-                signal = getattr(root.vb, relay_signal_name)
-                handler = getattr(vb, handler_name)
-                signal.connect(handler)
+                        AxisItem | None,
+                    ],
+                    None,
+                ] = getattr(vb, ev_name)
+
+                sub_handlers: list[Callable] = self.relay_handlers[ev_name]
+
+                # on the first registry of a relayed event we pop the
+                # root's handler and override it to a custom broadcaster
+                # routine.
+                if not sub_handlers:
+
+                    src_handler = getattr(
+                        root.vb,
+                        ev_name,
+                    )
+
+                    def broadcast(
+                        ev: 'QEvent',
+
+                        # TODO: drop this viewbox specific input and
+                        # allow a predicate to be passed in by user.
+                        axis: 'Optional[int]' = None,
+
+                        *,
+
+                        # these are bound in by the ``partial`` below
+                        # and ensure a unique broadcaster per event.
+                        ev_name: str = None,
+                        src_handler: Callable = None,
+                        relayed_from: 'ViewBox' = None,
+
+                        # remaining inputs the source handler expects
+                        **kwargs,
+
+                    ) -> None:
+                        '''
+                        Broadcast signal or event: this is a source
+                        event which will be relayed to attached
+                        "relayee" plot item consumers.
+
+                        The event is accepted halting any further
+                        handlers from being triggered.
+
+                        Sequence is,
+                        - pre-relay to all consumers *first* - exactly
+                          like how a ``Signal.emit()`` blocks until all
+                          downstream relay handlers have run.
+                        - run the event's source handler event
+
+                        '''
+                        ev.accept()
+
+                        # broadcast first to relayees *first*. trigger
+                        # relay of event to all consumers **before**
+                        # processing/consumption in the source handler.
+                        relayed_handlers = self.relay_handlers[ev_name]
+
+                        assert getattr(vb, ev_name).__name__ == ev_name
+
+                        # TODO: generalize as an input predicate
+                        if axis is None:
+                            for handler in relayed_handlers:
+                                handler(
+                                    ev,
+                                    axis=axis,
+                                    **kwargs,
+                                )
+
+                        # run "source" widget's handler last
+                        src_handler(
+                            ev,
+                            axis=axis,
+                        )
+
+                    # dynamic handler override on the publisher plot
+                    setattr(
+                        root.vb,
+                        ev_name,
+                        partial(
+                            broadcast,
+                            ev_name=ev_name,
+                            src_handler=src_handler
+                        ),
+                    )
+
+                else:
+                    assert getattr(root.vb, ev_name)
+                    assert relayee_handler not in sub_handlers
+
+                # append relayed-to widget's handler to relay table
+                sub_handlers.append(relayee_handler)
 
         # link dim-axes to root if requested by user.
-        # TODO: solve more-then-wanted scaled panning on click drag
-        # which seems to be due to broadcast. So we probably need to
-        # disable broadcast when axes are linked in a particular
-        # dimension?
         for dim in link_axes:
             # link x and y axes to new view box such that the top level
             # viewbox propagates to the root (and whatever other
             # plotitem overlays that have been added).
             vb.linkView(dim, root.vb)
 
-        # make overlaid viewbox impossible to focus since the top
-        # level should handle all input and relay to overlays.
-        # NOTE: this was solved with the `setZValue()` above!
+        # => NOTE: in order to prevent "more-then-linear" scaled
+        # panning moves on (for eg. click-drag) certain range change
+        # signals (i.e. ``.sigXRangeChanged``), the user needs to be
+        # careful that any broadcasted ``relay_events`` are are short
+        # circuited in sub-handlers (aka relayee's) implementations. As
+        # an example if a ``ViewBox.mouseDragEvent`` is broadcasted, the
+        # overlayed implementations need to be sure they either don't
+        # also link the x-axes (by not providing ``link_axes=(0,)``
+        # above) or that the relayee ``.mouseDragEvent()`` handlers are
+        # ready to "``return`` early" in the case that
+        # ``.sigXRangeChanged`` is emitted as part of linked axes.
+        # For more details on such signalling mechanics peek in
+        # ``ViewBox.linkView()``.
 
-        # TODO: we will probably want to add a "focus" api such that
-        # a new "top level" ``PlotItem`` can be selected dynamically
-        # (and presumably the axes dynamically sorted to match).
+        # make overlaid viewbox impossible to focus since the top level
+        # should handle all input and relay to overlays. Note that the
+        # "root" plot item gettingn interaction priority is configured
+        # with the ``.setZValue()`` during init.
         vb.setFlag(
             vb.GraphicsItemFlag.ItemIsFocusable,
             False
         )
         vb.setFocusPolicy(Qt.NoFocus)
 
+        # => TODO: add a "focus" api for switching the "top level"
+        # ``PlotItem`` dynamically.
+
         # append-compose into the layout all axes from this plot
-        self.layout.insert(index, plotitem)
+        self.layout.insert_plotitem(index, plotitem)
 
         plotitem.setGeometry(root.vb.sceneBoundingRect())
 
@@ -579,25 +494,6 @@ class PlotItemOverlay:
         root.vb.setFocus()
         assert root.vb.focusWidget()
 
-    # XXX: do we need this? Why would you build then destroy?
-    def remove_plotitem(self, plotItem: PlotItem) -> None:
-        '''
-        Remove this ``PlotItem`` from the overlayed set making not shown
-        and unable to accept input.
-
-        '''
-        ...
-
-    # TODO: i think this would be super hot B)
-    def focus_item(self, plotitem: PlotItem) -> PlotItem:
-        '''
-        Apply focus to a contained PlotItem thus making it the "top level"
-        item in the overlay able to accept peripheral's input from the user
-        and responsible for zoom and panning control via its ``ViewBox``.
-
-        '''
-        ...
-
     def get_axis(
         self,
         plot: PlotItem,
@@ -630,8 +526,9 @@ class PlotItemOverlay:
 
         return axes
 
-    # TODO: i guess we need this if you want to detach existing plots
-    # dynamically? XXX: untested as of now.
+    # XXX: untested as of now.
+    # TODO: need this as part of selecting a different root/source
+    # plot to rewire interaction event broadcast dynamically.
     def _disconnect_all(
         self,
         plotitem: PlotItem,
@@ -646,3 +543,22 @@ class PlotItemOverlay:
             disconnected.append(sig)
 
         return disconnected
+
+    # XXX: do we need this? Why would you build then destroy?
+    # def remove_plotitem(self, plotItem: PlotItem) -> None:
+    #     '''
+    #     Remove this ``PlotItem`` from the overlayed set making not shown
+    #     and unable to accept input.
+
+    #     '''
+    #     ...
+
+    # TODO: i think this would be super hot B)
+    # def focus_plotitem(self, plotitem: PlotItem) -> PlotItem:
+    #     '''
+    #     Apply focus to a contained PlotItem thus making it the "top level"
+    #     item in the overlay able to accept peripheral's input from the user
+    #     and responsible for zoom and panning control via its ``ViewBox``.
+
+    #     '''
+    #     ...

piker/ui/_search.py

@@ -665,9 +665,9 @@ class SearchWidget(QtWidgets.QWidget):
 
         log.info(f'Requesting symbol: {symbol}.{provider}')
 
-        await godw.load_symbol(
+        await godw.load_symbols(
             provider,
-            symbol,
+            [symbol],
             'info',
         )
 

piker/ui/cli.py

@@ -46,8 +46,10 @@ def _kivy_import_hack():
 @click.argument('name', nargs=1, required=True)
 @click.pass_obj
 def monitor(config, rate, name, dhost, test, tl):
-    """Start a real-time watchlist UI
-    """
+    '''
+    Start a real-time watchlist UI
+
+    '''
     # global opts
     brokermod = config['brokermods'][0]
     loglevel = config['loglevel']
@@ -70,8 +72,12 @@ def monitor(config, rate, name, dhost, test, tl):
         ) as portal:
             # run app "main"
             await _async_main(
-                name, portal, tickers,
-                brokermod, rate, test=test,
+                name,
+                portal,
+                tickers,
+                brokermod,
+                rate,
+                test=test,
             )
 
     tractor.run(
@@ -122,7 +128,7 @@ def optschain(config, symbol, date, rate, test):
 @cli.command()
 @click.option(
     '--profile',
-    '-p',
+    # '-p',
     default=None,
     help='Enable pyqtgraph profiling'
 )
@@ -131,9 +137,14 @@ def optschain(config, symbol, date, rate, test):
     is_flag=True,
     help='Enable tractor debug mode'
 )
-@click.argument('symbol', required=True)
+@click.argument('symbols', nargs=-1, required=True)
 @click.pass_obj
-def chart(config, symbol, profile, pdb):
+def chart(
+    config,
+    symbols: list[str],
+    profile,
+    pdb: bool,
+):
     '''
     Start a real-time chartng UI
 
@@ -144,14 +155,16 @@ def chart(config, symbol, profile, pdb):
         _profile._pg_profile = True
         _profile.ms_slower_then = float(profile)
 
+    # Qt UI entrypoint
     from ._app import _main
 
-    if '.' not in symbol:
-        click.echo(click.style(
-            f'symbol: {symbol} must have a {symbol}.<provider> suffix',
-            fg='red',
-        ))
-        return
+    for symbol in symbols:
+        if '.' not in symbol:
+            click.echo(click.style(
+                f'symbol: {symbol} must have a {symbol}.<provider> suffix',
+                fg='red',
+            ))
+            return
 
 
     # global opts
@@ -159,8 +172,9 @@ def chart(config, symbol, profile, pdb):
     tractorloglevel = config['tractorloglevel']
     pikerloglevel = config['loglevel']
 
+
     _main(
-        sym=symbol,
+        syms=symbols,
         brokernames=brokernames,
         piker_loglevel=pikerloglevel,
         tractor_kwargs={
@@ -170,5 +184,6 @@ def chart(config, symbol, profile, pdb):
             'enable_modules': [
                 'piker.clearing._client'
             ],
+            'registry_addr': config.get('registry_addr'),
         },
     )

tests/conftest.py

@@ -14,15 +14,6 @@ def pytest_addoption(parser):
                      help="Use a practice API account")
 
 
-@pytest.fixture(scope='session', autouse=True)
-def loglevel(request):
-    orig = tractor.log._default_loglevel
-    level = tractor.log._default_loglevel = request.config.option.loglevel
-    log.get_console_log(level)
-    yield level
-    tractor.log._default_loglevel = orig
-
-
 @pytest.fixture(scope='session')
 def test_config():
     dirname = os.path.dirname

tests/test_feeds.py

@@ -0,0 +1,65 @@
+'''
+Data feed layer APIs, performance, msg throttling.
+
+'''
+from pprint import pprint
+
+import pytest
+import trio
+from piker import (
+    open_piker_runtime,
+    open_feed,
+)
+from piker.data import ShmArray
+
+
+@pytest.mark.parametrize(
+    'fqsns',
+    [
+        ['btcusdt.binance']
+    ],
+    ids=lambda param: f'fqsns={param}',
+)
+def test_basic_rt_feed(
+    fqsns: list[str],
+):
+    '''
+    Start a real-time data feed for provided fqsn and pull
+    a few quotes then simply shut down.
+
+    '''
+    async def main():
+        async with (
+            open_piker_runtime('test_basic_rt_feed'),
+            open_feed(
+                fqsns,
+                loglevel='info',
+
+                # TODO: ensure throttle rate is applied
+                # limit to at least display's FPS
+                # avoiding needless Qt-in-guest-mode context switches
+                # tick_throttle=_quote_throttle_rate,
+
+            ) as feed
+        ):
+            for fqin in fqsns:
+                assert feed.symbols[fqin]
+
+            ohlcv: ShmArray = feed.rt_shm
+            hist_ohlcv: ShmArray = feed.hist_shm
+
+            count: int = 0
+            async for quotes in feed.stream:
+
+                # print quote msg, rt and history
+                # buffer values on console.
+                pprint(quotes)
+                pprint(ohlcv.array[-1])
+                pprint(hist_ohlcv.array[-1])
+
+                if count >= 100:
+                    break
+
+                count += 1
+
+    trio.run(main)