Official service-mngr to `tractor.hilevel` move

Such that we maintain that subsys in the actor-runtime repo (with
hopefully an extensive test suite XD).

Port deats,
- rewrite `open_service_mngr()` as a thin wrapper that delegates into
  the new `tractor.hilevel.open_service_mngr()` but with maintenance of
  the `Services` class-singleton for now.
- port `.service._daemon` usage to the new
  `ServiceMngr.start_service_ctx()` a rename from
  `.start_service_task()` which is now likely destined for the soon
  supported `tractor.trionics.TaskMngr` nursery extension.
- ref the new `ServiceMngr.an: ActorNursery` instance var name.

Other,
- always enable the `tractor.pause_from_sync()` support via `greenback`
  whenever `debug_mode` is set at `pikerd` init.

piker/brokers/__init__.py

@@ -50,7 +50,7 @@ __brokers__: list[str] = [
     'binance',
     'ib',
     'kraken',
-    'kucoin'
+    'kucoin',
 
     # broken but used to work
     # 'questrade',
@@ -71,7 +71,7 @@ def get_brokermod(brokername: str) -> ModuleType:
     Return the imported broker module by name.
 
     '''
-    module = import_module('.' + brokername, 'piker.brokers')
+    module: ModuleType = import_module('.' + brokername, 'piker.brokers')
     # we only allow monkeying because it's for internal keying
     module.name = module.__name__.split('.')[-1]
     return module

piker/brokers/_daemon.py

@@ -23,6 +23,7 @@ from __future__ import annotations
 from contextlib import (
     asynccontextmanager as acm,
 )
+from functools import partial
 from types import ModuleType
 from typing import (
     TYPE_CHECKING,
@@ -190,14 +191,17 @@ def broker_init(
 
 
 async def spawn_brokerd(
-
     brokername: str,
     loglevel: str | None = None,
 
     **tractor_kwargs,
 
 ) -> bool:
+    '''
+    Spawn a `brokerd.<backendname>` subactor service daemon
+    using `pikerd`'s service mngr.
 
+    '''
     from piker.service._util import log  # use service mngr log
     log.info(f'Spawning {brokername} broker daemon')
 
@@ -217,27 +221,35 @@ async def spawn_brokerd(
 
     # ask `pikerd` to spawn a new sub-actor and manage it under its
     # actor nursery
-    from piker.service import Services
+    from piker.service import (
-
+        get_service_mngr,
-    dname: str = tractor_kwargs.pop('name')  # f'brokerd.{brokername}'
+        ServiceMngr,
-    portal = await Services.actor_n.start_actor(
-        dname,
-        enable_modules=_data_mods + tractor_kwargs.pop('enable_modules'),
-        debug_mode=Services.debug_mode,
-        **tractor_kwargs
     )
-
+    dname: str = tractor_kwargs.pop('name')  # f'brokerd.{brokername}'
-    # NOTE: the service mngr expects an already spawned actor + its
+    mngr: ServiceMngr = get_service_mngr()
-    # portal ref in order to do non-blocking setup of brokerd
+    ctx: tractor.Context = await mngr.start_service(
-    # service nursery.
+        daemon_name=dname,
-    await Services.start_service_task(
+        ctx_ep=partial(
-        dname,
-        portal,
-
             # signature of target root-task endpoint
             daemon_fixture_ep,
+
+            # passed to daemon_fixture_ep(**kwargs)
             brokername=brokername,
             loglevel=loglevel,
+        ),
+        debug_mode=mngr.debug_mode,
+        loglevel=loglevel,
+        enable_modules=(
+            _data_mods
+            +
+            tractor_kwargs.pop('enable_modules')
+        ),
+        **tractor_kwargs
+    )
+    assert (
+        not ctx.cancel_called
+        and ctx.portal  # parent side
+        and dname in ctx.chan.uid  # subactor is named as desired
     )
     return True
 
@@ -262,8 +274,7 @@ 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/_util.py

@@ -18,10 +18,11 @@
 Handy cross-broker utils.
 
 """
+from __future__ import annotations
 from functools import partial
 
 import json
-import asks
+import httpx
 import logging
 
 from ..log import (
@@ -60,11 +61,11 @@ class NoData(BrokerError):
     def __init__(
         self,
         *args,
-        info: dict,
+        info: dict|None = None,
 
     ) -> None:
         super().__init__(*args)
-        self.info: dict = info
+        self.info: dict|None = info
 
         # when raised, machinery can check if the backend
         # set a "frame size" for doing datetime calcs.
@@ -90,16 +91,18 @@ class DataThrottle(BrokerError):
 
 
 def resproc(
-    resp: asks.response_objects.Response,
+    resp: httpx.Response,
     log: logging.Logger,
     return_json: bool = True,
     log_resp: bool = False,
 
-) -> asks.response_objects.Response:
+) -> httpx.Response:
-    """Process response and return its json content.
+    '''
+    Process response and return its json content.
 
     Raise the appropriate error on non-200 OK responses.
-    """
+
+    '''
     if not resp.status_code == 200:
         raise BrokerError(resp.body)
     try:

piker/brokers/binance/api.py

@@ -25,14 +25,13 @@ from __future__ import annotations
 from collections import ChainMap
 from contextlib import (
     asynccontextmanager as acm,
+    AsyncExitStack,
 )
 from datetime import datetime
 from pprint import pformat
 from typing import (
     Any,
     Callable,
-    Hashable,
-    Sequence,
     Type,
 )
 import hmac
@@ -43,8 +42,7 @@ import trio
 from pendulum import (
     now,
 )
-import asks
+import httpx
-from rapidfuzz import process as fuzzy
 import numpy as np
 
 from piker import config
@@ -54,6 +52,7 @@ from piker.clearing._messages import (
 from piker.accounting import (
     Asset,
     digits_to_dec,
+    MktPair,
 )
 from piker.types import Struct
 from piker.data import (
@@ -69,7 +68,6 @@ from .venues import (
     PAIRTYPES,
     Pair,
     MarketType,
-
     _spot_url,
     _futes_url,
     _testnet_futes_url,
@@ -79,19 +77,18 @@ from .venues import (
 log = get_logger('piker.brokers.binance')
 
 
-def get_config() -> dict:
+def get_config() -> dict[str, Any]:
-
     conf: dict
     path: Path
     conf, path = config.load(
         conf_name='brokers',
         touch_if_dne=True,
     )
-
+    section: dict = conf.get('binance')
-    section = conf.get('binance')
-
     if not section:
-        log.warning(f'No config section found for binance in {path}')
+        log.warning(
+            f'No config section found for binance in {path}'
+        )
         return {}
 
     return section
@@ -147,7 +144,7 @@ def binance_timestamp(
 
 class Client:
     '''
-    Async ReST API client using ``trio`` + ``asks`` B)
+    Async ReST API client using `trio` + `httpx` B)
 
     Supports all of the spot, margin and futures endpoints depending
     on method.
@@ -156,10 +153,17 @@ class Client:
     def __init__(
         self,
 
+        venue_sessions: dict[
+            str,  # venue key
+            tuple[httpx.AsyncClient, str]  # session, eps path
+        ],
+        conf: dict[str, Any],
         # TODO: change this to `Client.[mkt_]venue: MarketType`?
         mkt_mode: MarketType = 'spot',
 
     ) -> None:
+        self.conf = conf
+
         # build out pair info tables for each market type
         # and wrap in a chain-map view for search / query.
         self._spot_pairs: dict[str, Pair] = {}  # spot info table
@@ -186,44 +190,13 @@ class Client:
         # market symbols for use by search. See `.exch_info()`.
         self._pairs: ChainMap[str, Pair] = ChainMap()
 
-        # spot EPs sesh
-        self._sesh = asks.Session(connections=4)
-        self._sesh.base_location: str = _spot_url
-        # spot testnet
-        self._test_sesh: asks.Session = asks.Session(connections=4)
-        self._test_sesh.base_location: str = _testnet_spot_url
-
-        # margin and extended spot endpoints session.
-        self._sapi_sesh = asks.Session(connections=4)
-        self._sapi_sesh.base_location: str = _spot_url
-
-        # futes EPs sesh
-        self._fapi_sesh = asks.Session(connections=4)
-        self._fapi_sesh.base_location: str = _futes_url
-        # futes testnet
-        self._test_fapi_sesh: asks.Session = asks.Session(connections=4)
-        self._test_fapi_sesh.base_location: str = _testnet_futes_url
-
         # global client "venue selection" mode.
         # set this when you want to switch venues and not have to
         # specify the venue for the next request.
         self.mkt_mode: MarketType = mkt_mode
 
-        # per 8
+        # per-mkt-venue API client table
-        self.venue_sesh: dict[
+        self.venue_sesh = venue_sessions
-            str,  # venue key
-            tuple[asks.Session, str]  # session, eps path
-        ] = {
-            'spot': (self._sesh, '/api/v3/'),
-            'spot_testnet': (self._test_sesh, '/fapi/v1/'),
-
-            'margin': (self._sapi_sesh, '/sapi/v1/'),
-
-            'usdtm_futes': (self._fapi_sesh, '/fapi/v1/'),
-            'usdtm_futes_testnet': (self._test_fapi_sesh, '/fapi/v1/'),
-
-            # 'futes_coin': self._dapi,  # TODO
-        }
 
         # lookup for going from `.mkt_mode: str` to the config
         # subsection `key: str`
@@ -238,40 +211,6 @@ class Client:
             'futes': ['usdtm_futes'],
         }
 
-        # for creating API keys see,
-        # https://www.binance.com/en/support/faq/how-to-create-api-keys-on-binance-360002502072
-        self.conf: dict = get_config()
-
-        for key, subconf in self.conf.items():
-            if api_key := subconf.get('api_key', ''):
-                venue_keys: list[str] = self.confkey2venuekeys[key]
-
-                venue_key: str
-                sesh: asks.Session
-                for venue_key in venue_keys:
-                    sesh, _ = self.venue_sesh[venue_key]
-
-                    api_key_header: dict = {
-                        # taken from official:
-                        # https://github.com/binance/binance-futures-connector-python/blob/main/binance/api.py#L47
-                        "Content-Type": "application/json;charset=utf-8",
-
-                        # TODO: prolly should just always query and copy
-                        # in the real latest ver?
-                        "User-Agent": "binance-connector/6.1.6smbz6",
-                        "X-MBX-APIKEY": api_key,
-                    }
-                    sesh.headers.update(api_key_header)
-
-                    # if `.use_tesnet = true` in the config then
-                    # also add headers for the testnet session which
-                    # will be used for all order control
-                    if subconf.get('use_testnet', False):
-                        testnet_sesh, _ = self.venue_sesh[
-                            venue_key + '_testnet'
-                        ]
-                        testnet_sesh.headers.update(api_key_header)
-
     def _mk_sig(
         self,
         data: dict,
@@ -290,7 +229,6 @@ class Client:
                 'to define the creds for auth-ed endpoints!?'
             )
 
-
         # XXX: Info on security and authentification
         # https://binance-docs.github.io/apidocs/#endpoint-security-type
         if not (api_secret := subconf.get('api_secret')):
@@ -319,7 +257,7 @@ class Client:
         params: dict,
 
         method: str = 'get',
-        venue: str | None = None,  # if None use `.mkt_mode` state
+        venue: str|None = None,  # if None use `.mkt_mode` state
         signed: bool = False,
         allow_testnet: bool = False,
 
@@ -330,8 +268,9 @@ class Client:
         - /fapi/v3/ USD-M FUTURES, or
         - /api/v3/ SPOT/MARGIN
 
-        account/market endpoint request depending on either passed in `venue: str`
+        account/market endpoint request depending on either passed in
-        or the current setting `.mkt_mode: str` setting, default `'spot'`.
+        `venue: str` or the current setting `.mkt_mode: str` setting,
+        default `'spot'`.
 
 
         Docs per venue API:
@@ -360,9 +299,6 @@ class Client:
                 venue=venue_key,
             )
 
-        sesh: asks.Session
-        path: str
-
         # Check if we're configured to route order requests to the
         # venue equivalent's testnet.
         use_testnet: bool = False
@@ -387,11 +323,12 @@ class Client:
             # ctl machinery B)
             venue_key += '_testnet'
 
-        sesh, path = self.venue_sesh[venue_key]
+        client: httpx.AsyncClient
-
+        path: str
-        meth: Callable = getattr(sesh, method)
+        client, path = self.venue_sesh[venue_key]
+        meth: Callable = getattr(client, method)
         resp = await meth(
-            path=path + endpoint,
+            url=path + endpoint,
             params=params,
             timeout=float('inf'),
         )
@@ -433,7 +370,15 @@ class Client:
                 item['filters'] = filters
 
             pair_type: Type = PAIRTYPES[venue]
+            try:
                 pair: Pair = pair_type(**item)
+            except Exception as e:
+                e.add_note(
+                    "\nDon't panic, prolly stupid binance changed their symbology schema again..\n"
+                    'Check out their API docs here:\n\n'
+                    'https://binance-docs.github.io/apidocs/spot/en/#exchange-information'
+                )
+                raise
             pair_table[pair.symbol.upper()] = pair
 
             # update an additional top-level-cross-venue-table
@@ -528,7 +473,9 @@ class Client:
 
         '''
         pair_table: dict[str, Pair] = self._venue2pairs[
-            venue or self.mkt_mode
+            venue
+            or
+            self.mkt_mode
         ]
         if (
             expiry
@@ -547,9 +494,9 @@ class Client:
             venues: list[str] = [venue]
 
         # batch per-venue download of all exchange infos
-        async with trio.open_nursery() as rn:
+        async with trio.open_nursery() as tn:
             for ven in venues:
-                rn.start_soon(
+                tn.start_soon(
                     self._cache_pairs,
                     ven,
                 )
@@ -602,11 +549,11 @@ class Client:
 
     ) -> dict[str, Any]:
 
-        fq_pairs: dict = await self.exch_info()
+        fq_pairs: dict[str, Pair] = await self.exch_info()
 
         # TODO: cache this list like we were in
         # `open_symbol_search()`?
-        keys: list[str] = list(fq_pairs)
+        # keys: list[str] = list(fq_pairs)
 
         return match_from_pairs(
             pairs=fq_pairs,
@@ -614,9 +561,20 @@ class Client:
             score_cutoff=50,
         )
 
+    def pair2venuekey(
+        self,
+        pair: Pair,
+    ) -> str:
+        return {
+            'USDTM': 'usdtm_futes',
+            'SPOT': 'spot',
+            # 'COINM': 'coin_futes',
+            # ^-TODO-^ bc someone might want it..?
+        }[pair.venue]
+
     async def bars(
         self,
-        symbol: str,
+        mkt: MktPair,
 
         start_dt: datetime | None = None,
         end_dt: datetime | None = None,
@@ -646,16 +604,20 @@ class Client:
         start_time = binance_timestamp(start_dt)
         end_time = binance_timestamp(end_dt)
 
+        bs_pair: Pair = self._pairs[mkt.bs_fqme.upper()]
+
         # https://binance-docs.github.io/apidocs/spot/en/#kline-candlestick-data
         bars = await self._api(
             'klines',
             params={
-                'symbol': symbol.upper(),
+                # NOTE: always query using their native symbology!
+                'symbol': mkt.bs_mktid.upper(),
                 'interval': '1m',
                 'startTime': start_time,
                 'endTime': end_time,
                 'limit': limit
             },
+            venue=self.pair2venuekey(bs_pair),
             allow_testnet=False,
         )
         new_bars: list[tuple] = []
@@ -972,17 +934,148 @@ class Client:
         await self.close_listen_key(key)
 
 
-@acm
+_venue_urls: dict[str, str] = {
-async def get_client() -> Client:
+    'spot': (
+        _spot_url,
+        '/api/v3/',
+    ),
+    'spot_testnet': (
+        _testnet_spot_url,
+        '/fapi/v1/'
+    ),
+    # margin and extended spot endpoints session.
+    # TODO: did this ever get implemented fully?
+    # 'margin': (
+    #     _spot_url,
+    #     '/sapi/v1/'
+    # ),
 
-    client = Client()
+    'usdtm_futes': (
-    await client.exch_info()
+        _futes_url,
-    log.info(
+        '/fapi/v1/',
-        f'{client} in {client.mkt_mode} mode: caching exchange infos..\n'
+    ),
-        'Cached multi-market pairs:\n'
+
-        f'spot: {len(client._spot_pairs)}\n'
+    'usdtm_futes_testnet': (
-        f'usdtm_futes: {len(client._ufutes_pairs)}\n'
+        _testnet_futes_url,
-        f'Total: {len(client._pairs)}\n'
+        '/fapi/v1/',
+    ),
+
+    # TODO: for anyone who actually needs it ;P
+    # 'coin_futes': ()
+}
+
+
+def init_api_keys(
+    client: Client,
+    conf: dict[str, Any],
+) -> None:
+    '''
+    Set up per-venue API keys each http client according to the user's
+    `brokers.conf`.
+
+    For ex, to use spot-testnet and live usdt futures APIs:
+
+    ```toml
+        [binance]
+        # spot test net
+        spot.use_testnet = true
+        spot.api_key = '<spot_api_key_from_binance_account>'
+        spot.api_secret = '<spot_api_key_password>'
+
+        # futes live
+        futes.use_testnet = false
+        accounts.usdtm = 'futes'
+        futes.api_key = '<futes_api_key_from_binance>'
+        futes.api_secret = '<futes_api_key_password>''
+
+        # if uncommented will use the built-in paper engine and not
+        # connect to `binance` API servers for order ctl.
+        # accounts.paper = 'paper'
+    ```
+
+    '''
+    for key, subconf in conf.items():
+        if api_key := subconf.get('api_key', ''):
+            venue_keys: list[str] = client.confkey2venuekeys[key]
+
+            venue_key: str
+            client: httpx.AsyncClient
+            for venue_key in venue_keys:
+                client, _ = client.venue_sesh[venue_key]
+
+                api_key_header: dict = {
+                    # taken from official:
+                    # https://github.com/binance/binance-futures-connector-python/blob/main/binance/api.py#L47
+                    "Content-Type": "application/json;charset=utf-8",
+
+                    # TODO: prolly should just always query and copy
+                    # in the real latest ver?
+                    "User-Agent": "binance-connector/6.1.6smbz6",
+                    "X-MBX-APIKEY": api_key,
+                }
+                client.headers.update(api_key_header)
+
+                # if `.use_tesnet = true` in the config then
+                # also add headers for the testnet session which
+                # will be used for all order control
+                if subconf.get('use_testnet', False):
+                    testnet_sesh, _ = client.venue_sesh[
+                        venue_key + '_testnet'
+                    ]
+                    testnet_sesh.headers.update(api_key_header)
+
+
+@acm
+async def get_client(
+    mkt_mode: MarketType = 'spot',
+) -> Client:
+    '''
+    Construct an single `piker` client which composes multiple underlying venue
+    specific API clients both for live and test networks.
+
+    '''
+    venue_sessions: dict[
+        str,  # venue key
+        tuple[httpx.AsyncClient, str]  # session, eps path
+    ] = {}
+    async with AsyncExitStack() as client_stack:
+        for name, (base_url, path) in _venue_urls.items():
+            api: httpx.AsyncClient = await client_stack.enter_async_context(
+                httpx.AsyncClient(
+                    base_url=base_url,
+                    # headers={},
+
+                    # TODO: is there a way to numerate this?
+                    # https://www.python-httpx.org/advanced/clients/#why-use-a-client
+                    # connections=4
+                )
+            )
+            venue_sessions[name] = (
+                api,
+                path,
             )
 
+        conf: dict[str, Any] = get_config()
+        # for creating API keys see,
+        # https://www.binance.com/en/support/faq/how-to-create-api-keys-on-binance-360002502072
+        client = Client(
+            venue_sessions=venue_sessions,
+            conf=conf,
+            mkt_mode=mkt_mode,
+        )
+        init_api_keys(
+            client=client,
+            conf=conf,
+        )
+        fq_pairs: dict[str, Pair] = await client.exch_info()
+        assert fq_pairs
+        log.info(
+            f'Loaded multi-venue `Client` in mkt_mode={client.mkt_mode!r}\n\n'
+            f'Symbology Summary:\n'
+            f'------ - ------\n'
+            f'spot: {len(client._spot_pairs)}\n'
+            f'usdtm_futes: {len(client._ufutes_pairs)}\n'
+            '------ - ------\n'
+            f'total: {len(client._pairs)}\n'
+        )
         yield client

piker/brokers/binance/broker.py

@@ -264,15 +264,20 @@ async def open_trade_dialog(
     # do a open_symcache() call.. though maybe we can hide
     # this in a new async version of open_account()?
     async with open_cached_client('binance') as client:
-        subconf: dict = client.conf[venue_name]
+        subconf: dict|None = client.conf.get(venue_name)
-        use_testnet = subconf.get('use_testnet', False)
 
         # XXX: if no futes.api_key or spot.api_key has been set we
         # always fall back to the paper engine!
-        if not subconf.get('api_key'):
+        if (
+            not subconf
+            or
+            not subconf.get('api_key')
+        ):
             await ctx.started('paper')
             return
 
+        use_testnet: bool = subconf.get('use_testnet', False)
+
     async with (
         open_cached_client('binance') as client,
     ):

piker/brokers/binance/feed.py

@@ -48,6 +48,7 @@ import tractor
 
 from piker.brokers import (
     open_cached_client,
+    NoData,
 )
 from piker._cacheables import (
     async_lifo_cache,
@@ -252,24 +253,30 @@ async def open_history_client(
             else:
                 client.mkt_mode = 'spot'
 
-            # NOTE: always query using their native symbology!
+            array: np.ndarray = await client.bars(
-            mktid: str = mkt.bs_mktid
+                mkt=mkt,
-            array = await client.bars(
-                mktid,
                 start_dt=start_dt,
                 end_dt=end_dt,
             )
+            if array.size == 0:
+                raise NoData(
+                    f'No frame for {start_dt} -> {end_dt}\n'
+                )
+
             times = array['time']
-            if (
+            if not times.any():
-                end_dt is None
+                raise ValueError(
-            ):
+                    'Bad frame with null-times?\n\n'
-                inow = round(time.time())
+                    f'{times}'
+                )
+
+            if end_dt is None:
+                inow: int = round(time.time())
                 if (inow - times[-1]) > 60:
                     await tractor.pause()
 
             start_dt = from_timestamp(times[0])
             end_dt = from_timestamp(times[-1])
-
             return array, start_dt, end_dt
 
         yield get_ohlc, {'erlangs': 3, 'rate': 3}

piker/brokers/binance/venues.py

@@ -137,10 +137,12 @@ class SpotPair(Pair, frozen=True):
     quoteOrderQtyMarketAllowed: bool
     isSpotTradingAllowed: bool
     isMarginTradingAllowed: bool
+    otoAllowed: bool
 
     defaultSelfTradePreventionMode: str
     allowedSelfTradePreventionModes: list[str]
     permissions: list[str]
+    permissionSets: list[list[str]]
 
     # NOTE: see `.data._symcache.SymbologyCache.load()` for why
     ns_path: str = 'piker.brokers.binance:SpotPair'

piker/brokers/kraken/api.py

@@ -27,8 +27,8 @@ from typing import (
 )
 import time
 
+import httpx
 import pendulum
-import asks
 import numpy as np
 import urllib.parse
 import hashlib
@@ -60,6 +60,11 @@ log = get_logger('piker.brokers.kraken')
 
 # <uri>/<version>/
 _url = 'https://api.kraken.com/0'
+
+_headers: dict[str, str] = {
+    'User-Agent': 'krakenex/2.1.0 (+https://github.com/veox/python3-krakenex)'
+}
+
 # TODO: this is the only backend providing this right?
 # in which case we should drop it from the defaults and
 # instead make a custom fields descr in this module!
@@ -135,16 +140,15 @@ class Client:
     def __init__(
         self,
         config: dict[str, str],
+        httpx_client: httpx.AsyncClient,
+
         name: str = '',
         api_key: str = '',
         secret: str = ''
     ) -> None:
-        self._sesh = asks.Session(connections=4)
+
-        self._sesh.base_location = _url
+        self._sesh: httpx.AsyncClient = httpx_client
-        self._sesh.headers.update({
+
-            'User-Agent':
-                'krakenex/2.1.0 (+https://github.com/veox/python3-krakenex)'
-        })
         self._name = name
         self._api_key = api_key
         self._secret = secret
@@ -166,10 +170,9 @@ class Client:
         method: str,
         data: dict,
     ) -> dict[str, Any]:
-        resp = await self._sesh.post(
+        resp: httpx.Response = await self._sesh.post(
-            path=f'/public/{method}',
+            url=f'/public/{method}',
             json=data,
-            timeout=float('inf')
         )
         return resproc(resp, log)
 
@@ -180,18 +183,18 @@ class Client:
         uri_path: str
     ) -> dict[str, Any]:
         headers = {
-            'Content-Type':
+            'Content-Type': 'application/x-www-form-urlencoded',
-                'application/x-www-form-urlencoded',
+            'API-Key': self._api_key,
-            'API-Key':
+            'API-Sign': get_kraken_signature(
-                self._api_key,
+                uri_path,
-            'API-Sign':
+                data,
-                get_kraken_signature(uri_path, data, self._secret)
+                self._secret,
+            ),
         }
-        resp = await self._sesh.post(
+        resp: httpx.Response = await self._sesh.post(
-            path=f'/private/{method}',
+            url=f'/private/{method}',
             data=data,
             headers=headers,
-            timeout=float('inf')
         )
         return resproc(resp, log)
 
@@ -665,10 +668,19 @@ class Client:
 @acm
 async def get_client() -> Client:
 
-    conf = get_config()
+    conf: dict[str, Any] = get_config()
+    async with httpx.AsyncClient(
+        base_url=_url,
+        headers=_headers,
+
+        # TODO: is there a way to numerate this?
+        # https://www.python-httpx.org/advanced/clients/#why-use-a-client
+        # connections=4
+    ) as trio_client:
         if conf:
             client = Client(
                 conf,
+                httpx_client=trio_client,
 
                 # TODO: don't break these up and just do internal
                 # conf lookups instead..
@@ -677,7 +689,10 @@ async def get_client() -> Client:
                 secret=conf['secret']
             )
         else:
-        client = Client({})
+            client = Client(
+                conf={},
+                httpx_client=trio_client,
+            )
 
         # at startup, load all symbols, and asset info in
         # batch requests.

piker/brokers/kraken/broker.py

@@ -612,18 +612,18 @@ async def open_trade_dialog(
 
                 # enter relay loop
                 await handle_order_updates(
-                    client,
+                    client=client,
-                    ws,
+                    ws=ws,
-                    stream,
+                    ws_stream=stream,
-                    ems_stream,
+                    ems_stream=ems_stream,
-                    apiflows,
+                    apiflows=apiflows,
-                    ids,
+                    ids=ids,
-                    reqids2txids,
+                    reqids2txids=reqids2txids,
-                    acnt,
+                    acnt=acnt,
-                    api_trans,
+                    ledger=ledger,
-                    acctid,
+                    acctid=acctid,
-                    acc_name,
+                    acc_name=acc_name,
-                    token,
+                    token=token,
                 )
 
 
@@ -639,7 +639,8 @@ async def handle_order_updates(
 
     # transaction records which will be updated
     # on new trade clearing events (aka order "fills")
-    ledger_trans: dict[str, Transaction],
+    ledger: TransactionLedger,
+    # ledger_trans: dict[str, Transaction],
     acctid: str,
     acc_name: str,
     token: str,
@@ -699,7 +700,8 @@ async def handle_order_updates(
                     # if tid not in ledger_trans
                 }
                 for tid, trade in trades.items():
-                    assert tid not in ledger_trans
+                    # assert tid not in ledger_trans
+                    assert tid not in ledger
                     txid = trade['ordertxid']
                     reqid = trade.get('userref')
 
@@ -747,11 +749,17 @@ async def handle_order_updates(
                     client,
                     api_name_set='wsname',
                 )
-                ppmsgs = trades2pps(
+                ppmsgs: list[BrokerdPosition] = trades2pps(
-                    acnt,
+                    acnt=acnt,
-                    acctid,
+                    ledger=ledger,
-                    new_trans,
+                    acctid=acctid,
+                    new_trans=new_trans,
                 )
+                # ppmsgs = trades2pps(
+                #     acnt,
+                #     acctid,
+                #     new_trans,
+                # )
                 for pp_msg in ppmsgs:
                     await ems_stream.send(pp_msg)
 

piker/brokers/kucoin.py

@@ -16,10 +16,9 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 '''
-Kucoin broker backend
+Kucoin cex API backend.
 
 '''
-
 from contextlib import (
     asynccontextmanager as acm,
     aclosing,
@@ -42,7 +41,7 @@ import wsproto
 from uuid import uuid4
 
 from trio_typing import TaskStatus
-import asks
+import httpx
 from bidict import bidict
 import numpy as np
 import pendulum
@@ -212,8 +211,12 @@ def get_config() -> BrokerConfig | None:
 
 class Client:
 
-    def __init__(self) -> None:
+    def __init__(
-        self._config: BrokerConfig | None = get_config()
+        self,
+        httpx_client: httpx.AsyncClient,
+    ) -> None:
+        self._http: httpx.AsyncClient = httpx_client
+        self._config: BrokerConfig|None = get_config()
         self._pairs: dict[str, KucoinMktPair] = {}
         self._fqmes2mktids: bidict[str, str] = bidict()
         self._bars: list[list[float]] = []
@@ -227,18 +230,24 @@ class Client:
 
     ) -> dict[str, str | bytes]:
         '''
-        Generate authenticated request headers
+        Generate authenticated request headers:
+
         https://docs.kucoin.com/#authentication
+        https://www.kucoin.com/docs/basic-info/connection-method/authentication/creating-a-request
+        https://www.kucoin.com/docs/basic-info/connection-method/authentication/signing-a-message
 
         '''
-
         if not self._config:
             raise ValueError(
-                'No config found when trying to send authenticated request')
+                'No config found when trying to send authenticated request'
+            )
 
         str_to_sign = (
             str(int(time.time() * 1000))
-            + action + f'/api/{api}/{endpoint.lstrip("/")}'
+            +
+            action
+            +
+            f'/api/{api}/{endpoint.lstrip("/")}'
         )
 
         signature = base64.b64encode(
@@ -249,6 +258,7 @@ class Client:
             ).digest()
         )
 
+        # TODO: can we cache this between calls?
         passphrase = base64.b64encode(
             hmac.new(
                 self._config.key_secret.encode('utf-8'),
@@ -270,8 +280,10 @@ class Client:
         self,
         action: Literal['POST', 'GET'],
         endpoint: str,
+
         api: str = 'v2',
         headers: dict = {},
+
     ) -> Any:
         '''
         Generic request wrapper for Kucoin API
@@ -284,14 +296,19 @@ class Client:
                 api,
             )
 
-        api_url = f'https://api.kucoin.com/api/{api}/{endpoint}'
+        req_meth: Callable = getattr(
-
+            self._http,
-        res = await asks.request(action, api_url, headers=headers)
+            action.lower(),
-
+        )
-        json = res.json()
+        res = await req_meth(
-        if 'data' in json:
+            url=f'/{api}/{endpoint}',
-            return json['data']
+            headers=headers,
+        )
+        json: dict = res.json()
+        if (data := json.get('data')) is not None:
+            return data
         else:
+            api_url: str = self._http.base_url
             log.error(
                 f'Error making request to {api_url} ->\n'
                 f'{pformat(res)}'
@@ -311,7 +328,7 @@ class Client:
         '''
         token_type = 'private' if private else 'public'
         try:
-            data: dict[str, Any] | None = await self._request(
+            data: dict[str, Any]|None = await self._request(
                 'POST',
                 endpoint=f'bullet-{token_type}',
                 api='v1'
@@ -349,8 +366,8 @@ class Client:
             currencies: dict[str, Currency] = {}
             entries: list[dict] = await self._request(
                 'GET',
-                api='v1',
                 endpoint='currencies',
+                api='v1',
             )
             for entry in entries:
                 curr = Currency(**entry).copy()
@@ -366,7 +383,10 @@ class Client:
         dict[str, KucoinMktPair],
         bidict[str, KucoinMktPair],
     ]:
-        entries = await self._request('GET', 'symbols')
+        entries = await self._request(
+            'GET',
+            endpoint='symbols',
+        )
         log.info(f' {len(entries)} Kucoin market pairs fetched')
 
         pairs: dict[str, KucoinMktPair] = {}
@@ -567,10 +587,18 @@ def fqme_to_kucoin_sym(
 
 @acm
 async def get_client() -> AsyncGenerator[Client, None]:
-    client = Client()
+    '''
+    Load an API `Client` preconfigured from user settings
 
-    async with trio.open_nursery() as n:
+    '''
-        n.start_soon(client.get_mkt_pairs)
+    async with (
+        httpx.AsyncClient(
+            base_url=f'https://api.kucoin.com/api',
+        ) as trio_client,
+    ):
+        client = Client(httpx_client=trio_client)
+        async with trio.open_nursery() as tn:
+            tn.start_soon(client.get_mkt_pairs)
             await client.get_currencies()
 
         yield client

piker/cli/__init__.py

@@ -335,7 +335,7 @@ def services(config, tl, ports):
                 name='service_query',
                 loglevel=config['loglevel'] if tl else None,
             ),
-            tractor.get_arbiter(
+            tractor.get_registry(
                 host=host,
                 port=ports[0]
             ) as portal

piker/data/_sampling.py

@@ -25,6 +25,7 @@ from collections import (
     defaultdict,
 )
 from contextlib import asynccontextmanager as acm
+from functools import partial
 import time
 from typing import (
     Any,
@@ -42,7 +43,7 @@ from tractor.trionics import (
     maybe_open_nursery,
 )
 import trio
-from trio_typing import TaskStatus
+from trio import TaskStatus
 
 from .ticktools import (
     frame_ticks,
@@ -70,6 +71,7 @@ if TYPE_CHECKING:
 _default_delay_s: float = 1.0
 
 
+# TODO: use new `tractor.singleton_acm` API for this!
 class Sampler:
     '''
     Global sampling engine registry.
@@ -79,9 +81,9 @@ class Sampler:
 
     This non-instantiated type is meant to be a singleton within
     a `samplerd` actor-service spawned once by the user wishing to
-    time-step-sample (real-time) quote feeds, see
+    time-step-sample a (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
@@ -375,7 +377,10 @@ async def register_with_sampler(
                 assert Sampler.ohlcv_shms
 
             # unblock caller
-            await ctx.started(set(Sampler.ohlcv_shms.keys()))
+            await ctx.started(
+                # XXX bc msgpack only allows one array type!
+                list(Sampler.ohlcv_shms.keys())
+            )
 
             if open_index_stream:
                 try:
@@ -419,7 +424,6 @@ async def register_with_sampler(
 
 
 async def spawn_samplerd(
-
     loglevel: str | None = None,
     **extra_tractor_kwargs
 
@@ -429,7 +433,10 @@ async def spawn_samplerd(
     update and increment count write and stream broadcasting.
 
     '''
-    from piker.service import Services
+    from piker.service import (
+        get_service_mngr,
+        ServiceMngr,
+    )
 
     dname = 'samplerd'
     log.info(f'Spawning `{dname}`')
@@ -437,26 +444,33 @@ async def spawn_samplerd(
     # singleton lock creation of ``samplerd`` since we only ever want
     # one daemon per ``pikerd`` proc tree.
     # TODO: make this built-into the service api?
-    async with Services.locks[dname + '_singleton']:
+    mngr: ServiceMngr = get_service_mngr()
+    already_started: bool = dname in mngr.service_tasks
 
-        if dname not in Services.service_tasks:
+    async with mngr._locks[dname + '_singleton']:
+        ctx: Context = await mngr.start_service(
+            daemon_name=dname,
+            ctx_ep=partial(
+                register_with_sampler,
+                period_s=1,
+                sub_for_broadcasts=False,
+            ),
+            debug_mode=mngr.debug_mode,  # set by pikerd flag
 
-            portal = await Services.actor_n.start_actor(
+            # proxy-through to tractor
-                dname,
             enable_modules=[
                 'piker.data._sampling',
             ],
             loglevel=loglevel,
-                debug_mode=Services.debug_mode,  # set by pikerd flag
             **extra_tractor_kwargs
         )
-
+        if not already_started:
-            await Services.start_service_task(
+            assert (
-                dname,
+                ctx
-                portal,
+                and
-                register_with_sampler,
+                ctx.portal
-                period_s=1,
+                and
-                sub_for_broadcasts=False,
+                not ctx.cancel_called
             )
             return True
 
@@ -889,6 +903,7 @@ async def uniform_rate_send(
             # to consumers which crash or lose network connection.
             # I.e. we **DO NOT** want to crash and propagate up to
             # ``pikerd`` these kinds of errors!
+            trio.EndOfChannel,
             trio.ClosedResourceError,
             trio.BrokenResourceError,
             ConnectionResetError,

piker/service/__init__.py

@@ -30,7 +30,11 @@ Actor runtime primtives and (distributed) service APIs for,
   => TODO: maybe to (re)move elsewhere?
 
 '''
-from ._mngr import Services as Services
+from ._mngr import (
+    get_service_mngr as get_service_mngr,
+    open_service_mngr as open_service_mngr,
+    ServiceMngr as ServiceMngr,
+)
 from ._registry import (
     _tractor_kwargs as _tractor_kwargs,
     _default_reg_addr as _default_reg_addr,

piker/service/_actor_runtime.py

@@ -21,7 +21,6 @@
 from __future__ import annotations
 import os
 from typing import (
-    Optional,
     Any,
     ClassVar,
 )
@@ -30,13 +29,13 @@ from contextlib import (
 )
 
 import tractor
-import trio
 
 from ._util import (
     get_console_log,
 )
 from ._mngr import (
-    Services,
+    open_service_mngr,
+    ServiceMngr,
 )
 from ._registry import (  # noqa
     _tractor_kwargs,
@@ -59,7 +58,7 @@ async def open_piker_runtime(
     registry_addrs: list[tuple[str, int]] = [],
 
     enable_modules: list[str] = [],
-    loglevel: Optional[str] = None,
+    loglevel: str|None = None,
 
     # XXX NOTE XXX: you should pretty much never want debug mode
     # for data daemons when running in production.
@@ -69,7 +68,7 @@ async def open_piker_runtime(
     # and spawn the service tree distributed per that.
     start_method: str = 'trio',
 
-    tractor_runtime_overrides: dict | None = None,
+    tractor_runtime_overrides: dict|None = None,
     **tractor_kwargs,
 
 ) -> tuple[
@@ -119,6 +118,10 @@ async def open_piker_runtime(
                 # spawn other specialized daemons I think?
                 enable_modules=enable_modules,
 
+                # TODO: how to configure this?
+                # keep it on by default if debug mode is set?
+                # maybe_enable_greenback=debug_mode,
+
                 **tractor_kwargs,
             ) as actor,
 
@@ -167,12 +170,13 @@ async def open_pikerd(
 
     **kwargs,
 
-) -> Services:
+) -> ServiceMngr:
     '''
-    Start a root piker daemon with an indefinite lifetime.
+    Start a root piker daemon actor (aka `pikerd`) with an indefinite
+    lifetime.
 
-    A root actor nursery is created which can be used to create and keep
+    A root actor-nursery is created which can be used to spawn and
-    alive underling services (see below).
+    supervise underling service sub-actors (see below).
 
     '''
     # NOTE: for the root daemon we always enable the root
@@ -199,8 +203,6 @@ async def open_pikerd(
             root_actor,
             reg_addrs,
         ),
-        tractor.open_nursery() as actor_nursery,
-        trio.open_nursery() as service_nursery,
     ):
         for addr in reg_addrs:
             if addr not in root_actor.accept_addrs:
@@ -209,25 +211,17 @@ async def open_pikerd(
                     'Maybe you have another daemon already running?'
                 )
 
-        # assign globally for future daemon/task creation
+        mngr: ServiceMngr
-        Services.actor_n = actor_nursery
+        async with open_service_mngr(
-        Services.service_n = service_nursery
+            debug_mode=debug_mode,
-        Services.debug_mode = debug_mode
+        ) as mngr:
-
+            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: Optional[str] = None,
+#     loglevel: str|None = None,
 #     **kwargs,
 
 # ) -> None:
@@ -256,7 +250,7 @@ async def maybe_open_pikerd(
     loglevel: str | None = None,
     **kwargs,
 
-) -> tractor._portal.Portal | ClassVar[Services]:
+) -> tractor._portal.Portal | ClassVar[ServiceMngr]:
     '''
     If no ``pikerd`` daemon-root-actor can be found start it and
     yield up (we should probably figure out returning a portal to self

piker/service/_ahab.py

@@ -49,7 +49,7 @@ from requests.exceptions import (
     ReadTimeout,
 )
 
-from ._mngr import Services
+from ._mngr import ServiceMngr
 from ._util import (
     log,  # sub-sys logger
     get_console_log,
@@ -453,7 +453,7 @@ async def open_ahabd(
 
 @acm
 async def start_ahab_service(
-    services: Services,
+    services: ServiceMngr,
     service_name: str,
 
     # endpoint config passed as **kwargs
@@ -549,7 +549,8 @@ async def start_ahab_service(
         log.warning('Failed to cancel root permsed container')
 
     except (
-        trio.MultiError,
+        # trio.MultiError,
+        ExceptionGroup,
     ) as err:
         for subexc in err.exceptions:
             if isinstance(subexc, PermissionError):

piker/service/_daemon.py

@@ -26,14 +26,17 @@ from typing import (
 from contextlib import (
     asynccontextmanager as acm,
 )
+from collections import defaultdict
 
 import tractor
+import trio
 
 from ._util import (
     log,  # sub-sys logger
 )
 from ._mngr import (
-    Services,
+    get_service_mngr,
+    ServiceMngr,
 )
 from ._actor_runtime import maybe_open_pikerd
 from ._registry import find_service
@@ -41,15 +44,14 @@ from ._registry import find_service
 
 @acm
 async def maybe_spawn_daemon(
-
     service_name: str,
     service_task_target: Callable,
-
     spawn_args: dict[str, Any],
 
     loglevel: str | None = None,
     singleton: bool = False,
 
+    _locks = defaultdict(trio.Lock),
     **pikerd_kwargs,
 
 ) -> tractor.Portal:
@@ -67,7 +69,7 @@ async def maybe_spawn_daemon(
     '''
     # serialize access to this section to avoid
     # 2 or more tasks racing to create a daemon
-    lock = Services.locks[service_name]
+    lock = _locks[service_name]
     await lock.acquire()
 
     async with find_service(
@@ -102,6 +104,12 @@ async def maybe_spawn_daemon(
         # service task for that actor.
         started: bool
         if pikerd_portal is None:
+
+            # await tractor.pause()
+            if tractor_kwargs.get('debug_mode', False):
+                from tractor.devx._debug import maybe_init_greenback
+                await maybe_init_greenback()
+
             started = await service_task_target(
                 loglevel=loglevel,
                 **spawn_args,
@@ -132,7 +140,65 @@ async def maybe_spawn_daemon(
         async with tractor.wait_for_actor(service_name) as portal:
             lock.release()
             yield portal
-            await portal.cancel_actor()
+            # --- ---- ---
+            # XXX NOTE XXX
+            # --- ---- ---
+            # DO NOT PUT A `portal.cancel_actor()` here (as was prior)!
+            #
+            # Doing so will cause an "out-of-band" ctxc
+            # (`tractor.ContextCancelled`) to be raised inside the
+            # `ServiceMngr.open_context_in_task()`'s call to
+            # `ctx.wait_for_result()` AND the internal self-ctxc
+            # "graceful capture" WILL NOT CATCH IT!
+            #
+            # This can cause certain types of operations to raise
+            # that ctxc BEFORE THEY `return`, resulting in
+            # a "false-negative" ctxc being raised when really
+            # nothing actually failed, other then our semantic
+            # "failure" to suppress an expected, graceful,
+            # self-cancel scenario..
+            #
+            # bUt wHy duZ It WorK lIKe dis..
+            # ------------------------------
+            # from the perspective of the `tractor.Context` this
+            # cancel request was conducted "out of band" since
+            # `Context.cancel()` was never called and thus the
+            # `._cancel_called: bool` was never set. Despite the
+            # remote `.canceller` being set to `pikerd` (i.e. the
+            # same `Actor.uid` of the raising service-mngr task) the
+            # service-task's ctx itself was never marked as having
+            # requested cancellation and thus still raises the ctxc
+            # bc it was unaware of any such request.
+            #
+            # How to make grokin these cases easier tho?
+            # ------------------------------------------
+            # Because `Portal.cancel_actor()` was called it requests
+            # "full-`Actor`-runtime-cancellation" of it's peer
+            # process which IS NOT THE SAME as a single inter-actor
+            # RPC task cancelling its local context with a remote
+            # peer `Task` in that same peer process.
+            #
+            # ?TODO? It might be better if we do one (or all) of the
+            # following:
+            #
+            # -[ ] at least set a special message for the
+            #    `ContextCancelled` when raised locally by the
+            #    unaware ctx task such that we check for the
+            #    `.canceller` being *our `Actor`* and in the case
+            #    where `Context._cancel_called == False` we specially
+            #    note that this is likely an "out-of-band"
+            #    runtime-cancel request triggered by some call to
+            #    `Portal.cancel_actor()`, possibly even reporting the
+            #    exact LOC of that caller by tracking it inside our
+            #    portal-type?
+            # -[ ] possibly add another field `ContextCancelled` like
+            #    maybe a,
+            #    `.request_type: Literal['os', 'proc', 'actor',
+            #    'ctx']` type thing which would allow immediately
+            #    being able to tell what kind of cancellation caused
+            #    the unexpected ctxc?
+            # -[ ] REMOVE THIS COMMENT, once we've settled on how to
+            #     better augment `tractor` to be more explicit on this!
 
 
 async def spawn_emsd(
@@ -147,26 +213,25 @@ async def spawn_emsd(
     """
     log.info('Spawning emsd')
 
-    portal = await Services.actor_n.start_actor(
+    smngr: ServiceMngr = get_service_mngr()
+    portal = await smngr.an.start_actor(
         'emsd',
         enable_modules=[
             'piker.clearing._ems',
             'piker.clearing._client',
         ],
         loglevel=loglevel,
-        debug_mode=Services.debug_mode,  # set by pikerd flag
+        debug_mode=smngr.debug_mode,  # set by pikerd flag
         **extra_tractor_kwargs
     )
 
     # non-blocking setup of clearing service
     from ..clearing._ems import _setup_persistent_emsd
 
-    await Services.start_service_task(
+    await smngr.start_service_ctx(
-        'emsd',
+        name='emsd',
-        portal,
+        portal=portal,
-
+        ctx_fn=_setup_persistent_emsd,
-        # signature of target root-task endpoint
-        _setup_persistent_emsd,
         loglevel=loglevel,
     )
     return True

piker/service/_mngr.py

@@ -18,148 +18,36 @@
 daemon-service management API.
 
 """
-from collections import defaultdict
+from contextlib import (
-from typing import (
+    asynccontextmanager as acm,
-    Callable,
-    Any,
 )
 
-import trio
-from trio_typing import TaskStatus
 import tractor
-from tractor import (
+from tractor.hilevel import (
-    current_actor,
+    ServiceMngr,
-    ContextCancelled,
+    # open_service_mngr as _open_service_mngr,
-    Context,
+    get_service_mngr as get_service_mngr,
-    Portal,
 )
+# 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
 
-from ._util import (
+Services: ServiceMngr|None = None
-    log,  # sub-sys logger
-)
 
+@acm
+async def open_service_mngr(
+    **kwargs,
+) -> ServiceMngr:
 
-# TODO: we need remote wrapping and a general soln:
+    global Services
-# - factor this into a ``tractor.highlevel`` extension # pack for the
+    async with tractor.hilevel.open_service_mngr(
-#   library.
+        **kwargs,
-# - wrap a "remote api" wherein you can get a method proxy
+    ) as mngr:
-#   to the pikerd actor for starting services remotely!
+        # Services = proxy(mngr)
-# - prolly rename this to ActorServicesNursery since it spawns
+        Services = mngr
-#   new actors and supervises them to completion?
+        yield mngr
-class Services:
+        Services = None
-
-    actor_n: tractor._supervise.ActorNursery
-    service_n: trio.Nursery
-    debug_mode: bool  # tractor sub-actor debug mode flag
-    service_tasks: dict[
-        str,
-        tuple[
-            trio.CancelScope,
-            Portal,
-            trio.Event,
-        ]
-    ] = {}
-    locks = defaultdict(trio.Lock)
-
-    @classmethod
-    async def start_service_task(
-        self,
-        name: str,
-        portal: Portal,
-        target: Callable,
-        allow_overruns: bool = False,
-        **ctx_kwargs,
-
-    ) -> (trio.CancelScope, Context):
-        '''
-        Open a context in a service sub-actor, add to a stack
-        that gets unwound at ``pikerd`` teardown.
-
-        This allows for allocating long-running sub-services in our main
-        daemon and explicitly controlling their lifetimes.
-
-        '''
-        async def open_context_in_task(
-            task_status: TaskStatus[
-                tuple[
-                    trio.CancelScope,
-                    trio.Event,
-                    Any,
-                ]
-            ] = trio.TASK_STATUS_IGNORED,
-
-        ) -> Any:
-
-            with trio.CancelScope() as cs:
-
-                async with portal.open_context(
-                    target,
-                    allow_overruns=allow_overruns,
-                    **ctx_kwargs,
-
-                ) as (ctx, first):
-
-                    # unblock once the remote context has started
-                    complete = trio.Event()
-                    task_status.started((cs, complete, first))
-                    log.info(
-                        f'`pikerd` service {name} started with value {first}'
-                    )
-                    try:
-                        # wait on any context's return value
-                        # and any final portal result from the
-                        # sub-actor.
-                        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 (await portal.result(), ctx_res)
-                    except ContextCancelled as ctxe:
-                        canceller: tuple[str, str] = ctxe.canceller
-                        our_uid: tuple[str, str] = current_actor().uid
-                        if (
-                            canceller != portal.channel.uid
-                            and
-                            canceller != our_uid
-                        ):
-                            log.cancel(
-                                f'Actor-service {name} was remotely cancelled?\n'
-                                f'remote canceller: {canceller}\n'
-                                f'Keeping {our_uid} alive, ignoring sub-actor cancel..\n'
-                            )
-                        else:
-                            raise
-
-
-
-                    finally:
-                        await portal.cancel_actor()
-                        complete.set()
-                        self.service_tasks.pop(name)
-
-        cs, complete, first = await self.service_n.start(open_context_in_task)
-
-        # store the cancel scope and portal for later cancellation or
-        # retstart if needed.
-        self.service_tasks[name] = (cs, portal, complete)
-
-        return cs, first
-
-    @classmethod
-    async def cancel_service(
-        self,
-        name: str,
-
-    ) -> Any:
-        '''
-        Cancel the service task and actor for the given ``name``.
-
-        '''
-        log.info(f'Cancelling `pikerd` service {name}')
-        cs, portal, complete = self.service_tasks[name]
-        cs.cancel()
-        await complete.wait()
-        assert name not in self.service_tasks, \
-            f'Serice task for {name} not terminated?'

piker/service/elastic.py

@@ -21,11 +21,13 @@ from typing import (
     TYPE_CHECKING,
 )
 
+# TODO: oof, needs to be changed to `httpx`!
 import asks
 
 if TYPE_CHECKING:
     import docker
     from ._ahab import DockerContainer
+    from . import ServiceMngr
 
 from ._util import log  # sub-sys logger
 from ._util import (
@@ -127,7 +129,7 @@ def start_elasticsearch(
 
 @acm
 async def start_ahab_daemon(
-    service_mngr: Services,
+    service_mngr: ServiceMngr,
     user_config: dict | None = None,
     loglevel: str | None = None,
 

piker/service/marketstore.py

@@ -53,7 +53,7 @@ import pendulum
 # import purerpc
 
 from ..data.feed import maybe_open_feed
-from . import Services
+from . import ServiceMngr
 from ._util import (
     log,  # sub-sys logger
     get_console_log,
@@ -233,7 +233,7 @@ def start_marketstore(
 
 @acm
 async def start_ahab_daemon(
-    service_mngr: Services,
+    service_mngr: ServiceMngr,
     user_config: dict | None = None,
     loglevel: str | None = None,
 

piker/tsp/__init__.py

@@ -458,13 +458,15 @@ async def start_backfill(
                     'bf_until <- last_start_dt:\n'
                     f'{backfill_until_dt} <- {last_start_dt}\n'
                 )
-
+                # UGH: what's a better way?
-                # ugh, what's a better way?
+                # TODO: backends are responsible for being correct on
-                # TODO: fwiw, we probably want a way to signal a throttle
+                # this right!?
-                # condition (eg. with ib) so that we can halt the
+                # -[ ] in the `ib` case we could maybe offer some way
-                # request loop until the condition is resolved?
+                #     to halt the request loop until the condition is
-                if timeframe > 1:
+                #     resolved or should the backend be entirely in
-                    await tractor.pause()
+                #     charge of solving such faults? yes, right?
+                # if timeframe > 1:
+                #     await tractor.pause()
                 return
 
             assert (
@@ -578,6 +580,7 @@ async def start_backfill(
                     '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.

tests/conftest.py

@@ -10,7 +10,7 @@ from piker import (
     config,
 )
 from piker.service import (
-    Services,
+    get_service_mngr,
 )
 from piker.log import get_console_log
 
@@ -129,7 +129,7 @@ async def _open_test_pikerd(
         ) as service_manager,
     ):
         # this proc/actor is the pikerd
-        assert service_manager is Services
+        assert service_manager is get_service_mngr()
 
         async with tractor.wait_for_actor(
             'pikerd',

tests/test_ems.py

@@ -26,7 +26,7 @@ import pytest
 import tractor
 from uuid import uuid4
 
-from piker.service import Services
+from piker.service import ServiceMngr
 from piker.log import get_logger
 from piker.clearing._messages import (
     Order,
@@ -158,7 +158,7 @@ def load_and_check_pos(
 
 
 def test_ems_err_on_bad_broker(
-    open_test_pikerd: Services,
+    open_test_pikerd: ServiceMngr,
     loglevel: str,
 ):
     async def load_bad_fqme():

tests/test_services.py

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