Merge pull request 'jsonrpc_err_in_rent' (#41 ) from jsonrpc_err_in_rent into gitea_feats

Reviewed-on: #41
Merge pull request 'tsp_gaps: fixes for fault-less OHLCV time-series loads' (#35 ) from tsp_gaps into gitea_feats
2025-02-21 21:21:02 +00:00 · 2025-02-21 20:46:37 +00:00 · 2025-02-19 17:05:13 -05:00 · 2025-02-19 17:05:13 -05:00 · 2025-02-19 17:05:13 -05:00 · 2025-02-19 17:01:24 -05:00
37 changed files with 3228 additions and 4150 deletions
--- a/README.rst
+++ b/README.rst
@ -1,162 +1,161 @@
 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 for real-time
+``piker`` is a broker agnostic, next-gen FOSS toolset and runtime for
-computational trading targeted at `hardcore Linux users <comp_trader>`_ .
+real-time computational trading targeted at `hardcore Linux users
 <comp_trader>`_ .
-we use as much bleeding edge tech as possible including (but not limited to):
+we use much bleeding edge tech including (but not limited to):
 - latest python for glue_
- trio_ & tractor_ for our distributed, multi-core, real-time streaming
+- uv_ for packaging and distribution
-  `structured concurrency`_ runtime B)
+- trio_ & tractor_ for our distributed `structured concurrency`_ runtime
- Qt_ for pristine high performance UIs
+- Qt_ for pristine low latency UIs
- pyqtgraph_ for real-time charting
+- pyqtgraph_ (which we've extended) for real-time charting and graphics
- ``polars`` ``numpy`` and ``numba`` for `fast numerics`_
+- ``polars`` ``numpy`` and ``numba`` for redic `fast numerics`_
- `apache arrow and parquet`_ for time series history management
+- `apache arrow and parquet`_ for time-series storage
  persistence and sharing
 - (prototyped) techtonicdb_ for L2 book storage
-.. |travis| image:: https://img.shields.io/travis/pikers/piker/master.svg
+potential projects we might integrate with soon,
-    :target: https://travis-ci.org/pikers/piker
+
 - (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/
 .. _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/
-.. _comp_trader: https://jfaleiro.wordpress.com/2019/10/09/computational-trader/
+.. _techtonicdb: https://github.com/0b01/tectonicdb
-focus and features:
+focus and feats:
-*******************
+****************
- 100% federated: your code, your hardware, your data feeds, your broker fills.
+fitting with these tenets, we're always open to new
- zero web: low latency, native software that doesn't try to re-invent the OS
+framework/lib/service interop suggestions and ideas!
 - 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.
-fitting with these tenets, we're always open to new framework suggestions and ideas.
+- **100% federated**:
  your code, your hardware, your data feeds, your broker fills.
-building the best looking, most reliable, keyboard friendly trading
+- **zero web**:
-platform is the dream; join the cause.
+  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.
-sane install with `poetry`
+a sane install with `uv`
-**************************
+************************
-TODO!
+bc why install with `python` when you can faster with `rust` ::
-
+    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 develop.nix
+    nix-shell default.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
-install wild-west style via `pip`
+start a chart
-*********************************
+*************
-``piker`` is currently under heavy pre-alpha development and as such
+run a realtime OHLCV chart stand-alone::
 should be cloned from this repo and hacked on directly.
-for a development install::
+    piker -l info chart btcusdt.spot.binance xmrusdt.spot.kraken
-    git clone git@github.com:pikers/piker.git
+this runs a chart UI (with 1m sampled OHLCV) and shows 2 spot markets from 2 diff cexes
-    cd piker
+overlayed on the same graph. Use of `piker` without first starting
-    virtualenv env
+a daemon (`pikerd` - see below) means there is an implicit spawning of the
-    source ./env/bin/activate
+multi-actor-runtime (implemented as a `tractor` app).
-    pip install -r requirements.txt -e .
+
 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
-check out our charts
+spawn a daemon standalone
-********************
+*************************
-bet you weren't expecting this from the foss::
+we call the root actor-process the ``pikerd``. it can be (and is
-
+recommended normally to be) started separately from the ``piker
-    piker -l info -b kraken -b binance chart btcusdt.binance --pdb
+chart`` program::
 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,
-connect your chart::
+- 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.
-    piker -l info -b kraken -b binance chart xmrusdt.binance --pdb
+*using* an actor-service (aka micro-daemon) manager which dynamically
 supervises various sub-subsystems-as-services throughout the ``piker``
 runtime-stack.
 now you can (implicitly) connect your chart::
-enjoy persistent real-time data feeds tied to daemon lifetime. the next
+    piker chart btcusdt.spot.binance
-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
+since ``pikerd`` was started separately you can now enjoy a persistent
-kill ``pikerd``.
+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``.
 if anyone asks you what this project is about
 *********************************************
-you don't talk about it.
+you don't talk about it; just use it.
 how do i get involved?
@ -166,6 +165,15 @@ enter the matrix.
 how come there ain't that many docs
 ***********************************
-suck it up, learn the code; no one is trying to sell you on anything.
+i mean we want/need them but building the core right has been higher
-also, we need lotsa help so if you want to start somewhere and can't
+prio then marketting (and likely will stay that way Bp).
-necessarily write serious code, this might be the place for you!
+
 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!
--- a/default.nix
+++ b/default.nix
@ -1,82 +1,134 @@
 with (import <nixpkgs> {});
 with python312Packages;
 let
  glibStorePath = lib.getLib glib;
-  qtpyStorePath = lib.getLib qtpy;
+  zlibStorePath = lib.getLib zlib;
-  pyqt6StorePath = lib.getLib pyqt6;
+  zstdStorePath = lib.getLib zstd;
-  pyqt6SipStorePath = lib.getLib pyqt6-sip;
+  dbusStorePath = lib.getLib dbus;
  libGLStorePath = lib.getLib libGL;
  freetypeStorePath = lib.getLib freetype;
  qt6baseStorePath = lib.getLib qt6.qtbase;
-  rapidfuzzStorePath = lib.getLib rapidfuzz;
+  fontconfigStorePath = lib.getLib fontconfig;
-  qdarkstyleStorePath = lib.getLib qdarkstyle;
+  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;
 in
 stdenv.mkDerivation {
-  name = "piker-qt6-poetry-shell";
+  name = "piker-qt6-uv";
  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
-    poetry-core
+    python312Packages.uv
-    qdarkstyle
+    python312Packages.qdarkstyle
-    rapidfuzz
+    python312Packages.rapidfuzz
-    pyqt6
+    python312Packages.pyqt6
-    qtpy
+    python312Packages.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}"
+    QTBASE_PATH="${qt6baseStorePath}/lib"
-    echo "qtbase path:    $QTBASE_PATH"
+    QT_PLUGIN_PATH="$QTBASE_PATH/qt-6/plugins"
-    echo ""
+    QT_QPA_PLATFORM_PLUGIN_PATH="$QT_PLUGIN_PATH/platforms"
    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 ""
-    # Maybe create venv & install deps
+    LIB_GCC_PATH="${libgcc.lib}/lib"
-    poetry install --with uis
+    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"
-    # Use pyqt6 from System, patch activate script
+    XCB_UTIL_CURSOR_PATH="${xcbutilcursorStorePath}/lib"
-    ACTIVATE_SCRIPT_PATH="$(poetry env info --path)/bin/activate"
+    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"
-    export RPDFUZZ_PATH="${rapidfuzzStorePath}/lib/python3.12/site-packages"
+    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$QTBASE_PATH"
-    export QDRKSTYLE_PATH="${qdarkstyleStorePath}/lib/python3.12/site-packages"
+    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$QT_PLUGIN_PATH"
-    export QTPY_PATH="${qtpyStorePath}/lib/python3.12/site-packages"
+    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$QT_QPA_PLATFORM_PLUGIN_PATH"
    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 ""
-    PATCH="export PYTHONPATH=\""
+    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="$PATCH\$RPDFUZZ_PATH"
+    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$XCB_UTIL_CURSOR_PATH"
-    PATCH="$PATCH:\$QDRKSTYLE_PATH"
+    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$XORG_LIB_X11_PATH"
-    PATCH="$PATCH:\$QTPY_PATH"
+    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$XORG_LIB_XCB_PATH"
-    PATCH="$PATCH:\$PYQT6_PATH"
+    LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$XORG_XCB_UTIL_IMAGE_PATH"
-    PATCH="$PATCH:\$PYQT6_SIP_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\""
+    export LD_LIBRARY_PATH
-    if grep -q "$PATCH" "$ACTIVATE_SCRIPT_PATH"; then
+    RPDFUZZ_PATH="${rapidfuzzStorePath}/lib/python3.12/site-packages"
-        echo "venv is already patched."
+    QDRKSTYLE_PATH="${qdarkstyleStorePath}/lib/python3.12/site-packages"
-    else
+    QTPY_PATH="${qtpyStorePath}/lib/python3.12/site-packages"
-        echo "patching $ACTIVATE_SCRIPT_PATH to use pyqt6 from nixos..."
+    PYQT6_PATH="${pyqt6StorePath}/lib/python3.12/site-packages"
-        sed -i "\$i$PATCH" $ACTIVATE_SCRIPT_PATH
+    PYQT6_SIP_PATH="${pyqt6SipStorePath}/lib/python3.12/site-packages"
-    fi
+
    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
    poetry shell
  '';
 }
--- a/piker/brokers/init.py
+++ b/piker/brokers/init.py
@ -51,7 +51,6 @@ __brokers__: list[str] = [
    'ib',
    'kraken',
    'kucoin',
    'deribit',
    # broken but used to work
    # 'questrade',
@ -62,6 +61,7 @@ __brokers__: list[str] = [
    # wstrade
    # iex
    # deribit
    # bitso
 ]
--- a/piker/brokers/_daemon.py
+++ b/piker/brokers/_daemon.py
@ -23,7 +23,6 @@ from __future__ import annotations
 from contextlib import (
    asynccontextmanager as acm,
 )
 from functools import partial
 from types import ModuleType
 from typing import (
    TYPE_CHECKING,
@ -61,13 +60,12 @@ 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
+    Allocate a actor-wide service nursery in ``brokerd``
-    feeds can be run in the background persistently by the broker
+    such that feeds can be run in the background persistently by
-    backend as needed.
+    the broker backend as needed.
    '''
    # NOTE: we only need to setup logging once (and only) here
@ -88,18 +86,6 @@ 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
@ -159,21 +145,18 @@ def broker_init(
      above.
    '''
-    brokermod: ModuleType = get_brokermod(brokername)
+    from ..brokers import get_brokermod
    brokermod = get_brokermod(brokername)
    modpath: str = brokermod.__name__
-    spawn_kws: dict = getattr(
+
    start_actor_kwargs['name'] = f'brokerd.{brokername}'
    start_actor_kwargs.update(
        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
@ -183,7 +166,8 @@ def broker_init(
    # lookup actor-enabled modules declared by the backend offering the
    # `brokerd` endpoint(s).
-    enabled: list[str] = [
+    enabled: list[str]
    enabled = start_actor_kwargs['enable_modules'] = [
        __name__,  # so that eps from THIS mod can be invoked
        modpath,
    ]
@ -195,13 +179,9 @@ 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 | datad_kwargs | spawn_kws,  # to `ActorNursery.start_actor()`
+        start_actor_kwargs,  # to `ActorNursery.start_actor()`
        # XXX see impl above; contains all (actor global)
        # setup/teardown expected in all `brokerd` actor instances.
@ -210,17 +190,14 @@ 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')
@ -234,41 +211,33 @@ 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 (
+    from piker.service import Services
-        get_service_mngr,
+
        ServiceMngr,
    )
    dname: str = tractor_kwargs.pop('name')  # f'brokerd.{brokername}'
-    mngr: ServiceMngr = get_service_mngr()
+    portal = await Services.actor_n.start_actor(
-    ctx: tractor.Context = await mngr.start_service(
+        dname,
-        daemon_name=dname,
+        enable_modules=_data_mods + tractor_kwargs.pop('enable_modules'),
-        ctx_ep=partial(
+        debug_mode=Services.debug_mode,
            # 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
    )
-    assert (
+
-        not ctx.cancel_called
+    # NOTE: the service mngr expects an already spawned actor + its
-        and ctx.portal  # parent side
+    # portal ref in order to do non-blocking setup of brokerd
-        and dname in ctx.chan.uid  # subactor is named as desired
+    # service nursery.
    await Services.start_service_task(
        dname,
        portal,
        # signature of target root-task endpoint
        daemon_fixture_ep,
        brokername=brokername,
        loglevel=loglevel,
    )
    return True
@ -293,7 +262,8 @@ async def maybe_spawn_brokerd(
    from piker.service import maybe_spawn_daemon
    async with maybe_spawn_daemon(
-        service_name=f'brokerd.{brokername}',
+
        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,6 +110,7 @@ class AggTrade(Struct, frozen=True):
 async def stream_messages(
    ws: NoBsWs,
 ) -> AsyncGenerator[NoBsWs, dict]:
    # TODO: match syntax here!
@ -220,6 +221,8 @@ 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,
@ -462,6 +465,8 @@ 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
@ -510,7 +515,6 @@ 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,7 +25,6 @@ from .api import (
    get_client,
 )
 from .feed import (
    get_mkt_info,
    open_history_client,
    open_symbol_search,
    stream_quotes,
@ -35,20 +34,15 @@ 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,59 +18,38 @@
 Deribit backend.
 '''
 from __future__ import annotations
 from contextlib import asynccontextmanager as acm
 from datetime import datetime
-from typing import (
+from typing import Any, Optional, Callable
    # Any,
    # Optional,
    Callable,
 )
 # from pprint import pformat
 import time
 import cryptofeed
 import trio
 from trio_typing import TaskStatus
-from pendulum import (
+import pendulum
-    from_timestamp,
+from rapidfuzz import process as fuzzy
 )
 import numpy as np
 import tractor
-from piker.accounting import (
+from piker.brokers import open_cached_client
-    Asset,
+from piker.log import get_logger, get_console_log
-    MktPair,
+from piker.data import ShmArray
-    unpack_fqme,
+from piker.brokers._util import (
-)
+    BrokerError,
 from piker.brokers import (
    open_cached_client,
    NoData,
    DataUnavailable,
 )
 from piker._cacheables import (
    async_lifo_cache,
 )
 from piker.log import (
    get_logger,
    mk_repr,
 )
 from piker.data.validate import FeedInit
 from cryptofeed import FeedHandler
 from cryptofeed.defines import (
    DERIBIT, L1_BOOK, TRADES, OPTION, CALL, PUT
 )
 from cryptofeed.symbols import Symbol
 from .api import (
-    Client,
+    Client, Trade,
-    # get_config,
+    get_config,
-    piker_sym_to_cb_sym,
+    str_to_cb_sym, piker_sym_to_cb_sym, cb_sym_to_deribit_inst,
    cb_sym_to_deribit_inst,
    str_to_cb_sym,
    maybe_open_price_feed
 )
 from .venues import (
    Pair,
    OptionPair,
    Trade,
 )
 _spawn_kwargs = {
    'infect_asyncio': True,
@ -85,215 +64,90 @@ 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(
-            timeframe: float,
+            end_dt: Optional[datetime] = None,
-            end_dt: datetime | None = None,
+            start_dt: Optional[datetime] = None,
            start_dt: datetime | None = None,
        ) -> tuple[
            np.ndarray,
            datetime,  # start
            datetime,  # end
        ]:
            if timeframe != 60:
                raise DataUnavailable('Only 1m bars are supported')
-            array: np.ndarray = await client.bars(
+            array = await client.bars(
-                mkt,
+                instrument,
                start_dt=start_dt,
                end_dt=end_dt,
            )
            if len(array) == 0:
-                if (
+                raise DataUnavailable
                    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 = from_timestamp(array[0]['time'])
+            start_dt = pendulum.from_timestamp(array[0]['time'])
-            end_dt = from_timestamp(array[-1]['time'])
+            end_dt = pendulum.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 (
+        yield get_ohlc, {'erlangs': 3, 'rate': 3}
            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
-    Open a live quote stream for the market set defined by `symbols`.
+    get_console_log(loglevel or tractor.current_actor().loglevel)
-    Internally this starts a `cryptofeed.FeedHandler` inside an `asyncio`-side
+    sym = symbols[0]
    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)
-        # build out init msgs according to latest spec
+        init_msgs = {
-        init_msgs.append(
+            # pass back token, and bool, signalling if we're the writer
-            FeedInit(
+            # and that history has been written
-                mkt_info=mkt,
+            sym: {
-            )
+                'symbol_info': {
-        )
+                    'asset_type': 'option',
-        # build `cryptofeed` feed-handle
+                    'price_tick_size': 0.0005
-        cf_sym: cryptofeed.Symbol = piker_sym_to_cb_sym(sym)
+                },
                'shm_write_opts': {'sum_tick_vml': False},
                'fqsn': sym,
            },
        }
-        from_cf: tractor.to_asyncio.LinkedTaskChannel
+        nsym = piker_sym_to_cb_sym(sym)
        async with maybe_open_price_feed(sym) as from_cf:
-            # load the "last trades" summary
+        async with maybe_open_price_feed(sym) as stream:
            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
-            # TODO, do we even need this or will the above always
+            cache = await client.cache_symbols()
            # work?
            # if not last_trades:
            #     await tractor.pause()
            #     async for typ, quote in from_cf:
            #         if typ == 'trade':
            #             last_trade = Trade(**(quote['data']))
            #             break
-            # else:
+            last_trades = (await client.last_trades(
-            last_trade = Trade(
+                cb_sym_to_deribit_inst(nsym), count=1)).trades
                **(last_trades[0])
            )
-            first_quote: dict = {
+            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 = {
                'symbol': sym,
                'last': last_trade.price,
                'brokerd_ts': last_trade.timestamp,
@ -304,84 +158,13 @@ async def stream_quotes(
                    'broker_ts': last_trade.timestamp
                }]
            }
-            task_status.started((
+            task_status.started((init_msgs,  first_quote))
                init_msgs,
                first_quote,
            ))
            feed_is_live.set()
-            # NOTE XXX, static for now!
+            async for typ, quote in stream:
-            # => since this only handles ONE mkt feed at a time we
+                topic = quote['symbol']
-            # don't need a lookup table to map interleaved quotes
+                await send_chan.send({topic: quote})
            # 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
@ -391,21 +174,12 @@ async def open_symbol_search(
    async with open_cached_client('deribit') as client:
        # load all symbols locally for fast search
-        # cache = client._pairs
+        cache = await client.cache_symbols()
        await ctx.started()
        async with ctx.open_stream() as stream:
-            pattern: str
+
            async for pattern in stream:
-
+                # repack in dict form
-                # NOTE: pattern fuzzy-matching is done within
+                await stream.send(
-                # the methd impl.
+                    await client.search_symbols(pattern))
                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
@ -1,196 +0,0 @@
 # 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,6 +1250,12 @@ 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,
@ -1367,7 +1373,9 @@ async def load_clients_for_trio(
    a ``tractor.to_asyncio.open_channel_from()``.
    '''
-    async with load_aio_clients() as accts2clients:
+    async with load_aio_clients(
        disconnect_on_exit=False,
    ) 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
+from piker.types import Struct  # NOTE, this is already a `tractor.msg.Struct`
 from piker.data import (
    def_iohlcv_fields,
    match_from_pairs,
@ -98,9 +98,18 @@ 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
@ -110,11 +119,10 @@ class KucoinMktPair(Struct, frozen=True):
    quoteIncrement: float
    quoteMaxSize: float
    quoteMinSize: float
    symbol: str  # our bs_mktid, kucoin's internal id
    feeCategory: int
    makerFeeCoefficient: float
    takerFeeCoefficient: float
    st: bool
    symbol: str  # our bs_mktid, kucoin's internal id
    takerFeeCoefficient: float
    tradingStartTime: float|None
 class AccountTrade(Struct, frozen=True):
@ -396,7 +404,13 @@ 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
            fqmes2mktids[
                item['name'].lower().replace('-', '')
            ] = pair.name
--- a/piker/clearing/README.rst
+++ b/piker/clearing/README.rst
@ -0,0 +1,49 @@
 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,11 +653,7 @@ 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,
@ -720,7 +716,7 @@ class Router(Struct):
            subs = self.subscribers[sub_key]
        sent_some: bool = False
-        for client_stream in subs.copy():
+        for client_stream in subs:
            try:
                await client_stream.send(msg)
                sent_some = True
@ -1014,10 +1010,6 @@ async def translate_and_relay_brokerd_events(
                status_msg.brokerd_msg = msg
                status_msg.src = msg.broker_details['name']
                if not status_msg.req:
                    # likely some order change state?
                    await tractor.pause()
                else:
                await router.client_broadcast(
                    status_msg.req.symbol,
                    status_msg,
--- 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_registry(
+            tractor.get_arbiter(
                host=host,
                port=ports[0]
            ) as portal
--- a/piker/config.py
+++ b/piker/config.py
@ -104,14 +104,15 @@ 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:
+    # if "pytest" in sys.modules:
-        import tractor
+    #     import tractor
-        actor = tractor.current_actor(err_on_no_runtime=False)
+    #     actor = tractor.current_actor(err_on_no_runtime=False)
-        if actor:  # runtime is up
+    #     if actor:  # runtime is up
-            rvs = tractor._state._runtime_vars
+    #         rvs = tractor._state._runtime_vars
-            testdirpath = Path(rvs['piker_vars']['piker_test_dir'])
+    #         import pdbp; pdbp.set_trace()
-            assert testdirpath.exists(), 'piker test harness might be borked!?'
+    #         testdirpath = Path(rvs['piker_vars']['piker_test_dir'])
-            app_name = str(testdirpath)
+    #         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,7 +25,6 @@ from collections import (
    defaultdict,
 )
 from contextlib import asynccontextmanager as acm
 from functools import partial
 import time
 from typing import (
    Any,
@ -43,7 +42,7 @@ from tractor.trionics import (
    maybe_open_nursery,
 )
 import trio
-from trio import TaskStatus
+from trio_typing import TaskStatus
 from .ticktools import (
    frame_ticks,
@ -54,9 +53,6 @@ from ._util import (
    get_console_log,
 )
 from ..service import maybe_spawn_daemon
 from piker.log import (
    mk_repr,
 )
 if TYPE_CHECKING:
    from ._sharedmem import (
@ -74,7 +70,6 @@ if TYPE_CHECKING:
 _default_delay_s: float = 1.0
 # TODO: use new `tractor.singleton_acm` API for this!
 class Sampler:
    '''
    Global sampling engine registry.
@ -84,9 +79,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 a (real-time) quote feeds, see
+    time-step-sample (real-time) quote feeds, see
-    `.service.maybe_open_samplerd()` and the below
+    ``.service.maybe_open_samplerd()`` and the below
-    `register_with_sampler()`.
+    ``register_with_sampler()``.
    '''
    service_nursery: None | trio.Nursery = None
@ -100,12 +95,6 @@ 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.
@ -269,15 +258,14 @@ class Sampler:
        subs: set
        last_ts, subs = pair
-        # NOTE, for debugging pub-sub issues
+        task = trio.lowlevel.current_task()
-        # task = trio.lowlevel.current_task()
+        log.debug(
-        # log.debug(
+            f'SUBS {self.subscribers}\n'
-        #     f'AlL-SUBS@{period_s!r}: {self.subscribers}\n'
+            f'PAIR {pair}\n'
-        #     f'PAIR: {pair}\n'
+            f'TASK: {task}: {id(task)}\n'
-        #     f'TASK: {task}: {id(task)}\n'
+            f'broadcasting {period_s} -> {last_ts}\n'
        #     f'broadcasting {period_s} -> {last_ts}\n'
            # f'consumers: {subs}'
-        # )
+        )
        borked: set[MsgStream] = set()
        sent: set[MsgStream] = set()
        while True:
@ -294,11 +282,12 @@ class Sampler:
                        await stream.send(msg)
                        sent.add(stream)
-                    except self.bcast_errors as err:
+                    except (
                        trio.BrokenResourceError,
                        trio.ClosedResourceError
                    ):
                        log.error(
-                            f'Connection dropped for IPC ctx\n'
+                            f'{stream._ctx.chan.uid} dropped connection'
                            f'{stream._ctx}\n\n'
                            f'Due to {type(err)}'
                        )
                        borked.add(stream)
                else:
@ -386,10 +375,7 @@ async def register_with_sampler(
                assert Sampler.ohlcv_shms
            # unblock caller
-            await ctx.started(
+            await ctx.started(set(Sampler.ohlcv_shms.keys()))
                # XXX bc msgpack only allows one array type!
                list(Sampler.ohlcv_shms.keys())
            )
            if open_index_stream:
                try:
@ -433,6 +419,7 @@ async def register_with_sampler(
 async def spawn_samplerd(
    loglevel: str | None = None,
    **extra_tractor_kwargs
@ -442,10 +429,7 @@ async def spawn_samplerd(
    update and increment count write and stream broadcasting.
    '''
-    from piker.service import (
+    from piker.service import Services
        get_service_mngr,
        ServiceMngr,
    )
    dname = 'samplerd'
    log.info(f'Spawning `{dname}`')
@ -453,33 +437,26 @@ 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?
-    mngr: ServiceMngr = get_service_mngr()
+    async with Services.locks[dname + '_singleton']:
    already_started: bool = dname in mngr.service_tasks
-    async with mngr._locks[dname + '_singleton']:
+        if dname not in Services.service_tasks:
        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
+            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
            )
-        if not already_started:
+
-            assert (
+            await Services.start_service_task(
-                ctx
+                dname,
-                and
+                portal,
-                ctx.portal
+                register_with_sampler,
-                and
+                period_s=1,
-                not ctx.cancel_called
+                sub_for_broadcasts=False,
            )
            return True
@ -584,7 +561,8 @@ async def open_sample_stream(
 async def sample_and_broadcast(
-    bus: _FeedsBus,
+
    bus: _FeedsBus,  # noqa
    rt_shm: ShmArray,
    hist_shm: ShmArray,
    quote_stream: trio.abc.ReceiveChannel,
@ -604,32 +582,11 @@ 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)
-        # XXX WARNING XXX only enable for debugging bc ow can cost
+        # TODO: ``numba`` this!
        # 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
@ -702,21 +659,6 @@ 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
@ -814,6 +756,7 @@ async def sample_and_broadcast(
 async def uniform_rate_send(
    rate: float,
    quote_stream: trio.abc.ReceiveChannel,
    stream: MsgStream,
@ -946,7 +889,6 @@ 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,17 +387,22 @@ 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 n,
+        trio.open_nursery() as tn,
        open_autorecon_ws(
            url=url,
            msg_recv_timeout=msg_recv_timeout,
@ -486,20 +491,15 @@ 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 = msg['id']
+                        req_id: int = error['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}')
-        n.start_soon(recv_task)
+        tn.start_soon(recv_task)
        yield json_rpc
-        n.cancel_scope.cancel()
+        tn.cancel_scope.cancel()
--- a/piker/data/feed.py
+++ b/piker/data/feed.py
@ -540,10 +540,7 @@ 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(
+        bus._subscribers.setdefault(bs_fqme, set())
            bs_fqme,
            set(),
        )
    # sync feed subscribers with flume handles
    await ctx.started(
--- a/piker/log.py
+++ b/piker/log.py
@ -18,11 +18,7 @@
 Log like a forester!
 """
 import logging
 import reprlib
 import json
 from typing import (
    Callable,
 )
 import tractor
 from pygments import (
@ -88,29 +84,3 @@ 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,11 +30,7 @@ Actor runtime primtives and (distributed) service APIs for,
  => TODO: maybe to (re)move elsewhere?
 '''
-from ._mngr import (
+from ._mngr import Services as Services
    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,6 +21,7 @@
 from __future__ import annotations
 import os
 from typing import (
    Optional,
    Any,
    ClassVar,
 )
@ -29,13 +30,13 @@ from contextlib import (
 )
 import tractor
 import trio
 from ._util import (
    get_console_log,
 )
 from ._mngr import (
-    open_service_mngr,
+    Services,
    ServiceMngr,
 )
 from ._registry import (  # noqa
    _tractor_kwargs,
@ -58,7 +59,7 @@ async def open_piker_runtime(
    registry_addrs: list[tuple[str, int]] = [],
    enable_modules: list[str] = [],
-    loglevel: str|None = None,
+    loglevel: Optional[str] = None,
    # XXX NOTE XXX: you should pretty much never want debug mode
    # for data daemons when running in production.
@ -118,10 +119,6 @@ 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,
@ -170,13 +167,12 @@ async def open_pikerd(
    **kwargs,
-) -> ServiceMngr:
+) -> Services:
    '''
-    Start a root piker daemon actor (aka `pikerd`) with an indefinite
+    Start a root piker daemon with an indefinite lifetime.
    lifetime.
-    A root actor-nursery is created which can be used to spawn and
+    A root actor nursery is created which can be used to create and keep
-    supervise underling service sub-actors (see below).
+    alive underling services (see below).
    '''
    # NOTE: for the root daemon we always enable the root
@ -203,6 +199,8 @@ 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:
@ -211,17 +209,25 @@ async def open_pikerd(
                    'Maybe you have another daemon already running?'
                )
-        mngr: ServiceMngr
+        # assign globally for future daemon/task creation
-        async with open_service_mngr(
+        Services.actor_n = actor_nursery
-            debug_mode=debug_mode,
+        Services.service_n = service_nursery
-        ) as mngr:
+        Services.debug_mode = debug_mode
-            yield mngr
+
        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()
 # TODO: do we even need this?
 # @acm
 # async def maybe_open_runtime(
-#     loglevel: str|None = None,
+#     loglevel: Optional[str] = None,
 #     **kwargs,
 # ) -> None:
@ -250,7 +256,7 @@ async def maybe_open_pikerd(
    loglevel: str | None = None,
    **kwargs,
-) -> tractor._portal.Portal | ClassVar[ServiceMngr]:
+) -> tractor._portal.Portal | ClassVar[Services]:
    '''
    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 ServiceMngr
+from ._mngr import Services
 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: ServiceMngr,
+    services: Services,
    service_name: str,
    # endpoint config passed as **kwargs
@ -549,8 +549,7 @@ 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,17 +26,14 @@ 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 (
-    get_service_mngr,
+    Services,
    ServiceMngr,
 )
 from ._actor_runtime import maybe_open_pikerd
 from ._registry import find_service
@ -44,14 +41,15 @@ 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:
@ -69,7 +67,7 @@ async def maybe_spawn_daemon(
    '''
    # serialize access to this section to avoid
    # 2 or more tasks racing to create a daemon
-    lock = _locks[service_name]
+    lock = Services.locks[service_name]
    await lock.acquire()
    async with find_service(
@ -134,65 +132,7 @@ 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(
@ -207,22 +147,21 @@ async def spawn_emsd(
    """
    log.info('Spawning emsd')
-    smngr: ServiceMngr = get_service_mngr()
+    portal = await Services.actor_n.start_actor(
    portal = await smngr.actor_n.start_actor(
        'emsd',
        enable_modules=[
            'piker.clearing._ems',
            'piker.clearing._client',
        ],
        loglevel=loglevel,
-        debug_mode=smngr.debug_mode,  # set by pikerd flag
+        debug_mode=Services.debug_mode,  # set by pikerd flag
        **extra_tractor_kwargs
    )
    # non-blocking setup of clearing service
    from ..clearing._ems import _setup_persistent_emsd
-    await smngr.start_service_task(
+    await Services.start_service_task(
        'emsd',
        portal,
--- a/piker/service/_mngr.py
+++ b/piker/service/_mngr.py
@ -18,29 +18,16 @@
 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 msgspec
 import tractor
 import trio
-from trio import TaskStatus
+from trio_typing import TaskStatus
 import tractor
 from tractor import (
    ActorNursery,
    current_actor,
    ContextCancelled,
    Context,
@ -52,130 +39,6 @@ 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.
@ -183,46 +46,31 @@ def get_service_mngr() -> ServiceMngr:
 #   to the pikerd actor for starting services remotely!
 # - prolly rename this to ActorServicesNursery since it spawns
 #   new actors and supervises them to completion?
-@dataclass
+class Services:
 class ServiceMngr:
 # class ServiceMngr(msgspec.Struct):
    '''
    A multi-subactor-as-service manager.
-    Spawn, supervise and monitor service/daemon subactors in a SC
+    actor_n: tractor._supervise.ActorNursery
    process tree.
    '''
    actor_n: ActorNursery
    service_n: trio.Nursery
-    debug_mode: bool = False # tractor sub-actor debug mode flag
+    debug_mode: bool  # tractor sub-actor debug mode flag
    service_tasks: dict[
        str,
        tuple[
            trio.CancelScope,
            Context,
            Portal,
            trio.Event,
        ]
-    ] = field(default_factory=dict)
+    ] = {}
-
+    locks = defaultdict(trio.Lock)
    # 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, Any):
+    ) -> (trio.CancelScope, Context):
        '''
        Open a context in a service sub-actor, add to a stack
        that gets unwound at ``pikerd`` teardown.
@ -235,7 +83,6 @@ class ServiceMngr:
            task_status: TaskStatus[
                tuple[
                    trio.CancelScope,
                    Context,
                    trio.Event,
                    Any,
                ]
@ -243,87 +90,64 @@ class ServiceMngr:
        ) -> 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,
-                    ) as (ctx, started):
+                ) as (ctx, first):
                    # unblock once the remote context has started
                    complete = trio.Event()
-                        task_status.started((
+                    task_status.started((cs, complete, first))
                            cs,
                            ctx,
                            complete,
                            started,
                        ))
                    log.info(
-                            f'`pikerd` service {name} started with value {started}'
+                        f'`pikerd` service {name} started with value {first}'
                    )
                    try:
                        # wait on any context's return value
                        # and any final portal result from the
                        # sub-actor.
-                        ctx_res: Any = await ctx.wait_for_result()
+                        ctx_res: Any = await ctx.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 (
+                        return (await portal.result(), ctx_res)
                            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
+                            canceller != portal.channel.uid
                            and
                            canceller != our_uid
                        ):
                            log.cancel(
-                            f'Actor-service `{name}` was remotely cancelled by a peer?\n'
+                                f'Actor-service {name} was remotely cancelled?\n'
-
+                                f'remote canceller: {canceller}\n'
-                            # TODO: this would be a good spot to use
+                                f'Keeping {our_uid} alive, ignoring sub-actor cancel..\n'
                            # a respawn feature Bo
                            f'-> Keeping `pikerd` service manager alive despite this inter-peer cancel\n\n'
                            f'cancellee: {portal.chan.uid}\n'
                            f'canceller: {canceller}\n'
                            )
                        else:
                            raise
                    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(
+        cs, complete, first = await self.service_n.start(open_context_in_task)
            open_context_in_task
        )
        # store the cancel scope and portal for later cancellation or
        # retstart if needed.
-        self.service_tasks[name] = (cs, sub_ctx, portal, complete)
+        self.service_tasks[name] = (cs, portal, complete)
        return cs, sub_ctx, started
        return cs, first
    @classmethod
    async def cancel_service(
        self,
        name: str,
@ -334,80 +158,8 @@ class ServiceMngr:
        '''
        log.info(f'Cancelling `pikerd` service {name}')
-        cs, sub_ctx, portal, complete = self.service_tasks[name]
+        cs, portal, complete = self.service_tasks[name]
-
+        cs.cancel()
        # cs.cancel()
        await sub_ctx.cancel()
        await complete.wait()
-
+        assert name not in self.service_tasks, \
        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,13 +21,11 @@ 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 (
@ -129,7 +127,7 @@ def start_elasticsearch(
@acm
 async def start_ahab_daemon(
-    service_mngr: ServiceMngr,
+    service_mngr: Services,
    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 ServiceMngr
+from . import Services
 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: ServiceMngr,
+    service_mngr: Services,
    user_config: dict | None = None,
    loglevel: str | None = None,
--- a/piker/storage/cli.py
+++ b/piker/storage/cli.py
@ -386,6 +386,8 @@ def ldshm(
            open_annot_ctl() as actl,
        ):
            shm_df: pl.DataFrame | None = None
            tf2aids: dict[float, dict] = {}
            for (
                shmfile,
                shm,
@ -526,16 +528,17 @@ 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.
-                    await tractor.pause()
+                    assert not diff
                    # 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,11 +162,12 @@ class NativeStorageClient:
    def index_files(self):
        for path in self._datadir.iterdir():
            if (
-                path.name in {'borked', 'expired',}
+                path.is_dir()
                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,16 +620,19 @@ async def start_backfill(
                    f'{next_start_dt} -> {last_start_dt}'
                )
-                # always drop the src asset token for
+                # NOTE, 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,6 +616,18 @@ 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,
@ -626,7 +638,6 @@ 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
 ]:
@ -639,19 +650,22 @@ def dedupe(
    '''
    wdts: pl.DataFrame = with_dts(src_df)
-    # maybe sort on any time field
+    deduped = wdts
    if sort:
        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'],
+        # subset=['dt'],
        subset=['time'],
        maintain_order=True,
    )
    # maybe sort on any time field
    if sort:
        deduped = deduped.sort(by='time')
        # TODO: detect out-of-order segments which were corrected!
        # -[ ] report in log msg
        # -[ ] possibly return segment sections which were moved?
    diff: int = (
        wdts.height
        -
--- a/poetry.lock
+++ b/poetry.lock
--- a/pyproject.toml
+++ b/pyproject.toml
@ -15,121 +15,17 @@
 # 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 = ["poetry-core"]
+requires = ["hatchling"]
-build-backend = "poetry.core.masonry.api"
+build-backend = "hatchling.build"
 # ------ - ------
 [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.0.alpha0.dev0"
 description = "trading gear for hackers"
 authors = ["Tyler Goodlet <goodboy_foss@protonmail.com>"]
 license = "AGPLv3"
 readme = "README.rst"
 # ------ - ------
 [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"
 tractor = {path = "../tractor", develop = true}
 websockets = "12.0"
 [tool.poetry.dependencies.asyncvnc]
 git = 'https://github.com/pikers/asyncvnc.git'
 branch = 'main'
 [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
 # ------ - ------
 # 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]
 name = "piker"
 version = "0.1.0a0dev0"
 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"
 readme = "README.rst"
 keywords = [
    "async",
    "trading",
@ -138,15 +34,100 @@ keywords=[
    "charting",
 ]
 classifiers = [
-  'Development Status :: 3 - Alpha',
+    "Development Status :: 3 - Alpha",
    "License :: OSI Approved :: GNU Affero General Public License v3 or later (AGPLv3+)",
-  'Operating System :: POSIX :: Linux',
+    "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 :: Financial and Insurance Industry",
-  'Intended Audience :: Science/Research',
+    "Intended Audience :: Science/Research",
-  'Intended Audience :: Developers',
+    "Intended Audience :: Developers",
-  'Intended Audience :: Education',
+    "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'
    # 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
 ]
 [dependency-groups]
 # TODO: 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",
 ]
 [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
@ -0,0 +1,93 @@
 # 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 (
-    get_service_mngr,
+    Services,
 )
 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 get_service_mngr()
+        assert service_manager is Services
        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 ServiceMngr
+from piker.service import Services
 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: ServiceMngr,
+    open_test_pikerd: Services,
    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,
-    ServiceMngr,
+    Services,
 )
 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: ServiceMngr
+        services: Services
        async with (
            open_test_pikerd(
--- a/uv.lock
+++ b/uv.lock
Author	SHA1	Message	Date
goodboy	5e371f1d73	Merge pull request 'jsonrpc_err_in_rent' (#41 ) from jsonrpc_err_in_rent into gitea_feats Reviewed-on: #41	2025-02-21 21:21:02 +00:00
goodboy	6c221bb348	Merge pull request 'tsp_gaps: fixes for fault-less OHLCV time-series loads' (#35 ) from tsp_gaps into gitea_feats Reviewed-on: #35	2025-02-21 20:46:37 +00:00
Tyler Goodlet	e391c896f8	Mk jsronrpc's underlying ws timeout `float('inf')` Since currently we're only using this IPC subsys for `deribit`, and generally speaking we're primarly supporting options markets (which are fairly "slow moving"), flip to a default of NOT resetting the `NoBsWs` on timeout since doing so normally breaks the jsron-rpc IPC session. Without a proper `fixture` passed to `open_autorecon_ws()` (which we should eventually implement!!) relying on a timeout-to-reset more or less will just cause breakage issues - a proper reconnect sequence must be implemented before using that feature. Deats, - expose and proxy through the `msg_recv_timeout` from `open_jsonrpc_session()` into the underlying `open_autorecon_ws()` call.	2025-02-19 17:05:13 -05:00
Tyler Goodlet	5633f5614d	Doc-n-clean `.data._web_bs.open_jsonrpc_session()` Add a doc-string reflecting recent refinements, drop all the old hook params, rename `n: trio.Nursery` -> `tn` for "task nursery" fitting with code base's naming style.	2025-02-19 17:05:13 -05:00
Tyler Goodlet	76735189de	data._web_bs: try to raise jsonrpc errors in parent task	2025-02-19 17:05:13 -05:00
Tyler Goodlet	d49608f74e	Refine history gap/termination signalling Namely handling backends which do not provide a default "frame size-duration" in their init-config by making the backfiller guess the value based on the first frame received. Deats, - adjust `start_backfill()` to take a more explicit `def_frame_duration: Duration` expected to be unpacked from any backend hist init-config by the `tsdb_backfill()` caller which now also computes a value from the first received frame when the config section isn't provided. - in `start_backfill()` we now always expect the `def_frame_duration` input and always decrement the query range by this value whenever a `NoData` is raised by the provider-backend paired with an explicit `log.warning()` about the handling. - also relay any `DataUnavailable.args[0]` message from the provider in the handler. - repair "gap reporting" which checks for expected frame duration vs. that received with much better humanized logging on the missing segment using `pendulum.Interval/Duration.in_words()` output.	2025-02-19 17:01:24 -05:00
Tyler Goodlet	bf0ac93aa3	Only use `frame_types` if delivered during enter The `open_history_client()` provider endpoint can optionally deliver a `frame_types: dict[int, pendulum.Duration]` subsection in its `config: dict[str, dict]` (as was implemented with the `ib` backend). This allows the `tsp` backfilling machinery to use this "recommended frame duration" to subtract from the `last_start_dt` any time a `NoData` gap is signalled by the `get_hist()` call allowing gaps to be ignored safely without missing history by knowing the next earliest dt we can query from using the `end_dt`. However, currently all crypto$ providers haven't implemented this feat yet.. As such only try to use the `frame_types` feature if provided when handling `NoData` conditions inside `tsp.start_backfill()` and otherwise raise as normal.	2025-02-19 17:01:24 -05:00
Tyler Goodlet	d7179d47b0	`.tsp._anal`: add (unused) `detect_vlm_gaps()`	2025-02-19 17:01:24 -05:00
Tyler Goodlet	c390e87536	`.storage.cli`: collect gap-markup-aids into `tf2aids: dict` prior to pause for introspection	2025-02-19 17:01:24 -05:00
Tyler Goodlet	5e4a6d61c7	Ignore any non-`.parquet` files under `.config/piker/nativedb/` subdir	2025-02-19 17:01:24 -05:00
Tyler Goodlet	3caaa30b03	Mask no-data pause, add perps to no-`/src`-in-fqme asset set Was orig for debugging an issue with `kucoin` i think but definitely shouldn't be left in XD Also add `'perpetual_future'` to the `.start_backfill()` input literal set since we don't expect the 'btc/usd.perp.binance' for now.	2025-02-19 17:01:24 -05:00
goodboy	1e3942fdc2	Merge pull request 'Add `ruff` to deps, bump `uv.lock`' (#32 ) from add_ruff_linter into gitea_feats Reviewed-on: #32	2025-02-17 19:55:32 +00:00
Nelson Torres	49ea380503	Add new `ruff.toml` config file Based on the default provided in their [docs](https://docs.astral.sh/ruff/configuration/) and migrating previous config from the prior `poetry`-verion of our `pyproject.toml`	2025-02-17 14:48:10 -05:00
Tyler Goodlet	933f169938	Add/reorg back some content from `poetry` old config	2025-02-14 13:47:02 -05:00
Nelson Torres	51337052a4	Remove legacy `poetry` config content from pyproject.toml	2025-02-14 15:01:29 -03:00
Tyler Goodlet	8abe55dcc6	Add `ruff` to deps, bump `uv.lock` Such that we start encouraging devs to lint code they touch and hopefully we include a pass as part of our tests/CI eventually B) Also, mk local `tractor` install `--editable` since without it being a locally hackable repo it's kinda pointless to install from the local fs Xp	2025-02-13 21:20:11 -05:00
goodboy	c933f2ad56	Merge pull request 'kucoin_and_binance_fix' (#9 ) from kucoin_and_binance_fix into gitea_feats Reviewed-on: #9	2025-02-13 19:40:50 +00:00
Tyler Goodlet	00108010c9	Mask `pytest` detection block in `piker.config` Seems to be some kinda super weird env bug since we moved to using `uv`? When it triggers it also seems to cause a pretty fundamental crash that not only breaks `tractor.devx._debug` stuff but also seems to get us in a perma-hang state where no SIGINT or other sys sig will be able to kill the root proc!?!? TODO, a `gitea` issue to track so we can fix the fundamental problem as well as transitive fault in `tractor`'s core which seems to be due to the error taking place during a sub-actor's module import phase which prevents the runtime from booting fully and then the proc getting stuck in a real gnarly SIG-state..	2025-02-13 13:32:11 -05:00
Tyler Goodlet	8a4901c517	`.binance.feed`: moar type fixes, drop `rapidfuzz`	2025-02-13 12:35:41 -05:00
Tyler Goodlet	d7f6a5ab63	Update to latest `KucoinMktPair` spec	2025-02-12 18:08:40 -05:00
Tyler Goodlet	e0fdabf651	Use `../tractor` srcs in editable mode?	2025-02-12 15:14:30 -05:00
Tyler Goodlet	cb88dfc9da	`kucoin`: repair live quotes streaming.. This must have broke at some point during the new `MktPair` and thus `.fqme: str` updates; mas-o-menos the symbol key in the quote-msg-`dict` was NOT set to the `MktPair.bs_fqme: str` value and thus wasn't being processed by the downstream sampling and feed subsys. So fix that as well as a few other refinements, - set the `topic: mkt.bs_fqme` in quote msgs obvi. - drop the "wait for first clearing vlm" quote poll loop; going to fix the sampler to handle a `first_quote` without a `'last'` key. - add some typing around calls to `get_mkt_info()`. - rename `stream_messages()` -> `iter_normed_quotes()`.	2025-02-12 15:05:22 -05:00
Nelson Torres	bb41dd6d18	Deleted settlePlan field from binance FutesPair.	2025-02-12 15:05:22 -05:00
Nelson Torres	99e90129ad	Added missing fields for kucoin. feeCategory, makerFeeCoefficient, takerFeeCoefficient and st.	2025-02-12 15:05:22 -05:00
Tyler Goodlet	cceb7a37b9	Lel, forgot to add a `SPOT` venue for `binance`..	2025-02-12 15:05:22 -05:00
goodboy	5382815b2d	Merge pull request 'uv migration and default.nix for qt6' (#17 ) from uv_migration into gitea_feats Reviewed-on: #17	2025-02-12 20:04:02 +00:00
Tyler Goodlet	cb1ba8a05f	Further root readme bump, factor `.clearing` content In line with our move to `uv` and recent `nix` support update a bunch of the summary content and factor out the order-control section to a new `.piker.clearing` readme file with embedded todos therein.	2025-02-12 15:01:51 -05:00
Nelson Torres	6c65ec4d3b	Readme update: - uv replace poetry - update nix-shell command	2025-02-12 13:05:25 -03:00
Nelson Torres	12e371b027	uv migration	2025-02-12 11:19:34 -03:00
goodboy	014bd58db4	Merge pull request 'go_httpx' (#2 ) from go_httpx into gitea_feats Landed-in: #2	2025-02-12 13:01:19 +00:00
Tyler Goodlet	844544ed8e	Port binance to `httpx` Like other backends use the `AsyncClient` for all venue specific client-sessions but change to allocating them inside `get_client()` using an `AsyncExitStack` and inserting directly in the `Client.venue_sesh: dict` table during init. Supporting impl tweaks: - remove most of the API client session building logic and instead make `Client.__init__()` take in a `venue_sessions: dict` (set it to `.venue_sesh`) and `conf: dict`, instead opting to do the http client configuration inside `get_client()` since all that code only needs to be run once. \|_load config inside `get_client()` once. \|_move session token creation into a new util func `init_api_keys()` and also call it from `get_client()` factory; toss in an ex. toml section config to the doc string. - define `_venue_urls: dict[str, str]` (content taken from old static `.venue_sesh` dict) at module level and feed them as `base_url: str` inputs to the client create loop. - adjust all call sigs in httpx-sesh-using methods, namely just `._api()`. - do a `.exch_info()` call in `get_client()` to cache the symbology set. Unrelated changes for various other outstanding buggers: - to get futures feeds correctly loading when selected from search (like 'XMRUSDT.USDTM.PERP'), expect a `MktPair` input to `Client.bars()` such that the exact venue-key can be looked up (via a new `.pair2venuekey()` meth) and then passed to `._api()`. - adjust `.broker.open_trade_dialog()` to failover to paper engine when there's no `api_key` key set for the `subconf` venue-key.	2025-02-11 16:27:28 -05:00
Nelson Torres	f479252d26	Added note to exception when missing field in SpotPair class	2025-02-11 16:27:28 -05:00
Nelson Torres	033ef2e35e	Added new fields to SpotPair class in venues	2025-02-11 16:27:28 -05:00
Tyler Goodlet	2cdece244c	binance: raise `NoData` on null hist arrays Like we do with other history backends to indicate lack of a data set. This avoids any raise that will will bring down the backloader task with some downstream error. Raise a `ValueError` on no time index for now.	2025-02-11 16:27:28 -05:00
Tyler Goodlet	018694bbdb	Woops, `data` can be an empty list XD	2025-02-11 16:27:28 -05:00
Tyler Goodlet	128a2d507f	Woops, fix missing `api_url` ref in error log	2025-02-11 16:27:28 -05:00
Tyler Goodlet	430650a6a7	Change type-annots to use `httpx.Response`	2025-02-11 16:27:28 -05:00
Tyler Goodlet	1da3cf5698	Port `kucoin` backend to `httpx`	2025-02-11 16:27:28 -05:00
Tyler Goodlet	a348603fc4	Port `kraken` backend to `httpx`	2025-02-11 16:27:28 -05:00
goodboy	86047824d8	Merge pull request '`.brokers.ib` random fixes-n-improvements from various other dev branches..' (#27 ) from ib_refinements into gitea_feats Merged-in: #27	2025-02-11 21:26:20 +00:00
Tyler Goodlet	cb92abbc38	ib: add connect status info emit	2025-02-11 14:56:17 -05:00
Tyler Goodlet	70332e375b	ib: `.api` mod and log-fmt cleaning About time we tidy'd a buncha status logging in this backend.. particularly for boot-up where there's lots of client-try-connect poll looping with account detection from the user config. `.api.Client` pprint and logging fmt improvements: - add `Client.__repr__()` which shows the minimally useful set of info from the underlying `.ib: IB` as well as a new `.acnts: list[str]` of the account aliases defined in the user's `brokers.toml`. - mk `.bars()` define a comprehensive `query_info: str` with all the request deats but only display if there's a problem with the response data. - mk `.get_config()` report both the config file path and the acnt aliases (NOT the actual account #s). - move all `.load_aio_clients()` client poll loop requests do `log.runtime()` statuses, only falling through to a `.warning()` when the loop fails to connect the client to the spec-ed API-gw addr, and \|_ don't allow loading accounts for which the user has not defined an alias in `brokers.toml::[ib]`; raise a value-error in such cases with a message indicating how to mod the config. \|_ only `log.info()` about acnts if some were loaded.. Other mod logging de-noising: - better status fmting in `.symbols.open_symbol_search()` with `repr(Client)`. - for `.feed.stream_quotes()` first quote reporting use `.runtime()`.	2025-02-11 14:56:17 -05:00
Tyler Goodlet	4940aabe05	ib: warn about mkt precision cuckups that `Contract`s clearly deliver wrong..	2025-02-11 14:56:17 -05:00
Tyler Goodlet	4f9998e9fb	ib: mask out trade and vlm rates for now	2025-02-11 14:56:17 -05:00
Tyler Goodlet	c92a236196	ib: more trade record edge case handling - timestamps came as `'date'`-keyed from 2022 and before but now are `'datetime'`.. - some symbols seem to have no commission field, so handle that.. - when no `'price'` field found return `None` from `norm_trade()`. - add a warn log on mid-fill commission updates.	2025-02-11 14:56:17 -05:00
goodboy	e4cd1f85f6	Merge pull request 'pyqt6' (#3 ) from pyqt6 into gitea_feats Reviewed-on: #3 (well by fomo anyway..)	2025-02-11 17:25:03 +00:00