Exclude crypto futes from `without_src` sym key

Extend the `col_sym_key` asset-type check in `start_backfill()`
to also exclude crypto-denominated futures (where `src` is
`'crypto_currency'` and `dst` is `'future'`) from the
`without_src=True` fqme path.

Also in `.brokers.binance` backend (it being the guilty culprit in the
discovery of this bug; and why i touched styling this code),

- reformat `make_sub()` fn sig to multiline style in
  `.binance.feed`.
- add backtick around `dict` in `make_sub()` docstring.
- reformat `or` conditionals to multiline style in
  `.binance.feed.get_mkt_info()`.

(this commit msg was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code

piker/brokers/binance/feed.py

@@ -203,9 +203,13 @@ async def stream_messages(
                 yield 'trade', piker_quote
 
 
-def make_sub(pairs: list[str], sub_name: str, uid: int) -> dict[str, str]:
+def make_sub(
+    pairs: list[str],
+    sub_name: str,
+    uid: int,
+) -> dict[str, str]:
     '''
-    Create a request subscription packet dict.
+    Create a request subscription packet `dict`.
 
     - spot:
       https://binance-docs.github.io/apidocs/spot/en/#live-subscribing-unsubscribing-to-streams
@@ -332,7 +336,8 @@ async def get_mkt_info(
         # TODO: handle coinm futes which have a margin asset that
         # is some crypto token!
         # https://binance-docs.github.io/apidocs/delivery/en/#exchange-information
-        or 'btc' in venue_lower
+        or
+        'btc' in venue_lower
     ):
         return None
 
@@ -343,12 +348,14 @@ async def get_mkt_info(
 
     if (
         venue
-        and 'spot' not in venue_lower
+        and
+        'spot' not in venue_lower
 
         # XXX: catch all in case user doesn't know which
         # venue they want (usdtm vs. coinm) and we can choose
         # a default (via config?) once we support coin-m APIs.
-        or 'perp' in venue_lower
+        or
+        'perp' in venue_lower
     ):
         if not mkt_mode:
             mkt_mode: str = f'{venue_lower}_futes'

piker/brokers/deribit/api.py

@@ -586,7 +586,7 @@ async def open_price_feed(
                 fh,
                 instrument
             )
-        ) as (first, chan):
+        ) as (chan, first):
             yield chan
 
 
@@ -653,7 +653,7 @@ async def open_order_feed(
                 fh,
                 instrument
             )
-        ) as (first, chan):
+        ) as (chan, first):
             yield chan
 
 

piker/brokers/deribit/feed.py

@@ -32,7 +32,7 @@ import tractor
 
 from piker.brokers import open_cached_client
 from piker.log import get_logger, get_console_log
-from piker.data import ShmArray
+from tractor.ipc._shm import ShmArray
 from piker.brokers._util import (
     BrokerError,
     DataUnavailable,

piker/brokers/ib/api.py

@@ -1529,7 +1529,7 @@ async def open_client_proxies() -> tuple[
             # TODO: maybe this should be the default in tractor?
             key=tractor.current_actor().uid,
 
-        ) as (cache_hit, (clients, _)),
+        ) as (cache_hit, (_, clients)),
 
         AsyncExitStack() as stack
     ):
@@ -1718,7 +1718,7 @@ async def open_client_proxy(
             open_aio_client_method_relay,
             client=client,
             event_consumers=event_table,
-        ) as (first, chan),
+        ) as (chan, first),
 
         trionics.collapse_eg(),  # loose-ify
         trio.open_nursery() as relay_tn,

piker/brokers/ib/broker.py

@@ -514,8 +514,8 @@ async def open_trade_event_stream(
         recv_trade_updates,
         client=client,
     ) as (
-        _,  # first pushed val
         trade_event_stream,
+        _,  # first pushed val
     ):
         task_status.started(trade_event_stream)
         # block forever to keep session trio-asyncio session

piker/brokers/ib/feed.py

@@ -989,7 +989,7 @@ async def open_aio_quote_stream(
         symbol=symbol,
         contract=contract,
 
-    ) as (contract, from_aio):
+    ) as (from_aio, contract):
 
         assert contract
 

piker/data/__init__.py

@@ -23,13 +23,13 @@ sharing live streams over a network.
 
 """
 from .ticktools import iterticks
-from ._sharedmem import (
+from tractor.ipc._shm import (
-    maybe_open_shm_array,
-    attach_shm_array,
-    open_shm_array,
-    get_shm_token,
     ShmArray,
+    get_shm_token,
+    open_shm_ndarray as open_shm_array,
+    attach_shm_ndarray as attach_shm_array,
 )
+from ._sharedmem import maybe_open_shm_array
 from ._source import (
     def_iohlcv_fields,
     def_ohlcv_fields,

piker/data/_formatters.py

@@ -28,9 +28,7 @@ from msgspec import field
 import numpy as np
 from numpy.lib import recfunctions as rfn
 
-from ._sharedmem import (
+from tractor.ipc._shm import ShmArray
-    ShmArray,
-)
 from ._pathops import (
     path_arrays_from_ohlc,
 )

piker/data/_sampling.py

@@ -55,9 +55,7 @@ from ._util import (
 from ..service import maybe_spawn_daemon
 
 if TYPE_CHECKING:
-    from ._sharedmem import (
+    from tractor.ipc._shm import ShmArray
-        ShmArray,
-    )
     from .feed import (
         _FeedsBus,
         Sub,
@@ -378,16 +376,16 @@ async def register_with_sampler(
                 # feed_is_live.is_set()
                 # ^TODO? pass it in instead?
             ):
-                from ._sharedmem import (
+                from tractor.ipc._shm import (
-                    attach_shm_array,
+                    attach_shm_ndarray,
-                    _Token,
+                    NDToken,
                 )
                 for period in shms_by_period:
 
                     # load and register shm handles
                     shm_token_msg = shms_by_period[period]
-                    shm = attach_shm_array(
+                    shm = attach_shm_ndarray(
-                        _Token.from_msg(shm_token_msg),
+                        NDToken.from_msg(shm_token_msg),
                         readonly=False,
                     )
                     shms_by_period[period] = shm

piker/data/_sharedmem.py

@@ -1,661 +1,106 @@
 # 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
+# This program is free software: you can redistribute it
-# it under the terms of the GNU Affero General Public License as published by
+# and/or modify it under the terms of the GNU Affero General
-# the Free Software Foundation, either version 3 of the License, or
+# Public License as published by the Free Software
-# (at your option) any later version.
+# 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,
+# This program is distributed in the hope that it will be
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# useful, but WITHOUT ANY WARRANTY; without even the implied
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
-# GNU Affero General Public License for more details.
+# PURPOSE.  See the GNU Affero General Public License for
+# more details.
 
-# You should have received a copy of the GNU Affero General Public License
+# You should have received a copy of the GNU Affero General
-# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+# Public License along with this program.  If not, see
+# <https://www.gnu.org/licenses/>.
 
-"""
+'''
-NumPy compatible shared memory buffers for real-time IPC streaming.
+Piker-specific shared memory helpers.
 
-"""
+Thin shim providing piker-only wrappers around
-from __future__ import annotations
+``tractor.ipc._shm``; all core types and functions
-from sys import byteorder
+are now imported directly from tractor throughout
-import time
+the codebase.
-from typing import Optional
-from multiprocessing.shared_memory import SharedMemory, _USE_POSIX
 
-if _USE_POSIX:
+'''
-    from _posixshmem import shm_unlink
-
-# import msgspec
 import numpy as np
-from numpy.lib import recfunctions as rfn
+
-import tractor
+from tractor.ipc._shm import (
+    NDToken,
+    ShmArray,
+    _known_tokens,
+    _make_token as _tractor_make_token,
+    open_shm_ndarray,
+    attach_shm_ndarray,
+)
 
 from ._util import log
-from ._source import def_iohlcv_fields
-from piker.types import Struct
-
-
-def cuckoff_mantracker():
-    '''
-    Disable all ``multiprocessing``` "resource tracking" machinery since
-    it's an absolute multi-threaded mess of non-SC madness.
-
-    '''
-    from multiprocessing import resource_tracker as mantracker
-
-    # Tell the "resource tracker" thing to fuck off.
-    class ManTracker(mantracker.ResourceTracker):
-        def register(self, name, rtype):
-            pass
-
-        def unregister(self, name, rtype):
-            pass
-
-        def ensure_running(self):
-            pass
-
-    # "know your land and know your prey"
-    # https://www.dailymotion.com/video/x6ozzco
-    mantracker._resource_tracker = ManTracker()
-    mantracker.register = mantracker._resource_tracker.register
-    mantracker.ensure_running = mantracker._resource_tracker.ensure_running
-    mantracker.unregister = mantracker._resource_tracker.unregister
-    mantracker.getfd = mantracker._resource_tracker.getfd
-
-
-cuckoff_mantracker()
-
-
-class SharedInt:
-    """Wrapper around a single entry shared memory array which
-    holds an ``int`` value used as an index counter.
-
-    """
-    def __init__(
-        self,
-        shm: SharedMemory,
-    ) -> None:
-        self._shm = shm
-
-    @property
-    def value(self) -> int:
-        return int.from_bytes(self._shm.buf, byteorder)
-
-    @value.setter
-    def value(self, value) -> None:
-        self._shm.buf[:] = value.to_bytes(self._shm.size, byteorder)
-
-    def destroy(self) -> None:
-        if _USE_POSIX:
-            # We manually unlink to bypass all the "resource tracker"
-            # nonsense meant for non-SC systems.
-            name = self._shm.name
-            try:
-                shm_unlink(name)
-            except FileNotFoundError:
-                # might be a teardown race here?
-                log.warning(f'Shm for {name} already unlinked?')
-
-
-class _Token(Struct, frozen=True):
-    '''
-    Internal represenation of a shared memory "token"
-    which can be used to key a system wide post shm entry.
-
-    '''
-    shm_name: str  # this servers as a "key" value
-    shm_first_index_name: str
-    shm_last_index_name: str
-    dtype_descr: tuple
-    size: int  # in struct-array index / row terms
-
-    @property
-    def dtype(self) -> np.dtype:
-        return np.dtype(list(map(tuple, self.dtype_descr))).descr
-
-    def as_msg(self):
-        return self.to_dict()
-
-    @classmethod
-    def from_msg(cls, msg: dict) -> _Token:
-        if isinstance(msg, _Token):
-            return msg
-
-        # TODO: native struct decoding
-        # return _token_dec.decode(msg)
-
-        msg['dtype_descr'] = tuple(map(tuple, msg['dtype_descr']))
-        return _Token(**msg)
-
-
-# _token_dec = msgspec.msgpack.Decoder(_Token)
-
-# TODO: this api?
-# _known_tokens = tractor.ActorVar('_shm_tokens', {})
-# _known_tokens = tractor.ContextStack('_known_tokens', )
-# _known_tokens = trio.RunVar('shms', {})
-
-# process-local store of keys to tokens
-_known_tokens = {}
-
-
-def get_shm_token(key: str) -> _Token:
-    """Convenience func to check if a token
-    for the provided key is known by this process.
-    """
-    return _known_tokens.get(key)
 
 
 def _make_token(
     key: str,
     size: int,
-    dtype: Optional[np.dtype] = None,
+    dtype: np.dtype|None = None,
-) -> _Token:
+) -> NDToken:
     '''
-    Create a serializable token that can be used
+    Wrap tractor's ``_make_token()`` with piker's
-    to access a shared array.
+    default dtype fallback to ``def_iohlcv_fields``.
 
     '''
-    dtype = def_iohlcv_fields if dtype is None else dtype
+    from ._source import def_iohlcv_fields
-    return _Token(
+    dtype = (
-        shm_name=key,
+        def_iohlcv_fields
-        shm_first_index_name=key + "_first",
+        if dtype is None
-        shm_last_index_name=key + "_last",
+        else dtype
-        dtype_descr=tuple(np.dtype(dtype).descr),
-        size=size,
     )
-
+    return _tractor_make_token(
-
-class ShmArray:
-    '''
-    A shared memory ``numpy`` (compatible) array API.
-
-    An underlying shared memory buffer is allocated based on
-    a user specified ``numpy.ndarray``. This fixed size array
-    can be read and written to by pushing data both onto the "front"
-    or "back" of a set index range. The indexes for the "first" and
-    "last" index are themselves stored in shared memory (accessed via
-    ``SharedInt`` interfaces) values such that multiple processes can
-    interact with the same array using a synchronized-index.
-
-    '''
-    def __init__(
-        self,
-        shmarr: np.ndarray,
-        first: SharedInt,
-        last: SharedInt,
-        shm: SharedMemory,
-        # readonly: bool = True,
-    ) -> None:
-        self._array = shmarr
-
-        # indexes for first and last indices corresponding
-        # to fille data
-        self._first = first
-        self._last = last
-
-        self._len = len(shmarr)
-        self._shm = shm
-        self._post_init: bool = False
-
-        # pushing data does not write the index (aka primary key)
-        dtype = shmarr.dtype
-        if dtype.fields:
-            self._write_fields = list(shmarr.dtype.fields.keys())[1:]
-        else:
-            self._write_fields = None
-
-    # TODO: ringbuf api?
-
-    @property
-    def _token(self) -> _Token:
-        return _Token(
-            shm_name=self._shm.name,
-            shm_first_index_name=self._first._shm.name,
-            shm_last_index_name=self._last._shm.name,
-            dtype_descr=tuple(self._array.dtype.descr),
-            size=self._len,
-        )
-
-    @property
-    def token(self) -> dict:
-        """Shared memory token that can be serialized and used by
-        another process to attach to this array.
-        """
-        return self._token.as_msg()
-
-    @property
-    def index(self) -> int:
-        return self._last.value % self._len
-
-    @property
-    def array(self) -> np.ndarray:
-        '''
-        Return an up-to-date ``np.ndarray`` view of the
-        so-far-written data to the underlying shm buffer.
-
-        '''
-        a = self._array[self._first.value:self._last.value]
-
-        # first, last = self._first.value, self._last.value
-        # a = self._array[first:last]
-
-        # TODO: eventually comment this once we've not seen it in the
-        # wild in a long time..
-        # XXX: race where first/last indexes cause a reader
-        # to load an empty array..
-        if len(a) == 0 and self._post_init:
-            raise RuntimeError('Empty array race condition hit!?')
-
-        return a
-
-    def ustruct(
-        self,
-        fields: Optional[list[str]] = None,
-
-        # type that all field values will be cast to
-        # in the returned view.
-        common_dtype: np.dtype = float,
-
-    ) -> np.ndarray:
-
-        array = self._array
-
-        if fields:
-            selection = array[fields]
-            # fcount = len(fields)
-        else:
-            selection = array
-            # fcount = len(array.dtype.fields)
-
-        # XXX: manual ``.view()`` attempt that also doesn't work.
-        # uview = selection.view(
-        #     dtype='<f16',
-        # ).reshape(-1, 4, order='A')
-
-        # assert len(selection) == len(uview)
-
-        u = rfn.structured_to_unstructured(
-            selection,
-            # dtype=float,
-            copy=True,
-        )
-
-        # unstruct = np.ndarray(u.shape, dtype=a.dtype, buffer=shm.buf)
-        # array[:] = a[:]
-        return u
-        # return ShmArray(
-        #     shmarr=u,
-        #     first=self._first,
-        #     last=self._last,
-        #     shm=self._shm
-        # )
-
-    def last(
-        self,
-        length: int = 1,
-
-    ) -> np.ndarray:
-        '''
-        Return the last ``length``'s worth of ("row") entries from the
-        array.
-
-        '''
-        return self.array[-length:]
-
-    def push(
-        self,
-        data: np.ndarray,
-
-        field_map: Optional[dict[str, str]] = None,
-        prepend: bool = False,
-        update_first: bool = True,
-        start: int | None = None,
-
-    ) -> int:
-        '''
-        Ring buffer like "push" to append data
-        into the buffer and return updated "last" index.
-
-        NB: no actual ring logic yet to give a "loop around" on overflow
-        condition, lel.
-
-        '''
-        length = len(data)
-
-        if prepend:
-            index = (start or self._first.value) - length
-
-            if index < 0:
-                raise ValueError(
-                    f'Array size of {self._len} was overrun during prepend.\n'
-                    f'You have passed {abs(index)} too many datums.'
-                )
-
-        else:
-            index = start if start is not None else self._last.value
-
-        end = index + length
-
-        if field_map:
-            src_names, dst_names = zip(*field_map.items())
-        else:
-            dst_names = src_names = self._write_fields
-
-        try:
-            self._array[
-                list(dst_names)
-            ][index:end] = data[list(src_names)][:]
-
-            # NOTE: there was a race here between updating
-            # the first and last indices and when the next reader
-            # tries to access ``.array`` (which due to the index
-            # overlap will be empty). Pretty sure we've fixed it now
-            # but leaving this here as a reminder.
-            if (
-                prepend
-                and update_first
-                and length
-            ):
-                assert index < self._first.value
-
-            if (
-                index < self._first.value
-                and update_first
-            ):
-                assert prepend, 'prepend=True not passed but index decreased?'
-                self._first.value = index
-
-            elif not prepend:
-                self._last.value = end
-
-            self._post_init = True
-            return end
-
-        except ValueError as err:
-            if field_map:
-                raise
-
-            # should raise if diff detected
-            self.diff_err_fields(data)
-            raise err
-
-    def diff_err_fields(
-        self,
-        data: np.ndarray,
-    ) -> None:
-        # reraise with any field discrepancy
-        our_fields, their_fields = (
-            set(self._array.dtype.fields),
-            set(data.dtype.fields),
-        )
-
-        only_in_ours = our_fields - their_fields
-        only_in_theirs = their_fields - our_fields
-
-        if only_in_ours:
-            raise TypeError(
-                f"Input array is missing field(s): {only_in_ours}"
-            )
-        elif only_in_theirs:
-            raise TypeError(
-                f"Input array has unknown field(s): {only_in_theirs}"
-            )
-
-    # TODO: support "silent" prepends that don't update ._first.value?
-    def prepend(
-        self,
-        data: np.ndarray,
-    ) -> int:
-        end = self.push(data, prepend=True)
-        assert end
-
-    def close(self) -> None:
-        self._first._shm.close()
-        self._last._shm.close()
-        self._shm.close()
-
-    def destroy(self) -> None:
-        if _USE_POSIX:
-            # We manually unlink to bypass all the "resource tracker"
-            # nonsense meant for non-SC systems.
-            shm_unlink(self._shm.name)
-
-        self._first.destroy()
-        self._last.destroy()
-
-    def flush(self) -> None:
-        # TODO: flush to storage backend like markestore?
-        ...
-
-
-def open_shm_array(
-    size: int,
-    key: str | None = None,
-    dtype: np.dtype | None = None,
-    append_start_index: int | None = None,
-    readonly: bool = False,
-
-) -> ShmArray:
-    '''Open a memory shared ``numpy`` using the standard library.
-
-    This call unlinks (aka permanently destroys) the buffer on teardown
-    and thus should be used from the parent-most accessor (process).
-
-    '''
-    # create new shared mem segment for which we
-    # have write permission
-    a = np.zeros(size, dtype=dtype)
-    a['index'] = np.arange(len(a))
-
-    shm = SharedMemory(
-        name=key,
-        create=True,
-        size=a.nbytes
-    )
-    array = np.ndarray(
-        a.shape,
-        dtype=a.dtype,
-        buffer=shm.buf
-    )
-    array[:] = a[:]
-    array.setflags(write=int(not readonly))
-
-    token = _make_token(
         key=key,
         size=size,
         dtype=dtype,
     )
 
-    # create single entry arrays for storing an first and last indices
-    first = SharedInt(
-        shm=SharedMemory(
-            name=token.shm_first_index_name,
-            create=True,
-            size=4,  # std int
-        )
-    )
-
-    last = SharedInt(
-        shm=SharedMemory(
-            name=token.shm_last_index_name,
-            create=True,
-            size=4,  # std int
-        )
-    )
-
-    # start the "real-time" updated section after 3-days worth of 1s
-    # sampled OHLC. this allows appending up to a days worth from
-    # tick/quote feeds before having to flush to a (tsdb) storage
-    # backend, and looks something like,
-    # -------------------------
-    # |              |        i
-    # _________________________
-    # <-------------> <------->
-    #  history         real-time
-    #
-    # Once fully "prepended", the history section will leave the
-    # ``ShmArray._start.value: int = 0`` and the yet-to-be written
-    # real-time section will start at ``ShmArray.index: int``.
-
-    # this sets the index to nearly 2/3rds into the the length of
-    # the buffer leaving at least a "days worth of second samples"
-    # for the real-time section.
-    if append_start_index is None:
-        append_start_index = round(size * 0.616)
-
-    last.value = first.value = append_start_index
-
-    shmarr = ShmArray(
-        array,
-        first,
-        last,
-        shm,
-    )
-
-    assert shmarr._token == token
-    _known_tokens[key] = shmarr.token
-
-    # "unlink" created shm on process teardown by
-    # pushing teardown calls onto actor context stack
-    stack = tractor.current_actor(
-        err_on_no_runtime=False,
-    ).lifetime_stack
-    if stack:
-        stack.callback(shmarr.close)
-        stack.callback(shmarr.destroy)
-
-    return shmarr
-
-
-def attach_shm_array(
-    token: tuple[str, str, tuple[str, str]],
-    readonly: bool = True,
-
-) -> ShmArray:
-    '''
-    Attach to an existing shared memory array previously
-    created by another process using ``open_shared_array``.
-
-    No new shared mem is allocated but wrapper types for read/write
-    access are constructed.
-
-    '''
-    token = _Token.from_msg(token)
-    key = token.shm_name
-
-    if key in _known_tokens:
-        assert _Token.from_msg(_known_tokens[key]) == token, "WTF"
-
-    # XXX: ugh, looks like due to the ``shm_open()`` C api we can't
-    # actually place files in a subdir, see discussion here:
-    # https://stackoverflow.com/a/11103289
-
-    # attach to array buffer and view as per dtype
-    _err: Optional[Exception] = None
-    for _ in range(3):
-        try:
-            shm = SharedMemory(
-                name=key,
-                create=False,
-            )
-            break
-        except OSError as oserr:
-            _err = oserr
-            time.sleep(0.1)
-    else:
-        if _err:
-            raise _err
-
-    shmarr = np.ndarray(
-        (token.size,),
-        dtype=token.dtype,
-        buffer=shm.buf
-    )
-    shmarr.setflags(write=int(not readonly))
-
-    first = SharedInt(
-        shm=SharedMemory(
-            name=token.shm_first_index_name,
-            create=False,
-            size=4,  # std int
-        ),
-    )
-    last = SharedInt(
-        shm=SharedMemory(
-            name=token.shm_last_index_name,
-            create=False,
-            size=4,  # std int
-        ),
-    )
-
-    # make sure we can read
-    first.value
-
-    sha = ShmArray(
-        shmarr,
-        first,
-        last,
-        shm,
-    )
-    # read test
-    sha.array
-
-    # Stash key -> token knowledge for future queries
-    # via `maybe_opepn_shm_array()` but only after we know
-    # we can attach.
-    if key not in _known_tokens:
-        _known_tokens[key] = token
-
-    # "close" attached shm on actor teardown
-    if (actor := tractor.current_actor(
-        err_on_no_runtime=False,
-    )):
-        actor.lifetime_stack.callback(sha.close)
-
-    return sha
-
 
 def maybe_open_shm_array(
     key: str,
     size: int,
-    dtype: np.dtype | None = None,
+    dtype: np.dtype|None = None,
-    append_start_index: int | None = None,
+    append_start_index: int|None = None,
     readonly: bool = False,
     **kwargs,
-
 ) -> tuple[ShmArray, bool]:
     '''
-    Attempt to attach to a shared memory block using a "key" lookup
+    Attempt to attach to a shared memory block
-    to registered blocks in the users overall "system" registry
+    using a "key" lookup to registered blocks in
-    (presumes you don't have the block's explicit token).
+    the user's overall "system" registry (presumes
+    you don't have the block's explicit token).
 
-    This function is meant to solve the problem of discovering whether
+    This is a thin wrapper around tractor's
-    a shared array token has been allocated or discovered by the actor
+    ``maybe_open_shm_ndarray()`` preserving piker's
-    running in **this** process. Systems where multiple actors may seek
+    historical defaults (``readonly=False``,
-    to access a common block can use this function to attempt to acquire
+    ``append_start_index=None``).
-    a token as discovered by the actors who have previously stored
-    a "key" -> ``_Token`` map in an actor local (aka python global)
-    variable.
 
-    If you know the explicit ``_Token`` for your memory segment instead
+    If you know the explicit ``NDToken`` for your
-    use ``attach_shm_array``.
+    memory segment instead use
+    ``tractor.ipc._shm.attach_shm_ndarray()``.
 
     '''
     try:
         # see if we already know this key
         token = _known_tokens[key]
         return (
-            attach_shm_array(
+            attach_shm_ndarray(
                 token=token,
                 readonly=readonly,
             ),
             False,
         )
     except KeyError:
-        log.debug(f"Could not find {key} in shms cache")
+        log.debug(
+            f'Could not find {key} in shms cache'
+        )
         if dtype:
             token = _make_token(
                 key,
@@ -663,9 +108,18 @@ def maybe_open_shm_array(
                 dtype=dtype,
             )
             try:
-                return attach_shm_array(token=token, **kwargs), False
+                return (
+                    attach_shm_ndarray(
+                        token=token,
+                        **kwargs,
+                    ),
+                    False,
+                )
             except FileNotFoundError:
-                log.debug(f"Could not attach to shm with token {token}")
+                log.debug(
+                    f'Could not attach to shm'
+                    f' with token {token}'
+                )
 
         # This actor does not know about memory
         # associated with the provided "key".
@@ -673,7 +127,7 @@ def maybe_open_shm_array(
         # to fail if a block has been allocated
         # on the OS by someone else.
         return (
-            open_shm_array(
+            open_shm_ndarray(
                 key=key,
                 size=size,
                 dtype=dtype,
@@ -683,18 +137,20 @@ def maybe_open_shm_array(
             True,
         )
 
+
 def try_read(
-    array: np.ndarray
+    array: np.ndarray,
-
+) -> np.ndarray|None:
-) -> Optional[np.ndarray]:
     '''
-    Try to read the last row from a shared mem array or ``None``
+    Try to read the last row from a shared mem
-    if the array read returns a zero-length array result.
+    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
+    Can be used to check for backfilling race
-    is currently being (re-)written by a writer actor but the reader is
+    conditions where an array is currently being
-    unaware and reads during the window where the first and last indexes
+    (re-)written by a writer actor but the reader
-    are being updated.
+    is unaware and reads during the window where
+    the first and last indexes are being updated.
 
     '''
     try:
@@ -702,14 +158,13 @@ def try_read(
     except IndexError:
         # XXX: race condition with backfilling shm.
         #
-        # the underlying issue is that a backfill (aka prepend) and subsequent
+        # the underlying issue is that a backfill
-        # shm array first/last index update could result in an empty array
+        # (aka prepend) and subsequent shm array
-        # read here since the indices may be updated in such a way that
+        # first/last index update could result in an
-        # a read delivers an empty array (though it seems like we
+        # empty array read here since the indices may
-        # *should* be able to prevent that?). also, as and alt and
+        # be updated in such a way that a read delivers
-        # something we need anyway, maybe there should be some kind of
+        # an empty array (though it seems like we
-        # signal that a prepend is taking place and this consumer can
+        # *should* be able to prevent that?).
-        # respond (eg. redrawing graphics) accordingly.
 
-        # the array read was emtpy
+        # the array read was empty
         return None

piker/data/feed.py

@@ -973,9 +973,6 @@ async def open_feed(
                     # assert flume.mkt.fqme == fqme
                     feed.flumes[fqme] = flume
 
-                    # TODO: do we need this?
-                    flume.feed = feed
-
                     # attach and cache shm handles
                     rt_shm = flume.rt_shm
                     assert rt_shm

piker/data/flows.py

@@ -22,25 +22,19 @@ real-time data processing data-structures.
 
 """
 from __future__ import annotations
-from typing import (
-    TYPE_CHECKING,
-)
 
 import tractor
 import pendulum
 import numpy as np
 
 from piker.types import Struct
-from ._sharedmem import (
+from tractor.ipc._shm import (
-    attach_shm_array,
     ShmArray,
-    _Token,
+    NDToken,
+    attach_shm_ndarray,
 )
 from piker.accounting import MktPair
 
-if TYPE_CHECKING:
-    from piker.data.feed import Feed
-
 
 class Flume(Struct):
     '''
@@ -64,11 +58,11 @@ class Flume(Struct):
     '''
     mkt: MktPair
     first_quote: dict
-    _rt_shm_token: _Token
+    _rt_shm_token: NDToken
 
     # optional since some data flows won't have a "downsampled" history
     # buffer/stream (eg. FSPs).
-    _hist_shm_token: _Token | None = None
+    _hist_shm_token: NDToken|None = None
 
     # private shm refs loaded dynamically from tokens
     _hist_shm: ShmArray | None = None
@@ -80,15 +74,11 @@ class Flume(Struct):
     izero_rt: int = 0
     throttle_rate: int | None = None
 
-    # TODO: do we need this really if we can pull the `Portal` from
-    # ``tractor``'s internals?
-    feed: Feed|None = None
-
     @property
     def rt_shm(self) -> ShmArray:
 
         if self._rt_shm is None:
-            self._rt_shm = attach_shm_array(
+            self._rt_shm = attach_shm_ndarray(
                 token=self._rt_shm_token,
                 readonly=self._readonly,
             )
@@ -104,7 +94,7 @@ class Flume(Struct):
             )
 
         if self._hist_shm is None:
-            self._hist_shm = attach_shm_array(
+            self._hist_shm = attach_shm_ndarray(
                 token=self._hist_shm_token,
                 readonly=self._readonly,
             )
@@ -156,7 +146,6 @@ class Flume(Struct):
         # will get instead some kind of msg-compat version
         # that it can load.
         msg.pop('stream')
-        msg.pop('feed')
         msg.pop('_rt_shm')
         msg.pop('_hist_shm')
 

piker/fsp/_api.py

@@ -37,12 +37,12 @@ import numpy as np
 import tractor
 from tractor.msg import NamespacePath
 
-from ..data._sharedmem import (
+from tractor.ipc._shm import (
     ShmArray,
-    maybe_open_shm_array,
+    NDToken,
-    attach_shm_array,
+    attach_shm_ndarray,
-    _Token,
 )
+from ..data._sharedmem import maybe_open_shm_array
 from ..log import get_logger
 
 log = get_logger(__name__)
@@ -78,8 +78,8 @@ class Fsp:
     # + the consuming fsp *to* the consumers output
     # shm flow.
     _flow_registry: dict[
-        tuple[_Token, str],
+        tuple[NDToken, str],
-        tuple[_Token, Optional[ShmArray]],
+        tuple[NDToken, Optional[ShmArray]],
     ] = {}
 
     def __init__(
@@ -148,7 +148,7 @@ class Fsp:
                 # times as possible as per:
                 # - https://github.com/pikers/piker/issues/359
                 # - https://github.com/pikers/piker/issues/332
-                maybe_array := attach_shm_array(dst_token)
+                maybe_array := attach_shm_ndarray(dst_token)
             )
 
         return maybe_array

piker/fsp/_engine.py

@@ -40,7 +40,7 @@ from ..log import (
 )
 from .. import data
 from ..data.flows import Flume
-from ..data._sharedmem import ShmArray
+from tractor.ipc._shm import ShmArray
 from ..data._sampling import (
     _default_delay_s,
     open_sample_stream,
@@ -49,7 +49,7 @@ from ..accounting import MktPair
 from ._api import (
     Fsp,
     _load_builtins,
-    _Token,
+    NDToken,
 )
 from ..toolz import Profiler
 
@@ -414,7 +414,7 @@ async def cascade(
     dst_flume_addr: dict,
     ns_path: NamespacePath,
 
-    shm_registry: dict[str, _Token],
+    shm_registry: dict[str, NDToken],
 
     zero_on_step: bool = False,
     loglevel: str|None = None,
@@ -465,9 +465,9 @@ async def cascade(
     # not sure how else to do it.
     for (token, fsp_name, dst_token) in shm_registry:
         Fsp._flow_registry[(
-            _Token.from_msg(token),
+            NDToken.from_msg(token),
             fsp_name,
-        )] = _Token.from_msg(dst_token), None
+        )] = NDToken.from_msg(dst_token), None
 
     fsp: Fsp = reg.get(
         NamespacePath(ns_path)

piker/fsp/_momo.py

@@ -25,7 +25,7 @@ from numba import jit, float64, optional, int64
 
 from ._api import fsp
 from ..data import iterticks
-from ..data._sharedmem import ShmArray
+from tractor.ipc._shm import ShmArray
 
 
 @jit(

piker/fsp/_volume.py

@@ -21,7 +21,7 @@ from tractor.trionics._broadcast import AsyncReceiver
 
 from ._api import fsp
 from ..data import iterticks
-from ..data._sharedmem import ShmArray
+from tractor.ipc._shm import ShmArray
 from ._momo import _wma
 from ..log import get_logger
 

piker/storage/cli.py

@@ -37,9 +37,7 @@ import typer
 
 from piker.service import open_piker_runtime
 from piker.cli import cli
-from piker.data import (
+from tractor.ipc._shm import ShmArray
-    ShmArray,
-)
 from piker import tsp
 from . import log
 from . import (
@@ -294,6 +292,11 @@ def ldshm(
                             f'Something is wrong with time period for {shm}:\n{times}'
                         )
                 period_s: float = float(max(d1, d2, med))
+                log.info(
+                    f'Processing shm buffer:\n'
+                    f'  file: {shmfile.name}\n'
+                    f'  period: {period_s}s\n'
+                )
 
                 null_segs: tuple = tsp.get_null_segs(
                     frame=shm.array,

piker/storage/nativedb.py

@@ -64,10 +64,8 @@ from pendulum import (
 
 from piker import config
 from piker import tsp
-from piker.data import (
+from tractor.ipc._shm import ShmArray
-    def_iohlcv_fields,
+from piker.data import def_iohlcv_fields
-    ShmArray,
-)
 from piker.log import get_logger
 from . import TimeseriesNotFound
 

piker/tsp/_anal.py

@@ -276,14 +276,41 @@ def get_null_segs(
         absi_zdiff: np.ndarray = np.diff(absi_zeros)
 
         if zero_t.size < 2:
-            try:
+            idx: int = zero_t['index'][0]
-                breakpoint()
+            idx_before: int = idx - 1
-            except RuntimeError:
+            idx_after: int = idx + 1
-                # XXX, if greenback not active from
+            index = frame['index']
-                # piker store ldshm cmd..
+            before_cond = idx_before <= index
-                log.exception(
+            after_cond = index <= idx_after
-                    "Can't debug single-sample null!\n"
+            bars: np.ndarray = frame[
-                )
+                before_cond
+                &
+                after_cond
+            ]
+            time: np.ndarray = bars['time']
+            from pendulum import (
+                from_timestamp,
+                Interval,
+            )
+            gap: Interval = (
+                from_timestamp(time[-1])
+                -
+                from_timestamp(time[0])
+            )
+            log.warning(
+                f'Single OHLCV-bar null-segment detected??\n'
+                f'gap -> {gap}\n'
+            )
+
+            # ^^XXX, if you want to debug the above bar-gap^^
+            # try:
+            #     breakpoint()
+            # except RuntimeError:
+            #     # XXX, if greenback not active from
+            #     # piker store ldshm cmd..
+            #     log.exception(
+            #         "Can't debug single-sample null!\n"
+            #     )
 
             return None
 

piker/tsp/_annotate.py

@@ -30,6 +30,11 @@ import tractor
 
 from piker.data._formatters import BGM
 from piker.storage import log
+from piker.toolz.profile import (
+    Profiler,
+    pg_profile_enabled,
+    ms_slower_then,
+)
 from piker.ui._style import get_fonts
 
 if TYPE_CHECKING:
@@ -92,12 +97,22 @@ async def markup_gaps(
     # gap's duration.
     show_txt: bool = False,
 
+    # A/B comparison: render individual arrows alongside batch
+    # for visual comparison
+    show_individual_arrows: bool = False,
+
 ) -> dict[int, dict]:
     '''
     Remote annotate time-gaps in a dt-fielded ts (normally OHLC)
     with rectangles.
 
     '''
+    profiler = Profiler(
+        msg=f'markup_gaps() for {gaps.height} gaps',
+        disabled=False,
+        ms_threshold=0.0,
+    )
+
     # XXX: force chart redraw FIRST to ensure PlotItem coordinate
     # system is properly initialized before we position annotations!
     # Without this, annotations may be misaligned on first creation
@@ -106,6 +121,19 @@ async def markup_gaps(
         fqme=fqme,
         timeframe=timeframe,
     )
+    profiler('first `.redraw()` before annot creation')
+
+    log.info(
+        f'markup_gaps() called:\n'
+        f'  fqme: {fqme}\n'
+        f'  timeframe: {timeframe}s\n'
+        f'  gaps.height: {gaps.height}\n'
+    )
+
+    # collect all annotation specs for batch submission
+    rect_specs: list[dict] = []
+    arrow_specs: list[dict] = []
+    text_specs: list[dict] = []
 
     aids: dict[int] = {}
     for i in range(gaps.height):
@@ -217,56 +245,38 @@ async def markup_gaps(
         #     1: 'wine',  # down-gap
         # }[sgn]
 
-        rect_kwargs: dict[str, Any] = dict(
+        # collect rect spec (no fqme/timeframe, added by batch
-            fqme=fqme,
+        # API)
-            timeframe=timeframe,
+        rect_spec: dict[str, Any] = dict(
+            meth='set_view_pos',
             start_pos=lc,
             end_pos=ro,
             color=color,
+            update_label=False,
             start_time=start_time,
             end_time=end_time,
         )
+        rect_specs.append(rect_spec)
 
-        # add up/down rects
-        aid: int|None = await actl.add_rect(**rect_kwargs)
-        if aid is None:
-            log.error(
-                f'Failed to add rect for,\n'
-                f'{rect_kwargs!r}\n'
-                f'\n'
-                f'Skipping to next gap!\n'
-            )
-            continue
-
-        assert aid
-        aids[aid] = rect_kwargs
         direction: str = (
             'down' if down_gap
             else 'up'
         )
-        # TODO! mk this a `msgspec.Struct` which we deserialize
+
-        # on the server side!
+        # collect arrow spec
-        # XXX: send timestamp for server-side index lookup
-        # to ensure alignment with current shm state
         gap_time: float = row['time'][0]
-        arrow_kwargs: dict[str, Any] = dict(
+        arrow_spec: dict[str, Any] = dict(
-            fqme=fqme,
-            timeframe=timeframe,
             x=iend,  # fallback if timestamp lookup fails
             y=cls,
             time=gap_time,  # for server-side index lookup
             color=color,
             alpha=169,
             pointing=direction,
-            # TODO: expose these as params to markup_gaps()?
             headLen=10,
             headWidth=2.222,
             pxMode=True,
         )
-
+        arrow_specs.append(arrow_spec)
-        aid: int = await actl.add_arrow(
-            **arrow_kwargs
-        )
 
         # add duration label to RHS of arrow
         if up_gap:
@@ -278,15 +288,12 @@ async def markup_gaps(
             assert flat
             anchor = (0, 0)  # up from bottom
 
-        # use a slightly smaller font for gap label txt.
+        # collect text spec if enabled
-        font, small_font = get_fonts()
-        font_size: int = small_font.px_size - 1
-        assert isinstance(font_size, int)
-
         if show_txt:
-            text_aid: int = await actl.add_text(
+            font, small_font = get_fonts()
-                fqme=fqme,
+            font_size: int = small_font.px_size - 1
-                timeframe=timeframe,
+
+            text_spec: dict[str, Any] = dict(
                 text=gap_label,
                 x=iend + 1,  # fallback if timestamp lookup fails
                 y=cls,
@@ -295,12 +302,46 @@ async def markup_gaps(
                 anchor=anchor,
                 font_size=font_size,
             )
-            aids[text_aid] = {'text': gap_label}
+            text_specs.append(text_spec)
 
-    # tell chart to redraw all its
+    # submit all annotations in single batch IPC msg
-    # graphics view layers Bo
+    log.info(
+        f'Submitting batch annotations:\n'
+        f'  rects: {len(rect_specs)}\n'
+        f'  arrows: {len(arrow_specs)}\n'
+        f'  texts: {len(text_specs)}\n'
+    )
+    profiler('built all annotation specs')
+
+    result: dict[str, list[int]] = await actl.add_batch(
+        fqme=fqme,
+        timeframe=timeframe,
+        rects=rect_specs,
+        arrows=arrow_specs,
+        texts=text_specs,
+        show_individual_arrows=show_individual_arrows,
+    )
+    profiler('batch `.add_batch()` IPC call complete')
+
+    # build aids dict from batch results
+    for aid in result['rects']:
+        aids[aid] = {'type': 'rect'}
+    for aid in result['arrows']:
+        aids[aid] = {'type': 'arrow'}
+    for aid in result['texts']:
+        aids[aid] = {'type': 'text'}
+
+    log.info(
+        f'Batch submission complete: {len(aids)} annotation(s) '
+        f'created'
+    )
+    profiler('built aids result dict')
+
+    # tell chart to redraw all its graphics view layers
     await actl.redraw(
         fqme=fqme,
         timeframe=timeframe,
     )
+    profiler('final `.redraw()` after annot creation')
+
     return aids

piker/tsp/_history.py

@@ -59,11 +59,12 @@ from piker.brokers import NoData
 from piker.accounting import (
     MktPair,
 )
-from piker.log import get_logger
+from piker.log import (
-from ..data._sharedmem import (
+    get_logger,
-    maybe_open_shm_array,
+    get_console_log,
-    ShmArray,
 )
+from tractor.ipc._shm import ShmArray
+from ..data._sharedmem import maybe_open_shm_array
 from piker.data._source import (
     def_iohlcv_fields,
 )
@@ -737,12 +738,21 @@ async def start_backfill(
                 # including the dst[/src] source asset token. SO,
                 # 'tsla.nasdaq.ib' over 'tsla/usd.nasdaq.ib' for
                 # historical reasons ONLY.
-                if mkt.dst.atype not in {
+                if (
-                    'crypto',
+                    mkt.dst.atype not in {
-                    'crypto_currency',
+                        'crypto',
-                    'fiat',  # a "forex pair"
+                        'crypto_currency',
-                    'perpetual_future',  # stupid "perps" from cex land
+                        'fiat',  # a "forex pair"
-                }:
+                        'perpetual_future',  # stupid "perps" from cex land
+                    }
+                    and not (
+                        mkt.src.atype == 'crypto_currency'
+                        and
+                        mkt.dst.atype in {
+                            'future',
+                        }
+                    )
+                ):
                     col_sym_key: str = mkt.get_fqme(
                         delim_char='',
                         without_src=True,
@@ -1386,6 +1396,10 @@ async def manage_history(
         engages.
 
     '''
+    get_console_log(
+        name=__name__,
+        level=loglevel,
+    )
     # TODO: is there a way to make each shm file key
     # actor-tree-discovery-addr unique so we avoid collisions
     # when doing tests which also allocate shms for certain instruments

piker/ui/_annotate.py

@@ -24,8 +24,11 @@ from pyqtgraph import (
     Point,
     functions as fn,
     Color,
+    GraphicsObject,
 )
+from pyqtgraph.Qt import internals
 import numpy as np
+import pyqtgraph as pg
 
 from piker.ui.qt import (
     QtCore,
@@ -35,6 +38,10 @@ from piker.ui.qt import (
     QRectF,
     QGraphicsPathItem,
 )
+from piker.ui._style import hcolor
+from piker.log import get_logger
+
+log = get_logger(__name__)
 
 
 def mk_marker_path(
@@ -104,7 +111,7 @@ def mk_marker_path(
 
 class LevelMarker(QGraphicsPathItem):
     '''
-    An arrow marker path graphich which redraws itself
+    An arrow marker path graphic which redraws itself
     to the specified view coordinate level on each paint cycle.
 
     '''
@@ -251,9 +258,9 @@ def qgo_draw_markers(
 
 ) -> float:
     '''
-    Paint markers in ``pg.GraphicsItem`` style by first
+    Paint markers in ``pg.GraphicsItem`` style by first removing the
-    removing the view transform for the painter, drawing the markers
+    view transform for the painter, drawing the markers in scene
-    in scene coords, then restoring the view coords.
+    coords, then restoring the view coords.
 
     '''
     # paint markers in native coordinate system
@@ -295,3 +302,449 @@ def qgo_draw_markers(
 
     p.setTransform(orig_tr)
     return max(sizes)
+
+
+class GapAnnotations(GraphicsObject):
+    '''
+    Batch-rendered gap annotations using Qt's efficient drawing
+    APIs.
+
+    Instead of creating individual `QGraphicsItem` instances per
+    gap (which is very slow for 1000+ gaps), this class stores all
+    gap rectangles and arrows in numpy-backed arrays and renders
+    them in single batch paint calls.
+
+    Performance: ~1000x faster than individual items for large gap
+    counts.
+
+    Based on patterns from:
+    - `pyqtgraph.BarGraphItem` (batch rect rendering)
+    - `pyqtgraph.ScatterPlotItem` (fragment rendering)
+    - `piker.ui._curve.FlowGraphic` (single path pattern)
+
+    '''
+    def __init__(
+        self,
+        gap_specs: list[dict],
+        array: np.ndarray|None = None,
+        color: str = 'dad_blue',
+        alpha: int = 169,
+        arrow_size: float = 10.0,
+        fqme: str|None = None,
+        timeframe: float|None = None,
+    ) -> None:
+        '''
+        gap_specs: list of dicts with keys:
+            - start_pos: (x, y) tuple for left corner of rect
+            - end_pos: (x, y) tuple for right corner of rect
+            - arrow_x: x position for arrow
+            - arrow_y: y position for arrow
+            - pointing: 'up' or 'down' for arrow direction
+            - start_time: (optional) timestamp for repositioning
+            - end_time: (optional) timestamp for repositioning
+
+        array: optional OHLC numpy array for repositioning on
+               backfill updates (when abs-index changes)
+
+        fqme: symbol name for these gaps (for logging/debugging)
+
+        timeframe: period in seconds that these gaps were
+                   detected on (used to skip reposition when
+                   called with wrong timeframe's array)
+
+        '''
+        super().__init__()
+        self._gap_specs = gap_specs
+        self._array = array
+        self._fqme = fqme
+        self._timeframe = timeframe
+        n_gaps = len(gap_specs)
+
+        # shared pen/brush matching original SelectRect/ArrowItem style
+        base_color = pg.mkColor(hcolor(color))
+
+        # rect pen: base color, fully opaque for outline
+        self._rect_pen = pg.mkPen(base_color, width=1)
+
+        # rect brush: base color with alpha=66 (SelectRect default)
+        rect_fill = pg.mkColor(hcolor(color))
+        rect_fill.setAlpha(66)
+        self._rect_brush = pg.functions.mkBrush(rect_fill)
+
+        # arrow pen: same as rects
+        self._arrow_pen = pg.mkPen(base_color, width=1)
+
+        # arrow brush: base color with user-specified alpha (default 169)
+        arrow_fill = pg.mkColor(hcolor(color))
+        arrow_fill.setAlpha(alpha)
+        self._arrow_brush = pg.functions.mkBrush(arrow_fill)
+
+        # allocate rect array using Qt's efficient storage
+        self._rectarray = internals.PrimitiveArray(
+            QtCore.QRectF,
+            4,
+        )
+        self._rectarray.resize(n_gaps)
+        rect_memory = self._rectarray.ndarray()
+
+        # fill rect array from gap specs
+        for (
+            i,
+            spec,
+        ) in enumerate(gap_specs):
+            (
+                start_x,
+                start_y,
+            ) = spec['start_pos']
+            (
+                end_x,
+                end_y,
+            ) = spec['end_pos']
+
+            # QRectF expects (x, y, width, height)
+            rect_memory[i, 0] = start_x
+            rect_memory[i, 1] = min(start_y, end_y)
+            rect_memory[i, 2] = end_x - start_x
+            rect_memory[i, 3] = abs(end_y - start_y)
+
+        # build single QPainterPath for all arrows
+        self._arrow_path = QtGui.QPainterPath()
+        self._arrow_size = arrow_size
+
+        for spec in gap_specs:
+            arrow_x = spec['arrow_x']
+            arrow_y = spec['arrow_y']
+            pointing = spec['pointing']
+
+            # create arrow polygon
+            if pointing == 'down':
+                # arrow points downward
+                arrow_poly = QtGui.QPolygonF([
+                    QPointF(arrow_x, arrow_y),  # tip
+                    QPointF(
+                        arrow_x - arrow_size/2,
+                        arrow_y - arrow_size,
+                    ),  # left
+                    QPointF(
+                        arrow_x + arrow_size/2,
+                        arrow_y - arrow_size,
+                    ),  # right
+                ])
+            else:  # up
+                # arrow points upward
+                arrow_poly = QtGui.QPolygonF([
+                    QPointF(arrow_x, arrow_y),  # tip
+                    QPointF(
+                        arrow_x - arrow_size/2,
+                        arrow_y + arrow_size,
+                    ),  # left
+                    QPointF(
+                        arrow_x + arrow_size/2,
+                        arrow_y + arrow_size,
+                    ),  # right
+                ])
+
+            self._arrow_path.addPolygon(arrow_poly)
+            self._arrow_path.closeSubpath()
+
+        # cache bounding rect
+        self._br: QRectF|None = None
+
+    def boundingRect(self) -> QRectF:
+        '''
+        Compute bounding rect from rect array and arrow path.
+
+        '''
+        if self._br is not None:
+            return self._br
+
+        # get rect bounds
+        rect_memory = self._rectarray.ndarray()
+        if len(rect_memory) == 0:
+            self._br = QRectF()
+            return self._br
+
+        x_min = rect_memory[:, 0].min()
+        y_min = rect_memory[:, 1].min()
+        x_max = (rect_memory[:, 0] + rect_memory[:, 2]).max()
+        y_max = (rect_memory[:, 1] + rect_memory[:, 3]).max()
+
+        # expand for arrow path
+        arrow_br = self._arrow_path.boundingRect()
+        x_min = min(x_min, arrow_br.left())
+        y_min = min(y_min, arrow_br.top())
+        x_max = max(x_max, arrow_br.right())
+        y_max = max(y_max, arrow_br.bottom())
+
+        self._br = QRectF(
+            x_min,
+            y_min,
+            x_max - x_min,
+            y_max - y_min,
+        )
+        return self._br
+
+    def paint(
+        self,
+        p: QtGui.QPainter,
+        opt: QtWidgets.QStyleOptionGraphicsItem,
+        w: QtWidgets.QWidget,
+    ) -> None:
+        '''
+        Batch render all rects and arrows in minimal paint calls.
+
+        '''
+        # draw all rects in single batch call (data coordinates)
+        p.setPen(self._rect_pen)
+        p.setBrush(self._rect_brush)
+        drawargs = self._rectarray.drawargs()
+        p.drawRects(*drawargs)
+
+        # draw arrows in scene/pixel coordinates so they maintain
+        # size regardless of zoom level
+        orig_tr = p.transform()
+        p.resetTransform()
+
+        # rebuild arrow path in scene coordinates
+        arrow_path_scene = QtGui.QPainterPath()
+
+        # arrow geometry matching pg.ArrowItem defaults
+        # headLen=10, headWidth=2.222
+        # headWidth is the half-width (center to edge distance)
+        head_len = self._arrow_size
+        head_width = head_len * 0.2222  # 2.222 at size=10
+
+        for spec in self._gap_specs:
+            if 'arrow_x' not in spec:
+                continue
+
+            arrow_x = spec['arrow_x']
+            arrow_y = spec['arrow_y']
+            pointing = spec['pointing']
+
+            # transform data coords to scene coords
+            scene_pt = orig_tr.map(QPointF(arrow_x, arrow_y))
+            sx = scene_pt.x()
+            sy = scene_pt.y()
+
+            # create arrow polygon in scene/pixel coords
+            # matching pg.ArrowItem geometry but rotated for up/down
+            if pointing == 'down':
+                # tip points downward (negative y direction)
+                arrow_poly = QtGui.QPolygonF([
+                    QPointF(sx, sy),  # tip
+                    QPointF(
+                        sx - head_width,
+                        sy - head_len,
+                    ),  # left base
+                    QPointF(
+                        sx + head_width,
+                        sy - head_len,
+                    ),  # right base
+                ])
+            else:  # up
+                # tip points upward (positive y direction)
+                arrow_poly = QtGui.QPolygonF([
+                    QPointF(sx, sy),  # tip
+                    QPointF(
+                        sx - head_width,
+                        sy + head_len,
+                    ),  # left base
+                    QPointF(
+                        sx + head_width,
+                        sy + head_len,
+                    ),  # right base
+                ])
+
+            arrow_path_scene.addPolygon(arrow_poly)
+            arrow_path_scene.closeSubpath()
+
+        p.setPen(self._arrow_pen)
+        p.setBrush(self._arrow_brush)
+        p.drawPath(arrow_path_scene)
+
+        # restore original transform
+        p.setTransform(orig_tr)
+
+    def reposition(
+        self,
+        array: np.ndarray|None = None,
+        fqme: str|None = None,
+        timeframe: float|None = None,
+    ) -> None:
+        '''
+        Reposition all annotations based on timestamps.
+
+        Used when viz is updated (eg during backfill) and abs-index
+        range changes - we need to lookup new indices from timestamps.
+
+        '''
+        # skip reposition if timeframe doesn't match
+        # (e.g., 1s gaps being repositioned with 60s array)
+        if (
+            timeframe is not None
+            and
+            self._timeframe is not None
+            and
+            timeframe != self._timeframe
+        ):
+            log.debug(
+                f'Skipping reposition for {self._fqme} gaps:\n'
+                f'  gap timeframe: {self._timeframe}s\n'
+                f'  array timeframe: {timeframe}s\n'
+            )
+            return
+
+        if array is None:
+            array = self._array
+
+        if array is None:
+            log.warning(
+                'GapAnnotations.reposition() called but no array '
+                'provided'
+            )
+            return
+
+        # collect all unique timestamps we need to lookup
+        timestamps: set[float] = set()
+        for spec in self._gap_specs:
+            if spec.get('start_time') is not None:
+                timestamps.add(spec['start_time'])
+            if spec.get('end_time') is not None:
+                timestamps.add(spec['end_time'])
+            if spec.get('time') is not None:
+                timestamps.add(spec['time'])
+
+        # vectorized timestamp -> row lookup using binary search
+        time_to_row: dict[float, dict] = {}
+        if timestamps:
+            import numpy as np
+            time_arr = array['time']
+            ts_array = np.array(list(timestamps))
+
+            search_indices = np.searchsorted(
+                time_arr,
+                ts_array,
+            )
+
+            # vectorized bounds check and exact match verification
+            valid_mask = (
+                (search_indices < len(array))
+                & (time_arr[search_indices] == ts_array)
+            )
+
+            valid_indices = search_indices[valid_mask]
+            valid_timestamps = ts_array[valid_mask]
+            matched_rows = array[valid_indices]
+
+            time_to_row = {
+                float(ts): {
+                    'index': float(row['index']),
+                    'open': float(row['open']),
+                    'close': float(row['close']),
+                }
+                for ts, row in zip(
+                    valid_timestamps,
+                    matched_rows,
+                )
+            }
+
+        # rebuild rect array from gap specs with new indices
+        rect_memory = self._rectarray.ndarray()
+
+        for (
+            i,
+            spec,
+        ) in enumerate(self._gap_specs):
+            start_time = spec.get('start_time')
+            end_time = spec.get('end_time')
+
+            if (
+                start_time is None
+                or end_time is None
+            ):
+                continue
+
+            start_row = time_to_row.get(start_time)
+            end_row = time_to_row.get(end_time)
+
+            if (
+                start_row is None
+                or end_row is None
+            ):
+                log.warning(
+                    f'Timestamp lookup failed for gap[{i}] during '
+                    f'reposition:\n'
+                    f'  fqme: {fqme}\n'
+                    f'  timeframe: {timeframe}s\n'
+                    f'  start_time: {start_time}\n'
+                    f'  end_time: {end_time}\n'
+                    f'  array time range: '
+                    f'{array["time"][0]} -> {array["time"][-1]}\n'
+                )
+                continue
+
+            start_idx = start_row['index']
+            end_idx = end_row['index']
+            start_close = start_row['close']
+            end_open = end_row['open']
+
+            from_idx: float = 0.16 - 0.06
+            start_x = start_idx + 1 - from_idx
+            end_x = end_idx + from_idx
+
+            # update rect in array
+            rect_memory[i, 0] = start_x
+            rect_memory[i, 1] = min(start_close, end_open)
+            rect_memory[i, 2] = end_x - start_x
+            rect_memory[i, 3] = abs(end_open - start_close)
+
+        # rebuild arrow path with new indices
+        self._arrow_path.clear()
+
+        for spec in self._gap_specs:
+            time_val = spec.get('time')
+            if time_val is None:
+                continue
+
+            arrow_row = time_to_row.get(time_val)
+            if arrow_row is None:
+                continue
+
+            arrow_x = arrow_row['index']
+            arrow_y = arrow_row['close']
+            pointing = spec['pointing']
+
+            # create arrow polygon
+            if pointing == 'down':
+                arrow_poly = QtGui.QPolygonF([
+                    QPointF(arrow_x, arrow_y),
+                    QPointF(
+                        arrow_x - self._arrow_size/2,
+                        arrow_y - self._arrow_size,
+                    ),
+                    QPointF(
+                        arrow_x + self._arrow_size/2,
+                        arrow_y - self._arrow_size,
+                    ),
+                ])
+            else:  # up
+                arrow_poly = QtGui.QPolygonF([
+                    QPointF(arrow_x, arrow_y),
+                    QPointF(
+                        arrow_x - self._arrow_size/2,
+                        arrow_y + self._arrow_size,
+                    ),
+                    QPointF(
+                        arrow_x + self._arrow_size/2,
+                        arrow_y + self._arrow_size,
+                    ),
+                ])
+
+            self._arrow_path.addPolygon(arrow_poly)
+            self._arrow_path.closeSubpath()
+
+        # invalidate bounding rect cache
+        self._br = None
+        self.prepareGeometryChange()
+        self.update()

piker/ui/_axes.py

@@ -75,6 +75,9 @@ class Axis(pg.AxisItem):
         self.pi = plotitem
         self._dpi_font = _font
 
+        # store for later recalculation on zoom
+        self._typical_max_str = typical_max_str
+
         self.setTickFont(_font.font)
         font_size = self._dpi_font.font.pixelSize()
 
@@ -156,6 +159,41 @@ class Axis(pg.AxisItem):
     def size_to_values(self) -> None:
         pass
 
+    def update_fonts(self, font: DpiAwareFont) -> None:
+        '''Update font and recalculate axis sizing after zoom change.'''
+        # IMPORTANT: tell Qt we're about to change geometry
+        self.prepareGeometryChange()
+
+        self._dpi_font = font
+        self.setTickFont(font.font)
+        font_size = font.font.pixelSize()
+
+        # recalculate text offset based on new font size
+        text_offset = None
+        if self.orientation in ('bottom',):
+            text_offset = floor(0.25 * font_size)
+        elif self.orientation in ('left', 'right'):
+            text_offset = floor(font_size / 2)
+
+        if text_offset:
+            self.setStyle(tickTextOffset=text_offset)
+
+        # recalculate bounding rect with new font
+        # Note: typical_max_str should be stored from init
+        if not hasattr(self, '_typical_max_str'):
+            self._typical_max_str = '100 000.000  '  # fallback default
+        self.typical_br = font._qfm.boundingRect(self._typical_max_str)
+
+        # Update PyQtGraph's internal text size tracking
+        # This is critical - PyQtGraph uses these internally for auto-expand
+        if self.orientation in ['left', 'right']:
+            self.textWidth = self.typical_br.width()
+        else:
+            self.textHeight = self.typical_br.height()
+
+        # resize axis to fit new font - this triggers PyQtGraph's auto-expand
+        self.size_to_values()
+
     def txt_offsets(self) -> tuple[int, int]:
         return tuple(self.style['tickTextOffset'])
 
@@ -256,7 +294,14 @@ class PriceAxis(Axis):
         self._min_tick = size
 
     def size_to_values(self) -> None:
-        self.setWidth(self.typical_br.width())
+        # Call PyQtGraph's internal width update mechanism
+        # This respects autoExpandTextSpace and updates min/max constraints
+        self._updateWidth()
+        # tell Qt our preferred size changed so layout recalculates
+        self.updateGeometry()
+        # force parent plot item to recalculate its layout
+        if self.pi and hasattr(self.pi, 'updateGeometry'):
+            self.pi.updateGeometry()
 
     # XXX: drop for now since it just eats up h space
 
@@ -300,7 +345,14 @@ class DynamicDateAxis(Axis):
     }
 
     def size_to_values(self) -> None:
-        self.setHeight(self.typical_br.height() + 1)
+        # Call PyQtGraph's internal height update mechanism
+        # This respects autoExpandTextSpace and updates min/max constraints
+        self._updateHeight()
+        # tell Qt our preferred size changed so layout recalculates
+        self.updateGeometry()
+        # force parent plot item to recalculate its layout
+        if self.pi and hasattr(self.pi, 'updateGeometry'):
+            self.pi.updateGeometry()
 
     def _indexes_to_timestrs(
         self,

piker/ui/_chart.py

@@ -49,7 +49,7 @@ from ._cursor import (
     Cursor,
     ContentsLabel,
 )
-from ..data._sharedmem import ShmArray
+from tractor.ipc._shm import ShmArray
 from ._ohlc import BarItems
 from ._curve import (
     Curve,

piker/ui/_dataviz.py

@@ -42,9 +42,7 @@ from numpy import (
 import pyqtgraph as pg
 
 from piker.ui.qt import QLineF
-from ..data._sharedmem import (
+from tractor.ipc._shm import ShmArray
-    ShmArray,
-)
 from ..data.flows import Flume
 from ..data._formatters import (
     IncrementalFormatter,

piker/ui/_editors.py

@@ -168,7 +168,7 @@ class ArrowEditor(Struct):
         '''
         uid: str = arrow._uid
         arrows: list[pg.ArrowItem] = self._arrows[uid]
-        log.info(
+        log.debug(
             f'Removing arrow from views\n'
             f'uid: {uid!r}\n'
             f'{arrow!r}\n'
@@ -286,7 +286,9 @@ class LineEditor(Struct):
             for line in lines:
                 line.show_labels()
                 line.hide_markers()
-                log.debug(f'Level active for level: {line.value()}')
+                log.debug(
+                    f'Line active @ level: {line.value()!r}'
+                )
                 # TODO: other flashy things to indicate the order is active
 
         return lines
@@ -329,7 +331,11 @@ class LineEditor(Struct):
                     if line in hovered:
                         hovered.remove(line)
 
-                    log.debug(f'deleting {line} with oid: {uuid}')
+                    log.debug(
+                        f'Deleting level-line\n'
+                        f'line: {line!r}\n'
+                        f'oid: {uuid!r}\n'
+                    )
                     line.delete()
 
                     # make sure the xhair doesn't get left off
@@ -337,7 +343,11 @@ class LineEditor(Struct):
                     cursor.show_xhair()
 
         else:
-            log.warning(f'Could not find line for {line}')
+            log.warning(
+                f'Could not find line for removal ??\n'
+                f'\n'
+                f'{line!r}\n'
+            )
 
         return lines
 
@@ -569,11 +579,11 @@ class SelectRect(QtWidgets.QGraphicsRectItem):
         if update_label:
             self.init_label(view_rect)
 
-        print(
+        log.debug(
-            'SelectRect modify:\n'
+            f'SelectRect modify,\n'
             f'QRectF: {view_rect}\n'
-            f'start_pos: {start_pos}\n'
+            f'start_pos: {start_pos!r}\n'
-            f'end_pos: {end_pos}\n'
+            f'end_pos: {end_pos!r}\n'
         )
         self.show()
 
@@ -640,8 +650,11 @@ class SelectRect(QtWidgets.QGraphicsRectItem):
             dmn=dmn,
         ))
 
-        # print(f'x2, y2: {(x2, y2)}')
+        # tracing
-        # print(f'xmn, ymn: {(xmn, ymx)}')
+        # log.info(
+        #     f'x2, y2: {(x2, y2)}\n'
+        #     f'xmn, ymn: {(xmn, ymx)}\n'
+        # )
 
         label_anchor = Point(
             xmx + 2,

piker/ui/_exec.py

@@ -203,6 +203,9 @@ def run_qtractor(
     if is_windows:
         window.configure_to_desktop()
 
+    # install global keyboard shortcuts for UI zoom
+    window.install_global_zoom_filter()
+
     # actually render to screen
     window.show()
     app.exec_()

piker/ui/_forms.py

@@ -124,6 +124,13 @@ class Edit(QLineEdit):
         self.sizeHint()
         self.update()
 
+    def update_fonts(self, font: DpiAwareFont) -> None:
+        '''Update font and recalculate widget size.'''
+        self.dpi_font = font
+        self.setFont(font.font)
+        # tell Qt our size hint changed so it recalculates layout
+        self.updateGeometry()
+
     def focus(self) -> None:
         self.selectAll()
         self.show()
@@ -241,6 +248,14 @@ class Selection(QComboBox):
         icon_size = round(h * 0.75)
         self.setIconSize(QSize(icon_size, icon_size))
 
+    def update_fonts(self, font: DpiAwareFont) -> None:
+        '''Update font and recalculate widget size.'''
+        self.setFont(font.font)
+        # recalculate heights with new font
+        self.resize()
+        # tell Qt our size hint changed so it recalculates layout
+        self.updateGeometry()
+
     def set_items(
         self,
         keys: list[str],
@@ -431,6 +446,39 @@ class FieldsForm(QWidget):
         self.fields[key] = select
         return select
 
+    def update_fonts(self) -> None:
+        '''Update font sizes after zoom change.'''
+        from ._style import _font, _font_small
+
+        # update stored font size
+        self._font_size = _font_small.px_size - 2
+
+        # update all labels
+        for name, label in self.labels.items():
+            if hasattr(label, 'update_font'):
+                label.update_font(_font.font, self._font_size - 1)
+
+        # update all fields (edits, selects)
+        for key, field in self.fields.items():
+            # first check for our custom update_fonts method (Edit, Selection)
+            if hasattr(field, 'update_fonts'):
+                field.update_fonts(_font)
+            # then handle stylesheet updates for those without custom methods
+            elif hasattr(field, 'setStyleSheet'):
+                # regenerate stylesheet with new font size
+                field.setStyleSheet(
+                    f"""QLineEdit {{
+                        color : {hcolor('gunmetal')};
+                        font-size : {self._font_size}px;
+                    }}
+                    """
+                )
+                field.setFont(_font.font)
+
+            # for Selection widgets that need style updates
+            if hasattr(field, 'set_style'):
+                field.set_style(color='gunmetal', font_size=self._font_size)
+
 
 async def handle_field_input(
 
@@ -633,6 +681,37 @@ class FillStatusBar(QProgressBar):
         self.setRange(0, int(slots))
         self.setValue(value)
 
+    def update_fonts(self, font_size: int) -> None:
+        '''Update font size after zoom change.'''
+        from ._style import _font_small
+
+        self.font_size = font_size
+        # regenerate stylesheet with new font size
+        self.setStyleSheet(
+            f"""
+            QProgressBar {{
+
+                text-align: center;
+
+                font-size : {self.font_size - 2}px;
+
+                background-color: {hcolor('papas_special')};
+                color : {hcolor('papas_special')};
+
+                border: {self.border_px}px solid {hcolor('default_light')};
+                border-radius: 2px;
+            }}
+            QProgressBar::chunk {{
+
+                background-color: {hcolor('default_spotlight')};
+                color: {hcolor('bracket')};
+
+                border-radius: 2px;
+            }}
+            """
+        )
+        self.setFont(_font_small.font)
+
 
 def mk_fill_status_bar(
 

piker/ui/_fsp.py

@@ -44,14 +44,12 @@ from piker.fsp import (
     dolla_vlm,
     flow_rates,
 )
-from piker.data import (
+from tractor.ipc._shm import (
-    Flume,
     ShmArray,
+    NDToken,
 )
-from piker.data._sharedmem import (
+from piker.data import Flume
-    _Token,
+from piker.data._sharedmem import try_read
-    try_read,
-)
 from piker.log import get_logger
 from piker.toolz import Profiler
 from piker.types import Struct
@@ -382,7 +380,7 @@ class FspAdmin:
             tuple,
             tuple[tractor.MsgStream, ShmArray]
         ] = {}
-        self._flow_registry: dict[_Token, str] = {}
+        self._flow_registry: dict[NDToken, str] = {}
 
         # TODO: make this a `.src_flume` and add
         # a `dst_flume`?

piker/ui/_label.py

@@ -334,3 +334,19 @@ class FormatLabel(QLabel):
         out = self.fmt_str.format(**fields)
         self.setText(out)
         return out
+
+    def update_font(
+        self,
+        font: QtGui.QFont,
+        font_size: int,
+        font_color: str = 'default_lightest',
+    ) -> None:
+        '''Update font after zoom change.'''
+        self.setStyleSheet(
+            f"""QLabel {{
+                color : {hcolor(font_color)};
+                font-size : {font_size}px;
+            }}
+            """
+        )
+        self.setFont(font)

piker/ui/_lines.py

@@ -38,7 +38,6 @@ from piker.ui.qt import (
     QtGui,
     QGraphicsPathItem,
     QStyleOptionGraphicsItem,
-    QGraphicsItem,
     QGraphicsScene,
     QWidget,
     QPointF,

piker/ui/_position.py

@@ -178,6 +178,26 @@ class SettingsPane:
     # encompasing high level namespace
     order_mode: OrderMode | None = None  # typing: ignore # noqa
 
+    def update_fonts(self) -> None:
+        '''Update font sizes after zoom change.'''
+        from ._style import _font_small
+
+        # update form fields
+        if self.form and hasattr(self.form, 'update_fonts'):
+            self.form.update_fonts()
+
+        # update fill status bar
+        if self.fill_bar and hasattr(self.fill_bar, 'update_fonts'):
+            self.fill_bar.update_fonts(_font_small.px_size)
+
+        # update labels with new fonts
+        if self.step_label:
+            self.step_label.setFont(_font_small.font)
+        if self.pnl_label:
+            self.pnl_label.setFont(_font_small.font)
+        if self.limit_label:
+            self.limit_label.setFont(_font_small.font)
+
     def set_accounts(
         self,
         names: list[str],

piker/ui/_remote_ctl.py

@@ -22,6 +22,7 @@ a chart from some other actor.
 from __future__ import annotations
 from contextlib import (
     asynccontextmanager as acm,
+    contextmanager as cm,
     AsyncExitStack,
 )
 from functools import partial
@@ -46,6 +47,7 @@ from piker.log import get_logger
 from piker.types import Struct
 from piker.service import find_service
 from piker.brokers import SymbolNotFound
+from piker.toolz import Profiler
 from piker.ui.qt import (
     QGraphicsItem,
 )
@@ -98,6 +100,8 @@ def rm_annot(
     annot: ArrowEditor|SelectRect|pg.TextItem
 ) -> bool:
     global _editors
+    from piker.ui._annotate import GapAnnotations
+
     match annot:
         case pg.ArrowItem():
             editor = _editors[annot._uid]
@@ -122,9 +126,35 @@ def rm_annot(
                 scene.removeItem(annot)
             return True
 
+        case GapAnnotations():
+            scene = annot.scene()
+            if scene:
+                scene.removeItem(annot)
+            return True
+
     return False
 
 
+@cm
+def no_qt_updates(*items):
+    '''
+    Disable Qt widget/item updates during context to batch
+    render operations and only trigger single repaint on exit.
+
+    Accepts both QWidgets and QGraphicsItems.
+
+    '''
+    for item in items:
+        if hasattr(item, 'setUpdatesEnabled'):
+            item.setUpdatesEnabled(False)
+    try:
+        yield
+    finally:
+        for item in items:
+            if hasattr(item, 'setUpdatesEnabled'):
+                item.setUpdatesEnabled(True)
+
+
 async def serve_rc_annots(
     ipc_key: str,
     annot_req_stream: MsgStream,
@@ -429,6 +459,333 @@ async def serve_rc_annots(
                 aids.add(aid)
                 await annot_req_stream.send(aid)
 
+            case {
+                'cmd': 'batch',
+                'fqme': fqme,
+                'timeframe': timeframe,
+                'rects': list(rect_specs),
+                'arrows': list(arrow_specs),
+                'texts': list(text_specs),
+                'show_individual_arrows': bool(show_individual_arrows),
+            }:
+                # batch submission handler - process multiple
+                # annotations in single IPC round-trip
+                ds: DisplayState = _dss[fqme]
+                try:
+                    chart: ChartPlotWidget = {
+                        60: ds.hist_chart,
+                        1: ds.chart,
+                    }[timeframe]
+                except KeyError:
+                    msg: str = (
+                        f'No chart for timeframe={timeframe}s, '
+                        f'skipping batch annotation'
+                    )
+                    log.error(msg)
+                    await annot_req_stream.send({'error': msg})
+                    continue
+
+                cv: ChartView = chart.cv
+                viz: Viz = chart.get_viz(fqme)
+                shm = viz.shm
+                arr = shm.array
+
+                result: dict[str, list[int]] = {
+                    'rects': [],
+                    'arrows': [],
+                    'texts': [],
+                }
+
+                profiler = Profiler(
+                    msg=(
+                        f'Batch annotate {len(rect_specs)} gaps '
+                        f'on {fqme}@{timeframe}s'
+                    ),
+                    disabled=False,
+                    delayed=False,
+                )
+
+                aids_set: set[int] = ctxs[ipc_key][1]
+
+                # build unified gap_specs for GapAnnotations class
+                from piker.ui._annotate import GapAnnotations
+
+                gap_specs: list[dict] = []
+                n_gaps: int = max(
+                    len(rect_specs),
+                    len(arrow_specs),
+                )
+                profiler('setup batch annot creation')
+
+                # collect all unique timestamps for vectorized lookup
+                timestamps: list[float] = []
+                for rect_spec in rect_specs:
+                    if start_time := rect_spec.get('start_time'):
+                        timestamps.append(start_time)
+                    if end_time := rect_spec.get('end_time'):
+                        timestamps.append(end_time)
+                for arrow_spec in arrow_specs:
+                    if time_val := arrow_spec.get('time'):
+                        timestamps.append(time_val)
+
+                profiler('collect `timestamps: list` complet!')
+
+                # build timestamp -> row mapping using binary search
+                # O(m log n) instead of O(n*m) with np.isin
+                time_to_row: dict[float, dict] = {}
+                if timestamps:
+                    import numpy as np
+                    time_arr = arr['time']
+                    ts_array = np.array(timestamps)
+
+                    # binary search for each timestamp in sorted time array
+                    search_indices = np.searchsorted(
+                        time_arr,
+                        ts_array,
+                    )
+
+                    profiler('`np.searchsorted()` complete!')
+
+                    # vectorized bounds check and exact match verification
+                    valid_mask = (
+                        (search_indices < len(arr))
+                        & (time_arr[search_indices] == ts_array)
+                    )
+
+                    # get all valid indices and timestamps
+                    valid_indices = search_indices[valid_mask]
+                    valid_timestamps = ts_array[valid_mask]
+
+                    # use fancy indexing to get all rows at once
+                    matched_rows = arr[valid_indices]
+
+                    # extract fields to plain arrays BEFORE dict building
+                    indices_arr = matched_rows['index'].astype(float)
+                    opens_arr = matched_rows['open'].astype(float)
+                    closes_arr = matched_rows['close'].astype(float)
+
+                    profiler('extracted field arrays')
+
+                    # build dict from plain arrays (much faster)
+                    time_to_row: dict[float, dict] = {
+                        float(ts): {
+                            'index': idx,
+                            'open': opn,
+                            'close': cls,
+                        }
+                        for (
+                            ts,
+                            idx,
+                            opn,
+                            cls,
+                        ) in zip(
+                            valid_timestamps,
+                            indices_arr,
+                            opens_arr,
+                            closes_arr,
+                        )
+                    }
+
+                    profiler('`time_to_row` creation complete!')
+
+                profiler(f'built timestamp lookup for {len(timestamps)} times')
+
+                # build gap_specs from rect+arrow specs
+                for i in range(n_gaps):
+                    gap_spec: dict = {}
+
+                    # get rect spec for this gap
+                    if i < len(rect_specs):
+                        rect_spec: dict = rect_specs[i].copy()
+                        start_time = rect_spec.get('start_time')
+                        end_time = rect_spec.get('end_time')
+
+                        if (
+                            start_time is not None
+                            and end_time is not None
+                        ):
+                            # lookup from pre-built mapping
+                            start_row = time_to_row.get(start_time)
+                            end_row = time_to_row.get(end_time)
+
+                            if (
+                                start_row is None
+                                or end_row is None
+                            ):
+                                log.warning(
+                                    f'Timestamp lookup failed for '
+                                    f'gap[{i}], skipping'
+                                )
+                                continue
+
+                            start_idx = start_row['index']
+                            end_idx = end_row['index']
+                            start_close = start_row['close']
+                            end_open = end_row['open']
+
+                            from_idx: float = 0.16 - 0.06
+                            gap_spec['start_pos'] = (
+                                start_idx + 1 - from_idx,
+                                start_close,
+                            )
+                            gap_spec['end_pos'] = (
+                                end_idx + from_idx,
+                                end_open,
+                            )
+                            gap_spec['start_time'] = start_time
+                            gap_spec['end_time'] = end_time
+                            gap_spec['color'] = rect_spec.get(
+                                'color',
+                                'dad_blue',
+                            )
+
+                    # get arrow spec for this gap
+                    if i < len(arrow_specs):
+                        arrow_spec: dict = arrow_specs[i].copy()
+                        x: float = float(arrow_spec.get('x', 0))
+                        y: float = float(arrow_spec.get('y', 0))
+                        time_val: float|None = arrow_spec.get('time')
+
+                        # timestamp-based index lookup (only for x, NOT y!)
+                        # y is already set to the PREVIOUS bar's close
+                        if time_val is not None:
+                            arrow_row = time_to_row.get(time_val)
+                            if arrow_row is not None:
+                                x = arrow_row['index']
+                                # NOTE: do NOT update y! it's the
+                                # previous bar's close, not current
+                            else:
+                                log.warning(
+                                    f'Arrow timestamp {time_val} not '
+                                    f'found for gap[{i}], using x={x}'
+                                )
+
+                        gap_spec['arrow_x'] = x
+                        gap_spec['arrow_y'] = y
+                        gap_spec['time'] = time_val
+                        gap_spec['pointing'] = arrow_spec.get(
+                            'pointing',
+                            'down',
+                        )
+                        gap_spec['alpha'] = arrow_spec.get('alpha', 169)
+
+                    gap_specs.append(gap_spec)
+
+                profiler(f'built {len(gap_specs)} gap_specs')
+
+                # create single GapAnnotations item for all gaps
+                if gap_specs:
+                    gaps_item = GapAnnotations(
+                        gap_specs=gap_specs,
+                        array=arr,
+                        color=gap_specs[0].get('color', 'dad_blue'),
+                        alpha=gap_specs[0].get('alpha', 169),
+                        arrow_size=10.0,
+                        fqme=fqme,
+                        timeframe=timeframe,
+                    )
+                    chart.plotItem.addItem(gaps_item)
+
+                    # register single item for repositioning
+                    aid: int = id(gaps_item)
+                    annots[aid] = gaps_item
+                    aids_set.add(aid)
+                    result['rects'].append(aid)
+                    profiler(
+                        f'created GapAnnotations item for {len(gap_specs)} '
+                        f'gaps'
+                    )
+
+                    # A/B comparison: optionally create individual arrows
+                    # alongside batch for visual comparison
+                    if show_individual_arrows:
+                        godw = chart.linked.godwidget
+                        arrows: ArrowEditor = ArrowEditor(godw=godw)
+                        for i, spec in enumerate(gap_specs):
+                            if 'arrow_x' not in spec:
+                                continue
+
+                            aid_str: str = str(uuid4())
+                            arrow: pg.ArrowItem = arrows.add(
+                                plot=chart.plotItem,
+                                uid=aid_str,
+                                x=spec['arrow_x'],
+                                y=spec['arrow_y'],
+                                pointing=spec['pointing'],
+                                color='bracket',  # different color
+                                alpha=spec.get('alpha', 169),
+                                headLen=10.0,
+                                headWidth=2.222,
+                                pxMode=True,
+                            )
+                            arrow._abs_x = spec['arrow_x']
+                            arrow._abs_y = spec['arrow_y']
+
+                            annots[aid_str] = arrow
+                            _editors[aid_str] = arrows
+                            aids_set.add(aid_str)
+                            result['arrows'].append(aid_str)
+
+                        profiler(
+                            f'created {len(gap_specs)} individual arrows '
+                            f'for comparison'
+                        )
+
+                # handle text items separately (less common, keep
+                # individual items)
+                n_texts: int = 0
+                for text_spec in text_specs:
+                    kwargs: dict = text_spec.copy()
+                    text: str = kwargs.pop('text')
+                    x: float = float(kwargs.pop('x'))
+                    y: float = float(kwargs.pop('y'))
+                    time_val: float|None = kwargs.pop('time', None)
+
+                    # timestamp-based index lookup
+                    if time_val is not None:
+                        matches = arr[arr['time'] == time_val]
+                        if len(matches) > 0:
+                            x = float(matches[0]['index'])
+                            y = float(matches[0]['close'])
+
+                    color = kwargs.pop('color', 'dad_blue')
+                    anchor = kwargs.pop('anchor', (0, 1))
+                    font_size = kwargs.pop('font_size', None)
+
+                    text_item: pg.TextItem = pg.TextItem(
+                        text,
+                        color=hcolor(color),
+                        anchor=anchor,
+                    )
+
+                    if font_size is None:
+                        from ._style import get_fonts
+                        font, font_small = get_fonts()
+                        font_size = font_small.px_size - 1
+
+                    qfont: QFont = text_item.textItem.font()
+                    qfont.setPixelSize(font_size)
+                    text_item.setFont(qfont)
+
+                    text_item.setPos(float(x), float(y))
+                    chart.plotItem.addItem(text_item)
+
+                    text_item._abs_x = float(x)
+                    text_item._abs_y = float(y)
+
+                    aid: str = str(uuid4())
+                    annots[aid] = text_item
+                    aids_set.add(aid)
+                    result['texts'].append(aid)
+                    n_texts += 1
+
+                profiler(
+                    f'created text annotations: {n_texts} texts'
+                )
+                profiler.finish()
+
+                await annot_req_stream.send(result)
+
             case {
                 'cmd': 'remove',
                 'aid': int(aid)|str(aid),
@@ -471,10 +828,26 @@ async def serve_rc_annots(
                 # XXX: reposition all annotations to ensure they
                 # stay aligned with viz data after reset (eg during
                 # backfill when abs-index range changes)
+                chart: ChartPlotWidget = {
+                    60: ds.hist_chart,
+                    1: ds.chart,
+                }[timeframe]
+                viz: Viz = chart.get_viz(fqme)
+                arr = viz.shm.array
+
                 n_repositioned: int = 0
                 for aid, annot in annots.items():
+                    # GapAnnotations batch items have .reposition()
+                    if hasattr(annot, 'reposition'):
+                        annot.reposition(
+                            array=arr,
+                            fqme=fqme,
+                            timeframe=timeframe,
+                        )
+                        n_repositioned += 1
+
                     # arrows and text items use abs x,y coords
-                    if (
+                    elif (
                         hasattr(annot, '_abs_x')
                         and
                         hasattr(annot, '_abs_y')
@@ -539,12 +912,21 @@ async def remote_annotate(
         finally:
             # ensure all annots for this connection are deleted
             # on any final teardown
+            profiler = Profiler(
+                msg=f'Annotation teardown for ctx {ctx.cid}',
+                disabled=False,
+                ms_threshold=0.0,
+            )
             (_ctx, aids) = _ctxs[ctx.cid]
             assert _ctx is ctx
+            profiler(f'got {len(aids)} aids to remove')
+
             for aid in aids:
                 annot: QGraphicsItem = _annots[aid]
                 assert rm_annot(annot)
 
+            profiler(f'removed all {len(aids)} annotations')
+
 
 class AnnotCtl(Struct):
     '''
@@ -746,6 +1128,64 @@ class AnnotCtl(Struct):
                 )
             return aid
 
+    async def add_batch(
+        self,
+        fqme: str,
+        timeframe: float,
+        rects: list[dict]|None = None,
+        arrows: list[dict]|None = None,
+        texts: list[dict]|None = None,
+        show_individual_arrows: bool = False,
+
+        from_acm: bool = False,
+
+    ) -> dict[str, list[int]]:
+        '''
+        Batch submit multiple annotations in single IPC msg for
+        much faster remote annotation vs. per-annot round-trips.
+
+        Returns dict of annotation IDs:
+        {
+            'rects': [aid1, aid2, ...],
+            'arrows': [aid3, aid4, ...],
+            'texts': [aid5, aid6, ...],
+        }
+
+        '''
+        ipc: MsgStream = self._get_ipc(fqme)
+        with trio.fail_after(10):
+            await ipc.send({
+                'fqme': fqme,
+                'cmd': 'batch',
+                'timeframe': timeframe,
+                'rects': rects or [],
+                'arrows': arrows or [],
+                'texts': texts or [],
+                'show_individual_arrows': show_individual_arrows,
+            })
+            result: dict = await ipc.receive()
+            match result:
+                case {'error': str(msg)}:
+                    log.error(msg)
+                    return {
+                        'rects': [],
+                        'arrows': [],
+                        'texts': [],
+                    }
+
+            # register all AIDs with their IPC streams
+            for aid_list in result.values():
+                for aid in aid_list:
+                    self._ipcs[aid] = ipc
+                    if not from_acm:
+                        self._annot_stack.push_async_callback(
+                            partial(
+                                self.remove,
+                                aid,
+                            )
+                        )
+            return result
+
     async def add_text(
         self,
         fqme: str,
@@ -881,3 +1321,14 @@ async def open_annot_ctl(
                         _annot_stack=annots_stack,
                     )
                     yield client
+
+                    # client exited, measure teardown time
+                    teardown_profiler = Profiler(
+                        msg='Client AnnotCtl teardown',
+                        disabled=False,
+                        ms_threshold=0.0,
+                    )
+                    teardown_profiler('exiting annots_stack')
+
+                teardown_profiler('annots_stack exited')
+                teardown_profiler('exiting gather_contexts')

piker/ui/_search.py

@@ -174,6 +174,13 @@ class CompleterView(QTreeView):
 
         self.setStyleSheet(f"font: {size}px")
 
+    def update_fonts(self) -> None:
+        '''Update font sizes after zoom change.'''
+        self.set_font_size(_font.px_size)
+        self.setIndentation(_font.px_size)
+        self.setFont(_font.font)
+        self.updateGeometry()
+
     def resize_to_results(
         self,
         w: float | None = 0,
@@ -630,6 +637,29 @@ class SearchWidget(QtWidgets.QWidget):
             | align_flag.AlignLeft,
         )
 
+    def update_fonts(self) -> None:
+        '''Update font sizes after zoom change.'''
+        # regenerate label stylesheet with new font size
+        self.label.setStyleSheet(
+            f"""QLabel {{
+                color : {hcolor('default_lightest')};
+                font-size : {_font.px_size - 2}px;
+            }}
+            """
+        )
+        self.label.setFont(_font.font)
+
+        # update search bar and view fonts
+        if hasattr(self.bar, 'update_fonts'):
+            self.bar.update_fonts(_font)
+        elif hasattr(self.bar, 'setFont'):
+            self.bar.setFont(_font.font)
+
+        if hasattr(self.view, 'update_fonts'):
+            self.view.update_fonts()
+
+        self.updateGeometry()
+
     def focus(self) -> None:
         self.show()
         self.bar.focus()

piker/ui/_style.py

@@ -79,9 +79,13 @@ class DpiAwareFont:
         self._font_inches: float = None
         self._screen = None
 
-    def _set_qfont_px_size(self, px_size: int) -> None:
+    def _set_qfont_px_size(
-        self._qfont.setPixelSize(px_size)
+        self,
+        px_size: int,
+    ) -> int:
+        self._qfont.setPixelSize(int(px_size))
         self._qfm = QtGui.QFontMetrics(self._qfont)
+        return self.px_size
 
     @property
     def screen(self) -> QtGui.QScreen:
@@ -124,17 +128,22 @@ class DpiAwareFont:
 
         return size
 
-    def configure_to_dpi(self, screen: QtGui.QScreen | None = None):
+    def configure_to_dpi(
+        self,
+        screen: QtGui.QScreen | None = None,
+        zoom_level: float = 1.0,
+    ) -> int:
         '''
-        Set an appropriately sized font size depending on the screen DPI.
+        Set an appropriately sized font size depending on the screen DPI
+        or scale the size according to `zoom_level`.
 
-        If we end up needing to generalize this more here there are resources
+        If we end up needing to generalize this more here there are
-        listed in the script in ``snippets/qt_screen_info.py``.
+        resources listed in the script in
+        ``snippets/qt_screen_info.py``.
 
         '''
         if self._font_size is not None:
-            self._set_qfont_px_size(self._font_size)
+            return self._set_qfont_px_size(self._font_size * zoom_level)
-            return
 
         # NOTE: if no font size set either in the [ui] section of the
         # config or not yet computed from our magic scaling calcs,
@@ -153,7 +162,7 @@ class DpiAwareFont:
             ldpi = pdpi
 
         mx_dpi = max(pdpi, ldpi)
-        mn_dpi = min(pdpi, ldpi)
+        # mn_dpi = min(pdpi, ldpi)
         scale = round(ldpi/pdpi, ndigits=2)
 
         if mx_dpi <= 97:  # for low dpi use larger font sizes
@@ -162,7 +171,7 @@ class DpiAwareFont:
         else:  # hidpi use smaller font sizes
             inches = _font_sizes['hi'][self._font_size_calc_key]
 
-        dpi = mn_dpi
+        # dpi = mn_dpi
 
         mult = 1.0
 
@@ -197,24 +206,25 @@ class DpiAwareFont:
         # always going to hit that error in range mapping from inches:
         # float to px size: int.
         self._font_inches = inches
-        font_size = math.floor(inches * dpi)
+        font_size = math.floor(inches * pdpi)
+
+        # apply zoom level multiplier
+        font_size = int(font_size * zoom_level)
 
         log.debug(
             f"screen:{screen.name()}\n"
             f"pDPI: {pdpi}, lDPI: {ldpi}, scale: {scale}\n"
+            f"zoom_level: {zoom_level}\n"
             f"\nOur best guess font size is {font_size}\n"
         )
         # apply the size
-        self._set_qfont_px_size(font_size)
+        return self._set_qfont_px_size(font_size)
 
     def boundingRect(self, value: str) -> QtCore.QRectF:
-
+        if self.screen is None:
-        screen = self.screen
-        if screen is None:
             raise RuntimeError("You must call .configure_to_dpi() first!")
 
-        unscaled_br = self._qfm.boundingRect(value)
+        unscaled_br: QtCore.QRectF = self._qfm.boundingRect(value)
-
         return QtCore.QRectF(
             0,
             0,
@@ -228,12 +238,22 @@ _font = DpiAwareFont()
 _font_small = DpiAwareFont(_font_size_key='small')
 
 
-def _config_fonts_to_screen() -> None:
+def _config_fonts_to_screen(
-    'configure global DPI aware font sizes'
+    zoom_level: float = 1.0
+) -> int:
+    '''
+    Configure global DPI aware font size(s).
 
+    If `zoom_level` is provided we apply it to auto-calculated
+    DPI-aware font.
+
+    Return the new `DpiAwareFont.px_size`.
+
+    '''
     global _font, _font_small
-    _font.configure_to_dpi()
+    _font.configure_to_dpi(zoom_level=zoom_level)
-    _font_small.configure_to_dpi()
+    _font_small.configure_to_dpi(zoom_level=zoom_level)
+    return _font.px_size
 
 
 def get_fonts() -> tuple[

piker/ui/_window.py

@@ -18,6 +18,7 @@
 Qt main window singletons and stuff.
 
 """
+from __future__ import annotations
 import os
 import signal
 import time
@@ -38,15 +39,107 @@ from piker.ui.qt import (
     QScreen,
     QCloseEvent,
     QSettings,
+    QEvent,
+    QObject,
 )
 from ..log import get_logger
-from ._style import _font_small, hcolor
+from . import _style
+from ._style import (
+    _font_small,
+    hcolor,
+)
 from ._widget import GodWidget
 
 
 log = get_logger(__name__)
 
 
+class GlobalZoomEventFilter(QObject):
+    '''
+    Application-level event filter for global UI zoom shortcuts.
+
+    This filter intercepts keyboard events BEFORE they reach widgets,
+    allowing us to implement global UI zoom shortcuts that take precedence
+    over widget-specific shortcuts.
+
+    Shortcuts:
+    - Ctrl+Shift+Plus/Equal: Zoom in
+    - Ctrl+Shift+Minus: Zoom out
+    - Ctrl+Shift+0: Reset zoom
+
+    '''
+    def __init__(self, main_window: MainWindow):
+        super().__init__()
+        self.main_window = main_window
+
+    def eventFilter(self, obj: QObject, event: QEvent) -> bool:
+        '''
+        Filter keyboard events for global zoom shortcuts.
+
+        Returns True to filter out (consume) the event, False to pass through.
+
+        '''
+        if event.type() == QEvent.Type.KeyPress:
+            key = event.key()
+            mods = event.modifiers()
+
+            # Mask out the KeypadModifier which Qt sometimes adds
+            mods = mods & ~Qt.KeyboardModifier.KeypadModifier
+
+            # Check if we have Ctrl+Shift (both required)
+            has_ctrl = bool(
+                mods
+                &
+                Qt.KeyboardModifier.ControlModifier
+            )
+            _has_shift = bool(
+                mods
+                &
+                Qt.KeyboardModifier.ShiftModifier
+            )
+
+            # Only handle UI zoom if BOTH Ctrl and Shift are pressed
+            # For Plus key: user presses Cmd+Shift+Equal (which makes Plus)
+            # For Minus key: user presses Cmd+Shift+Minus
+            if (
+                has_ctrl
+                # and
+                # has_shift
+            ):
+                # Zoom in: Ctrl+Shift+Plus
+                # Note: Plus key usually comes as Key_Equal with Shift modifier
+                if key in (
+                    Qt.Key.Key_Plus,
+                    Qt.Key.Key_Equal,
+                ):
+                    self.main_window.zoom_in()
+                    return True  # consume event
+
+                # Zoom out: Ctrl+Shift+Minus
+                # Note: On some keyboards Shift+Minus produces '_' (Underscore)
+                elif key in (
+                    Qt.Key.Key_Minus,
+                    Qt.Key.Key_Underscore,
+                ):
+                    self.main_window.zoom_out()
+                    return True  # consume event
+
+                # Reset zoom: Ctrl+Shift+0
+                # Note: On some keyboards Shift+0 produces ')' (ParenRight)
+                elif key in (
+                    Qt.Key.Key_0,
+                    Qt.Key.Key_ParenRight,
+                ):
+                    self.main_window.reset_zoom()
+                    return True  # consume event
+
+            # Pass through if only Ctrl (no Shift) - this goes to chart zoom
+            # Pass through all other events too
+            return False
+
+        return False
+
+
 class MultiStatus:
 
     bar: QStatusBar
@@ -189,6 +282,24 @@ class MainWindow(QMainWindow):
             self.restoreGeometry(geometry)
             log.debug('Restored window geometry from previous session')
 
+        # zoom level for UI scaling (1.0 = 100%, 1.5 = 150%, etc)
+        # Change this value to set the default startup zoom level
+        self._zoom_level: float = 1.0  # Start at 100% (normal)
+        self._min_zoom: float = 0.5
+        self._max_zoom: float = 3.0  # Reduced from 10.0 to prevent extreme cropping
+        self._zoom_step: float = 0.2  # 20% per keypress
+
+        # event filter for global zoom shortcuts
+        self._zoom_filter: GlobalZoomEventFilter | None = None
+
+    def install_global_zoom_filter(self) -> None:
+        '''Install application-level event filter for global UI zoom shortcuts.'''
+        if self._zoom_filter is None:
+            self._zoom_filter = GlobalZoomEventFilter(self)
+            app = QApplication.instance()
+            app.installEventFilter(self._zoom_filter)
+            log.info('Installed global zoom shortcuts: Ctrl+Shift+Plus/Minus/0')
+
     @property
     def mode_label(self) -> QLabel:
 
@@ -357,6 +468,201 @@ class MainWindow(QMainWindow):
         self.godwidget.on_win_resize(event)
         event.accept()
 
+    def zoom_in(self) -> None:
+        '''
+        Increase overall UI-widgets zoom level by scaling it the
+        global font sizes.
+
+        '''
+        new_zoom: float = min(
+            self._zoom_level + self._zoom_step,
+            self._max_zoom,
+        )
+        if new_zoom != self._zoom_level:
+            self._zoom_level = new_zoom
+            font_size: int = self._apply_zoom()
+            log.info(
+                f'Zoomed in UI\n'
+                f'zoom_step: {self._zoom_step!r}\n'
+                f'zoom_level(%): {self._zoom_level:.1%}\n'
+                f'font_size: {font_size!r}'
+            )
+
+    def zoom_out(self) -> float:
+        '''
+        Decrease UI zoom level.
+
+        '''
+        new_zoom: float = max(self._zoom_level - self._zoom_step, self._min_zoom)
+        if new_zoom != self._zoom_level:
+            self._zoom_level = new_zoom
+            font_size: int = self._apply_zoom()
+            log.info(
+                f'Zoomed out UI\n'
+                f'zoom_step: {self._zoom_step!r}\n'
+                f'zoom_level(%): {self._zoom_level:.1%}\n'
+                f'font_size: {font_size!r}'
+            )
+
+        return new_zoom
+
+    def reset_zoom(self) -> None:
+        '''
+        Reset UI zoom to 100%.
+
+        '''
+        if self._zoom_level != 1.0:
+            self._zoom_level = 1.0
+            font_size: int = self._apply_zoom()
+            log.info(
+                f'Reset zoom level\n'
+                f'zoom_step: {self._zoom_step!r}\n'
+                f'zoom_level(%): {self._zoom_level:.1%}\n'
+                f'font_size: {font_size!r}'
+            )
+
+        return self._zoom_level
+
+    def _apply_zoom(self) -> int:
+        '''
+        Apply current zoom level to all UI elements.
+
+        '''
+        # reconfigure fonts with zoom multiplier
+        font_size: int = _style._config_fonts_to_screen(
+            zoom_level=self._zoom_level
+        )
+
+        # update status bar styling with new font size
+        if self._status_bar:
+            sb = self.statusBar()
+            sb.setStyleSheet((
+                f"color : {hcolor('gunmetal')};"
+                f"background : {hcolor('default_dark')};"
+                f"font-size : {_style._font_small.px_size}px;"
+                "padding : 0px;"
+            ))
+
+        # force update of mode label if it exists
+        if self._status_label:
+            self._status_label.setFont(_style._font_small.font)
+
+        # update godwidget and its children
+        if self.godwidget:
+            # update search widget if it exists
+            if hasattr(self.godwidget, 'search') and self.godwidget.search:
+                self.godwidget.search.update_fonts()
+
+            # update order mode panes in all chart views
+            self._update_chart_order_panes()
+
+            # recursively update all other widgets with stylesheets
+            self._refresh_widget_fonts(self.godwidget)
+            self.godwidget.update()
+
+        return font_size
+
+    def _update_chart_order_panes(self) -> None:
+        '''
+        Update order entry panels in all charts.
+
+        '''
+        if not self.godwidget:
+            return
+
+        # iterate through all linked splits (hist and rt)
+        for splits_name in ['hist_linked', 'rt_linked']:
+            splits = getattr(self.godwidget, splits_name, None)
+            if not splits:
+                continue
+
+            # get main chart
+            chart = getattr(splits, 'chart', None)
+            if chart:
+                # update axes
+                self._update_chart_axes(chart)
+
+                # update order pane
+                if hasattr(chart, 'view'):
+                    view = chart.view
+                    if hasattr(view, 'order_mode') and view.order_mode:
+                        order_mode = view.order_mode
+                        if hasattr(order_mode, 'pane') and order_mode.pane:
+                            order_mode.pane.update_fonts()
+
+            # also check subplots
+            subplots = getattr(splits, 'subplots', {})
+            for name, subplot_chart in subplots.items():
+                # update subplot axes
+                self._update_chart_axes(subplot_chart)
+
+                # update subplot order pane
+                if hasattr(subplot_chart, 'view'):
+                    subplot_view = subplot_chart.view
+                    if hasattr(subplot_view, 'order_mode') and subplot_view.order_mode:
+                        subplot_order_mode = subplot_view.order_mode
+                        if hasattr(subplot_order_mode, 'pane') and subplot_order_mode.pane:
+                            subplot_order_mode.pane.update_fonts()
+
+            # resize all sidepanes to match main chart's sidepane width
+            # this ensures volume/subplot sidepanes match the main chart
+            if splits and hasattr(splits, 'resize_sidepanes'):
+                splits.resize_sidepanes()
+
+    def _update_chart_axes(self, chart) -> None:
+        '''Update axis fonts and sizing for a chart.'''
+        from . import _style
+
+        # update price axis (right side)
+        if hasattr(chart, 'pi') and chart.pi:
+            plot_item = chart.pi
+            # get all axes from plot item
+            for axis_name in ['left', 'right', 'bottom', 'top']:
+                axis = plot_item.getAxis(axis_name)
+                if axis and hasattr(axis, 'update_fonts'):
+                    axis.update_fonts(_style._font)
+
+            # force plot item to recalculate its entire layout
+            plot_item.updateGeometry()
+
+            # force chart widget to update
+            if hasattr(chart, 'updateGeometry'):
+                chart.updateGeometry()
+
+            # trigger a full scene update
+            if hasattr(chart, 'update'):
+                chart.update()
+
+    def _refresh_widget_fonts(self, widget: QWidget) -> None:
+        '''
+        Recursively update font sizes in all child widgets.
+
+        This handles widgets that have font-size hardcoded in their stylesheets.
+        '''
+        from . import _style
+
+        # recursively process all children
+        for child in widget.findChildren(QWidget):
+            # skip widgets that have their own update_fonts method (handled separately)
+            if hasattr(child, 'update_fonts'):
+                continue
+
+            # update child's stylesheet if it has font-size
+            child_stylesheet = child.styleSheet()
+            if child_stylesheet and 'font-size' in child_stylesheet:
+                # for labels and simple widgets, regenerate stylesheet
+                # this is a heuristic - may need refinement
+                try:
+                    child.setFont(_style._font.font)
+                except (AttributeError, RuntimeError):
+                    pass
+
+            # update child's font
+            try:
+                child.setFont(_style._font.font)
+            except (AttributeError, RuntimeError):
+                pass
+
 
 # singleton app per actor
 _qt_win: QMainWindow = None

piker/ui/qt.py

@@ -42,6 +42,7 @@ from PyQt6.QtCore import (
     QSize,
     QModelIndex,
     QItemSelectionModel,
+    QObject,
     pyqtBoundSignal,
     pyqtRemoveInputHook,
     QSettings,

pyproject.toml

@@ -106,7 +106,7 @@ default-groups = [
 
 [dependency-groups]
 uis = [
-  "pyqtgraph",
+  "pyqtgraph >= 0.14.0",
   "qdarkstyle >=3.0.2, <4.0.0",
   "pyqt6 >=6.7.0, <7.0.0",
 
@@ -193,9 +193,12 @@ include = ["piker"]
 
 
 [tool.uv.sources]
-pyqtgraph = { git = "https://github.com/pikers/pyqtgraph.git" }
 tomlkit = { git = "https://github.com/pikers/tomlkit.git", branch ="piker_pin" }
 pyvnc = { git = "https://github.com/regulad/pyvnc.git" }
+# pyqtgraph = { git = "https://github.com/pyqtgraph/pyqtgraph.git", branch = 'master' }
+# pyqtgraph = { path = '../pyqtgraph', editable = true }
+# ?TODO, resync our fork?
+# pyqtgraph = { git = "https://github.com/pikers/pyqtgraph.git" }
 
 # to get fancy next-cmd/suggestion feats prior to 0.22.2 B)
 # https://github.com/xonsh/xonsh/pull/6037
@@ -203,9 +206,8 @@ pyvnc = { git = "https://github.com/regulad/pyvnc.git" }
 # xonsh = { git = 'https://github.com/xonsh/xonsh.git', branch = 'main' }
 
 # XXX since, we're like, always hacking new shite all-the-time. Bp
-tractor = { git = "https://github.com/goodboy/tractor.git", branch ="piker_pin" }
+tractor = { git = "https://github.com/goodboy/tractor.git", branch ="main" }
 # tractor = { git = "https://pikers.dev/goodboy/tractor", branch = "piker_pin" }
-# tractor = { git = "https://pikers.dev/goodboy/tractor", branch = "main" }
 # ------ goodboy ------
 # hackin dev-envs, usually there's something new he's hackin in..
 # tractor = { path = "../tractor", editable = true }

snippets/calc_ppi.xsh

@@ -0,0 +1,64 @@
+#!env xonsh
+'''
+Compute the pxs-per-inch (PPI) naively for the local DE.
+
+NOTE, currently this only supports the `sway`-TWM on wayland.
+
+!TODO!
+- [ ] support Xorg (and possibly other OSs as well?
+- [ ] conver this to pure py code, dropping the `.xsh` specifics
+     instead for `subprocess` API calls?
+- [ ] possibly unify all this with `./qt_screen_info.py` as part of
+     a "PPI config wizard" or something, but more then likely we'll
+     have lib-ified version inside modden/piker by then?
+
+'''
+
+import math
+import json
+
+# XXX, xonsh part using "subprocess mode"
+disp_infos: list[dict] = json.loads($(wlr-randr --json))
+lappy: dict = disp_infos[0]
+
+dims: dict[str, int] = lappy['physical_size']
+w_cm: int = dims['width']
+h_cm: int = dims['height']
+
+# cm per inch
+cpi: float = 25.4
+
+# compute "diagonal" size (aka hypot)
+diag_inches: float = math.sqrt((h_cm/cpi)**2 + (w_cm/cpi)**2)
+
+# compute reso-hypot / inches-hypot
+hi_res: dict[str, float|bool] = lappy['modes'][0]
+w_px: int = hi_res['width']
+h_px: int = hi_res['height']
+
+diag_pxs: float = math.sqrt(h_px**2 + w_px**2)
+unscaled_ppi: float = diag_pxs/diag_inches
+
+# retrieve TWM info on the display (including scaling info)
+sway_disp_info: dict = json.loads($(swaymsg -r -t get_outputs))[0]
+scale: float = sway_disp_info['scale']
+
+print(
+    f'output: {sway_disp_info["name"]!r}\n'
+    f'--- DIMENSIONS ---\n'
+    f'w_cm: {w_cm!r}\n'
+    f'h_cm: {h_cm!r}\n'
+    f'w_px: {w_px!r}\n'
+    f'h_cm: {h_px!r}\n'
+    f'\n'
+    f'--- DIAGONALS ---\n'
+    f'diag_inches: {diag_inches!r}\n'
+    f'diag_pxs: {diag_pxs!r}\n'
+    f'\n'
+    f'--- PPI-related-info ---\n'
+    f'(DE reported) scale: {scale!r}\n'
+    f'unscaled PPI: {unscaled_ppi!r}\n'
+    f'|_ =sqrt(h_px**2 + w_px**2) / sqrt(h_in**2 + w_in**2)\n'
+    f'scaled PPI: {unscaled_ppi/scale!r}\n'
+    f'|_ =unscaled_ppi/scale\n'
+)

snippets/qt_screen_info.py

@@ -31,8 +31,8 @@ Resource list for mucking with DPIs on multiple screens:
 - https://doc.qt.io/qt-5/qguiapplication.html#screenAt
 
 '''
+import os
 
-from pyqtgraph import QtGui
 from PyQt6 import (
     QtCore,
     QtWidgets,
@@ -43,6 +43,11 @@ from PyQt6.QtCore import (
     QSize,
     QRect,
 )
+from pyqtgraph import QtGui
+
+
+# https://doc.qt.io/qt-6/highdpi.html#environment-variable-reference
+os.environ['QT_USE_PHYSICAL_DPI'] = '1'
 
 # Proper high DPI scaling is available in Qt >= 5.6.0. This attibute
 # must be set before creating the application
@@ -58,13 +63,22 @@ if hasattr(Qt, 'AA_UseHighDpiPixmaps'):
         True,
     )
 
+# NOTE, inherits `QGuiApplication`
+# https://doc.qt.io/qt-6/qapplication.html
+# https://doc.qt.io/qt-6/qguiapplication.html
 app = QtWidgets.QApplication([])
+#
+# ^TODO? various global DPI settings?
+# [ ] DPI rounding policy,
+# - https://doc.qt.io/qt-6/qt.html#HighDpiScaleFactorRoundingPolicy-enum
+# - https://doc.qt.io/qt-6/qguiapplication.html#setHighDpiScaleFactorRoundingPolicy
+
 window = QtWidgets.QMainWindow()
 main_widget = QtWidgets.QWidget()
 window.setCentralWidget(main_widget)
 window.show()
 
-pxr: float = main_widget.devicePixelRatioF()
+_main_pxr: float = main_widget.devicePixelRatioF()
 
 # explicitly get main widget and primary displays
 current_screen: QtGui.QScreen = app.screenAt(
@@ -77,7 +91,13 @@ for screen in app.screens():
     name: str = screen.name()
     model: str = screen.model().rstrip()
     size: QSize = screen.size()
-    geo: QRect = screen.availableGeometry()
+    geo: QRect = screen.geometry()
+
+    # device-pixel-ratio
+    # https://doc.qt.io/qt-6/highdpi.html
+    pxr: float = screen.devicePixelRatio()
+
+    unscaled_size: QSize = pxr * size
     phydpi: float = screen.physicalDotsPerInch()
     logdpi: float = screen.logicalDotsPerInch()
     is_primary: bool = screen is primary_screen
@@ -88,11 +108,12 @@ for screen in app.screens():
         f'|_primary: {is_primary}\n'
         f' _current: {is_current}\n'
         f' _model: {model}\n'
-        f' _screen size: {size}\n'
+        f' _size: {size}\n'
-        f' _screen geometry: {geo}\n'
+        f' _geometry: {geo}\n'
-        f' _devicePixelRationF(): {pxr}\n'
+        f' _devicePixelRatio(): {pxr}\n'
-        f' _physical dpi: {phydpi}\n'
+        f' _unscaled-size: {unscaled_size!r}\n'
-        f' _logical dpi: {logdpi}\n'
+        f' _physical-dpi: {phydpi}\n'
+        f' _logical-dpi: {logdpi}\n'
     )
 
 # app-wide font info
@@ -110,8 +131,8 @@ str_w: int = str_br.width()
 
 print(
     f'------ global font settings ------\n'
-    f'font dpi: {fontdpi}\n'
+    f'font dpi: {fontdpi!r}\n'
-    f'font height: {font_h}\n'
+    f'font height: {font_h!r}\n'
-    f'string bounding rect: {str_br}\n'
+    f'string bounding rect: {str_br!r}\n'
-    f'string width : {str_w}\n'
+    f'string width : {str_w!r}\n'
 )

tests/test_dpi_font.py

@@ -0,0 +1,36 @@
+import pytest
+from piker.ui._style import DpiAwareFont
+
+
+class MockScreen:
+    def __init__(self, pdpi, ldpi, name="MockScreen"):
+        self._pdpi = pdpi
+        self._ldpi = ldpi
+        self._name = name
+
+    def physicalDotsPerInch(self):
+        return self._pdpi
+
+    def logicalDotsPerInch(self):
+        return self._ldpi
+
+    def name(self):
+        return self._name
+
+
+@pytest.mark.parametrize(
+    "pdpi, ldpi, expected_px",
+    [
+        (96, 96, 9),       # normal DPI
+        (169, 96, 15),     # HiDPI
+        (120, 96, 10),     # mid-DPI
+    ]
+)
+def test_font_px_size(pdpi, ldpi, expected_px):
+    font = DpiAwareFont()
+    font.configure_to_dpi(screen=MockScreen(pdpi, ldpi))
+
+    px = font.px_size
+    print(f"{pdpi}x{ldpi} DPI -> Computed pixel size: {px}")
+
+    assert px == expected_px

uv.lock

@@ -1034,7 +1034,7 @@ requires-dist = [
     { name = "tomli", specifier = ">=2.0.1,<3.0.0" },
     { name = "tomli-w", specifier = ">=1.0.0,<2.0.0" },
     { name = "tomlkit", git = "https://github.com/pikers/tomlkit.git?branch=piker_pin" },
-    { name = "tractor", git = "https://github.com/goodboy/tractor.git?branch=piker_pin" },
+    { name = "tractor", git = "https://github.com/goodboy/tractor.git?branch=main" },
     { name = "trio", specifier = ">=0.27" },
     { name = "trio-typing", specifier = ">=0.10.0" },
     { name = "trio-util", specifier = ">=0.7.0,<0.8.0" },
@@ -1055,7 +1055,7 @@ dev = [
     { name = "prompt-toolkit", specifier = "==3.0.40" },
     { name = "pyperclip", specifier = ">=1.9.0" },
     { name = "pyqt6", specifier = ">=6.7.0,<7.0.0" },
-    { name = "pyqtgraph", git = "https://github.com/pikers/pyqtgraph.git" },
+    { name = "pyqtgraph", specifier = ">=0.14.0" },
     { name = "pytest" },
     { name = "qdarkstyle", specifier = ">=3.0.2,<4.0.0" },
     { name = "rapidfuzz", specifier = ">=3.2.0,<4.0.0" },
@@ -1073,7 +1073,7 @@ repl = [
 testing = [{ name = "pytest" }]
 uis = [
     { name = "pyqt6", specifier = ">=6.7.0,<7.0.0" },
-    { name = "pyqtgraph", git = "https://github.com/pikers/pyqtgraph.git" },
+    { name = "pyqtgraph", specifier = ">=0.14.0" },
     { name = "qdarkstyle", specifier = ">=3.0.2,<4.0.0" },
     { name = "rapidfuzz", specifier = ">=3.2.0,<4.0.0" },
 ]
@@ -1365,11 +1365,15 @@ wheels = [
 
 [[package]]
 name = "pyqtgraph"
-version = "0.12.3"
+version = "0.14.0"
-source = { git = "https://github.com/pikers/pyqtgraph.git#373f9561ea8ec4fef9b4e8bdcdd4bbf372dd6512" }
+source = { registry = "https://pypi.org/simple" }
 dependencies = [
+    { name = "colorama" },
     { name = "numpy" },
 ]
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/32/36/4c242f81fdcbfa4fb62a5645f6af79191f4097a0577bd5460c24f19cc4ef/pyqtgraph-0.14.0-py3-none-any.whl", hash = "sha256:7abb7c3e17362add64f8711b474dffac5e7b0e9245abdf992e9a44119b7aa4f5", size = 1924755, upload-time = "2025-11-16T19:43:22.251Z" },
+]
 
 [[package]]
 name = "pyreadline3"
@@ -1676,7 +1680,7 @@ wheels = [
 [[package]]
 name = "tractor"
 version = "0.1.0a6.dev0"
-source = { git = "https://github.com/goodboy/tractor.git?branch=piker_pin#566a17ba92469000a01fd18c709ea3e336e72796" }
+source = { git = "https://github.com/goodboy/tractor.git?branch=main#e77198bb64f0467a50e251ed140daee439752354" }
 dependencies = [
     { name = "bidict" },
     { name = "cffi" },