Group bcast errors as `Sampler.bcast_errors`

A new class var `tuple[Exception]` such that the err set can be reffed externally as needed for catching other similar pub-sub/IPC failures in other (related) real-time sub-systems. Also added some now-masked logging for debugging live-feed stream reading issues that should ONLY be used for debugging since they'll greatly degrade HFT perf. Used the new `log.mk_repr()` stuff (that one day we should prolly pull from `modden` as a dep) for pretty console emissions.
.binance.feed: fix `byfque` key-type, drop `rapidfuzz` import
2025-02-11 11:31:13 -05:00 · 2025-02-11 11:13:14 -05:00 · 2025-02-11 11:01:52 -05:00 · 2024-11-22 14:58:30 -05:00 · 2024-11-19 21:39:02 -05:00 · 2024-11-19 21:37:20 -05:00
37 changed files with 4154 additions and 3232 deletions
--- a/README.rst
+++ b/README.rst
@ -1,161 +1,162 @@
 piker
 -----
-trading gear for hackers
+trading gear for hackers.

 |gh_actions|

 .. |gh_actions| image:: https://img.shields.io/endpoint.svg?url=https%3A%2F%2Factions-badge.atrox.dev%2Fpikers%2Fpiker%2Fbadge&style=popout-square
    :target: https://actions-badge.atrox.dev/piker/pikers/goto

-``piker`` is a broker agnostic, next-gen FOSS toolset and runtime for
-real-time computational trading targeted at `hardcore Linux users
-<comp_trader>`_ .
+``piker`` is a broker agnostic, next-gen FOSS toolset for real-time
+computational trading targeted at `hardcore Linux users <comp_trader>`_ .

-we use much bleeding edge tech including (but not limited to):
+we use as much bleeding edge tech as possible including (but not limited to):

 - latest python for glue_
- uv_ for packaging and distribution
- trio_ & tractor_ for our distributed `structured concurrency`_ runtime
- Qt_ for pristine low latency UIs
- pyqtgraph_ (which we've extended) for real-time charting and graphics
- ``polars`` ``numpy`` and ``numba`` for redic `fast numerics`_
- `apache arrow and parquet`_ for time-series storage
+- trio_ & tractor_ for our distributed, multi-core, real-time streaming
+  `structured concurrency`_ runtime B)
+- Qt_ for pristine high performance UIs
+- pyqtgraph_ for real-time charting
+- ``polars`` ``numpy`` and ``numba`` for `fast numerics`_
+- `apache arrow and parquet`_ for time series history management
+  persistence and sharing
+- (prototyped) techtonicdb_ for L2 book storage

-potential projects we might integrate with soon,
-
- (already prototyped in ) techtonicdb_ for L2 book storage
-
-.. _comp_trader: https://jfaleiro.wordpress.com/2019/10/09/computational-trader/
-.. _glue: https://numpy.org/doc/stable/user/c-info.python-as-glue.html#using-python-as-glue
-.. _uv: https://docs.astral.sh/uv/
+.. |travis| image:: https://img.shields.io/travis/pikers/piker/master.svg
+    :target: https://travis-ci.org/pikers/piker
 .. _trio: https://github.com/python-trio/trio
 .. _tractor: https://github.com/goodboy/tractor
 .. _structured concurrency: https://trio.discourse.group/
+.. _marketstore: https://github.com/alpacahq/marketstore
+.. _techtonicdb: https://github.com/0b01/tectonicdb
 .. _Qt: https://www.qt.io/
 .. _pyqtgraph: https://github.com/pyqtgraph/pyqtgraph
+.. _glue: https://numpy.org/doc/stable/user/c-info.python-as-glue.html#using-python-as-glue
 .. _apache arrow and parquet: https://arrow.apache.org/faq/
 .. _fast numerics: https://zerowithdot.com/python-numpy-and-pandas-performance/
-.. _techtonicdb: https://github.com/0b01/tectonicdb
+.. _comp_trader: https://jfaleiro.wordpress.com/2019/10/09/computational-trader/


-focus and feats:
-****************
-fitting with these tenets, we're always open to new
-framework/lib/service interop suggestions and ideas!
+focus and features:
+*******************
+- 100% federated: your code, your hardware, your data feeds, your broker fills.
+- zero web: low latency, native software that doesn't try to re-invent the OS
+- maximal **privacy**: prevent brokers and mms from knowing your
+  planz; smack their spreads with dark volume.
+- zero clutter: modal, context oriented UIs that echew minimalism, reduce
+  thought noise and encourage un-emotion.
+- first class parallelism: built from the ground up on next-gen structured concurrency
+  primitives.
+- traders first: broker/exchange/asset-class agnostic
+- systems grounded: real-time financial signal processing that will
+  make any queuing or DSP eng juice their shorts.
+- non-tina UX: sleek, powerful keyboard driven interaction with expected use in tiling wms
+- data collaboration: every process and protocol is multi-host scalable.
+- fight club ready: zero interest in adoption by suits; no corporate friendly license, ever.

- **100% federated**:
-  your code, your hardware, your data feeds, your broker fills.
+fitting with these tenets, we're always open to new framework suggestions and ideas.

- **zero web**:
-  low latency as a prime objective, native UIs and modern IPC
-  protocols without trying to re-invent the "OS-as-an-app"..
-
- **maximal privacy**:
-  prevent brokers and mms from knowing your planz; smack their
-  spreads with dark volume from a VPN tunnel.
-
- **zero clutter**:
-  modal, context oriented UIs that echew minimalism, reduce thought
-  noise and encourage un-emotion.
-
- **first class parallelism**:
-  built from the ground up on a next-gen structured concurrency
-  supervision sys.
-
- **traders first**:
-  broker/exchange/venue/asset-class/money-sys agnostic
-
- **systems grounded**:
-  real-time financial signal processing (fsp) that will make any
-  queuing or DSP eng juice their shorts.
-
- **non-tina UX**:
-  sleek, powerful keyboard driven interaction with expected use in
-  tiling wms (or maybe even a DDE).
-
- **data collab at scale**:
-  every actor-process and protocol is multi-host aware.
-
- **fight club ready**:
-  zero interest in adoption by suits; no corporate friendly license,
-  ever.
-
-building the hottest looking, fastest, most reliable, keyboard
-friendly FOSS trading platform is the dream; join the cause.
+building the best looking, most reliable, keyboard friendly trading
+platform is the dream; join the cause.


-a sane install with `uv`
-************************
-bc why install with `python` when you can faster with `rust` ::
+sane install with `poetry`
+**************************
+TODO!

-    uv lock
+
+rigorous install on ``nixos`` using ``poetry2nix``
+**************************************************
+TODO!


 hacky install on nixos
 **********************
-``NixOS`` is our core devs' distro of choice for which we offer
+`NixOS` is our core devs' distro of choice for which we offer
 a stringently defined development shell envoirment that can be loaded with::

-    nix-shell default.nix
+    nix-shell develop.nix
+
+this will setup the required python environment to run piker, make sure to
+run::
+
+    pip install -r requirements.txt -e .
+
+once after loading the shell


-start a chart
-*************
-run a realtime OHLCV chart stand-alone::
+install wild-west style via `pip`
+*********************************
+``piker`` is currently under heavy pre-alpha development and as such
+should be cloned from this repo and hacked on directly.

-    piker -l info chart btcusdt.spot.binance xmrusdt.spot.kraken
+for a development install::

-this runs a chart UI (with 1m sampled OHLCV) and shows 2 spot markets from 2 diff cexes
-overlayed on the same graph. Use of `piker` without first starting
-a daemon (`pikerd` - see below) means there is an implicit spawning of the
-multi-actor-runtime (implemented as a `tractor` app).
-
-For additional subsystem feats available through our chart UI see the
-various sub-readmes:
-
- order control using a mouse-n-keyboard UX B)
- cross venue market-pair (what most call "symbol") search, select, overlay Bo
- financial-signal-processing (`piker.fsp`) write-n-reload to sub-chart BO
- src-asset derivatives scan for anal, like the infamous "max pain" XO
+    git clone git@github.com:pikers/piker.git
+    cd piker
+    virtualenv env
+    source ./env/bin/activate
+    pip install -r requirements.txt -e .


-spawn a daemon standalone
-*************************
-we call the root actor-process the ``pikerd``. it can be (and is
-recommended normally to be) started separately from the ``piker
-chart`` program::
+check out our charts
+********************
+bet you weren't expecting this from the foss::
+
+    piker -l info -b kraken -b binance chart btcusdt.binance --pdb
+
+
+this runs the main chart (currently with 1m sampled OHLC) in in debug
+mode and you can practice paper trading using the following
+micro-manual:
+
+``order_mode`` (
+    edge triggered activation by any of the following keys,
+    ``mouse-click`` on y-level to submit at that price
+    ):
+
+    - ``f``/ ``ctl-f`` to stage buy
+    - ``d``/ ``ctl-d`` to stage sell
+    - ``a`` to stage alert
+
+
+``search_mode`` (
+    ``ctl-l`` or ``ctl-space`` to open,
+    ``ctl-c`` or ``ctl-space`` to close
+    ) :
+
+    - begin typing to have symbol search automatically lookup
+      symbols from all loaded backend (broker) providers
+    - arrow keys and mouse click to navigate selection
+    - vi-like ``ctl-[hjkl]`` for navigation
+
+
+you can also configure your position allocation limits from the
+sidepane.
+
+
+run in distributed mode
+***********************
+start the service manager and data feed daemon in the background and
+connect to it::

    pikerd -l info --pdb

-the daemon does nothing until a ``piker``-client (like ``piker
-chart``) connects and requests some particular sub-system. for
-a connecting chart ``pikerd`` will spawn and manage at least,

- a data-feed daemon: ``datad`` which does all the work of comms with
-  the backend provider (in this case the ``binance`` cex).
- a paper-trading engine instance, ``paperboi.binance``, (if no live
-  account has been configured) which allows for auto/manual order
-  control against the live quote stream.
+connect your chart::

-*using* an actor-service (aka micro-daemon) manager which dynamically
-supervises various sub-subsystems-as-services throughout the ``piker``
-runtime-stack.
+    piker -l info -b kraken -b binance chart xmrusdt.binance --pdb

-now you can (implicitly) connect your chart::

-    piker chart btcusdt.spot.binance
-
-since ``pikerd`` was started separately you can now enjoy a persistent
-real-time data stream tied to the daemon-tree's lifetime. i.e. the next
-time you spawn a chart it will obviously not only load much faster
-(since the underlying ``datad.binance`` is left running with its
-in-memory IPC data structures) but also the data-feed and any order
-mgmt states should be persistent until you finally cancel ``pikerd``.
+enjoy persistent real-time data feeds tied to daemon lifetime. the next
+time you spawn a chart it will load much faster since the data feed has
+been cached and is now always running live in the background until you
+kill ``pikerd``.


 if anyone asks you what this project is about
 *********************************************
-you don't talk about it; just use it.
+you don't talk about it.


 how do i get involved?
@ -165,15 +166,6 @@ enter the matrix.

 how come there ain't that many docs
 ***********************************
-i mean we want/need them but building the core right has been higher
-prio then marketting (and likely will stay that way Bp).
-
-soo, suck it up bc,
-
- no one is trying to sell you on anything
- learning the code base is prolly way more valuable
- the UI/UXs are intended to be "intuitive" for any hacker..
-
-we obviously need tonz help so if you want to start somewhere and
-can't necessarily write "advanced" concurrent python/rust code, this
-helping document literally anything might be the place for you!
+suck it up, learn the code; no one is trying to sell you on anything.
+also, we need lotsa help so if you want to start somewhere and can't
+necessarily write serious code, this might be the place for you!
--- a/default.nix
+++ b/default.nix
@ -1,134 +1,82 @@
 with (import <nixpkgs> {});
+with python312Packages;
 let
  glibStorePath = lib.getLib glib;
-  zlibStorePath = lib.getLib zlib;
-  zstdStorePath = lib.getLib zstd;
-  dbusStorePath = lib.getLib dbus;
-  libGLStorePath = lib.getLib libGL;
-  freetypeStorePath = lib.getLib freetype;
+  qtpyStorePath = lib.getLib qtpy;
+  pyqt6StorePath = lib.getLib pyqt6;
+  pyqt6SipStorePath = lib.getLib pyqt6-sip;
  qt6baseStorePath = lib.getLib qt6.qtbase;
-  fontconfigStorePath = lib.getLib fontconfig;
-  libxkbcommonStorePath = lib.getLib libxkbcommon;
-  xcbutilcursorStorePath = lib.getLib xcb-util-cursor;
-
-  qtpyStorePath = lib.getLib python312Packages.qtpy;
-  pyqt6StorePath = lib.getLib python312Packages.pyqt6;
-  pyqt6SipStorePath = lib.getLib python312Packages.pyqt6-sip;
-  rapidfuzzStorePath = lib.getLib python312Packages.rapidfuzz;
-  qdarkstyleStorePath = lib.getLib python312Packages.qdarkstyle;
-
-  xorgLibX11StorePath = lib.getLib xorg.libX11;
-  xorgLibxcbStorePath = lib.getLib xorg.libxcb;
-  xorgxcbutilwmStorePath = lib.getLib xorg.xcbutilwm;
-  xorgxcbutilimageStorePath = lib.getLib xorg.xcbutilimage;
-  xorgxcbutilerrorsStorePath = lib.getLib xorg.xcbutilerrors;
-  xorgxcbutilkeysymsStorePath = lib.getLib xorg.xcbutilkeysyms;
-  xorgxcbutilrenderutilStorePath = lib.getLib xorg.xcbutilrenderutil;
+  rapidfuzzStorePath = lib.getLib rapidfuzz;
+  qdarkstyleStorePath = lib.getLib qdarkstyle;
 in
 stdenv.mkDerivation {
-  name = "piker-qt6-uv";
+  name = "piker-qt6-poetry-shell";
  buildInputs = [
    # System requirements.
    glib
-    zlib
-    dbus
-    zstd
-    libGL
-    freetype
    qt6.qtbase
    libgcc.lib
-    fontconfig
-    libxkbcommon
-
-    # Xorg requirements
-    xcb-util-cursor
-    xorg.libxcb
-    xorg.libX11
-    xorg.xcbutilwm
-    xorg.xcbutilimage
-    xorg.xcbutilerrors
-    xorg.xcbutilkeysyms
-    xorg.xcbutilrenderutil

    # Python requirements.
    python312Full
-    python312Packages.uv
-    python312Packages.qdarkstyle
-    python312Packages.rapidfuzz
-    python312Packages.pyqt6
-    python312Packages.qtpy
+    poetry-core
+    qdarkstyle
+    rapidfuzz
+    pyqt6
+    qtpy
  ];
  src = null;
  shellHook = ''
    set -e

+    export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:${libgcc.lib}/lib:${glibStorePath}/lib
+
    # Set the Qt plugin path
    # export QT_DEBUG_PLUGINS=1

-    QTBASE_PATH="${qt6baseStorePath}/lib"
-    QT_PLUGIN_PATH="$QTBASE_PATH/qt-6/plugins"
-    QT_QPA_PLATFORM_PLUGIN_PATH="$QT_PLUGIN_PATH/platforms"
+    QTBASE_PATH="${qt6baseStorePath}"
+    echo "qtbase path:    $QTBASE_PATH"
+    echo ""
+    export QT_PLUGIN_PATH="$QTBASE_PATH/lib/qt-6/plugins"
+    export QT_QPA_PLATFORM_PLUGIN_PATH="$QT_PLUGIN_PATH/platforms"
+    echo "qt plugin path: $QT_PLUGIN_PATH"
+    echo ""

-    LIB_GCC_PATH="${libgcc.lib}/lib"
-    GLIB_PATH="${glibStorePath}/lib"
-    ZSTD_PATH="${zstdStorePath}/lib"
-    ZLIB_PATH="${zlibStorePath}/lib"
-    DBUS_PATH="${dbusStorePath}/lib"
-    LIBGL_PATH="${libGLStorePath}/lib"
-    FREETYPE_PATH="${freetypeStorePath}/lib"
-    FONTCONFIG_PATH="${fontconfigStorePath}/lib"
-    LIB_XKB_COMMON_PATH="${libxkbcommonStorePath}/lib"
+    # Maybe create venv & install deps
+    poetry install --with uis

-    XCB_UTIL_CURSOR_PATH="${xcbutilcursorStorePath}/lib"
-    XORG_LIB_X11_PATH="${xorgLibX11StorePath}/lib"
-    XORG_LIB_XCB_PATH="${xorgLibxcbStorePath}/lib"
-    XORG_XCB_UTIL_IMAGE_PATH="${xorgxcbutilimageStorePath}/lib"
-    XORG_XCB_UTIL_WM_PATH="${xorgxcbutilwmStorePath}/lib"
-    XORG_XCB_UTIL_RENDER_UTIL_PATH="${xorgxcbutilrenderutilStorePath}/lib"
-    XORG_XCB_UTIL_KEYSYMS_PATH="${xorgxcbutilkeysymsStorePath}/lib"
-    XORG_XCB_UTIL_ERRORS_PATH="${xorgxcbutilerrorsStorePath}/lib"
+    # Use pyqt6 from System, patch activate script
+    ACTIVATE_SCRIPT_PATH="$(poetry env info --path)/bin/activate"

-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$QTBASE_PATH"
-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$QT_PLUGIN_PATH"
-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$QT_QPA_PLATFORM_PLUGIN_PATH"
+    export RPDFUZZ_PATH="${rapidfuzzStorePath}/lib/python3.12/site-packages"
+    export QDRKSTYLE_PATH="${qdarkstyleStorePath}/lib/python3.12/site-packages"
+    export QTPY_PATH="${qtpyStorePath}/lib/python3.12/site-packages"
+    export PYQT6_PATH="${pyqt6StorePath}/lib/python3.12/site-packages"
+    export PYQT6_SIP_PATH="${pyqt6SipStorePath}/lib/python3.12/site-packages"
+    echo "rapidfuzz at:   $RPDFUZZ_PATH"
+    echo "qdarkstyle at:  $QDRKSTYLE_PATH"
+    echo "qtpy at:        $QTPY_PATH" 
+    echo "pyqt6 at:       $PYQT6_PATH"
+    echo "pyqt6-sip at:   $PYQT6_SIP_PATH"
+    echo ""

-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$LIB_GCC_PATH"
-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$DBUS_PATH"
-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$GLIB_PATH"
-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$ZLIB_PATH"
-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$ZSTD_PATH"
-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$LIBGL_PATH"
-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$FONTCONFIG_PATH"
-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$FREETYPE_PATH"
-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$LIB_XKB_COMMON_PATH"
+    PATCH="export PYTHONPATH=\""

-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$XCB_UTIL_CURSOR_PATH"
-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$XORG_LIB_X11_PATH"
-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$XORG_LIB_XCB_PATH"
-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$XORG_XCB_UTIL_IMAGE_PATH"
-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$XORG_XCB_UTIL_WM_PATH"
-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$XORG_XCB_UTIL_RENDER_UTIL_PATH"
-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$XORG_XCB_UTIL_KEYSYMS_PATH"
-    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$XORG_XCB_UTIL_ERRORS_PATH"
+    PATCH="$PATCH\$RPDFUZZ_PATH"
+    PATCH="$PATCH:\$QDRKSTYLE_PATH"
+    PATCH="$PATCH:\$QTPY_PATH"
+    PATCH="$PATCH:\$PYQT6_PATH"
+    PATCH="$PATCH:\$PYQT6_SIP_PATH"

-    export LD_LIBRARY_PATH
+    PATCH="$PATCH\""

-    RPDFUZZ_PATH="${rapidfuzzStorePath}/lib/python3.12/site-packages"
-    QDRKSTYLE_PATH="${qdarkstyleStorePath}/lib/python3.12/site-packages"
-    QTPY_PATH="${qtpyStorePath}/lib/python3.12/site-packages"
-    PYQT6_PATH="${pyqt6StorePath}/lib/python3.12/site-packages"
-    PYQT6_SIP_PATH="${pyqt6SipStorePath}/lib/python3.12/site-packages"
-
-    PATCH="$PATCH:$RPDFUZZ_PATH"
-    PATCH="$PATCH:$QDRKSTYLE_PATH"
-    PATCH="$PATCH:$QTPY_PATH"
-    PATCH="$PATCH:$PYQT6_PATH"
-    PATCH="$PATCH:$PYQT6_SIP_PATH"
-
-    export PATCH
-
-    # Install deps
-    uv lock
+    if grep -q "$PATCH" "$ACTIVATE_SCRIPT_PATH"; then
+        echo "venv is already patched."
+    else
+        echo "patching $ACTIVATE_SCRIPT_PATH to use pyqt6 from nixos..."
+        sed -i "\$i$PATCH" $ACTIVATE_SCRIPT_PATH
+    fi

+    poetry shell
  '';
 }
--- a/piker/brokers/init.py
+++ b/piker/brokers/init.py
@ -51,6 +51,7 @@ __brokers__: list[str] = [
    'ib',
    'kraken',
    'kucoin',
+    'deribit',

    # broken but used to work
    # 'questrade',
@ -61,7 +62,6 @@ __brokers__: list[str] = [
    # wstrade
    # iex

-    # deribit
    # bitso
 ]

--- a/piker/brokers/_daemon.py
+++ b/piker/brokers/_daemon.py
@ -23,6 +23,7 @@ from __future__ import annotations
 from contextlib import (
    asynccontextmanager as acm,
 )
+from functools import partial
 from types import ModuleType
 from typing import (
    TYPE_CHECKING,
@ -60,12 +61,13 @@ async def _setup_persistent_brokerd(
    ctx: tractor.Context,
    brokername: str,
    loglevel: str | None = None,
+    debug_mode: bool = False,

 ) -> None:
    '''
-    Allocate a actor-wide service nursery in ``brokerd``
-    such that feeds can be run in the background persistently by
-    the broker backend as needed.
+    Allocate a actor-wide service nursery in `brokerd` such that
+    feeds can be run in the background persistently by the broker
+    backend as needed.

    '''
    # NOTE: we only need to setup logging once (and only) here
@ -86,6 +88,18 @@ async def _setup_persistent_brokerd(
    from piker.data import feed
    assert not feed._bus

+    if (
+        debug_mode
+        and
+        tractor.current_actor().is_infected_aio()
+    ):
+        # NOTE, whenever running `asyncio` in provider's actor
+        # runtime be sure we enabled `breakpoint()` support
+        # for non-`trio.Task` usage.
+        from tractor.devx._debug import maybe_init_greenback
+        await maybe_init_greenback()
+        # breakpoint()  # XXX, SHOULD WORK from `trio.Task`!
+
    # allocate a nursery to the bus for spawning background
    # tasks to service client IPC requests, normally
    # `tractor.Context` connections to explicitly required
@ -145,18 +159,21 @@ def broker_init(
      above.

    '''
-    from ..brokers import get_brokermod
-    brokermod = get_brokermod(brokername)
+    brokermod: ModuleType = get_brokermod(brokername)
    modpath: str = brokermod.__name__
-
-    start_actor_kwargs['name'] = f'brokerd.{brokername}'
-    start_actor_kwargs.update(
-        getattr(
-            brokermod,
-            '_spawn_kwargs',
-            {},
-        )
+    spawn_kws: dict = getattr(
+        brokermod,
+        '_spawn_kwargs',
+        {},
    )
+    # ^^ NOTE, here we pull any runtime parameters specific
+    # to spawning the sub-actor for the backend. For ex.
+    # both `ib` and `deribit` rely on,
+    #  `'infect_asyncio': True,` since they both
+    #  use `tractor`'s "infected `asyncio` mode"
+    #  for their libs but you could also do something like
+    #  `'debug_mode: True` which would be like passing
+    #  `--pdb` for just that provider backend.

    # XXX TODO: make this not so hacky/monkeypatched..
    # -> we need a sane way to configure the logging level for all
@ -166,8 +183,7 @@ def broker_init(

    # lookup actor-enabled modules declared by the backend offering the
    # `brokerd` endpoint(s).
-    enabled: list[str]
-    enabled = start_actor_kwargs['enable_modules'] = [
+    enabled: list[str] = [
        __name__,  # so that eps from THIS mod can be invoked
        modpath,
    ]
@ -179,9 +195,13 @@ def broker_init(
        subpath: str = f'{modpath}.{submodname}'
        enabled.append(subpath)

+    datad_kwargs: dict = {
+        'name': f'brokerd.{brokername}',
+        'enable_modules': enabled,
+    }
    return (
        brokermod,
-        start_actor_kwargs,  # to `ActorNursery.start_actor()`
+        start_actor_kwargs | datad_kwargs | spawn_kws,  # to `ActorNursery.start_actor()`

        # XXX see impl above; contains all (actor global)
        # setup/teardown expected in all `brokerd` actor instances.
@ -190,14 +210,17 @@ def broker_init(


 async def spawn_brokerd(
-
    brokername: str,
    loglevel: str | None = None,

    **tractor_kwargs,

 ) -> bool:
+    '''
+    Spawn a `brokerd.<backendname>` subactor service daemon
+    using `pikerd`'s service mngr.

+    '''
    from piker.service._util import log  # use service mngr log
    log.info(f'Spawning {brokername} broker daemon')

@ -211,33 +234,41 @@ async def spawn_brokerd(
        **tractor_kwargs,
    )

-    brokermod = get_brokermod(brokername)
-    extra_tractor_kwargs = getattr(brokermod, '_spawn_kwargs', {})
-    tractor_kwargs.update(extra_tractor_kwargs)
-
    # ask `pikerd` to spawn a new sub-actor and manage it under its
    # actor nursery
-    from piker.service import Services
-
+    from piker.service import (
+        get_service_mngr,
+        ServiceMngr,
+    )
    dname: str = tractor_kwargs.pop('name')  # f'brokerd.{brokername}'
-    portal = await Services.actor_n.start_actor(
-        dname,
-        enable_modules=_data_mods + tractor_kwargs.pop('enable_modules'),
-        debug_mode=Services.debug_mode,
+    mngr: ServiceMngr = get_service_mngr()
+    ctx: tractor.Context = await mngr.start_service(
+        daemon_name=dname,
+        ctx_ep=partial(
+            # signature of target root-task endpoint
+            daemon_fixture_ep,
+
+            # passed to daemon_fixture_ep(**kwargs)
+            brokername=brokername,
+            loglevel=loglevel,
+            debug_mode=mngr.debug_mode,
+        ),
+        debug_mode=mngr.debug_mode,
+        # ^TODO, allow overriding this per-daemon from client side?
+        # |_ it's already supported in `tractor` so..
+
+        loglevel=loglevel,
+        enable_modules=(
+            _data_mods
+            +
+            tractor_kwargs.pop('enable_modules')
+        ),
        **tractor_kwargs
    )
-
-    # NOTE: the service mngr expects an already spawned actor + its
-    # portal ref in order to do non-blocking setup of brokerd
-    # service nursery.
-    await Services.start_service_task(
-        dname,
-        portal,
-
-        # signature of target root-task endpoint
-        daemon_fixture_ep,
-        brokername=brokername,
-        loglevel=loglevel,
+    assert (
+        not ctx.cancel_called
+        and ctx.portal  # parent side
+        and dname in ctx.chan.uid  # subactor is named as desired
    )
    return True

@ -262,8 +293,7 @@ async def maybe_spawn_brokerd(
    from piker.service import maybe_spawn_daemon

    async with maybe_spawn_daemon(
-
-        f'brokerd.{brokername}',
+        service_name=f'brokerd.{brokername}',
        service_task_target=spawn_brokerd,
        spawn_args={
            'brokername': brokername,
--- a/piker/brokers/binance/feed.py
+++ b/piker/brokers/binance/feed.py
@ -110,7 +110,6 @@ class AggTrade(Struct, frozen=True):

 async def stream_messages(
    ws: NoBsWs,
-
 ) -> AsyncGenerator[NoBsWs, dict]:

    # TODO: match syntax here!
@ -221,8 +220,6 @@ def make_sub(pairs: list[str], sub_name: str, uid: int) -> dict[str, str]:
    }


-# TODO, why aren't frame resp `log.info()`s showing in upstream
-# code?!
@acm
 async def open_history_client(
    mkt: MktPair,
@ -465,8 +462,6 @@ async def stream_quotes(
    ):
        init_msgs: list[FeedInit] = []
        for sym in symbols:
-            mkt: MktPair
-            pair: Pair
            mkt, pair = await get_mkt_info(sym)

            # build out init msgs according to latest spec
@ -515,6 +510,7 @@ async def stream_quotes(

            # start streaming
            async for typ, quote in msg_gen:
+
                # period = time.time() - last
                # hz = 1/period if period else float('inf')
                # if hz > 60:
--- a/piker/brokers/deribit/init.py
+++ b/piker/brokers/deribit/init.py
@ -25,6 +25,7 @@ from .api import (
    get_client,
 )
 from .feed import (
+    get_mkt_info,
    open_history_client,
    open_symbol_search,
    stream_quotes,
@ -34,15 +35,20 @@ from .feed import (
    # open_trade_dialog,
    # norm_trade_records,
 # )
+from .venues import (
+    OptionPair,
+)

 log = get_logger(__name__)

 __all__ = [
    'get_client',
 #    'trades_dialogue',
+    'get_mkt_info',
    'open_history_client',
    'open_symbol_search',
    'stream_quotes',
+    'OptionPair',
 #    'norm_trade_records',
 ]

--- a/piker/brokers/deribit/api.py
+++ b/piker/brokers/deribit/api.py
--- a/piker/brokers/deribit/feed.py
+++ b/piker/brokers/deribit/feed.py
@ -18,38 +18,59 @@
 Deribit backend.

 '''
+from __future__ import annotations
 from contextlib import asynccontextmanager as acm
 from datetime import datetime
-from typing import Any, Optional, Callable
+from typing import (
+    # Any,
+    # Optional,
+    Callable,
+)
+# from pprint import pformat
 import time

+import cryptofeed
 import trio
 from trio_typing import TaskStatus
-import pendulum
-from rapidfuzz import process as fuzzy
+from pendulum import (
+    from_timestamp,
+)
 import numpy as np
 import tractor

-from piker.brokers import open_cached_client
-from piker.log import get_logger, get_console_log
-from piker.data import ShmArray
-from piker.brokers._util import (
-    BrokerError,
+from piker.accounting import (
+    Asset,
+    MktPair,
+    unpack_fqme,
+)
+from piker.brokers import (
+    open_cached_client,
+    NoData,
    DataUnavailable,
 )
-
-from cryptofeed import FeedHandler
-from cryptofeed.defines import (
-    DERIBIT, L1_BOOK, TRADES, OPTION, CALL, PUT
+from piker._cacheables import (
+    async_lifo_cache,
 )
-from cryptofeed.symbols import Symbol
+from piker.log import (
+    get_logger,
+    mk_repr,
+)
+from piker.data.validate import FeedInit
+

 from .api import (
-    Client, Trade,
-    get_config,
-    str_to_cb_sym, piker_sym_to_cb_sym, cb_sym_to_deribit_inst,
+    Client,
+    # get_config,
+    piker_sym_to_cb_sym,
+    cb_sym_to_deribit_inst,
+    str_to_cb_sym,
    maybe_open_price_feed
 )
+from .venues import (
+    Pair,
+    OptionPair,
+    Trade,
+)

 _spawn_kwargs = {
    'infect_asyncio': True,
@ -64,90 +85,215 @@ async def open_history_client(
    mkt: MktPair,
 ) -> tuple[Callable, int]:

-    fnstrument: str = mkt.bs_fqme
    # TODO implement history getter for the new storage layer.
    async with open_cached_client('deribit') as client:

+        pair: OptionPair = client._pairs[mkt.dst.name]
+        # XXX NOTE, the cuckers use ms !!!
+        creation_time_s: int = pair.creation_timestamp/1000
+
        async def get_ohlc(
-            end_dt: Optional[datetime] = None,
-            start_dt: Optional[datetime] = None,
+            timeframe: float,
+            end_dt: datetime | None = None,
+            start_dt: datetime | None = None,

        ) -> tuple[
            np.ndarray,
            datetime,  # start
            datetime,  # end
        ]:
+            if timeframe != 60:
+                raise DataUnavailable('Only 1m bars are supported')

-            array = await client.bars(
-                instrument,
+            array: np.ndarray = await client.bars(
+                mkt,
                start_dt=start_dt,
                end_dt=end_dt,
            )
            if len(array) == 0:
-                raise DataUnavailable
+                if (
+                    end_dt is None
+                ):
+                    raise DataUnavailable(
+                        'No history seems to exist yet?\n\n'
+                        f'{mkt}'
+                    )
+                elif (
+                    end_dt
+                    and
+                    end_dt.timestamp() < creation_time_s
+                ):
+                    # the contract can't have history
+                    # before it was created.
+                    pair_type_str: str = type(pair).__name__
+                    create_dt: datetime = from_timestamp(creation_time_s)
+                    raise DataUnavailable(
+                        f'No history prior to\n'
+                        f'`{pair_type_str}.creation_timestamp: int = '
+                        f'{pair.creation_timestamp}\n\n'
+                        f'------ deribit sux ------\n'
+                        f'WHICH IN "NORMAL PEOPLE WHO USE EPOCH TIME" form is,\n'
+                        f'creation_time_s: {creation_time_s}\n'
+                        f'create_dt: {create_dt}\n'
+                    )
+                raise NoData(
+                    f'No frame for {start_dt} -> {end_dt}\n'
+                )

-            start_dt = pendulum.from_timestamp(array[0]['time'])
-            end_dt = pendulum.from_timestamp(array[-1]['time'])
+            start_dt = from_timestamp(array[0]['time'])
+            end_dt = from_timestamp(array[-1]['time'])
+
+            times = array['time']
+            if not times.any():
+                raise ValueError(
+                    'Bad frame with null-times?\n\n'
+                    f'{times}'
+                )
+
+            if end_dt is None:
+                inow: int = round(time.time())
+                if (inow - times[-1]) > 60:
+                    await tractor.pause()

            return array, start_dt, end_dt

-        yield get_ohlc, {'erlangs': 3, 'rate': 3}
+        yield (
+            get_ohlc,
+            {  # backfill config
+                'erlangs': 3,
+                'rate': 3,
+            }
+        )
+
+
+@async_lifo_cache()
+async def get_mkt_info(
+    fqme: str,
+
+) -> tuple[MktPair, Pair|OptionPair] | None:
+
+    # uppercase since kraken bs_mktid is always upper
+    if 'deribit' not in fqme.lower():
+        fqme += '.deribit'
+
+    mkt_mode: str = ''
+    broker, mkt_ep, venue, expiry = unpack_fqme(fqme)
+
+    # NOTE: we always upper case all tokens to be consistent with
+    # binance's symbology style for pairs, like `BTCUSDT`, but in
+    # theory we could also just keep things lower case; as long as
+    # we're consistent and the symcache matches whatever this func
+    # returns, always!
+    expiry: str = expiry.upper()
+    venue: str = venue.upper()
+    # venue_lower: str = venue.lower()
+
+    mkt_mode: str = 'option'
+
+    async with open_cached_client(
+        'deribit',
+    ) as client:
+
+        assets: dict[str, Asset] = await client.get_assets()
+        pair_str: str = mkt_ep.lower()
+
+        pair: Pair = await client.exch_info(
+            sym=pair_str,
+        )
+        mkt_mode = pair.venue
+        client.mkt_mode = mkt_mode
+
+        dst: Asset | None = assets.get(pair.bs_dst_asset)
+        src: Asset | None = assets.get(pair.bs_src_asset)
+
+        mkt = MktPair(
+            dst=dst,
+            src=src,
+            price_tick=pair.price_tick,
+            size_tick=pair.size_tick,
+            bs_mktid=pair.symbol,
+            venue=mkt_mode,
+            broker='deribit',
+            _atype=mkt_mode,
+            _fqme_without_src=True,
+
+            # expiry=pair.expiry,
+            # XXX TODO, currently we don't use it since it's
+            # already "described" in the `OptionPair.symbol: str`
+            # and if we slap in the ISO repr it's kinda hideous..
+            # -[ ] figure out the best either std
+        )
+        return mkt, pair


 async def stream_quotes(
-
    send_chan: trio.abc.SendChannel,
    symbols: list[str],
    feed_is_live: trio.Event,
-    loglevel: str = None,

    # startup sync
    task_status: TaskStatus[tuple[dict, dict]] = trio.TASK_STATUS_IGNORED,

 ) -> None:
-    # XXX: required to propagate ``tractor`` loglevel to piker logging
-    get_console_log(loglevel or tractor.current_actor().loglevel)
+    '''
+    Open a live quote stream for the market set defined by `symbols`.

-    sym = symbols[0]
+    Internally this starts a `cryptofeed.FeedHandler` inside an `asyncio`-side
+    task and relays through L1 and `Trade` msgs here to our `trio.Task`.
+
+    '''
+    sym = symbols[0].split('.')[0]
+    init_msgs: list[FeedInit] = []
+
+    # multiline nested `dict` formatter (since rn quote-msgs are
+    # just that).
+    pfmt: Callable[[str], str] = mk_repr(
+        # so we can see `deribit`'s delightfully mega-long bs fields..
+        maxstring=100,
+    )

    async with (
        open_cached_client('deribit') as client,
        send_chan as send_chan
    ):
+        mkt: MktPair
+        pair: Pair
+        mkt, pair = await get_mkt_info(sym)

-        init_msgs = {
-            # pass back token, and bool, signalling if we're the writer
-            # and that history has been written
-            sym: {
-                'symbol_info': {
-                    'asset_type': 'option',
-                    'price_tick_size': 0.0005
-                },
-                'shm_write_opts': {'sum_tick_vml': False},
-                'fqsn': sym,
-            },
-        }
+        # build out init msgs according to latest spec
+        init_msgs.append(
+            FeedInit(
+                mkt_info=mkt,
+            )
+        )
+        # build `cryptofeed` feed-handle
+        cf_sym: cryptofeed.Symbol = piker_sym_to_cb_sym(sym)

-        nsym = piker_sym_to_cb_sym(sym)
+        from_cf: tractor.to_asyncio.LinkedTaskChannel
+        async with maybe_open_price_feed(sym) as from_cf:

-        async with maybe_open_price_feed(sym) as stream:
+            # load the "last trades" summary
+            last_trades_res: cryptofeed.LastTradesResult = await client.last_trades(
+                cb_sym_to_deribit_inst(cf_sym),
+                count=1,
+            )
+            last_trades: list[Trade] = last_trades_res.trades

-            cache = await client.cache_symbols()
+            # TODO, do we even need this or will the above always
+            # work?
+            # if not last_trades:
+            #     await tractor.pause()
+            #     async for typ, quote in from_cf:
+            #         if typ == 'trade':
+            #             last_trade = Trade(**(quote['data']))
+            #             break

-            last_trades = (await client.last_trades(
-                cb_sym_to_deribit_inst(nsym), count=1)).trades
+            # else:
+            last_trade = Trade(
+                **(last_trades[0])
+            )

-            if len(last_trades) == 0:
-                last_trade = None
-                async for typ, quote in stream:
-                    if typ == 'trade':
-                        last_trade = Trade(**(quote['data']))
-                        break
-
-            else:
-                last_trade = Trade(**(last_trades[0]))
-
-            first_quote = {
+            first_quote: dict = {
                'symbol': sym,
                'last': last_trade.price,
                'brokerd_ts': last_trade.timestamp,
@ -158,13 +304,84 @@ async def stream_quotes(
                    'broker_ts': last_trade.timestamp
                }]
            }
-            task_status.started((init_msgs,  first_quote))
+            task_status.started((
+                init_msgs,
+                first_quote,
+            ))

            feed_is_live.set()

-            async for typ, quote in stream:
-                topic = quote['symbol']
-                await send_chan.send({topic: quote})
+            # NOTE XXX, static for now!
+            # => since this only handles ONE mkt feed at a time we
+            # don't need a lookup table to map interleaved quotes
+            # from multiple possible mkt-pairs
+            topic: str = mkt.bs_fqme
+
+            # deliver until cancelled
+            async for typ, ref in from_cf:
+                match typ:
+                    case 'trade':
+                        trade: cryptofeed.types.Trade = ref
+
+                        # TODO, re-impl this according to teh ideal
+                        # fqme for opts that we choose!!
+                        bs_fqme: str = cb_sym_to_deribit_inst(
+                            str_to_cb_sym(trade.symbol)
+                        ).lower()
+
+                        piker_quote: dict = {
+                            'symbol': bs_fqme,
+                            'last': trade.price,
+                            'broker_ts': time.time(),
+                            # ^TODO, name this `brokerd/datad_ts` and
+                            # use `time.time_ns()` ??
+                            'ticks': [{
+                                'type': 'trade',
+                                'price': float(trade.price),
+                                'size': float(trade.amount),
+                                'broker_ts': trade.timestamp,
+                            }],
+                        }
+                        log.info(
+                            f'deribit {typ!r} quote for {sym!r}\n\n'
+                            f'{trade}\n\n'
+                            f'{pfmt(piker_quote)}\n'
+                        )
+
+                    case 'l1':
+                        book: cryptofeed.types.L1Book = ref
+
+                        # TODO, so this is where we can possibly change things
+                        # and instead lever the `MktPair.bs_fqme: str` output?
+                        bs_fqme: str = cb_sym_to_deribit_inst(
+                            str_to_cb_sym(book.symbol)
+                        ).lower()
+
+                        piker_quote: dict = {
+                            'symbol': bs_fqme,
+                            'ticks': [
+
+                                {'type': 'bid',
+                                 'price': float(book.bid_price),
+                                 'size': float(book.bid_size)},
+
+                                {'type': 'bsize',
+                                 'price': float(book.bid_price),
+                                 'size': float(book.bid_size),},
+
+                                {'type': 'ask',
+                                 'price': float(book.ask_price),
+                                 'size': float(book.ask_size),},
+
+                                {'type': 'asize',
+                                 'price': float(book.ask_price),
+                                 'size': float(book.ask_size),}
+                            ]
+                        }
+
+                await send_chan.send({
+                    topic: piker_quote,
+                })


@tractor.context
@ -174,12 +391,21 @@ async def open_symbol_search(
    async with open_cached_client('deribit') as client:

        # load all symbols locally for fast search
-        cache = await client.cache_symbols()
+        # cache = client._pairs
        await ctx.started()

        async with ctx.open_stream() as stream:
-
+            pattern: str
            async for pattern in stream:
-                # repack in dict form
-                await stream.send(
-                    await client.search_symbols(pattern))
+
+                # NOTE: pattern fuzzy-matching is done within
+                # the methd impl.
+                pairs: dict[str, Pair] = await client.search_symbols(
+                    pattern,
+                )
+                # repack in fqme-keyed table
+                byfqme: dict[str, Pair] = {}
+                for pair in pairs.values():
+                    byfqme[pair.bs_fqme] = pair
+
+                await stream.send(byfqme)
--- a/piker/brokers/deribit/venues.py
+++ b/piker/brokers/deribit/venues.py
@ -0,0 +1,196 @@
+# piker: trading gear for hackers
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+"""
+Per market data-type definitions and schemas types.
+
+"""
+from __future__ import annotations
+import pendulum
+from typing import (
+    Literal,
+    Optional,
+)
+from decimal import Decimal
+
+from piker.types import Struct
+
+
+# API endpoint paths by venue / sub-API
+_domain: str = 'deribit.com'
+_url = f'https://www.{_domain}'
+
+# WEBsocketz
+_ws_url: str = f'wss://www.{_domain}/ws/api/v2'
+
+# test nets
+_testnet_ws_url: str = f'wss://test.{_domain}/ws/api/v2'
+
+MarketType = Literal[
+    'option'
+]
+
+
+def get_api_eps(venue: MarketType) -> tuple[str, str]:
+    '''
+    Return API ep root paths per venue.
+
+    '''
+    return {
+        'option': (
+            _ws_url,
+        ),
+    }[venue]
+
+
+class Pair(Struct, frozen=True, kw_only=True):
+
+    symbol: str
+
+    # src
+    quote_currency: str # 'BTC'
+
+    # dst
+    base_currency: str # "BTC",
+
+    tick_size: float # 0.0001 # [{'above_price': 0.005, 'tick_size': 0.0005}]
+    tick_size_steps: list[dict[str, float]] 
+
+    @property
+    def price_tick(self) -> Decimal:
+        return Decimal(str(self.tick_size_steps[0]['above_price']))
+
+    @property
+    def size_tick(self) -> Decimal:
+        return Decimal(str(self.tick_size))
+
+    @property
+    def bs_fqme(self) -> str:
+        return f'{self.symbol}'
+
+    @property
+    def bs_mktid(self) -> str:
+        return f'{self.symbol}.{self.venue}'
+
+
+class OptionPair(Pair, frozen=True):
+
+    taker_commission: float # 0.0003
+    strike: float # 5000.0
+    settlement_period: str # 'day'
+    settlement_currency: str # "BTC",
+    rfq: bool # false
+    price_index: str # 'btc_usd'
+    option_type: str # 'call'
+    min_trade_amount: float # 0.1
+    maker_commission: float # 0.0003
+    kind: str # 'option'
+    is_active: bool # true
+    instrument_type: str # 'reversed'
+    instrument_name: str # 'BTC-1SEP24-55000-C'
+    instrument_id: int # 364671
+    expiration_timestamp: int # 1725177600000
+    creation_timestamp: int # 1724918461000
+    counter_currency: str # 'USD' 
+    contract_size: float # '1.0'
+    block_trade_tick_size: float # '0.0001'
+    block_trade_min_trade_amount: int # '25'
+    block_trade_commission: float # '0.003'
+
+    # NOTE: see `.data._symcache.SymbologyCache.load()` for why
+    ns_path: str = 'piker.brokers.deribit:OptionPair'
+
+    # TODO, impl this without the MM:SS part of
+    # the `'THH:MM:SS..'` etc..
+    @property
+    def expiry(self) -> str:
+        iso_date = pendulum.from_timestamp(
+            self.expiration_timestamp / 1000
+        ).isoformat()
+        return iso_date 
+
+    @property
+    def venue(self) -> str:
+        return f'{self.instrument_type}_option'
+
+    @property
+    def bs_fqme(self) -> str:
+        return f'{self.symbol}'
+
+    @property
+    def bs_src_asset(self) -> str:
+        return f'{self.quote_currency}'
+
+    @property
+    def bs_dst_asset(self) -> str:
+        return f'{self.symbol}'
+
+
+PAIRTYPES: dict[MarketType, Pair] = {
+    'option': OptionPair,
+}
+
+
+class JSONRPCResult(Struct):
+    id: int
+    usIn: int
+    usOut: int
+    usDiff: int
+    testnet: bool
+    jsonrpc: str = '2.0'
+    error: Optional[dict] = None
+    result: Optional[list[dict]] = None
+
+
+class JSONRPCChannel(Struct):
+    method: str
+    params: dict
+    jsonrpc: str = '2.0'
+
+
+class KLinesResult(Struct):
+    low: list[float]
+    cost: list[float]
+    high: list[float]
+    open: list[float]
+    close: list[float]
+    ticks: list[int]
+    status: str
+    volume: list[float]
+
+
+class Trade(Struct):
+    iv: float
+    price: float
+    amount: float
+    trade_id: str
+    contracts: float
+    direction: str
+    trade_seq: int
+    timestamp: int
+    mark_price: float
+    index_price: float
+    tick_direction: int
+    instrument_name: str
+    combo_id: Optional[str] = '',
+    combo_trade_id: Optional[int] = 0,
+    block_trade_id: Optional[str] = '',
+    block_trade_leg_count: Optional[int] = 0,
+
+
+class LastTradesResult(Struct):
+    trades: list[Trade]
+    has_more: bool
--- a/piker/brokers/ib/api.py
+++ b/piker/brokers/ib/api.py
@ -1250,12 +1250,6 @@ async def load_aio_clients(

        for i in range(connect_retries):
            try:
-                log.info(
-                    'Trying `ib_async` connect\n'
-                    f'{host}: {port}\n'
-                    f'clientId: {client_id}\n'
-                    f'timeout: {connect_timeout}\n'
-                )
                await ib.connectAsync(
                    host,
                    port,
@ -1373,9 +1367,7 @@ async def load_clients_for_trio(
    a ``tractor.to_asyncio.open_channel_from()``.

    '''
-    async with load_aio_clients(
-        disconnect_on_exit=False,
-    ) as accts2clients:
+    async with load_aio_clients() as accts2clients:

        to_trio.send_nowait(accts2clients)

--- a/piker/brokers/kucoin.py
+++ b/piker/brokers/kucoin.py
@ -62,7 +62,7 @@ from piker._cacheables import (
 )
 from piker.log import get_logger
 from piker.data.validate import FeedInit
-from piker.types import Struct  # NOTE, this is already a `tractor.msg.Struct`
+from piker.types import Struct
 from piker.data import (
    def_iohlcv_fields,
    match_from_pairs,
@ -98,18 +98,9 @@ class KucoinMktPair(Struct, frozen=True):
    def size_tick(self) -> Decimal:
        return Decimal(str(self.quoteMinSize))

-    callauctionFirstStageStartTime: None|float
-    callauctionIsEnabled: bool
-    callauctionPriceCeiling: float|None
-    callauctionPriceFloor: float|None
-    callauctionSecondStageStartTime: float|None
-    callauctionThirdStageStartTime: float|None
-
    enableTrading: bool
-    feeCategory: int
    feeCurrency: str
    isMarginEnabled: bool
-    makerFeeCoefficient: float
    market: str
    minFunds: float
    name: str
@ -119,10 +110,11 @@ class KucoinMktPair(Struct, frozen=True):
    quoteIncrement: float
    quoteMaxSize: float
    quoteMinSize: float
-    st: bool
    symbol: str  # our bs_mktid, kucoin's internal id
+    feeCategory: int
+    makerFeeCoefficient: float
    takerFeeCoefficient: float
-    tradingStartTime: float|None
+    st: bool


 class AccountTrade(Struct, frozen=True):
@ -404,13 +396,7 @@ class Client:
        pairs: dict[str, KucoinMktPair] = {}
        fqmes2mktids: bidict[str, str] = bidict()
        for item in entries:
-            try:
-                pair = pairs[item['name']] = KucoinMktPair(**item)
-            except TypeError as te:
-                raise TypeError(
-                    '`KucoinMktPair` and reponse fields do not match ??\n'
-                    f'{KucoinMktPair.fields_diff(item)}\n'
-                ) from te
+            pair = pairs[item['name']] = KucoinMktPair(**item)
            fqmes2mktids[
                item['name'].lower().replace('-', '')
            ] = pair.name
--- a/piker/clearing/README.rst
+++ b/piker/clearing/README.rst
@ -1,49 +0,0 @@
-piker.clearing
-______________
-trade execution-n-control subsys for both live and paper trading as
-well as algo-trading manual override/interaction across any backend
-broker and data provider.
-
-avail UIs
-*********
-
-order ctl
---------
-the `piker.clearing` subsys is exposed mainly though
-the `piker chart` GUI as a "chart trader" style UX and
-is automatically enabled whenever a chart is opened.
-
-.. ^TODO, more prose here!
-
-the "manual" order control features are exposed via the
-`piker.ui.order_mode` API and can pretty much always be
-used (at least) in simulated-trading mode, aka "paper"-mode, and
-the micro-manual is as follows:
-
-``order_mode`` (
-    edge triggered activation by any of the following keys,
-    ``mouse-click`` on y-level to submit at that price
-    ):
-
-    - ``f``/ ``ctl-f`` to stage buy
-    - ``d``/ ``ctl-d`` to stage sell
-    - ``a`` to stage alert
-
-
-``search_mode`` (
-    ``ctl-l`` or ``ctl-space`` to open,
-    ``ctl-c`` or ``ctl-space`` to close
-    ) :
-
-    - begin typing to have symbol search automatically lookup
-      symbols from all loaded backend (broker) providers
-    - arrow keys and mouse click to navigate selection
-    - vi-like ``ctl-[hjkl]`` for navigation
-
-
-position (pp) mgmt
------------------
-you can also configure your position allocation limits from the
-sidepane.
-
-.. ^TODO, explain and provide tut once more refined!
--- a/piker/clearing/_ems.py
+++ b/piker/clearing/_ems.py
@ -653,7 +653,11 @@ class Router(Struct):
            flume = feed.flumes[fqme]
            first_quote: dict = flume.first_quote
            book: DarkBook = self.get_dark_book(broker)
-            book.lasts[fqme]: float = float(first_quote['last'])
+
+            if not (last := first_quote.get('last')):
+                last: float = flume.rt_shm.array[-1]['close']
+
+            book.lasts[fqme]: float = float(last)

            async with self.maybe_open_brokerd_dialog(
                brokermod=brokermod,
@ -716,7 +720,7 @@ class Router(Struct):
            subs = self.subscribers[sub_key]

        sent_some: bool = False
-        for client_stream in subs:
+        for client_stream in subs.copy():
            try:
                await client_stream.send(msg)
                sent_some = True
@ -1010,10 +1014,14 @@ async def translate_and_relay_brokerd_events(
                status_msg.brokerd_msg = msg
                status_msg.src = msg.broker_details['name']

-                await router.client_broadcast(
-                    status_msg.req.symbol,
-                    status_msg,
-                )
+                if not status_msg.req:
+                    # likely some order change state?
+                    await tractor.pause()
+                else:
+                    await router.client_broadcast(
+                        status_msg.req.symbol,
+                        status_msg,
+                    )

                if status == 'closed':
                    log.info(f'Execution for {oid} is complete!')
--- a/piker/cli/init.py
+++ b/piker/cli/init.py
@ -335,7 +335,7 @@ def services(config, tl, ports):
                name='service_query',
                loglevel=config['loglevel'] if tl else None,
            ),
-            tractor.get_arbiter(
+            tractor.get_registry(
                host=host,
                port=ports[0]
            ) as portal
--- a/piker/config.py
+++ b/piker/config.py
@ -104,15 +104,14 @@ def get_app_dir(
    # `tractor`) with the testing dir and check for it whenever we
    # detect `pytest` is being used (which it isn't under normal
    # operation).
-    # if "pytest" in sys.modules:
-    #     import tractor
-    #     actor = tractor.current_actor(err_on_no_runtime=False)
-    #     if actor:  # runtime is up
-    #         rvs = tractor._state._runtime_vars
-    #         import pdbp; pdbp.set_trace()
-    #         testdirpath = Path(rvs['piker_vars']['piker_test_dir'])
-    #         assert testdirpath.exists(), 'piker test harness might be borked!?'
-    #         app_name = str(testdirpath)
+    if "pytest" in sys.modules:
+        import tractor
+        actor = tractor.current_actor(err_on_no_runtime=False)
+        if actor:  # runtime is up
+            rvs = tractor._state._runtime_vars
+            testdirpath = Path(rvs['piker_vars']['piker_test_dir'])
+            assert testdirpath.exists(), 'piker test harness might be borked!?'
+            app_name = str(testdirpath)

    if platform.system() == 'Windows':
        key = "APPDATA" if roaming else "LOCALAPPDATA"
--- a/piker/data/_sampling.py
+++ b/piker/data/_sampling.py
@ -25,6 +25,7 @@ from collections import (
    defaultdict,
 )
 from contextlib import asynccontextmanager as acm
+from functools import partial
 import time
 from typing import (
    Any,
@ -42,7 +43,7 @@ from tractor.trionics import (
    maybe_open_nursery,
 )
 import trio
-from trio_typing import TaskStatus
+from trio import TaskStatus

 from .ticktools import (
    frame_ticks,
@ -53,6 +54,9 @@ from ._util import (
    get_console_log,
 )
 from ..service import maybe_spawn_daemon
+from piker.log import (
+    mk_repr,
+)

 if TYPE_CHECKING:
    from ._sharedmem import (
@ -70,6 +74,7 @@ if TYPE_CHECKING:
 _default_delay_s: float = 1.0


+# TODO: use new `tractor.singleton_acm` API for this!
 class Sampler:
    '''
    Global sampling engine registry.
@ -79,9 +84,9 @@ class Sampler:

    This non-instantiated type is meant to be a singleton within
    a `samplerd` actor-service spawned once by the user wishing to
-    time-step-sample (real-time) quote feeds, see
-    ``.service.maybe_open_samplerd()`` and the below
-    ``register_with_sampler()``.
+    time-step-sample a (real-time) quote feeds, see
+    `.service.maybe_open_samplerd()` and the below
+    `register_with_sampler()`.

    '''
    service_nursery: None | trio.Nursery = None
@ -95,6 +100,12 @@ class Sampler:
    # history loading.
    incr_task_cs: trio.CancelScope | None = None

+    bcast_errors: tuple[Exception] = (
+        trio.BrokenResourceError,
+        trio.ClosedResourceError,
+        trio.EndOfChannel,
+    )
+
    # holds all the ``tractor.Context`` remote subscriptions for
    # a particular sample period increment event: all subscribers are
    # notified on a step.
@ -258,14 +269,15 @@ class Sampler:
        subs: set
        last_ts, subs = pair

-        task = trio.lowlevel.current_task()
-        log.debug(
-            f'SUBS {self.subscribers}\n'
-            f'PAIR {pair}\n'
-            f'TASK: {task}: {id(task)}\n'
-            f'broadcasting {period_s} -> {last_ts}\n'
-            # f'consumers: {subs}'
-        )
+        # NOTE, for debugging pub-sub issues
+        # task = trio.lowlevel.current_task()
+        # log.debug(
+        #     f'AlL-SUBS@{period_s!r}: {self.subscribers}\n'
+        #     f'PAIR: {pair}\n'
+        #     f'TASK: {task}: {id(task)}\n'
+        #     f'broadcasting {period_s} -> {last_ts}\n'
+        #     f'consumers: {subs}'
+        # )
        borked: set[MsgStream] = set()
        sent: set[MsgStream] = set()
        while True:
@ -282,12 +294,11 @@ class Sampler:
                        await stream.send(msg)
                        sent.add(stream)

-                    except (
-                        trio.BrokenResourceError,
-                        trio.ClosedResourceError
-                    ):
+                    except self.bcast_errors as err:
                        log.error(
-                            f'{stream._ctx.chan.uid} dropped connection'
+                            f'Connection dropped for IPC ctx\n'
+                            f'{stream._ctx}\n\n'
+                            f'Due to {type(err)}'
                        )
                        borked.add(stream)
                else:
@ -375,7 +386,10 @@ async def register_with_sampler(
                assert Sampler.ohlcv_shms

            # unblock caller
-            await ctx.started(set(Sampler.ohlcv_shms.keys()))
+            await ctx.started(
+                # XXX bc msgpack only allows one array type!
+                list(Sampler.ohlcv_shms.keys())
+            )

            if open_index_stream:
                try:
@ -419,7 +433,6 @@ async def register_with_sampler(


 async def spawn_samplerd(
-
    loglevel: str | None = None,
    **extra_tractor_kwargs

@ -429,7 +442,10 @@ async def spawn_samplerd(
    update and increment count write and stream broadcasting.

    '''
-    from piker.service import Services
+    from piker.service import (
+        get_service_mngr,
+        ServiceMngr,
+    )

    dname = 'samplerd'
    log.info(f'Spawning `{dname}`')
@ -437,26 +453,33 @@ async def spawn_samplerd(
    # singleton lock creation of ``samplerd`` since we only ever want
    # one daemon per ``pikerd`` proc tree.
    # TODO: make this built-into the service api?
-    async with Services.locks[dname + '_singleton']:
+    mngr: ServiceMngr = get_service_mngr()
+    already_started: bool = dname in mngr.service_tasks

-        if dname not in Services.service_tasks:
-
-            portal = await Services.actor_n.start_actor(
-                dname,
-                enable_modules=[
-                    'piker.data._sampling',
-                ],
-                loglevel=loglevel,
-                debug_mode=Services.debug_mode,  # set by pikerd flag
-                **extra_tractor_kwargs
-            )
-
-            await Services.start_service_task(
-                dname,
-                portal,
+    async with mngr._locks[dname + '_singleton']:
+        ctx: Context = await mngr.start_service(
+            daemon_name=dname,
+            ctx_ep=partial(
                register_with_sampler,
                period_s=1,
                sub_for_broadcasts=False,
+            ),
+            debug_mode=mngr.debug_mode,  # set by pikerd flag
+
+            # proxy-through to tractor
+            enable_modules=[
+                'piker.data._sampling',
+            ],
+            loglevel=loglevel,
+            **extra_tractor_kwargs
+        )
+        if not already_started:
+            assert (
+                ctx
+                and
+                ctx.portal
+                and
+                not ctx.cancel_called
            )
            return True

@ -561,8 +584,7 @@ async def open_sample_stream(


 async def sample_and_broadcast(
-
-    bus: _FeedsBus,  # noqa
+    bus: _FeedsBus,
    rt_shm: ShmArray,
    hist_shm: ShmArray,
    quote_stream: trio.abc.ReceiveChannel,
@ -582,11 +604,32 @@ async def sample_and_broadcast(

    overruns = Counter()

+    # NOTE, only used for debugging live-data-feed issues, though
+    # this should be resolved more correctly in the future using the
+    # new typed-msgspec feats of `tractor`!
+    #
+    # XXX, a multiline nested `dict` formatter (since rn quote-msgs
+    # are just that).
+    # pfmt: Callable[[str], str] = mk_repr()
+
    # iterate stream delivered by broker
    async for quotes in quote_stream:
-        # print(quotes)

-        # TODO: ``numba`` this!
+        # XXX WARNING XXX only enable for debugging bc ow can cost
+        # ALOT of perf with HF-feedz!!!
+        #
+        # log.info(
+        #     'Rx live quotes:\n'
+        #     f'{pfmt(quotes)}'
+        # )
+
+        # TODO,
+        # -[ ] `numba` or `cython`-nize this loop possibly?
+        #  |_alternatively could we do it in rust somehow by upacking
+        #    arrow msgs instead of using `msgspec`?
+        # -[ ] use `msgspec.Struct` support in new typed-msging from
+        #     `tractor` to ensure only allowed msgs are transmitted?
+        #
        for broker_symbol, quote in quotes.items():
            # TODO: in theory you can send the IPC msg *before* writing
            # to the sharedmem array to decrease latency, however, that
@ -659,6 +702,21 @@ async def sample_and_broadcast(
            sub_key: str = broker_symbol.lower()
            subs: set[Sub] = bus.get_subs(sub_key)

+            # TODO, figure out how to make this useful whilst
+            # incoporating feed "pausing" ..
+            #
+            # if not subs:
+            #     all_bs_fqmes: list[str] = list(
+            #         bus._subscribers.keys()
+            #     )
+            #     log.warning(
+            #         f'No subscribers for {brokername!r} live-quote ??\n'
+            #         f'broker_symbol: {broker_symbol}\n\n'
+
+            #         f'Maybe the backend-sys symbol does not match one of,\n'
+            #         f'{pfmt(all_bs_fqmes)}\n'
+            #     )
+
            # NOTE: by default the broker backend doesn't append
            # it's own "name" into the fqme schema (but maybe it
            # should?) so we have to manually generate the correct
@ -756,7 +814,6 @@ async def sample_and_broadcast(


 async def uniform_rate_send(
-
    rate: float,
    quote_stream: trio.abc.ReceiveChannel,
    stream: MsgStream,
@ -889,6 +946,7 @@ async def uniform_rate_send(
            # to consumers which crash or lose network connection.
            # I.e. we **DO NOT** want to crash and propagate up to
            # ``pikerd`` these kinds of errors!
+            trio.EndOfChannel,
            trio.ClosedResourceError,
            trio.BrokenResourceError,
            ConnectionResetError,
--- a/piker/data/_web_bs.py
+++ b/piker/data/_web_bs.py
@ -387,22 +387,17 @@ async def open_jsonrpc_session(
    # broken and never restored with wtv init sequence is required to
    # re-establish a working req-resp session.

+    # request_type: Optional[type] = None,
+    # request_hook: Optional[Callable] = None,
+    # error_hook: Optional[Callable] = None,
 ) -> Callable[[str, dict], dict]:
-    '''
-    Init a json-RPC-over-websocket connection to the provided `url`.

-    A `json_rpc: Callable[[str, dict], dict` is delivered to the
-    caller for sending requests and a bg-`trio.Task` handles
-    processing of response msgs including error reporting/raising in
-    the parent/caller task.
-
-    '''
    # NOTE, store all request msgs so we can raise errors on the
    # caller side!
    req_msgs: dict[int, dict] = {}

    async with (
-        trio.open_nursery() as tn,
+        trio.open_nursery() as n,
        open_autorecon_ws(
            url=url,
            msg_recv_timeout=msg_recv_timeout,
@ -491,15 +486,20 @@ async def open_jsonrpc_session(
                        'params': _,
                    }:
                        log.debug(f'Recieved\n{msg}')
+                        # if request_hook:
+                        #     await request_hook(request_type(**msg))

                    case {
                        'error': error
                    }:
+                        # if error_hook:
+                        #     await error_hook(response_type(**msg))
+
                        # retreive orig request msg, set error
                        # response in original "result" msg,
                        # THEN FINALLY set the event to signal caller
                        # to raise the error in the parent task.
-                        req_id: int = error['id']
+                        req_id: int = msg['id']
                        req_msg: dict = req_msgs[req_id]
                        result: dict = rpc_results[req_id]
                        result['error'] = error
@ -513,6 +513,6 @@ async def open_jsonrpc_session(
                    case _:
                        log.warning(f'Unhandled JSON-RPC msg!?\n{msg}')

-        tn.start_soon(recv_task)
+        n.start_soon(recv_task)
        yield json_rpc
-        tn.cancel_scope.cancel()
+        n.cancel_scope.cancel()
--- a/piker/data/feed.py
+++ b/piker/data/feed.py
@ -540,7 +540,10 @@ async def open_feed_bus(
        # subscription since the backend isn't (yet) expected to
        # append it's own name to the fqme, so we filter on keys
        # which *do not* include that name (e.g .ib) .
-        bus._subscribers.setdefault(bs_fqme, set())
+        bus._subscribers.setdefault(
+            bs_fqme,
+            set(),
+        )

    # sync feed subscribers with flume handles
    await ctx.started(
--- a/piker/log.py
+++ b/piker/log.py
@ -18,7 +18,11 @@
 Log like a forester!
 """
 import logging
+import reprlib
 import json
+from typing import (
+    Callable,
+)

 import tractor
 from pygments import (
@ -84,3 +88,29 @@ def colorize_json(
        # likeable styles: algol_nu, tango, monokai
        formatters.TerminalTrueColorFormatter(style=style)
    )
+
+
+# TODO, eventually defer to the version in `modden` once
+# it becomes a dep!
+def mk_repr(
+    **repr_kws,
+) -> Callable[[str], str]:
+    '''
+    Allocate and deliver a `repr.Repr` instance with provided input
+    settings using the std-lib's `reprlib` mod,
+     * https://docs.python.org/3/library/reprlib.html
+
+    ------ Ex. ------
+    An up to 6-layer-nested `dict` as multi-line:
+    - https://stackoverflow.com/a/79102479
+    - https://docs.python.org/3/library/reprlib.html#reprlib.Repr.maxlevel
+
+    '''
+    def_kws: dict[str, int] = dict(
+        indent=2,
+        maxlevel=6,  # recursion levels
+        maxstring=66,  # match editor line-len limit
+    )
+    def_kws |= repr_kws
+    reprr = reprlib.Repr(**def_kws)
+    return reprr.repr
--- a/piker/service/init.py
+++ b/piker/service/init.py
@ -30,7 +30,11 @@ Actor runtime primtives and (distributed) service APIs for,
  => TODO: maybe to (re)move elsewhere?

 '''
-from ._mngr import Services as Services
+from ._mngr import (
+    get_service_mngr as get_service_mngr,
+    open_service_mngr as open_service_mngr,
+    ServiceMngr as ServiceMngr,
+)
 from ._registry import (
    _tractor_kwargs as _tractor_kwargs,
    _default_reg_addr as _default_reg_addr,
--- a/piker/service/_actor_runtime.py
+++ b/piker/service/_actor_runtime.py
@ -21,7 +21,6 @@
 from __future__ import annotations
 import os
 from typing import (
-    Optional,
    Any,
    ClassVar,
 )
@ -30,13 +29,13 @@ from contextlib import (
 )

 import tractor
-import trio

 from ._util import (
    get_console_log,
 )
 from ._mngr import (
-    Services,
+    open_service_mngr,
+    ServiceMngr,
 )
 from ._registry import (  # noqa
    _tractor_kwargs,
@ -59,7 +58,7 @@ async def open_piker_runtime(
    registry_addrs: list[tuple[str, int]] = [],

    enable_modules: list[str] = [],
-    loglevel: Optional[str] = None,
+    loglevel: str|None = None,

    # XXX NOTE XXX: you should pretty much never want debug mode
    # for data daemons when running in production.
@ -69,7 +68,7 @@ async def open_piker_runtime(
    # and spawn the service tree distributed per that.
    start_method: str = 'trio',

-    tractor_runtime_overrides: dict | None = None,
+    tractor_runtime_overrides: dict|None = None,
    **tractor_kwargs,

 ) -> tuple[
@ -119,6 +118,10 @@ async def open_piker_runtime(
                # spawn other specialized daemons I think?
                enable_modules=enable_modules,

+                # TODO: how to configure this?
+                # keep it on by default if debug mode is set?
+                # maybe_enable_greenback=debug_mode,
+
                **tractor_kwargs,
            ) as actor,

@ -167,12 +170,13 @@ async def open_pikerd(

    **kwargs,

-) -> Services:
+) -> ServiceMngr:
    '''
-    Start a root piker daemon with an indefinite lifetime.
+    Start a root piker daemon actor (aka `pikerd`) with an indefinite
+    lifetime.

-    A root actor nursery is created which can be used to create and keep
-    alive underling services (see below).
+    A root actor-nursery is created which can be used to spawn and
+    supervise underling service sub-actors (see below).

    '''
    # NOTE: for the root daemon we always enable the root
@ -199,8 +203,6 @@ async def open_pikerd(
            root_actor,
            reg_addrs,
        ),
-        tractor.open_nursery() as actor_nursery,
-        trio.open_nursery() as service_nursery,
    ):
        for addr in reg_addrs:
            if addr not in root_actor.accept_addrs:
@ -209,25 +211,17 @@ async def open_pikerd(
                    'Maybe you have another daemon already running?'
                )

-        # assign globally for future daemon/task creation
-        Services.actor_n = actor_nursery
-        Services.service_n = service_nursery
-        Services.debug_mode = debug_mode
-
-        try:
-            yield Services
-
-        finally:
-            # TODO: is this more clever/efficient?
-            # if 'samplerd' in Services.service_tasks:
-            #     await Services.cancel_service('samplerd')
-            service_nursery.cancel_scope.cancel()
+        mngr: ServiceMngr
+        async with open_service_mngr(
+            debug_mode=debug_mode,
+        ) as mngr:
+            yield mngr


 # TODO: do we even need this?
 # @acm
 # async def maybe_open_runtime(
-#     loglevel: Optional[str] = None,
+#     loglevel: str|None = None,
 #     **kwargs,

 # ) -> None:
@ -256,7 +250,7 @@ async def maybe_open_pikerd(
    loglevel: str | None = None,
    **kwargs,

-) -> tractor._portal.Portal | ClassVar[Services]:
+) -> tractor._portal.Portal | ClassVar[ServiceMngr]:
    '''
    If no ``pikerd`` daemon-root-actor can be found start it and
    yield up (we should probably figure out returning a portal to self
--- a/piker/service/_ahab.py
+++ b/piker/service/_ahab.py
@ -49,7 +49,7 @@ from requests.exceptions import (
    ReadTimeout,
 )

-from ._mngr import Services
+from ._mngr import ServiceMngr
 from ._util import (
    log,  # sub-sys logger
    get_console_log,
@ -453,7 +453,7 @@ async def open_ahabd(

@acm
 async def start_ahab_service(
-    services: Services,
+    services: ServiceMngr,
    service_name: str,

    # endpoint config passed as **kwargs
@ -549,7 +549,8 @@ async def start_ahab_service(
        log.warning('Failed to cancel root permsed container')

    except (
-        trio.MultiError,
+        # trio.MultiError,
+        ExceptionGroup,
    ) as err:
        for subexc in err.exceptions:
            if isinstance(subexc, PermissionError):
--- a/piker/service/_daemon.py
+++ b/piker/service/_daemon.py
@ -26,14 +26,17 @@ from typing import (
 from contextlib import (
    asynccontextmanager as acm,
 )
+from collections import defaultdict

 import tractor
+import trio

 from ._util import (
    log,  # sub-sys logger
 )
 from ._mngr import (
-    Services,
+    get_service_mngr,
+    ServiceMngr,
 )
 from ._actor_runtime import maybe_open_pikerd
 from ._registry import find_service
@ -41,15 +44,14 @@ from ._registry import find_service

@acm
 async def maybe_spawn_daemon(
-
    service_name: str,
    service_task_target: Callable,
-
    spawn_args: dict[str, Any],

    loglevel: str | None = None,
    singleton: bool = False,

+    _locks = defaultdict(trio.Lock),
    **pikerd_kwargs,

 ) -> tractor.Portal:
@ -67,7 +69,7 @@ async def maybe_spawn_daemon(
    '''
    # serialize access to this section to avoid
    # 2 or more tasks racing to create a daemon
-    lock = Services.locks[service_name]
+    lock = _locks[service_name]
    await lock.acquire()

    async with find_service(
@ -132,7 +134,65 @@ async def maybe_spawn_daemon(
        async with tractor.wait_for_actor(service_name) as portal:
            lock.release()
            yield portal
-            await portal.cancel_actor()
+            # --- ---- ---
+            # XXX NOTE XXX
+            # --- ---- ---
+            # DO NOT PUT A `portal.cancel_actor()` here (as was prior)!
+            #
+            # Doing so will cause an "out-of-band" ctxc
+            # (`tractor.ContextCancelled`) to be raised inside the
+            # `ServiceMngr.open_context_in_task()`'s call to
+            # `ctx.wait_for_result()` AND the internal self-ctxc
+            # "graceful capture" WILL NOT CATCH IT!
+            #
+            # This can cause certain types of operations to raise
+            # that ctxc BEFORE THEY `return`, resulting in
+            # a "false-negative" ctxc being raised when really
+            # nothing actually failed, other then our semantic
+            # "failure" to suppress an expected, graceful,
+            # self-cancel scenario..
+            #
+            # bUt wHy duZ It WorK lIKe dis..
+            # ------------------------------
+            # from the perspective of the `tractor.Context` this
+            # cancel request was conducted "out of band" since
+            # `Context.cancel()` was never called and thus the
+            # `._cancel_called: bool` was never set. Despite the
+            # remote `.canceller` being set to `pikerd` (i.e. the
+            # same `Actor.uid` of the raising service-mngr task) the
+            # service-task's ctx itself was never marked as having
+            # requested cancellation and thus still raises the ctxc
+            # bc it was unaware of any such request.
+            #
+            # How to make grokin these cases easier tho?
+            # ------------------------------------------
+            # Because `Portal.cancel_actor()` was called it requests
+            # "full-`Actor`-runtime-cancellation" of it's peer
+            # process which IS NOT THE SAME as a single inter-actor
+            # RPC task cancelling its local context with a remote
+            # peer `Task` in that same peer process.
+            #
+            # ?TODO? It might be better if we do one (or all) of the
+            # following:
+            #
+            # -[ ] at least set a special message for the
+            #    `ContextCancelled` when raised locally by the
+            #    unaware ctx task such that we check for the
+            #    `.canceller` being *our `Actor`* and in the case
+            #    where `Context._cancel_called == False` we specially
+            #    note that this is likely an "out-of-band"
+            #    runtime-cancel request triggered by some call to
+            #    `Portal.cancel_actor()`, possibly even reporting the
+            #    exact LOC of that caller by tracking it inside our
+            #    portal-type?
+            # -[ ] possibly add another field `ContextCancelled` like
+            #    maybe a,
+            #    `.request_type: Literal['os', 'proc', 'actor',
+            #    'ctx']` type thing which would allow immediately
+            #    being able to tell what kind of cancellation caused
+            #    the unexpected ctxc?
+            # -[ ] REMOVE THIS COMMENT, once we've settled on how to
+            #     better augment `tractor` to be more explicit on this!


 async def spawn_emsd(
@ -147,21 +207,22 @@ async def spawn_emsd(
    """
    log.info('Spawning emsd')

-    portal = await Services.actor_n.start_actor(
+    smngr: ServiceMngr = get_service_mngr()
+    portal = await smngr.actor_n.start_actor(
        'emsd',
        enable_modules=[
            'piker.clearing._ems',
            'piker.clearing._client',
        ],
        loglevel=loglevel,
-        debug_mode=Services.debug_mode,  # set by pikerd flag
+        debug_mode=smngr.debug_mode,  # set by pikerd flag
        **extra_tractor_kwargs
    )

    # non-blocking setup of clearing service
    from ..clearing._ems import _setup_persistent_emsd

-    await Services.start_service_task(
+    await smngr.start_service_task(
        'emsd',
        portal,

--- a/piker/service/_mngr.py
+++ b/piker/service/_mngr.py
@ -18,16 +18,29 @@
 daemon-service management API.

 """
+from __future__ import annotations
+from contextlib import (
+    asynccontextmanager as acm,
+    # contextmanager as cm,
+)
 from collections import defaultdict
+from dataclasses import (
+    dataclass,
+    field,
+)
+import functools
+import inspect
 from typing import (
    Callable,
    Any,
 )

-import trio
-from trio_typing import TaskStatus
+import msgspec
 import tractor
+import trio
+from trio import TaskStatus
 from tractor import (
+    ActorNursery,
    current_actor,
    ContextCancelled,
    Context,
@ -39,6 +52,130 @@ from ._util import (
 )


+# TODO: implement a singleton deco-API for wrapping the below
+# factory's impl for general actor-singleton use?
+#
+# @singleton
+# async def open_service_mngr(
+#     **init_kwargs,
+# ) -> ServiceMngr:
+#     '''
+#     Note this function body is invoke IFF no existing singleton instance already
+#     exists in this proc's memory.
+
+#     '''
+#     # setup
+#     yield ServiceMngr(**init_kwargs)
+#     # teardown
+
+
+
+# TODO: singleton factory API instead of a class API
+@acm
+async def open_service_mngr(
+    *,
+    debug_mode: bool = False,
+
+    # impl deat which ensures a single global instance
+    _singleton: list[ServiceMngr|None] = [None],
+    **init_kwargs,
+
+) -> ServiceMngr:
+    '''
+    Open a multi-subactor-as-service-daemon tree supervisor.
+
+    The delivered `ServiceMngr` is a singleton instance for each
+    actor-process and is allocated on first open and never
+    de-allocated unless explicitly deleted by al call to
+    `del_service_mngr()`.
+
+    '''
+    # TODO: factor this an allocation into
+    # a `._mngr.open_service_mngr()` and put in the
+    # once-n-only-once setup/`.__aenter__()` part!
+    # -[ ] how to make this only happen on the `mngr == None` case?
+    #  |_ use `.trionics.maybe_open_context()` (for generic
+    #     async-with-style-only-once of the factory impl, though
+    #     what do we do for the allocation case?
+    #    / `.maybe_open_nursery()` (since for this specific case
+    #    it's simpler?) to activate
+    async with (
+        tractor.open_nursery() as an,
+        trio.open_nursery() as tn,
+    ):
+        # impl specific obvi..
+        init_kwargs.update({
+            'actor_n': an,
+            'service_n': tn,
+        })
+
+        mngr: ServiceMngr|None
+        if (mngr := _singleton[0]) is None:
+
+            log.info('Allocating a new service mngr!')
+            mngr = _singleton[0] = ServiceMngr(**init_kwargs)
+
+            # TODO: put into `.__aenter__()` section of
+            # eventual `@singleton_acm` API wrapper.
+            #
+            # assign globally for future daemon/task creation
+            mngr.actor_n = an
+            mngr.service_n = tn
+
+        else:
+            assert (
+                mngr.actor_n
+                and
+                mngr.service_tn
+            )
+            log.info(
+                'Using extant service mngr!\n\n'
+                f'{mngr!r}\n'  # it has a nice `.__repr__()` of services state
+            )
+
+        try:
+            # NOTE: this is a singleton factory impl specific detail
+            # which should be supported in the condensed
+            # `@singleton_acm` API?
+            mngr.debug_mode = debug_mode
+
+            yield mngr
+        finally:
+            # TODO: is this more clever/efficient?
+            # if 'samplerd' in mngr.service_tasks:
+            #     await mngr.cancel_service('samplerd')
+            tn.cancel_scope.cancel()
+
+
+
+def get_service_mngr() -> ServiceMngr:
+    '''
+    Try to get the singleton service-mngr for this actor presuming it
+    has already been allocated using,
+
+    .. code:: python
+
+        async with open_<@singleton_acm(func)>() as mngr`
+            ... this block kept open ...
+
+    If not yet allocated raise a `ServiceError`.
+
+    '''
+    # https://stackoverflow.com/a/12627202
+    # https://docs.python.org/3/library/inspect.html#inspect.Signature
+    maybe_mngr: ServiceMngr|None = inspect.signature(
+        open_service_mngr
+    ).parameters['_singleton'].default[0]
+
+    if maybe_mngr is None:
+        raise RuntimeError(
+            'Someone must allocate a `ServiceMngr` using\n\n'
+            '`async with open_service_mngr()` beforehand!!\n'
+        )
+
+    return maybe_mngr
+
+
 # TODO: we need remote wrapping and a general soln:
 # - factor this into a ``tractor.highlevel`` extension # pack for the
 #   library.
@ -46,31 +183,46 @@ from ._util import (
 #   to the pikerd actor for starting services remotely!
 # - prolly rename this to ActorServicesNursery since it spawns
 #   new actors and supervises them to completion?
-class Services:
+@dataclass
+class ServiceMngr:
+# class ServiceMngr(msgspec.Struct):
+    '''
+    A multi-subactor-as-service manager.

-    actor_n: tractor._supervise.ActorNursery
+    Spawn, supervise and monitor service/daemon subactors in a SC
+    process tree.
+
+    '''
+    actor_n: ActorNursery
    service_n: trio.Nursery
-    debug_mode: bool  # tractor sub-actor debug mode flag
+    debug_mode: bool = False # tractor sub-actor debug mode flag
+
    service_tasks: dict[
        str,
        tuple[
            trio.CancelScope,
+            Context,
            Portal,
            trio.Event,
        ]
-    ] = {}
-    locks = defaultdict(trio.Lock)
+    ] = field(default_factory=dict)
+
+    # internal per-service task mutexs
+    _locks = defaultdict(trio.Lock)

-    @classmethod
    async def start_service_task(
        self,
        name: str,
        portal: Portal,
+
+        # TODO: typevar for the return type of the target and then
+        # use it below for `ctx_res`?
        target: Callable,
+
        allow_overruns: bool = False,
        **ctx_kwargs,

-    ) -> (trio.CancelScope, Context):
+    ) -> (trio.CancelScope, Context, Any):
        '''
        Open a context in a service sub-actor, add to a stack
        that gets unwound at ``pikerd`` teardown.
@ -83,6 +235,7 @@ class Services:
            task_status: TaskStatus[
                tuple[
                    trio.CancelScope,
+                    Context,
                    trio.Event,
                    Any,
                ]
@ -90,64 +243,87 @@ class Services:

        ) -> Any:

+            # TODO: use the ctx._scope directly here instead?
+            # -[ ] actually what semantics do we expect for this
+            #   usage!?
            with trio.CancelScope() as cs:
+                try:
+                    async with portal.open_context(
+                        target,
+                        allow_overruns=allow_overruns,
+                        **ctx_kwargs,

-                async with portal.open_context(
-                    target,
-                    allow_overruns=allow_overruns,
-                    **ctx_kwargs,
+                    ) as (ctx, started):

-                ) as (ctx, first):
-
-                    # unblock once the remote context has started
-                    complete = trio.Event()
-                    task_status.started((cs, complete, first))
-                    log.info(
-                        f'`pikerd` service {name} started with value {first}'
-                    )
-                    try:
+                        # unblock once the remote context has started
+                        complete = trio.Event()
+                        task_status.started((
+                            cs,
+                            ctx,
+                            complete,
+                            started,
+                        ))
+                        log.info(
+                            f'`pikerd` service {name} started with value {started}'
+                        )
                        # wait on any context's return value
                        # and any final portal result from the
                        # sub-actor.
-                        ctx_res: Any = await ctx.result()
+                        ctx_res: Any = await ctx.wait_for_result()

                        # NOTE: blocks indefinitely until cancelled
                        # either by error from the target context
                        # function or by being cancelled here by the
                        # surrounding cancel scope.
-                        return (await portal.result(), ctx_res)
-                    except ContextCancelled as ctxe:
-                        canceller: tuple[str, str] = ctxe.canceller
-                        our_uid: tuple[str, str] = current_actor().uid
-                        if (
-                            canceller != portal.channel.uid
-                            and
-                            canceller != our_uid
-                        ):
-                            log.cancel(
-                                f'Actor-service {name} was remotely cancelled?\n'
-                                f'remote canceller: {canceller}\n'
-                                f'Keeping {our_uid} alive, ignoring sub-actor cancel..\n'
-                            )
-                        else:
-                            raise
+                        return (
+                            await portal.wait_for_result(),
+                            ctx_res,
+                        )

+                except ContextCancelled as ctxe:
+                    canceller: tuple[str, str] = ctxe.canceller
+                    our_uid: tuple[str, str] = current_actor().uid
+                    if (
+                        canceller != portal.chan.uid
+                        and
+                        canceller != our_uid
+                    ):
+                        log.cancel(
+                            f'Actor-service `{name}` was remotely cancelled by a peer?\n'

+                            # TODO: this would be a good spot to use
+                            # a respawn feature Bo
+                            f'-> Keeping `pikerd` service manager alive despite this inter-peer cancel\n\n'

-                    finally:
-                        await portal.cancel_actor()
-                        complete.set()
-                        self.service_tasks.pop(name)
+                            f'cancellee: {portal.chan.uid}\n'
+                            f'canceller: {canceller}\n'
+                        )
+                    else:
+                        raise

-        cs, complete, first = await self.service_n.start(open_context_in_task)
+                finally:
+                    # NOTE: the ctx MUST be cancelled first if we
+                    # don't want the above `ctx.wait_for_result()` to
+                    # raise a self-ctxc. WHY, well since from the ctx's
+                    # perspective the cancel request will have
+                    # arrived out-out-of-band at the `Actor.cancel()`
+                    # level, thus `Context.cancel_called == False`,
+                    # meaning `ctx._is_self_cancelled() == False`.
+                    # with trio.CancelScope(shield=True):
+                    # await ctx.cancel()
+                    await portal.cancel_actor()
+                    complete.set()
+                    self.service_tasks.pop(name)
+
+        cs, sub_ctx, complete, started = await self.service_n.start(
+            open_context_in_task
+        )

        # store the cancel scope and portal for later cancellation or
        # retstart if needed.
-        self.service_tasks[name] = (cs, portal, complete)
+        self.service_tasks[name] = (cs, sub_ctx, portal, complete)
+        return cs, sub_ctx, started

-        return cs, first
-
-    @classmethod
    async def cancel_service(
        self,
        name: str,
@ -158,8 +334,80 @@ class Services:

        '''
        log.info(f'Cancelling `pikerd` service {name}')
-        cs, portal, complete = self.service_tasks[name]
-        cs.cancel()
+        cs, sub_ctx, portal, complete = self.service_tasks[name]
+
+        # cs.cancel()
+        await sub_ctx.cancel()
        await complete.wait()
-        assert name not in self.service_tasks, \
-            f'Serice task for {name} not terminated?'
+
+        if name in self.service_tasks:
+            # TODO: custom err?
+            # raise ServiceError(
+            raise RuntimeError(
+                f'Serice task for {name} not terminated?'
+            )
+
+        # assert name not in self.service_tasks, \
+        #     f'Serice task for {name} not terminated?'
+
+    async def start_service(
+        self,
+        daemon_name: str,
+        ctx_ep: Callable,  # kwargs must `partial`-ed in!
+
+        debug_mode: bool = False,
+        **tractor_actor_kwargs,
+
+    ) -> Context:
+        '''
+        Start a "service" task in a new sub-actor (daemon) and manage it's lifetime
+        indefinitely.
+
+        Services can be cancelled/shutdown using `.cancel_service()`.
+
+        '''
+        entry: tuple|None = self.service_tasks.get(daemon_name)
+        if entry:
+            (cs, sub_ctx, portal, complete) = entry
+            return sub_ctx
+
+        if daemon_name not in self.service_tasks:
+            portal = await self.actor_n.start_actor(
+                daemon_name,
+                debug_mode=(  # maybe set globally during allocate
+                    debug_mode
+                    or
+                    self.debug_mode
+                ),
+                **tractor_actor_kwargs,
+            )
+            ctx_kwargs: dict[str, Any] = {}
+            if isinstance(ctx_ep, functools.partial):
+                ctx_kwargs: dict[str, Any] = ctx_ep.keywords
+                ctx_ep: Callable = ctx_ep.func
+
+            (cs, sub_ctx, started) = await self.start_service_task(
+                daemon_name,
+                portal,
+                ctx_ep,
+                **ctx_kwargs,
+            )
+
+            return sub_ctx
+
+
+# TODO:
+# -[ ] factor all the common shit from `.data._sampling`
+#   and `.brokers._daemon` into here / `ServiceMngr`
+#   in terms of allocating the `Portal` as part of the
+#   "service-in-subactor" starting!
+# -[ ] move to `tractor.hilevel._service`, import and use here!
+# NOTE: purposely leaks the ref to the mod-scope Bo
+# import tractor
+# from tractor.hilevel import (
+#     open_service_mngr,
+#     ServiceMngr,
+# )
+# mngr: ServiceMngr|None = None
+# with tractor.hilevel.open_service_mngr() as mngr:
+#     Services = proxy(mngr)
--- a/piker/service/elastic.py
+++ b/piker/service/elastic.py
@ -21,11 +21,13 @@ from typing import (
    TYPE_CHECKING,
 )

+# TODO: oof, needs to be changed to `httpx`!
 import asks

 if TYPE_CHECKING:
    import docker
    from ._ahab import DockerContainer
+    from . import ServiceMngr

 from ._util import log  # sub-sys logger
 from ._util import (
@ -127,7 +129,7 @@ def start_elasticsearch(

@acm
 async def start_ahab_daemon(
-    service_mngr: Services,
+    service_mngr: ServiceMngr,
    user_config: dict | None = None,
    loglevel: str | None = None,

--- a/piker/service/marketstore.py
+++ b/piker/service/marketstore.py
@ -53,7 +53,7 @@ import pendulum
 # import purerpc

 from ..data.feed import maybe_open_feed
-from . import Services
+from . import ServiceMngr
 from ._util import (
    log,  # sub-sys logger
    get_console_log,
@ -233,7 +233,7 @@ def start_marketstore(

@acm
 async def start_ahab_daemon(
-    service_mngr: Services,
+    service_mngr: ServiceMngr,
    user_config: dict | None = None,
    loglevel: str | None = None,

--- a/piker/storage/cli.py
+++ b/piker/storage/cli.py
@ -386,8 +386,6 @@ def ldshm(
            open_annot_ctl() as actl,
        ):
            shm_df: pl.DataFrame | None = None
-            tf2aids: dict[float, dict] = {}
-
            for (
                shmfile,
                shm,
@ -528,17 +526,16 @@ def ldshm(
                            new_df,
                            step_gaps,
                        )
+
                        # last chance manual overwrites in REPL
-                        # await tractor.pause()
+                        await tractor.pause()
                        assert aids
-                        tf2aids[period_s] = aids

                else:
                    # allow interaction even when no ts problems.
-                    assert not diff
+                    await tractor.pause()
+                    # assert not diff

-            await tractor.pause()
-            log.info('Exiting TSP shm anal-izer!')

            if shm_df is None:
                log.error(
--- a/piker/storage/nativedb.py
+++ b/piker/storage/nativedb.py
@ -162,12 +162,11 @@ class NativeStorageClient:
    def index_files(self):
        for path in self._datadir.iterdir():
            if (
-                path.is_dir()
+                path.name in {'borked', 'expired',}
                or
                '.parquet' not in str(path)
-                # or
-                # path.name in {'borked', 'expired',}
            ):
+                # ignore all non-apache files (for now)
                continue

            key: str = path.name.rstrip('.parquet')
--- a/piker/tsp/init.py
+++ b/piker/tsp/init.py
@ -620,19 +620,16 @@ async def start_backfill(
                    f'{next_start_dt} -> {last_start_dt}'
                )

-                # NOTE, always drop the src asset token for
+                # always drop the src asset token for
                # non-currency-pair like market types (for now)
-                #
-                # THAT IS, for now our table key schema is NOT
-                # 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 {
                    'crypto',
                    'crypto_currency',
                    'fiat',  # a "forex pair"
                    'perpetual_future',  # stupid "perps" from cex land
                }:
+                    # for now, our table key schema is not including
+                    # the dst[/src] source asset token.
                    col_sym_key: str = mkt.get_fqme(
                        delim_char='',
                        without_src=True,
--- a/piker/tsp/_anal.py
+++ b/piker/tsp/_anal.py
@ -616,18 +616,6 @@ def detect_price_gaps(
    # ])
    ...

-# TODO: probably just use the null_segs impl above?
-def detect_vlm_gaps(
-    df: pl.DataFrame,
-    col: str = 'volume',
-
-) -> pl.DataFrame:
-
-    vnull: pl.DataFrame = w_dts.filter(
-        pl.col(col) == 0
-    )
-    return vnull
-

 def dedupe(
    src_df: pl.DataFrame,
@ -638,6 +626,7 @@ def dedupe(

 ) -> tuple[
    pl.DataFrame,  # with dts
+    pl.DataFrame,  # gaps
    pl.DataFrame,  # with deduplicated dts (aka gap/repeat removal)
    int,  # len diff between input and deduped
 ]:
@ -650,22 +639,19 @@ def dedupe(
    '''
    wdts: pl.DataFrame = with_dts(src_df)

-    deduped = wdts
-
-    # remove duplicated datetime samples/sections
-    deduped: pl.DataFrame = wdts.unique(
-        # subset=['dt'],
-        subset=['time'],
-        maintain_order=True,
-    )
-
    # maybe sort on any time field
    if sort:
-        deduped = deduped.sort(by='time')
+        wdts = wdts.sort(by='time')
        # TODO: detect out-of-order segments which were corrected!
        # -[ ] report in log msg
        # -[ ] possibly return segment sections which were moved?

+    # remove duplicated datetime samples/sections
+    deduped: pl.DataFrame = wdts.unique(
+        subset=['dt'],
+        maintain_order=True,
+    )
+
    diff: int = (
        wdts.height
        -
--- a/poetry.lock
+++ b/poetry.lock
--- a/pyproject.toml
+++ b/pyproject.toml
@ -15,119 +15,138 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 [build-system]
-requires = ["hatchling"]
-build-backend = "hatchling.build"
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

-[project]
+# ------ - ------
+
+[tool.ruff.lint]
+# https://docs.astral.sh/ruff/settings/#lint_ignore
+ignore = []
+
+# https://docs.astral.sh/ruff/settings/#lint_per-file-ignores
+# "piker/ui/qt.py" = [
+#   "E402",
+#   'F401',  # unused imports (without __all__ or blah as blah)
+#   # "F841", # unused variable rules
+# ]
+# ignore-init-module-imports = false
+
+# ------ - ------
+
+[tool.poetry]
 name = "piker"
-version = "0.1.0a0dev0"
+version = "0.1.0.alpha0.dev0"
 description = "trading gear for hackers"
-authors = [{ name = "Tyler Goodlet", email = "goodboy_foss@protonmail.com" }]
-requires-python = ">=3.12, <3.13"
-license = "AGPL-3.0-or-later"
+authors = ["Tyler Goodlet <goodboy_foss@protonmail.com>"]
+license = "AGPLv3"
 readme = "README.rst"
-keywords = [
-    "async",
-    "trading",
-    "finance",
-    "quant",
-    "charting",
-]
-classifiers = [
-    "Development Status :: 3 - Alpha",
-    "License :: OSI Approved :: GNU Affero General Public License v3 or later (AGPLv3+)",
-    "Operating System :: POSIX :: Linux",
-    "Programming Language :: Python :: Implementation :: CPython",
-    "Programming Language :: Python :: 3 :: Only",
-    "Programming Language :: Python :: 3.11",
-    "Programming Language :: Python :: 3.12",
-    "Intended Audience :: Financial and Insurance Industry",
-    "Intended Audience :: Science/Research",
-    "Intended Audience :: Developers",
-    "Intended Audience :: Education",
-]
-dependencies = [
-    "async-generator >=1.10, <2.0.0",
-    "attrs >=23.1.0, <24.0.0",
-    "bidict >=0.22.1, <0.23.0",
-    "colorama >=0.4.6, <0.5.0",
-    "colorlog >=6.7.0, <7.0.0",
-    "ib-insync >=0.9.86, <0.10.0",
-    "numba >=0.59.0, <0.60.0",
-    "numpy >=1.25, <2.0",
-    "polars >=0.18.13, <0.19.0",
-    "pygments >=2.16.1, <3.0.0",
-    "rich >=13.5.2, <14.0.0",
-    "tomli >=2.0.1, <3.0.0",
-    "tomli-w >=1.0.0, <2.0.0",
-    "trio-util >=0.7.0, <0.8.0",
-    "trio-websocket >=0.10.3, <0.11.0",
-    "typer >=0.9.0, <1.0.0",
-    "rapidfuzz >=3.5.2, <4.0.0",
-    "pdbp >=1.5.0, <2.0.0",
-    "trio >=0.24, <0.25",
-    "pendulum >=3.0.0, <4.0.0",
-    "httpx >=0.27.0, <0.28.0",
-    "cryptofeed >=2.4.0, <3.0.0",
-    "pyarrow >=17.0.0, <18.0.0",
-    "websockets ==12.0",
-    "msgspec",
-    "tractor",
-    "asyncvnc",
-    "tomlkit",
-]

-[project.optional-dependencies]
-uis = [
-    # https://docs.astral.sh/uv/concepts/projects/dependencies/#optional-dependencies
-    # TODO: make sure the levenshtein shit compiles on nix..
-    # rapidfuzz = {extras = ["speedup"], version = "^0.18.0"}
-    "rapidfuzz >=3.2.0, <4.0.0",
-    "qdarkstyle >=3.0.2, <4.0.0",
-    "pyqt6 >=6.7.0, <7.0.0",
-    "pyqtgraph",
+# ------ - ------

-    # for consideration,
-    # - 'visidata'
+[tool.poetry.dependencies]
+async-generator = "^1.10"
+attrs = "^23.1.0"
+bidict = "^0.22.1"
+colorama = "^0.4.6"
+colorlog = "^6.7.0"
+ib-insync = "^0.9.86"
+msgspec = "^0.18.6"
+numba = "^0.59.0"
+numpy = "^1.25"
+polars = "^0.18.13"
+pygments = "^2.16.1"
+python = ">=3.11, <3.13"
+rich = "^13.5.2"
+# setuptools = "^68.0.0"
+tomli = "^2.0.1"
+tomli-w = "^1.0.0"
+trio-util = "^0.7.0"
+trio-websocket = "^0.10.3"
+typer = "^0.9.0"
+rapidfuzz = "^3.5.2"
+pdbp = "^1.5.0"
+trio = "^0.24"
+pendulum = "^3.0.0"
+httpx = "^0.27.0"
+cryptofeed = "^2.4.0"
+pyarrow = "^17.0.0"

-    # TODO: add an `--only daemon` group for running non-ui / pikerd
-    # service tree in distributed mode B)
-    # https://docs.astral.sh/uv/concepts/projects/dependencies/#optional-dependencies
-]
+tractor = {path = "../tractor", develop = true}
+websockets = "12.0"
+[tool.poetry.dependencies.asyncvnc]
+git = 'https://github.com/pikers/asyncvnc.git'
+branch = 'main'

-[dependency-groups]
-# TODO: a toolset that makes debugging a `pikerd` service (tree) easy
-# to hack on directly using more or less the local env:
+[tool.poetry.dependencies.tomlkit]
+develop = true
+git = 'https://github.com/pikers/tomlkit.git'
+branch = 'piker_pin'
+# path = "../tomlkit/"
+
+[tool.poetry.group.uis]
+optional = true
+[tool.poetry.group.uis.dependencies]
+# https://python-poetry.org/docs/managing-dependencies/#dependency-groups
+# TODO: make sure the levenshtein shit compiles on nix..
+# rapidfuzz = {extras = ["speedup"], version = "^0.18.0"}
+rapidfuzz = "^3.2.0"
+qdarkstyle = ">=3.0.2"
+pyqtgraph = { git = 'https://github.com/pikers/pyqtgraph.git' }
+
+# ------ - ------
+pyqt6 = "^6.7.0"
+
+[tool.poetry.group.dev]
+optional = true
+[tool.poetry.group.dev.dependencies]
+# testing / CI
+pytest = "^6.0.0"
+elasticsearch = "^8.9.0"
+xonsh = "^0.14.2"
+prompt-toolkit = "3.0.40"
+cython = "^3.0.0"
+greenback = "^1.1.1"
+
+# console ehancements and eventually remote debugging
+# extras/helpers.
+# TODO: add a toolset that makes debugging a `pikerd` service
+# (tree) easy to hack on directly using more or less the local env:
 # - xonsh + xxh
 # - rsyscall + pdbp
 # - actor runtime control console like BEAM/OTP
-#
-# console ehancements and eventually remote debugging extras/helpers.
-# use `uv --dev` to enable
-dev = [
-    "pytest >=6.0.0, <7.0.0",
-    "elasticsearch >=8.9.0, <9.0.0",
-    "xonsh >=0.14.2, <0.15.0",
-    "prompt-toolkit ==3.0.40",
-    "cython >=3.0.0, <4.0.0",
-    "greenback >=1.1.1, <2.0.0",
-    "ruff>=0.9.6",
+
+# ------ - ------
+
+# TODO: add an `--only daemon` group for running non-ui / pikerd
+# service tree in distributed mode B)
+# https://python-poetry.org/docs/managing-dependencies/#installing-group-dependencies
+# [tool.poetry.group.daemon.dependencies]
+
+[tool.poetry.scripts]
+piker = 'piker.cli:cli'
+pikerd = 'piker.cli:pikerd'
+ledger = 'piker.accounting.cli:ledger'
+
+
+[project]
+keywords=[
+  "async",
+  "trading",
+  "finance",
+  "quant",
+  "charting",
+]
+classifiers=[
+  'Development Status :: 3 - Alpha',
+  "License :: OSI Approved :: GNU Affero General Public License v3 or later (AGPLv3+)",
+  'Operating System :: POSIX :: Linux',
+  "Programming Language :: Python :: Implementation :: CPython",
+  "Programming Language :: Python :: 3 :: Only",
+  "Programming Language :: Python :: 3.11",
+  "Programming Language :: Python :: 3.12",
+  'Intended Audience :: Financial and Insurance Industry',
+  'Intended Audience :: Science/Research',
+  'Intended Audience :: Developers',
+  'Intended Audience :: Education',
 ]
-
-[project.scripts]
-piker = "piker.cli:cli"
-pikerd = "piker.cli:pikerd"
-ledger = "piker.accounting.cli:ledger"
-
-[tool.hatch.build.targets.sdist]
-include = ["piker"]
-
-[tool.hatch.build.targets.wheel]
-include = ["piker"]
-
-[tool.uv.sources]
-pyqtgraph = { git = "https://github.com/pikers/pyqtgraph.git" }
-asyncvnc = { git = "https://github.com/pikers/asyncvnc.git", branch = "main" }
-tomlkit = { git = "https://github.com/pikers/tomlkit.git", branch ="piker_pin" }
-msgspec = { git = "https://github.com/jcrist/msgspec.git" }
-tractor = { path = "../tractor", editable = true }
--- a/ruff.toml
+++ b/ruff.toml
@ -1,93 +0,0 @@
-# from default `ruff.toml` @
-# https://docs.astral.sh/ruff/configuration/
-
-# Exclude a variety of commonly ignored directories.
-exclude = [
-    ".bzr",
-    ".direnv",
-    ".eggs",
-    ".git",
-    ".git-rewrite",
-    ".hg",
-    ".ipynb_checkpoints",
-    ".mypy_cache",
-    ".nox",
-    ".pants.d",
-    ".pyenv",
-    ".pytest_cache",
-    ".pytype",
-    ".ruff_cache",
-    ".svn",
-    ".tox",
-    ".venv",
-    ".vscode",
-    "__pypackages__",
-    "_build",
-    "buck-out",
-    "build",
-    "dist",
-    "node_modules",
-    "site-packages",
-    "venv",
-]
-
-# Same as Black.
-line-length = 88
-indent-width = 4
-
-# Assume Python 3.9
-target-version = "py312"
-
-# ------ - ------
-# TODO, stop warnings around `anext()` builtin use?
-# tool.ruff.target-version = "py310"
-
-
-[lint]
-# Enable Pyflakes (`F`) and a subset of the pycodestyle (`E`)  codes by default.
-# Unlike Flake8, Ruff doesn't enable pycodestyle warnings (`W`) or
-# McCabe complexity (`C901`) by default.
-select = ["E4", "E7", "E9", "F"]
-ignore = []
-ignore-init-module-imports = false
-
-[lint.per-file-ignores]
-"piker/ui/qt.py" = [
-  "E402",
-  'F401',  # unused imports (without __all__ or blah as blah)
-  # "F841", # unused variable rules
-]
-
-# Allow fix for all enabled rules (when `--fix`) is provided.
-fixable = ["ALL"]
-unfixable = []
-
-# Allow unused variables when underscore-prefixed.
-dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
-
-[format]
-# Use single quotes in `ruff format`.
-quote-style = "single"
-
-# Like Black, indent with spaces, rather than tabs.
-indent-style = "space"
-
-# Like Black, respect magic trailing commas.
-skip-magic-trailing-comma = false
-
-# Like Black, automatically detect the appropriate line ending.
-line-ending = "auto"
-
-# Enable auto-formatting of code examples in docstrings. Markdown,
-# reStructuredText code/literal blocks and doctests are all supported.
-#
-# This is currently disabled by default, but it is planned for this
-# to be opt-out in the future.
-docstring-code-format = false
-
-# Set the line length limit used when formatting code snippets in
-# docstrings.
-#
-# This only has an effect when the `docstring-code-format` setting is
-# enabled.
-docstring-code-line-length = "dynamic"
--- a/tests/conftest.py
+++ b/tests/conftest.py
@ -10,7 +10,7 @@ from piker import (
    config,
 )
 from piker.service import (
-    Services,
+    get_service_mngr,
 )
 from piker.log import get_console_log

@ -129,7 +129,7 @@ async def _open_test_pikerd(
        ) as service_manager,
    ):
        # this proc/actor is the pikerd
-        assert service_manager is Services
+        assert service_manager is get_service_mngr()

        async with tractor.wait_for_actor(
            'pikerd',
--- a/tests/test_ems.py
+++ b/tests/test_ems.py
@ -26,7 +26,7 @@ import pytest
 import tractor
 from uuid import uuid4

-from piker.service import Services
+from piker.service import ServiceMngr
 from piker.log import get_logger
 from piker.clearing._messages import (
    Order,
@ -158,7 +158,7 @@ def load_and_check_pos(


 def test_ems_err_on_bad_broker(
-    open_test_pikerd: Services,
+    open_test_pikerd: ServiceMngr,
    loglevel: str,
 ):
    async def load_bad_fqme():
--- a/tests/test_services.py
+++ b/tests/test_services.py
@ -15,7 +15,7 @@ import tractor

 from piker.service import (
    find_service,
-    Services,
+    ServiceMngr,
 )
 from piker.data import (
    open_feed,
@ -44,7 +44,7 @@ def test_runtime_boot(
    async def main():
        port = 6666
        daemon_addr = ('127.0.0.1', port)
-        services: Services
+        services: ServiceMngr

        async with (
            open_test_pikerd(
--- a/uv.lock
+++ b/uv.lock