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

Reviewed-on: #41

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!

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
   '';
 }

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
 ]
 

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(
+
-        brokermod,
+    start_actor_kwargs['name'] = f'brokerd.{brokername}'
-        '_spawn_kwargs',
+    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,

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:

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',
 ]
 

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)

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

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)
 

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:
-            pair = pairs[item['name']] = KucoinMktPair(**item)
+            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

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!

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,14 +1010,10 @@ async def translate_and_relay_brokerd_events(
                 status_msg.brokerd_msg = msg
                 status_msg.src = msg.broker_details['name']
 
-                if not status_msg.req:
+                await router.client_broadcast(
-                    # likely some order change state?
+                    status_msg.req.symbol,
-                    await tractor.pause()
+                    status_msg,
-                else:
+                )
-                    await router.client_broadcast(
-                        status_msg.req.symbol,
-                        status_msg,
-                    )
 
                 if status == 'closed':
                     log.info(f'Execution for {oid} is complete!')

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

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"

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}'
-        #     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,
+            portal = await Services.actor_n.start_actor(
-            ctx_ep=partial(
+                dname,
+                enable_modules=[
+                    'piker.data._sampling',
+                ],
+                loglevel=loglevel,
+                debug_mode=Services.debug_mode,  # set by pikerd flag
+                **extra_tractor_kwargs
+            )
+
+            await Services.start_service_task(
+                dname,
+                portal,
                 register_with_sampler,
                 period_s=1,
                 sub_for_broadcasts=False,
-            ),
-            debug_mode=mngr.debug_mode,  # set by pikerd flag
-
-            # proxy-through to tractor
-            enable_modules=[
-                'piker.data._sampling',
-            ],
-            loglevel=loglevel,
-            **extra_tractor_kwargs
-        )
-        if not already_started:
-            assert (
-                ctx
-                and
-                ctx.portal
-                and
-                not ctx.cancel_called
             )
             return True
 
@@ -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,

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()

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(

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

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,

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.
@@ -68,7 +69,7 @@ async def open_piker_runtime(
     # and spawn the service tree distributed per that.
     start_method: str = 'trio',
 
-    tractor_runtime_overrides: dict|None = None,
+    tractor_runtime_overrides: dict | None = None,
     **tractor_kwargs,
 
 ) -> tuple[
@@ -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

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):

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,
 

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):
+                async with portal.open_context(
+                    target,
+                    allow_overruns=allow_overruns,
+                    **ctx_kwargs,
 
-                        # unblock once the remote context has started
+                ) as (ctx, first):
-                        complete = trio.Event()
+
-                        task_status.started((
+                    # unblock once the remote context has started
-                            cs,
+                    complete = trio.Event()
-                            ctx,
+                    task_status.started((cs, complete, first))
-                            complete,
+                    log.info(
-                            started,
+                        f'`pikerd` service {name} started with value {first}'
-                        ))
+                    )
-                        log.info(
+                    try:
-                            f'`pikerd` service {name} started with value {started}'
-                        )
                         # wait on any context's return value
                         # and any final portal result from the
                         # sub-actor.
-                        ctx_res: Any = await ctx.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(),
+                    except ContextCancelled as ctxe:
-                            ctx_res,
+                        canceller: tuple[str, str] = ctxe.canceller
-                        )
+                        our_uid: tuple[str, str] = current_actor().uid
+                        if (
+                            canceller != portal.channel.uid
+                            and
+                            canceller != our_uid
+                        ):
+                            log.cancel(
+                                f'Actor-service {name} was remotely cancelled?\n'
+                                f'remote canceller: {canceller}\n'
+                                f'Keeping {our_uid} alive, ignoring sub-actor cancel..\n'
+                            )
+                        else:
+                            raise
 
-                except ContextCancelled as ctxe:
-                    canceller: tuple[str, str] = ctxe.canceller
-                    our_uid: tuple[str, str] = current_actor().uid
-                    if (
-                        canceller != portal.chan.uid
-                        and
-                        canceller != our_uid
-                    ):
-                        log.cancel(
-                            f'Actor-service `{name}` was remotely cancelled by a peer?\n'
 
-                            # TODO: this would be a good spot to use
-                            # a respawn feature Bo
-                            f'-> Keeping `pikerd` service manager alive despite this inter-peer cancel\n\n'
 
-                            f'cancellee: {portal.chan.uid}\n'
+                    finally:
-                            f'canceller: {canceller}\n'
+                        await portal.cancel_actor()
-                        )
+                        complete.set()
-                    else:
+                        self.service_tasks.pop(name)
-                        raise
 
-                finally:
+        cs, complete, first = await self.service_n.start(open_context_in_task)
-                    # NOTE: the ctx MUST be cancelled first if we
-                    # don't want the above `ctx.wait_for_result()` to
-                    # raise a self-ctxc. WHY, well since from the ctx's
-                    # perspective the cancel request will have
-                    # arrived out-out-of-band at the `Actor.cancel()`
-                    # level, thus `Context.cancel_called == False`,
-                    # meaning `ctx._is_self_cancelled() == False`.
-                    # with trio.CancelScope(shield=True):
-                    # await ctx.cancel()
-                    await portal.cancel_actor()
-                    complete.set()
-                    self.service_tasks.pop(name)
-
-        cs, sub_ctx, complete, started = await self.service_n.start(
-            open_context_in_task
-        )
 
         # store the cancel scope and portal for later cancellation or
         # retstart if needed.
-        self.service_tasks[name] = (cs, 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:
+            f'Serice task for {name} not terminated?'
-            # 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)

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,
 

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,
 

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(

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')

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,

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
         -

pyproject.toml

@@ -15,138 +15,119 @@
 # 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"
 
-# ------ - ------
+[project]
-
-[tool.ruff.lint]
-# https://docs.astral.sh/ruff/settings/#lint_ignore
-ignore = []
-
-# https://docs.astral.sh/ruff/settings/#lint_per-file-ignores
-# "piker/ui/qt.py" = [
-#   "E402",
-#   'F401',  # unused imports (without __all__ or blah as blah)
-#   # "F841", # unused variable rules
-# ]
-# ignore-init-module-imports = false
-
-# ------ - ------
-
-[tool.poetry]
 name = "piker"
-version = "0.1.0.alpha0.dev0"
+version = "0.1.0a0dev0"
 description = "trading gear for hackers"
-authors = ["Tyler Goodlet <goodboy_foss@protonmail.com>"]
+authors = [{ name = "Tyler Goodlet", email = "goodboy_foss@protonmail.com" }]
-license = "AGPLv3"
+requires-python = ">=3.12, <3.13"
+license = "AGPL-3.0-or-later"
 readme = "README.rst"
+keywords = [
+    "async",
+    "trading",
+    "finance",
+    "quant",
+    "charting",
+]
+classifiers = [
+    "Development Status :: 3 - Alpha",
+    "License :: OSI Approved :: GNU Affero General Public License v3 or later (AGPLv3+)",
+    "Operating System :: POSIX :: Linux",
+    "Programming Language :: Python :: Implementation :: CPython",
+    "Programming Language :: Python :: 3 :: Only",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
+    "Intended Audience :: Financial and Insurance Industry",
+    "Intended Audience :: Science/Research",
+    "Intended Audience :: Developers",
+    "Intended Audience :: Education",
+]
+dependencies = [
+    "async-generator >=1.10, <2.0.0",
+    "attrs >=23.1.0, <24.0.0",
+    "bidict >=0.22.1, <0.23.0",
+    "colorama >=0.4.6, <0.5.0",
+    "colorlog >=6.7.0, <7.0.0",
+    "ib-insync >=0.9.86, <0.10.0",
+    "numba >=0.59.0, <0.60.0",
+    "numpy >=1.25, <2.0",
+    "polars >=0.18.13, <0.19.0",
+    "pygments >=2.16.1, <3.0.0",
+    "rich >=13.5.2, <14.0.0",
+    "tomli >=2.0.1, <3.0.0",
+    "tomli-w >=1.0.0, <2.0.0",
+    "trio-util >=0.7.0, <0.8.0",
+    "trio-websocket >=0.10.3, <0.11.0",
+    "typer >=0.9.0, <1.0.0",
+    "rapidfuzz >=3.5.2, <4.0.0",
+    "pdbp >=1.5.0, <2.0.0",
+    "trio >=0.24, <0.25",
+    "pendulum >=3.0.0, <4.0.0",
+    "httpx >=0.27.0, <0.28.0",
+    "cryptofeed >=2.4.0, <3.0.0",
+    "pyarrow >=17.0.0, <18.0.0",
+    "websockets ==12.0",
+    "msgspec",
+    "tractor",
+    "asyncvnc",
+    "tomlkit",
+]
 
-# ------ - ------
+[project.optional-dependencies]
+uis = [
+    # https://docs.astral.sh/uv/concepts/projects/dependencies/#optional-dependencies
+    # TODO: make sure the levenshtein shit compiles on nix..
+    # rapidfuzz = {extras = ["speedup"], version = "^0.18.0"}
+    "rapidfuzz >=3.2.0, <4.0.0",
+    "qdarkstyle >=3.0.2, <4.0.0",
+    "pyqt6 >=6.7.0, <7.0.0",
+    "pyqtgraph",
 
-[tool.poetry.dependencies]
+    # for consideration,
-async-generator = "^1.10"
+    # - 'visidata'
-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}
+    # TODO: add an `--only daemon` group for running non-ui / pikerd
-websockets = "12.0"
+    # service tree in distributed mode B)
-[tool.poetry.dependencies.asyncvnc]
+    # https://docs.astral.sh/uv/concepts/projects/dependencies/#optional-dependencies
-git = 'https://github.com/pikers/asyncvnc.git'
+]
-branch = 'main'
 
-[tool.poetry.dependencies.tomlkit]
+[dependency-groups]
-develop = true
+# TODO: a toolset that makes debugging a `pikerd` service (tree) easy
-git = 'https://github.com/pikers/tomlkit.git'
+# to hack on directly using more or less the local env:
-branch = 'piker_pin'
-# path = "../tomlkit/"
-
-[tool.poetry.group.uis]
-optional = true
-[tool.poetry.group.uis.dependencies]
-# https://python-poetry.org/docs/managing-dependencies/#dependency-groups
-# TODO: make sure the levenshtein shit compiles on nix..
-# rapidfuzz = {extras = ["speedup"], version = "^0.18.0"}
-rapidfuzz = "^3.2.0"
-qdarkstyle = ">=3.0.2"
-pyqtgraph = { git = 'https://github.com/pikers/pyqtgraph.git' }
-
-# ------ - ------
-pyqt6 = "^6.7.0"
-
-[tool.poetry.group.dev]
-optional = true
-[tool.poetry.group.dev.dependencies]
-# testing / CI
-pytest = "^6.0.0"
-elasticsearch = "^8.9.0"
-xonsh = "^0.14.2"
-prompt-toolkit = "3.0.40"
-cython = "^3.0.0"
-greenback = "^1.1.1"
-
-# console ehancements and eventually remote debugging
-# extras/helpers.
-# TODO: add a toolset that makes debugging a `pikerd` service
-# (tree) easy to hack on directly using more or less the local env:
 # - xonsh + xxh
 # - rsyscall + pdbp
 # - actor runtime control console like BEAM/OTP
-
+#
-# ------ - ------
+# console ehancements and eventually remote debugging extras/helpers.
-
+# use `uv --dev` to enable
-# TODO: add an `--only daemon` group for running non-ui / pikerd
+dev = [
-# service tree in distributed mode B)
+    "pytest >=6.0.0, <7.0.0",
-# https://python-poetry.org/docs/managing-dependencies/#installing-group-dependencies
+    "elasticsearch >=8.9.0, <9.0.0",
-# [tool.poetry.group.daemon.dependencies]
+    "xonsh >=0.14.2, <0.15.0",
-
+    "prompt-toolkit ==3.0.40",
-[tool.poetry.scripts]
+    "cython >=3.0.0, <4.0.0",
-piker = 'piker.cli:cli'
+    "greenback >=1.1.1, <2.0.0",
-pikerd = 'piker.cli:pikerd'
+    "ruff>=0.9.6",
-ledger = 'piker.accounting.cli:ledger'
-
-
-[project]
-keywords=[
-  "async",
-  "trading",
-  "finance",
-  "quant",
-  "charting",
-]
-classifiers=[
-  'Development Status :: 3 - Alpha',
-  "License :: OSI Approved :: GNU Affero General Public License v3 or later (AGPLv3+)",
-  'Operating System :: POSIX :: Linux',
-  "Programming Language :: Python :: Implementation :: CPython",
-  "Programming Language :: Python :: 3 :: Only",
-  "Programming Language :: Python :: 3.11",
-  "Programming Language :: Python :: 3.12",
-  'Intended Audience :: Financial and Insurance Industry',
-  'Intended Audience :: Science/Research',
-  'Intended Audience :: Developers',
-  'Intended Audience :: Education',
 ]
+
+[project.scripts]
+piker = "piker.cli:cli"
+pikerd = "piker.cli:pikerd"
+ledger = "piker.accounting.cli:ledger"
+
+[tool.hatch.build.targets.sdist]
+include = ["piker"]
+
+[tool.hatch.build.targets.wheel]
+include = ["piker"]
+
+[tool.uv.sources]
+pyqtgraph = { git = "https://github.com/pikers/pyqtgraph.git" }
+asyncvnc = { git = "https://github.com/pikers/asyncvnc.git", branch = "main" }
+tomlkit = { git = "https://github.com/pikers/tomlkit.git", branch ="piker_pin" }
+msgspec = { git = "https://github.com/jcrist/msgspec.git" }
+tractor = { path = "../tractor", editable = true }

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"

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',

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():

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(