Use `pytest` plugin now exposed by `tractor`

Using `tractor.trionics.collapse_eg()` as needed and doing
some renames, in similar style as elsewhere:
- `pcs` -> `rent_cs`,
- `n` -> `tn` for nursery handles,

Also,
- tweak the `._reconnect_forever()` while loop to use the
  (also) `trio`-internal
  `mc_state: trio._channel.MemoryChannelState = snd._state` instead
  of `snd._close` to poll for open send/receive consumer task counts
  since,
    1. it seems more reliable then using the `snd._closed`,
    2. there's no other way to access the info.. afaik?

- handle `ConnectionRejected` explicitly alongside handshake-errs as
  a retry case.
- add a base-exc handler which `.exception()` reports the reconnect
  attempt failure explicitly.
- drop some lingering `Optional` usage.

config/conf.toml

@@ -1,7 +1,9 @@
 [network]
-tsdb.backend = 'marketstore'
-tsdb.host = 'localhost'
-tsdb.grpc_port = 5995
+pikerd = [
+  '/ipv4/127.0.0.1/tcp/6116',  # std localhost daemon-actor tree
+  # '/uds/6116',  # TODO std uds socket file
+]
+
 
 [ui]
 # set custom font + size which will scale entire UI

examples/basic_order_bot.py

@@ -121,6 +121,7 @@ async def bot_main():
             # tick_throttle=10,
         ) as feed,
 
+        tractor.trionics.collapse_eg(),
         trio.open_nursery() as tn,
     ):
         assert accounts

piker/accounting/_pos.py

@@ -365,7 +365,11 @@ class Position(Struct):
         # added: bool = False
         tid: str = t.tid
         if tid in self._events:
-            log.warning(f'{t} is already added?!')
+            log.debug(
+                f'Txn is already added?\n'
+                f'\n'
+                f'{t}\n'
+            )
             # return added
 
         # TODO: apparently this IS possible with a dict but not
@@ -731,7 +735,7 @@ class Account(Struct):
                     else:
                         # TODO: we reallly need a diff set of
                         # loglevels/colors per subsys.
-                        log.warning(
+                        log.debug(
                             f'Recent position for {fqme} was closed!'
                         )
 

piker/accounting/calc.py

@@ -465,7 +465,7 @@ def ledger_to_dfs(
 
         df = dfs[key] = ldf.with_columns([
 
-            pl.cumsum('size').alias('cumsize'),
+            pl.cum_sum('size').alias('cumsize'),
 
             # amount of source asset "sent" (via buy txns in
             # the market) to acquire the dst asset, PER txn.
@@ -480,7 +480,7 @@ def ledger_to_dfs(
         ]).with_columns([
 
             # rolling balance in src asset units
-            (pl.col('dst_bot').cumsum() * -1).alias('src_balance'),
+            (pl.col('dst_bot').cum_sum() * -1).alias('src_balance'),
 
             # "position operation type" in terms of increasing the
             # amount in the dst asset (entering) or decreasing the
@@ -622,7 +622,7 @@ def ledger_to_dfs(
                 # cost that was included in the least-recently
                 # entered txn that is still part of the current CSi
                 # set.
-                # => we look up the cost-per-unit cumsum and apply
+                # => we look up the cost-per-unit cum_sum and apply
                 # if over the current txn size (by multiplication)
                 # and then reverse that previusly applied cost on
                 # the txn_cost for this record.

piker/brokers/__init__.py

@@ -98,14 +98,13 @@ async def open_cached_client(
     If one has not been setup do it and cache it.
 
     '''
-    brokermod: ModuleType = get_brokermod(brokername)
-
-    # TODO: make abstract or `typing.Protocol`
-    # client: Client
+    brokermod = get_brokermod(brokername)
     async with maybe_open_context(
         acm_func=brokermod.get_client,
         kwargs=kwargs,
+
     ) as (cache_hit, client):
+
         if cache_hit:
             log.runtime(f'Reusing existing {client}')
 

piker/brokers/binance/api.py

@@ -374,14 +374,9 @@ class Client:
                 pair: Pair = pair_type(**item)
             except Exception as e:
                 e.add_note(
-                    f'\n'
-                    f'New or removed field we need to codify!\n'
-                    f'pair-type: {pair_type!r}\n'
-                    f'\n'
-                    f"Don't panic, prolly stupid binance changed their symbology schema again..\n"
-                    f'Check out their API docs here:\n'
-                    f'\n'
-                    f'https://binance-docs.github.io/apidocs/spot/en/#exchange-information\n'
+                    "\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

piker/brokers/binance/broker.py

@@ -440,6 +440,7 @@ async def open_trade_dialog(
             #   - ledger: TransactionLedger
 
             async with (
+                tractor.trionics.collapse_eg(),
                 trio.open_nursery() as tn,
                 ctx.open_stream() as ems_stream,
             ):

piker/brokers/binance/feed.py

@@ -448,7 +448,6 @@ async def subscribe(
 
 
 async def stream_quotes(
-
     send_chan: trio.abc.SendChannel,
     symbols: list[str],
     feed_is_live: trio.Event,
@@ -460,6 +459,7 @@ async def stream_quotes(
 ) -> None:
 
     async with (
+        tractor.trionics.maybe_raise_from_masking_exc(),
         send_chan as send_chan,
         open_cached_client('binance') as client,
     ):

piker/brokers/binance/venues.py

@@ -97,13 +97,6 @@ class Pair(Struct, frozen=True, kw_only=True):
     baseAsset: str
     baseAssetPrecision: int
 
-    permissionSets: list[list[str]]
-
-    # https://developers.binance.com/docs/binance-spot-api-docs#2025-08-26
-    # will become non-optional 2025-08-28?
-    # https://developers.binance.com/docs/binance-spot-api-docs#future-changes
-    pegInstructionsAllowed: bool|None = None
-
     filters: dict[
         str,
         str | int | float,
@@ -149,11 +142,7 @@ class SpotPair(Pair, frozen=True):
     defaultSelfTradePreventionMode: str
     allowedSelfTradePreventionModes: list[str]
     permissions: list[str]
-
-    # can the paint botz creat liq gaps even easier on this asset?
-    # Bp
-    # https://developers.binance.com/docs/binance-spot-api-docs/faqs/order_amend_keep_priority
-    amendAllowed: bool
+    permissionSets: list[list[str]]
 
     # NOTE: see `.data._symcache.SymbologyCache.load()` for why
     ns_path: str = 'piker.brokers.binance:SpotPair'

piker/brokers/cli.py

@@ -471,15 +471,11 @@ def search(
 
     '''
     # global opts
-    brokermods: list[ModuleType] = list(config['brokermods'].values())
-
-    # TODO: this is coming from the `search --pdb` NOT from
-    # the `piker --pdb` XD ..
-    # -[ ] pull from the parent click ctx's values..dumdum
-    # assert pdb
+    brokermods = list(config['brokermods'].values())
 
     # define tractor entrypoint
     async def main(func):
+
         async with maybe_open_pikerd(
             loglevel=config['loglevel'],
             debug_mode=pdb,

piker/brokers/core.py

@@ -22,9 +22,7 @@ routines should be primitive data types where possible.
 """
 import inspect
 from types import ModuleType
-from typing import (
-    Any,
-)
+from typing import List, Dict, Any, Optional
 
 import trio
 
@@ -36,10 +34,8 @@ from ..accounting import MktPair
 
 
 async def api(brokername: str, methname: str, **kwargs) -> dict:
-    '''
-    Make (proxy through) a broker API call by name and return its result.
-
-    '''
+    """Make (proxy through) a broker API call by name and return its result.
+    """
     brokermod = get_brokermod(brokername)
     async with brokermod.get_client() as client:
         meth = getattr(client, methname, None)
@@ -66,14 +62,10 @@ async def api(brokername: str, methname: str, **kwargs) -> dict:
 
 async def stocks_quote(
     brokermod: ModuleType,
-    tickers: list[str]
-
-) -> dict[str, dict[str, Any]]:
-    '''
-    Return a `dict` of snapshot quotes for the provided input
-    `tickers`: a `list` of fqmes.
-
-    '''
+    tickers: List[str]
+) -> Dict[str, Dict[str, Any]]:
+    """Return quotes dict for ``tickers``.
+    """
     async with brokermod.get_client() as client:
         return await client.quote(tickers)
 
@@ -82,15 +74,13 @@ async def stocks_quote(
 async def option_chain(
     brokermod: ModuleType,
     symbol: str,
-    date: str|None = None,
-) -> dict[str, dict[str, dict[str, Any]]]:
-    '''
-    Return option chain for ``symbol`` for ``date``.
+    date: Optional[str] = None,
+) -> Dict[str, Dict[str, Dict[str, Any]]]:
+    """Return option chain for ``symbol`` for ``date``.
 
     By default all expiries are returned. If ``date`` is provided
     then contract quotes for that single expiry are returned.
-
-    '''
+    """
     async with brokermod.get_client() as client:
         if date:
             id = int((await client.tickers2ids([symbol]))[symbol])
@@ -108,7 +98,7 @@ async def option_chain(
 # async def contracts(
 #     brokermod: ModuleType,
 #     symbol: str,
-# ) -> dict[str, dict[str, dict[str, Any]]]:
+# ) -> Dict[str, Dict[str, Dict[str, Any]]]:
 #     """Return option contracts (all expiries) for ``symbol``.
 #     """
 #     async with brokermod.get_client() as client:
@@ -120,24 +110,15 @@ async def bars(
     brokermod: ModuleType,
     symbol: str,
     **kwargs,
-) -> dict[str, dict[str, dict[str, Any]]]:
-    '''
-    Return option contracts (all expiries) for ``symbol``.
-
-    '''
+) -> Dict[str, Dict[str, Dict[str, Any]]]:
+    """Return option contracts (all expiries) for ``symbol``.
+    """
     async with brokermod.get_client() as client:
         return await client.bars(symbol, **kwargs)
 
 
-async def search_w_brokerd(
-    name: str,
-    pattern: str,
-) -> dict:
+async def search_w_brokerd(name: str, pattern: str) -> dict:
 
-    # TODO: WHY NOT WORK!?!
-    # when we `step` through the next block?
-    # import tractor
-    # await tractor.pause()
     async with open_cached_client(name) as client:
 
         # TODO: support multiple asset type concurrent searches.
@@ -149,12 +130,12 @@ async def symbol_search(
     pattern: str,
     **kwargs,
 
-) -> dict[str, dict[str, dict[str, Any]]]:
+) -> Dict[str, Dict[str, Dict[str, Any]]]:
     '''
     Return symbol info from broker.
 
     '''
-    results: list[str] = []
+    results = []
 
     async def search_backend(
         brokermod: ModuleType
@@ -162,13 +143,6 @@ async def symbol_search(
 
         brokername: str = mod.name
 
-        # TODO: figure this the FUCK OUT
-        # -> ok so obvi in the root actor any async task that's
-        # spawned outside the main tractor-root-actor task needs to
-        # call this..
-        # await tractor.devx._debug.maybe_init_greenback()
-        # tractor.pause_from_sync()
-
         async with maybe_spawn_brokerd(
             mod.name,
             infect_asyncio=getattr(
@@ -188,6 +162,7 @@ async def symbol_search(
             ))
 
     async with trio.open_nursery() as n:
+
         for mod in brokermods:
             n.start_soon(search_backend, mod.name)
 
@@ -197,13 +172,11 @@ async def symbol_search(
 async def mkt_info(
     brokermod: ModuleType,
     fqme: str,
-
     **kwargs,
 
 ) -> MktPair:
     '''
-    Return the `piker.accounting.MktPair` info struct from a given
-    backend broker tradable src/dst asset pair.
+    Return MktPair info from broker including src and dst assets.
 
     '''
     async with open_cached_client(brokermod.name) as client:

piker/brokers/deribit/api.py

@@ -31,7 +31,7 @@ from typing import (
     Callable,
 )
 
-import pendulum
+from pendulum import now
 import trio
 from trio_typing import TaskStatus
 from rapidfuzz import process as fuzzy
@@ -39,6 +39,7 @@ import numpy as np
 from tractor.trionics import (
     broadcast_receiver,
     maybe_open_context
+    collapse_eg,
 )
 from tractor import to_asyncio
 # XXX WOOPS XD
@@ -432,6 +433,7 @@ async def get_client(
 ) -> Client:
 
     async with (
+        collapse_eg(),
         trio.open_nursery() as n,
         open_jsonrpc_session(
             _testnet_ws_url, dtype=JSONRPCResult) as json_rpc

piker/brokers/ib/_util.py

@@ -34,7 +34,6 @@ from piker.brokers._util import get_logger
 if TYPE_CHECKING:
     from .api import Client
     from ib_insync import IB
-    import i3ipc
 
 log = get_logger('piker.brokers.ib')
 
@@ -49,37 +48,6 @@ _reset_tech: Literal[
 ] = 'vnc'
 
 
-no_setup_msg:str = (
-    'No data reset hack test setup for {vnc_sockaddr}!\n'
-    'See config setup tips @\n'
-    'https://github.com/pikers/piker/tree/master/piker/brokers/ib'
-)
-
-
-def try_xdo_manual(
-    vnc_sockaddr: str,
-):
-    '''
-    Do the "manual" `xdo`-based screen switch + click
-    combo since apparently the `asyncvnc` client ain't workin..
-
-    Note this is only meant as a backup method for Xorg users,
-    ideally you can use a real vnc client and the `vnc_click_hack()`
-    impl!
-
-    '''
-    global _reset_tech
-    try:
-        i3ipc_xdotool_manual_click_hack()
-        _reset_tech = 'i3ipc_xdotool'
-        return True
-    except OSError:
-        log.exception(
-            no_setup_msg.format(vnc_sockaddr)
-        )
-        return False
-
-
 async def data_reset_hack(
     # vnc_host: str,
     client: Client,
@@ -122,9 +90,15 @@ async def data_reset_hack(
     vnc_port: int
     vnc_sockaddr: tuple[str] | None = client.conf.get('vnc_addrs')
 
+    no_setup_msg:str = (
+        f'No data reset hack test setup for {vnc_sockaddr}!\n'
+        'See config setup tips @\n'
+        'https://github.com/pikers/piker/tree/master/piker/brokers/ib'
+    )
+
     if not vnc_sockaddr:
         log.warning(
-            no_setup_msg.format(vnc_sockaddr)
+            no_setup_msg
             +
             'REQUIRES A `vnc_addrs: array` ENTRY'
         )
@@ -145,38 +119,27 @@ async def data_reset_hack(
                         port=vnc_port,
                     )
                 )
-            except (
-                OSError,  # no VNC server avail..
-                PermissionError,  # asyncvnc pw fail..
-            ):
+            except OSError:
+                if vnc_host != 'localhost':
+                    log.warning(no_setup_msg)
+                    return False
+
                 try:
                     import i3ipc  # noqa  (since a deps dynamic check)
                 except ModuleNotFoundError:
-                    log.warning(
-                        no_setup_msg.format(vnc_sockaddr)
-                    )
+                    log.warning(no_setup_msg)
                     return False
 
-                if vnc_host not in {
-                    'localhost',
-                    '127.0.0.1',
-                }:
-                    focussed, matches = i3ipc_fin_wins_titled()
-                    if not matches:
-                        log.warning(
-                            no_setup_msg.format(vnc_sockaddr)
-                        )
-                        return False
-                    else:
-                        try_xdo_manual(vnc_sockaddr)
-
-                # localhost but no vnc-client or it borked..
-                else:
-                    try_xdo_manual(vnc_sockaddr)
+                try:
+                    i3ipc_xdotool_manual_click_hack()
+                    _reset_tech = 'i3ipc_xdotool'
+                    return True
+                except OSError:
+                    log.exception(no_setup_msg)
+                    return False
 
         case 'i3ipc_xdotool':
-            try_xdo_manual(vnc_sockaddr)
-            # i3ipc_xdotool_manual_click_hack()
+            i3ipc_xdotool_manual_click_hack()
 
         case _ as tech:
             raise RuntimeError(f'{tech} is not supported for reset tech!?')
@@ -215,9 +178,9 @@ async def vnc_click_hack(
         host,
         port=port,
 
-        # TODO: doesn't work?
-        # see, https://github.com/barneygale/asyncvnc/issues/7
-        password='doggy',
+        # TODO: doesn't work see:
+        # https://github.com/barneygale/asyncvnc/issues/7
+        # password='ibcansmbz',
 
     ) as client:
 
@@ -231,103 +194,70 @@ async def vnc_click_hack(
         client.keyboard.press('Ctrl', 'Alt', key)  # keys are stacked
 
 
-def i3ipc_fin_wins_titled(
-    titles: list[str] = [
-        'Interactive Brokers',  # tws running in i3
-        'IB Gateway',  # gw running in i3
-        # 'IB',  # gw running in i3 (newer version?)
-
-        # !TODO, remote vnc instance
-        # -[ ] something in title (or other Con-props) that indicates
-        #   this is explicitly for ibrk sw?
-        #  |_[ ]  !can use modden spawn eventually!
-        'TigerVNC',
-        # 'vncviewer',  # the terminal..
-    ],
-) -> tuple[
-    i3ipc.Con,  # orig focussed win
-    list[tuple[str, i3ipc.Con]],  # matching wins by title
-]:
-    '''
-    Attempt to find a local-DE window titled with an entry in
-    `titles`.
-
-    If found deliver the current focussed window and all matching
-    `i3ipc.Con`s in a list.
-
-    '''
-    import i3ipc
-    ipc = i3ipc.Connection()
-
-    # TODO: might be worth offering some kinda api for grabbing
-    # the window id from the pid?
-    # https://stackoverflow.com/a/2250879
-    tree = ipc.get_tree()
-    focussed: i3ipc.Con = tree.find_focused()
-
-    matches: list[i3ipc.Con] = []
-    for name in titles:
-        results = tree.find_titled(name)
-        print(f'results for {name}: {results}')
-        if results:
-            con = results[0]
-            matches.append((
-                name,
-                con,
-            ))
-
-    return (
-        focussed,
-        matches,
-    )
-
-
-
 def i3ipc_xdotool_manual_click_hack() -> None:
     '''
     Do the data reset hack but expecting a local X-window using `xdotool`.
 
     '''
-    focussed, matches = i3ipc_fin_wins_titled()
-    orig_win_id = focussed.window
+    import i3ipc
+    i3 = i3ipc.Connection()
+
+    # TODO: might be worth offering some kinda api for grabbing
+    # the window id from the pid?
+    # https://stackoverflow.com/a/2250879
+    t = i3.get_tree()
+
+    orig_win_id = t.find_focused().window
+
+    # for tws
+    win_names: list[str] = [
+        'Interactive Brokers',  # tws running in i3
+        'IB Gateway',  # gw running in i3
+        # 'IB',  # gw running in i3 (newer version?)
+    ]
+
     try:
-        for name, con in matches:
-            print(f'Resetting data feed for {name}')
-            win_id = str(con.window)
-            w, h = con.rect.width, con.rect.height
+        for name in win_names:
+            results = t.find_titled(name)
+            print(f'results for {name}: {results}')
+            if results:
+                con = results[0]
+                print(f'Resetting data feed for {name}')
+                win_id = str(con.window)
+                w, h = con.rect.width, con.rect.height
 
-            # TODO: seems to be a few libs for python but not sure
-            # if they support all the sub commands we need, order of
-            # most recent commit history:
-            # https://github.com/rr-/pyxdotool
-            # https://github.com/ShaneHutter/pyxdotool
-            # https://github.com/cphyc/pyxdotool
+                # TODO: seems to be a few libs for python but not sure
+                # if they support all the sub commands we need, order of
+                # most recent commit history:
+                # https://github.com/rr-/pyxdotool
+                # https://github.com/ShaneHutter/pyxdotool
+                # https://github.com/cphyc/pyxdotool
 
-            # TODO: only run the reconnect (2nd) kc on a detected
-            # disconnect?
-            for key_combo, timeout in [
-                # only required if we need a connection reset.
-                # ('ctrl+alt+r', 12),
-                # data feed reset.
-                ('ctrl+alt+f', 6)
-            ]:
-                subprocess.call([
-                    'xdotool',
-                    'windowactivate', '--sync', win_id,
+                # TODO: only run the reconnect (2nd) kc on a detected
+                # disconnect?
+                for key_combo, timeout in [
+                    # only required if we need a connection reset.
+                    # ('ctrl+alt+r', 12),
+                    # data feed reset.
+                    ('ctrl+alt+f', 6)
+                ]:
+                    subprocess.call([
+                        'xdotool',
+                        'windowactivate', '--sync', win_id,
 
-                    # move mouse to bottom left of window (where
-                    # there should be nothing to click).
-                    'mousemove_relative', '--sync', str(w-4), str(h-4),
+                        # move mouse to bottom left of window (where
+                        # there should be nothing to click).
+                        'mousemove_relative', '--sync', str(w-4), str(h-4),
 
-                    # NOTE: we may need to stick a `--retry 3` in here..
-                    'click', '--window', win_id,
-                    '--repeat', '3', '1',
+                        # NOTE: we may need to stick a `--retry 3` in here..
+                        'click', '--window', win_id,
+                        '--repeat', '3', '1',
 
-                    # hackzorzes
-                    'key', key_combo,
-                    ],
-                    timeout=timeout,
-                )
+                        # hackzorzes
+                        'key', key_combo,
+                        ],
+                        timeout=timeout,
+                    )
 
         # re-activate and focus original window
         subprocess.call([

piker/brokers/ib/api.py

@@ -48,6 +48,7 @@ from bidict import bidict
 import trio
 import tractor
 from tractor import to_asyncio
+from tractor import trionics
 from pendulum import (
     from_timestamp,
     DateTime,
@@ -1369,8 +1370,8 @@ async def load_clients_for_trio(
     '''
     Pure async mngr proxy to ``load_aio_clients()``.
 
-    This is a bootstrap entrypoing to call from
-    a ``tractor.to_asyncio.open_channel_from()``.
+    This is a bootstrap entrypoint to call from
+    a `tractor.to_asyncio.open_channel_from()`.
 
     '''
     async with load_aio_clients(
@@ -1391,7 +1392,10 @@ async def open_client_proxies() -> tuple[
     async with (
         tractor.trionics.maybe_open_context(
             acm_func=tractor.to_asyncio.open_channel_from,
-            kwargs={'target': load_clients_for_trio},
+            kwargs={
+                'target': load_clients_for_trio,
+                # ^XXX, kwarg to `open_channel_from()`
+            },
 
             # lock around current actor task access
             # TODO: maybe this should be the default in tractor?
@@ -1584,7 +1588,8 @@ async def open_client_proxy(
             event_consumers=event_table,
         ) as (first, chan),
 
-        trio.open_nursery() as relay_n,
+        trionics.collapse_eg(),  # loose-ify
+        trio.open_nursery() as relay_tn,
     ):
 
         assert isinstance(first, Client)
@@ -1624,7 +1629,7 @@ async def open_client_proxy(
 
                     continue
 
-        relay_n.start_soon(relay_events)
+        relay_tn.start_soon(relay_events)
 
         yield proxy
 

piker/brokers/ib/broker.py

@@ -34,6 +34,7 @@ import trio
 from trio_typing import TaskStatus
 import tractor
 from tractor.to_asyncio import LinkedTaskChannel
+from tractor import trionics
 from ib_insync.contract import (
     Contract,
 )
@@ -407,7 +408,7 @@ async def update_and_audit_pos_msg(
 
     # TODO: make this a "propaganda" log level?
     if ibpos.avgCost != msg.avg_price:
-        log.warning(
+        log.debug(
             f'IB "FIFO" avg price for {msg.symbol} is DIFF:\n'
             f'ib: {ibfmtmsg}\n'
             '---------------------------\n'
@@ -738,7 +739,7 @@ async def open_trade_dialog(
                                 f'UNEXPECTED POSITION says IB => {msg.symbol}\n'
                                 'Maybe they LIQUIDATED YOU or your ledger is wrong?\n'
                             )
-                            log.error(logmsg)
+                            log.debug(logmsg)
 
             await ctx.started((
                 all_positions,
@@ -747,21 +748,22 @@ async def open_trade_dialog(
 
             async with (
                 ctx.open_stream() as ems_stream,
-                trio.open_nursery() as n,
+                trionics.collapse_eg(),
+                trio.open_nursery() as tn,
             ):
                 # relay existing open orders to ems
                 for msg in order_msgs:
                     await ems_stream.send(msg)
 
                 for client in set(aioclients.values()):
-                    trade_event_stream: LinkedTaskChannel = await n.start(
+                    trade_event_stream: LinkedTaskChannel = await tn.start(
                         open_trade_event_stream,
                         client,
                     )
 
                     # start order request handler **before** local trades
                     # event loop
-                    n.start_soon(
+                    tn.start_soon(
                         handle_order_requests,
                         ems_stream,
                         accounts_def,
@@ -769,7 +771,7 @@ async def open_trade_dialog(
                     )
 
                     # allocate event relay tasks for each client connection
-                    n.start_soon(
+                    tn.start_soon(
                         deliver_trade_events,
 
                         trade_event_stream,
@@ -1241,47 +1243,32 @@ async def deliver_trade_events(
                 # never relay errors for non-broker related issues
                 # https://interactivebrokers.github.io/tws-api/message_codes.html
                 code: int = err['error_code']
-                reason: str = err['reason']
-                reqid: str = str(err['reqid'])
-
-                # "Warning:" msg codes,
-                # https://interactivebrokers.github.io/tws-api/message_codes.html#warning_codes
-                # - 2109: 'Outside Regular Trading Hours'
-                if 'Warning:' in reason:
-                    log.warning(
-                        f'Order-API-warning: {code!r}\n'
-                        f'reqid: {reqid!r}\n'
-                        f'\n'
-                        f'{pformat(err)}\n'
-                        # ^TODO? should we just print the `reason`
-                        # not the full `err`-dict?
-                    )
-                    continue
-
-                # XXX known special (ignore) cases
-                elif code in {
-                    200,  # uhh.. ni idea
+                if code in {
+                    200,  # uhh
 
                     # hist pacing / connectivity
                     162,
                     165,
 
+                    # WARNING codes:
+                    # https://interactivebrokers.github.io/tws-api/message_codes.html#warning_codes
+                    # Attribute 'Outside Regular Trading Hours' is
+                    # " 'ignored based on the order type and
+                    # destination. PlaceOrder is now ' 'being
+                    # processed.',
+                    2109,
+
                     # XXX: lol this isn't even documented..
                     # 'No market data during competing live session'
                     1669,
                 }:
-                    log.error(
-                        f'Order-API-error which is non-cancel-causing ?!\n'
-                        f'\n'
-                        f'{pformat(err)}\n'
-                    )
                     continue
 
+                reqid: str = str(err['reqid'])
+                reason: str = err['reason']
+
                 if err['reqid'] == -1:
-                    log.error(
-                        f'TWS external order error ??\n'
-                        f'{pformat(err)}\n'
-                    )
+                    log.error(f'TWS external order error:\n{pformat(err)}')
 
                 flow: dict = dict(
                     flows.get(reqid)

piker/brokers/ib/feed.py

@@ -587,7 +587,7 @@ async def get_bars(
                         data_cs.cancel()
 
                     # spawn new data reset task
-                    data_cs, reset_done = await tn.start(
+                    data_cs, reset_done = await nurse.start(
                         partial(
                             wait_on_data_reset,
                             proxy,
@@ -607,11 +607,11 @@ async def get_bars(
     # such that simultaneous symbol queries don't try data resettingn
     # too fast..
     unset_resetter: bool = False
-    async with trio.open_nursery() as tn:
+    async with trio.open_nursery() as nurse:
 
         # start history request that we allow
         # to run indefinitely until a result is acquired
-        tn.start_soon(query)
+        nurse.start_soon(query)
 
         # start history reset loop which waits up to the timeout
         # for a result before triggering a data feed reset.
@@ -631,7 +631,7 @@ async def get_bars(
                 unset_resetter: bool = True
 
             # spawn new data reset task
-            data_cs, reset_done = await tn.start(
+            data_cs, reset_done = await nurse.start(
                 partial(
                     wait_on_data_reset,
                     proxy,
@@ -705,9 +705,7 @@ async def _setup_quote_stream(
         # to_trio, from_aio = trio.open_memory_channel(2**8)  # type: ignore
         def teardown():
             ticker.updateEvent.disconnect(push)
-            log.error(
-                f'Disconnected stream for `{symbol}`'
-            )
+            log.error(f"Disconnected stream for `{symbol}`")
             client.ib.cancelMktData(contract)
 
             # decouple broadcast mem chan
@@ -763,10 +761,7 @@ async def open_aio_quote_stream(
     symbol: str,
     contract: Contract | None = None,
 
-) -> (
-    trio.abc.Channel|  # iface
-    tractor.to_asyncio.LinkedTaskChannel  # actually
-):
+) -> trio.abc.ReceiveStream:
 
     from tractor.trionics import broadcast_receiver
     global _quote_streams
@@ -783,7 +778,6 @@ async def open_aio_quote_stream(
             yield from_aio
             return
 
-    from_aio: tractor.to_asyncio.LinkedTaskChannel
     async with tractor.to_asyncio.open_channel_from(
         _setup_quote_stream,
         symbol=symbol,
@@ -989,18 +983,17 @@ async def stream_quotes(
         )
         cs: trio.CancelScope | None = None
         startup: bool = True
-        iter_quotes: trio.abc.Channel
         while (
             startup
             or cs.cancel_called
         ):
             with trio.CancelScope() as cs:
                 async with (
-                    trio.open_nursery() as tn,
+                    trio.open_nursery() as nurse,
                     open_aio_quote_stream(
                         symbol=sym,
                         contract=con,
-                    ) as iter_quotes,
+                    ) as stream,
                 ):
                     # ugh, clear ticks since we've consumed them
                     # (ahem, ib_insync is stateful trash)
@@ -1028,9 +1021,9 @@ async def stream_quotes(
                         await rt_ev.wait()
                         cs.cancel()  # cancel called should now be set
 
-                    tn.start_soon(reset_on_feed)
+                    nurse.start_soon(reset_on_feed)
 
-                    async with aclosing(iter_quotes):
+                    async with aclosing(stream):
                         # if syminfo.get('no_vlm', False):
                         if not init_msg.shm_write_opts['has_vlm']:
 
@@ -1045,21 +1038,19 @@ async def stream_quotes(
                             # wait for real volume on feed (trading might be
                             # closed)
                             while True:
-                                ticker = await iter_quotes.receive()
+                                ticker = await stream.receive()
 
                                 # for a real volume contract we rait for
                                 # the first "real" trade to take place
                                 if (
                                     # not calc_price
                                     # and not ticker.rtTime
-                                    False
-                                    # not ticker.rtTime
+                                    not ticker.rtTime
                                 ):
                                     # spin consuming tickers until we
                                     # get a real market datum
                                     log.debug(f"New unsent ticker: {ticker}")
                                     continue
-
                                 else:
                                     log.debug("Received first volume tick")
                                     # ugh, clear ticks since we've
@@ -1075,18 +1066,13 @@ async def stream_quotes(
                             log.debug(f"First ticker received {quote}")
 
                         # tell data-layer spawner-caller that live
-                        # quotes are now active desptie not having
-                        # necessarily received a first vlm/clearing
-                        # tick.
-                        ticker = await iter_quotes.receive()
+                        # quotes are now streaming.
                         feed_is_live.set()
-                        fqme: str = quote['fqme']
-                        await send_chan.send({fqme: quote})
 
                         # last = time.time()
-                        async for ticker in iter_quotes:
+                        async for ticker in stream:
                             quote = normalize(ticker)
-                            fqme: str = quote['fqme']
+                            fqme = quote['fqme']
                             await send_chan.send({fqme: quote})
 
                             # ugh, clear ticks since we've consumed them

piker/brokers/kraken/api.py

@@ -34,7 +34,6 @@ import urllib.parse
 import hashlib
 import hmac
 import base64
-import tractor
 import trio
 
 from piker import config
@@ -373,7 +372,8 @@ class Client:
         #     1658347714, 'status': 'Success'}]}
 
         if xfers:
-            await tractor.pause()
+            import tractor
+            await tractor.pp()
 
         trans: dict[str, Transaction] = {}
         for entry in xfers:
@@ -501,8 +501,7 @@ class Client:
             for xkey, data in resp['result'].items():
 
                 # NOTE: always cache in pairs tables for faster lookup
-                with tractor.devx.maybe_open_crash_handler(): # as bxerr:
-                    pair = Pair(xname=xkey, **data)
+                pair = Pair(xname=xkey, **data)
 
                 # register the above `Pair` structs for all
                 # key-sets/monikers: a set of 4 (frickin) tables

piker/brokers/kraken/broker.py

@@ -175,8 +175,9 @@ async def handle_order_requests(
 
             case {
                 'account': 'kraken.spot' as account,
-                'action': 'buy'|'sell',
-            }:
+                'action': action,
+            } if action in {'buy', 'sell'}:
+
                 # validate
                 order = BrokerdOrder(**msg)
 
@@ -261,12 +262,6 @@ async def handle_order_requests(
                 } | extra
 
                 log.info(f'Submitting WS order request:\n{pformat(req)}')
-
-                # NOTE HOWTO, debug order requests
-                #
-                # if 'XRP' in pair:
-                #     await tractor.pause()
-
                 await ws.send_msg(req)
 
                 # placehold for sanity checking in relay loop
@@ -549,7 +544,7 @@ async def open_trade_dialog(
             # to be reloaded.
             balances: dict[str, float] = await client.get_balances()
 
-            await verify_balances(
+            verify_balances(
                 acnt,
                 src_fiat,
                 balances,
@@ -1090,8 +1085,6 @@ async def handle_order_updates(
                         f'Failed to {action} order {reqid}:\n'
                         f'{errmsg}'
                     )
-                    # if tractor._state.debug_mode():
-                    #     await tractor.pause()
 
                     symbol: str = 'N/A'
                     if chain := apiflows.get(reqid):

piker/brokers/kraken/symbols.py

@@ -21,6 +21,7 @@ Symbology defs and search.
 from decimal import Decimal
 
 import tractor
+from rapidfuzz import process as fuzzy
 
 from piker._cacheables import (
     async_lifo_cache,
@@ -40,13 +41,8 @@ from piker.accounting._mktinfo import (
 )
 
 
+# https://www.kraken.com/features/api#get-tradable-pairs
 class Pair(Struct):
-    '''
-    A tradable asset pair as schema-defined by,
-
-    https://docs.kraken.com/api/docs/rest-api/get-tradable-asset-pairs
-
-    '''
     xname: str  # idiotic bs_mktid equiv i guess?
     altname: str  # alternate pair name
     wsname: str  # WebSocket pair name (if available)
@@ -57,6 +53,7 @@ class Pair(Struct):
     lot: str  # volume lot size
 
     cost_decimals: int
+    costmin: float
     pair_decimals: int  # scaling decimal places for pair
     lot_decimals: int  # scaling decimal places for volume
 
@@ -82,7 +79,6 @@ class Pair(Struct):
     tick_size: float  # min price step size
     status: str
 
-    costmin: str|None = None  # XXX, only some mktpairs?
     short_position_limit: float = 0
     long_position_limit: float = float('inf')
 

piker/brokers/questrade.py

@@ -37,12 +37,6 @@ import tractor
 from async_generator import asynccontextmanager
 import numpy as np
 import wrapt
-
-# TODO, port to `httpx`/`trio-websocket` whenver i get back to
-# writing a proper ws-api streamer for this backend (since the data
-# feeds are free now) as per GH feat-req:
-# https://github.com/pikers/piker/issues/509
-#
 import asks
 
 from ..calc import humanize, percent_change

piker/clearing/_ems.py

@@ -42,6 +42,7 @@ from bidict import bidict
 import trio
 from trio_typing import TaskStatus
 import tractor
+from tractor import trionics
 
 from ._util import (
     log,  # sub-sys logger
@@ -500,7 +501,7 @@ class Router(Struct):
 
     '''
     # setup at actor spawn time
-    nursery: trio.Nursery
+    _tn: trio.Nursery
 
     # broker to book map
     books: dict[str, DarkBook] = {}
@@ -670,7 +671,7 @@ class Router(Struct):
                 # dark book clearing loop, also lives with parent
                 # daemon to allow dark order clearing while no
                 # client is connected.
-                self.nursery.start_soon(
+                self._tn.start_soon(
                     clear_dark_triggers,
                     self,
                     relay.brokerd_stream,
@@ -693,7 +694,7 @@ class Router(Struct):
 
                 # spawn a ``brokerd`` order control dialog stream
                 # that syncs lifetime with the parent `emsd` daemon.
-                self.nursery.start_soon(
+                self._tn.start_soon(
                     translate_and_relay_brokerd_events,
                     broker,
                     relay.brokerd_stream,
@@ -767,10 +768,12 @@ async def _setup_persistent_emsd(
 
     global _router
 
-    # open a root "service nursery" for the ``emsd`` actor
-    async with trio.open_nursery() as service_nursery:
-
-        _router = Router(nursery=service_nursery)
+    # open a root "service task-nursery" for the `emsd`-actor
+    async with (
+        trionics.collapse_eg(),
+        trio.open_nursery() as tn
+    ):
+        _router = Router(_tn=tn)
 
         # TODO: send back the full set of persistent
         # orders/execs?
@@ -1519,7 +1522,7 @@ async def maybe_open_trade_relays(
         loglevel: str = 'info',
     ):
 
-        fqme, relay, feed, client_ready = await _router.nursery.start(
+        fqme, relay, feed, client_ready = await _router._tn.start(
             _router.open_trade_relays,
             fqme,
             exec_mode,

piker/cli/__init__.py

@@ -134,86 +134,65 @@ def pikerd(
     Spawn the piker broker-daemon.
 
     '''
-    from tractor.devx import maybe_open_crash_handler
-    with maybe_open_crash_handler(pdb=pdb):
-        log = get_console_log(loglevel, name='cli')
+    # from tractor.devx import maybe_open_crash_handler
+    # with maybe_open_crash_handler(pdb=False):
+    log = get_console_log(loglevel, name='cli')
 
-        if pdb:
-            log.warning((
-                "\n"
-                "!!! YOU HAVE ENABLED DAEMON DEBUG MODE !!!\n"
-                "When a `piker` daemon crashes it will block the "
-                "task-thread until resumed from console!\n"
-                "\n"
+    if pdb:
+        log.warning((
+            "\n"
+            "!!! YOU HAVE ENABLED DAEMON DEBUG MODE !!!\n"
+            "When a `piker` daemon crashes it will block the "
+            "task-thread until resumed from console!\n"
+            "\n"
+        ))
+
+    # service-actor registry endpoint socket-address set
+    regaddrs: list[tuple[str, int]] = []
+
+    conf, _ = config.load(
+        conf_name='conf',
+    )
+    network: dict = conf.get('network')
+    if (
+        network is None
+        and not maddr
+    ):
+        regaddrs = [(
+            _default_registry_host,
+            _default_registry_port,
+        )]
+
+    else:
+        eps: dict = load_trans_eps(
+            network,
+            maddr,
+        )
+        for layers in eps['pikerd']:
+            regaddrs.append((
+                layers['ipv4']['addr'],
+                layers['tcp']['port'],
             ))
 
-        # service-actor registry endpoint socket-address set
-        regaddrs: list[tuple[str, int]] = []
+    from .. import service
 
-        conf, _ = config.load(
-            conf_name='conf',
-        )
-        network: dict = conf.get('network')
-        if (
-            network is None
-            and not maddr
+    async def main():
+        service_mngr: service.Services
+        async with (
+            service.open_pikerd(
+                registry_addrs=regaddrs,
+                loglevel=loglevel,
+                debug_mode=pdb,
+                enable_transports=['uds'],
+                # enable_transports=['tcp'],
+            ) as service_mngr,
         ):
-            regaddrs = [(
-                _default_registry_host,
-                _default_registry_port,
-            )]
+            assert service_mngr
+            # ?TODO? spawn all other sub-actor daemons according to
+            # multiaddress endpoint spec defined by user config
+            await trio.sleep_forever()
 
-        else:
-            eps: dict = load_trans_eps(
-                network,
-                maddr,
-            )
-            for layers in eps['pikerd']:
-                regaddrs.append((
-                    layers['ipv4']['addr'],
-                    layers['tcp']['port'],
-                ))
-
-        from .. import service
-
-        async def main():
-            service_mngr: service.Services
-
-            async with (
-                service.open_pikerd(
-                    registry_addrs=regaddrs,
-                    loglevel=loglevel,
-                    debug_mode=pdb,
-
-                ) as service_mngr,  # normally delivers a ``Services`` handle
-
-                # AsyncExitStack() as stack,
-            ):
-                # TODO: spawn all other sub-actor daemons according to
-                # multiaddress endpoint spec defined by user config
-                assert service_mngr
-
-                # if tsdb:
-                #     dname, conf = await stack.enter_async_context(
-                #         service.marketstore.start_ahab_daemon(
-                #             service_mngr,
-                #             loglevel=loglevel,
-                #         )
-                #     )
-                #     log.info(f'TSDB `{dname}` up with conf:\n{conf}')
-
-                # if es:
-                #     dname, conf = await stack.enter_async_context(
-                #         service.elastic.start_ahab_daemon(
-                #             service_mngr,
-                #             loglevel=loglevel,
-                #         )
-                #     )
-                #     log.info(f'DB `{dname}` up with conf:\n{conf}')
-
-                await trio.sleep_forever()
-
-        trio.run(main)
+    trio.run(main)
 
 
 @click.group(context_settings=config._context_defaults)
@@ -328,6 +307,10 @@ def services(config, tl, ports):
     if not ports:
         ports = [_default_registry_port]
 
+    addr = tractor._addr.wrap_address(
+        addr=(host, ports[0])
+    )
+
     async def list_services():
         nonlocal host
         async with (
@@ -336,15 +319,17 @@ def services(config, tl, ports):
                 loglevel=config['loglevel'] if tl else None,
             ),
             tractor.get_registry(
-                host=host,
-                port=ports[0]
+                addr=addr,
             ) as portal
         ):
-            registry = await portal.run_from_ns('self', 'get_registry')
+            registry = await portal.run_from_ns(
+                'self',
+                'get_registry',
+            )
             json_d = {}
             for key, socket in registry.items():
-                host, port = socket
-                json_d[key] = f'{host}:{port}'
+                json_d[key] = f'{socket}'
+
             click.echo(f"{colorize_json(json_d)}")
 
     trio.run(list_services)

piker/config.py

@@ -41,10 +41,13 @@ from .log import get_logger
 log = get_logger('broker-config')
 
 
-# XXX NOTE: taken from ``click`` since apparently they have some
-# super weirdness with sigint and sudo..no clue
-# we're probably going to slowly just modify it to our own version over
-# time..
+# XXX NOTE: taken from `click`
+# |_https://github.com/pallets/click/blob/main/src/click/utils.py#L449
+#
+# (since apparently they have some super weirdness with SIGINT and
+# sudo.. no clue we're probably going to slowly just modify it to our
+# own version over time..)
+#
 def get_app_dir(
     app_name: str,
     roaming: bool = True,
@@ -261,7 +264,7 @@ def load(
         MutableMapping,
     ] = tomllib.loads,
 
-    touch_if_dne: bool = False,
+    touch_if_dne: bool = True,
 
     **tomlkws,
 
@@ -270,7 +273,7 @@ def load(
     Load config file by name.
 
     If desired config is not in the top level piker-user config path then
-    pass the ``path: Path`` explicitly.
+    pass the `path: Path` explicitly.
 
     '''
     # create the $HOME/.config/piker dir if dne
@@ -285,7 +288,8 @@ def load(
 
     if (
         not path.is_file()
-        and touch_if_dne
+        and
+        touch_if_dne
     ):
         # only do a template if no path provided,
         # just touch an empty file with same name.

piker/data/_symcache.py

@@ -91,6 +91,18 @@ class SymbologyCache(Struct):
     # provided by the backend pkg.
     mktmaps: dict[str, MktPair] = field(default_factory=dict)
 
+    def pformat(self) -> str:
+        return (
+            f'<{type(self).__name__}(\n'
+            f' .mod: {self.mod!r}\n'
+            f' .assets: {len(self.assets)!r}\n'
+            f' .pairs: {len(self.pairs)!r}\n'
+            f' .mktmaps: {len(self.mktmaps)!r}\n'
+            f')>'
+        )
+
+    __repr__ = pformat
+
     def write_config(self) -> None:
 
         # put the backend's pair-struct type ref at the top

piker/data/_web_bs.py

@@ -27,7 +27,6 @@ from functools import partial
 from types import ModuleType
 from typing import (
     Any,
-    Optional,
     Callable,
     AsyncContextManager,
     AsyncGenerator,
@@ -35,6 +34,7 @@ from typing import (
 )
 import json
 
+import tractor
 import trio
 from trio_typing import TaskStatus
 from trio_websocket import (
@@ -167,7 +167,7 @@ async def _reconnect_forever(
 
     async def proxy_msgs(
         ws: WebSocketConnection,
-        pcs: trio.CancelScope,  # parent cancel scope
+        rent_cs: trio.CancelScope,  # parent cancel scope
     ):
         '''
         Receive (under `timeout` deadline) all msgs from from underlying
@@ -192,7 +192,7 @@ async def _reconnect_forever(
                         f'{url} connection bail with:'
                     )
                     await trio.sleep(0.5)
-                    pcs.cancel()
+                    rent_cs.cancel()
 
                     # go back to reonnect loop in parent task
                     return
@@ -204,7 +204,7 @@ async def _reconnect_forever(
                         f'{src_mod}\n'
                         'WS feed seems down and slow af.. reconnecting\n'
                     )
-                    pcs.cancel()
+                    rent_cs.cancel()
 
                     # go back to reonnect loop in parent task
                     return
@@ -228,7 +228,12 @@ async def _reconnect_forever(
     nobsws._connected = trio.Event()
     task_status.started()
 
-    while not snd._closed:
+    mc_state: trio._channel.MemoryChannelState = snd._state
+    while (
+        mc_state.open_receive_channels > 0
+        and
+        mc_state.open_send_channels > 0
+    ):
         log.info(
             f'{src_mod}\n'
             f'{url} trying (RE)CONNECT'
@@ -237,10 +242,11 @@ async def _reconnect_forever(
         ws: WebSocketConnection
         try:
             async with (
-                trio.open_nursery() as n,
                 open_websocket_url(url) as ws,
+                tractor.trionics.collapse_eg(),
+                trio.open_nursery() as tn,
             ):
-                cs = nobsws._cs = n.cancel_scope
+                cs = nobsws._cs = tn.cancel_scope
                 nobsws._ws = ws
                 log.info(
                     f'{src_mod}\n'
@@ -248,7 +254,7 @@ async def _reconnect_forever(
                 )
 
                 # begin relay loop to forward msgs
-                n.start_soon(
+                tn.start_soon(
                     proxy_msgs,
                     ws,
                     cs,
@@ -262,7 +268,7 @@ async def _reconnect_forever(
 
                     # TODO: should we return an explicit sub-cs
                     # from this fixture task?
-                    await n.start(
+                    await tn.start(
                         open_fixture,
                         fixture,
                         nobsws,
@@ -272,11 +278,23 @@ async def _reconnect_forever(
                 # to let tasks run **inside** the ws open block above.
                 nobsws._connected.set()
                 await trio.sleep_forever()
-        except HandshakeError:
+
+        except (
+            HandshakeError,
+            ConnectionRejected,
+        ):
             log.exception('Retrying connection')
+            await trio.sleep(0.5)  # throttle
 
-        # ws & nursery block ends
+        except BaseException as _berr:
+            berr = _berr
+            log.exception(
+                'Reconnect-attempt failed ??\n'
+            )
+            await trio.sleep(0.2)  # throttle
+            raise berr
 
+        #|_ws & nursery block ends
         nobsws._connected = trio.Event()
         if cs.cancelled_caught:
             log.cancel(
@@ -324,21 +342,25 @@ async def open_autorecon_ws(
     connetivity errors, or some user defined recv timeout.
 
     You can provide a ``fixture`` async-context-manager which will be
-    entered/exitted around each connection reset; eg. for (re)requesting
-    subscriptions without requiring streaming setup code to rerun.
+    entered/exitted around each connection reset; eg. for
+    (re)requesting subscriptions without requiring streaming setup
+    code to rerun.
 
     '''
     snd: trio.MemorySendChannel
     rcv: trio.MemoryReceiveChannel
     snd, rcv = trio.open_memory_channel(616)
 
-    async with trio.open_nursery() as n:
+    async with (
+        tractor.trionics.collapse_eg(),
+        trio.open_nursery() as tn
+    ):
         nobsws = NoBsWs(
             url,
             rcv,
             msg_recv_timeout=msg_recv_timeout,
         )
-        await n.start(
+        await tn.start(
             partial(
                 _reconnect_forever,
                 url,
@@ -351,11 +373,10 @@ async def open_autorecon_ws(
         await nobsws._connected.wait()
         assert nobsws._cs
         assert nobsws.connected()
-
         try:
             yield nobsws
         finally:
-            n.cancel_scope.cancel()
+            tn.cancel_scope.cancel()
 
 
 '''
@@ -368,8 +389,8 @@ of msgs over a `NoBsWs`.
 class JSONRPCResult(Struct):
     id: int
     jsonrpc: str = '2.0'
-    result: Optional[dict] = None
-    error: Optional[dict] = None
+    result: dict|None = None
+    error: dict|None = None
 
 
 @acm

piker/data/feed.py

@@ -357,7 +357,9 @@ async def allocate_persistent_feed(
 
     # yield back control to starting nursery once we receive either
     # some history or a real-time quote.
-    log.info(f'loading OHLCV history: {fqme}')
+    log.info(
+        f'loading OHLCV history: {fqme!r}\n'
+    )
     await some_data_ready.wait()
 
     flume = Flume(
@@ -728,7 +730,10 @@ class Feed(Struct):
                 async for msg in stream:
                     await tx.send(msg)
 
-        async with trio.open_nursery() as nurse:
+        async with (
+            tractor.trionics.collapse_eg(),
+            trio.open_nursery() as nurse
+        ):
             # spawn a relay task for each stream so that they all
             # multiplex to a common channel.
             for brokername in mods:

piker/service/_actor_runtime.py

@@ -107,17 +107,22 @@ async def open_piker_runtime(
         async with (
             tractor.open_root_actor(
 
-                # passed through to ``open_root_actor``
+                # passed through to `open_root_actor`
                 registry_addrs=registry_addrs,
                 name=name,
+                start_method=start_method,
                 loglevel=loglevel,
                 debug_mode=debug_mode,
-                start_method=start_method,
+
+                # XXX NOTE MEMBER DAT der's a perf hit yo!!
+                # https://greenback.readthedocs.io/en/latest/principle.html#performance
+                maybe_enable_greenback=True,
 
                 # TODO: eventually we should be able to avoid
                 # having the root have more then permissions to
                 # spawn other specialized daemons I think?
                 enable_modules=enable_modules,
+                hide_tb=False,
 
                 **tractor_kwargs,
             ) as actor,
@@ -257,7 +262,10 @@ async def maybe_open_pikerd(
     loglevel: str | None = None,
     **kwargs,
 
-) -> tractor._portal.Portal | ClassVar[Services]:
+) -> (
+    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
@@ -282,10 +290,11 @@ async def maybe_open_pikerd(
 
     registry_addrs: list[tuple[str, int]] = (
         registry_addrs
-        or [_default_reg_addr]
+        or
+        [_default_reg_addr]
     )
 
-    pikerd_portal: tractor.Portal | None
+    pikerd_portal: tractor.Portal|None
     async with (
         open_piker_runtime(
             name=query_name,

piker/service/_daemon.py

@@ -28,6 +28,7 @@ from contextlib import (
 )
 
 import tractor
+from trio.lowlevel import current_task
 
 from ._util import (
     log,  # sub-sys logger
@@ -70,69 +71,84 @@ async def maybe_spawn_daemon(
     lock = Services.locks[service_name]
     await lock.acquire()
 
-    async with find_service(
-        service_name,
-        registry_addrs=[('127.0.0.1', 6116)],
-    ) as portal:
-        if portal is not None:
-            lock.release()
-            yield portal
-            return
+    try:
+        async with find_service(
+            service_name,
+            registry_addrs=[('127.0.0.1', 6116)],
+        ) as portal:
+            if portal is not None:
+                lock.release()
+                yield portal
+                return
 
-    log.warning(
-        f"Couldn't find any existing {service_name}\n"
-        'Attempting to spawn new daemon-service..'
-    )
+        log.warning(
+            f"Couldn't find any existing {service_name}\n"
+            'Attempting to spawn new daemon-service..'
+        )
 
-    # ask root ``pikerd`` daemon to spawn the daemon we need if
-    # pikerd is not live we now become the root of the
-    # process tree
-    async with maybe_open_pikerd(
-        loglevel=loglevel,
-        **pikerd_kwargs,
+        # ask root ``pikerd`` daemon to spawn the daemon we need if
+        # pikerd is not live we now become the root of the
+        # process tree
+        async with maybe_open_pikerd(
+            loglevel=loglevel,
+            **pikerd_kwargs,
 
-    ) as pikerd_portal:
+        ) as pikerd_portal:
 
-        # we are the root and thus are `pikerd`
-        # so spawn the target service directly by calling
-        # the provided target routine.
-        # XXX: this assumes that the target is well formed and will
-        # do the right things to setup both a sub-actor **and** call
-        # the ``_Services`` api from above to start the top level
-        # service task for that actor.
-        started: bool
-        if pikerd_portal is None:
-            started = await service_task_target(
-                loglevel=loglevel,
-                **spawn_args,
+            # we are the root and thus are `pikerd`
+            # so spawn the target service directly by calling
+            # the provided target routine.
+            # XXX: this assumes that the target is well formed and will
+            # do the right things to setup both a sub-actor **and** call
+            # the ``_Services`` api from above to start the top level
+            # service task for that actor.
+            started: bool
+            if pikerd_portal is None:
+                started = await service_task_target(
+                    loglevel=loglevel,
+                    **spawn_args,
+                )
+
+            else:
+                # request a remote `pikerd` (service manager) to start the
+                # target daemon-task, the target can't return
+                # a non-serializable value since it is expected that service
+                # starting is non-blocking and the target task will persist
+                # running "under" or "within" the `pikerd` actor tree after
+                # the questing client disconnects. in other words this
+                # spawns a persistent daemon actor that continues to live
+                # for the lifespan of whatever the service manager inside
+                # `pikerd` says it should.
+                started = await pikerd_portal.run(
+                    service_task_target,
+                    loglevel=loglevel,
+                    **spawn_args,
+                )
+
+            if started:
+                log.info(f'Service {service_name} started!')
+
+            # block until we can discover (by IPC connection) to the newly
+            # spawned daemon-actor and then deliver the portal to the
+            # caller.
+            async with tractor.wait_for_actor(service_name) as portal:
+                lock.release()
+                yield portal
+                await portal.cancel_actor()
+
+    except BaseException as _err:
+        err = _err
+        if (
+            lock.locked()
+            and
+            lock.statistics().owner is current_task()
+        ):
+            log.exception(
+                f'Releasing stale lock after crash..?'
+                f'{err!r}\n'
             )
-
-        else:
-            # request a remote `pikerd` (service manager) to start the
-            # target daemon-task, the target can't return
-            # a non-serializable value since it is expected that service
-            # starting is non-blocking and the target task will persist
-            # running "under" or "within" the `pikerd` actor tree after
-            # the questing client disconnects. in other words this
-            # spawns a persistent daemon actor that continues to live
-            # for the lifespan of whatever the service manager inside
-            # `pikerd` says it should.
-            started = await pikerd_portal.run(
-                service_task_target,
-                loglevel=loglevel,
-                **spawn_args,
-            )
-
-        if started:
-            log.info(f'Service {service_name} started!')
-
-        # block until we can discover (by IPC connection) to the newly
-        # spawned daemon-actor and then deliver the portal to the
-        # caller.
-        async with tractor.wait_for_actor(service_name) as portal:
             lock.release()
-            yield portal
-            await portal.cancel_actor()
+            raise err
 
 
 async def spawn_emsd(

piker/service/_mngr.py

@@ -109,7 +109,7 @@ class Services:
                         # wait on any context's return value
                         # and any final portal result from the
                         # sub-actor.
-                        ctx_res: Any = await ctx.result()
+                        ctx_res: Any = await ctx.wait_for_result()
 
                         # NOTE: blocks indefinitely until cancelled
                         # either by error from the target context

piker/service/_registry.py

@@ -101,13 +101,15 @@ async def open_registry(
 
     if (
         not tractor.is_root_process()
-        and not Registry.addrs
+        and
+        not Registry.addrs
     ):
         Registry.addrs.extend(actor.reg_addrs)
 
     if (
         ensure_exists
-        and not Registry.addrs
+        and
+        not Registry.addrs
     ):
         raise RuntimeError(
             f"`{uid}` registry should already exist but doesn't?"
@@ -146,7 +148,7 @@ async def find_service(
     | list[Portal]
     | None
 ):
-
+    # try:
     reg_addrs: list[tuple[str, int]]
     async with open_registry(
         addrs=(
@@ -157,22 +159,39 @@ async def find_service(
             or Registry.addrs
         ),
     ) as reg_addrs:
-        log.info(f'Scanning for service `{service_name}`')
 
-        maybe_portals: list[Portal] | Portal | None
+        log.info(
+            f'Scanning for service {service_name!r}'
+        )
 
         # attach to existing daemon by name if possible
+        maybe_portals: list[Portal]|Portal|None
         async with tractor.find_actor(
             service_name,
             registry_addrs=reg_addrs,
             only_first=first_only,  # if set only returns single ref
         ) as maybe_portals:
             if not maybe_portals:
+                # log.info(
+                print(
+                    f'Could NOT find service {service_name!r} -> {maybe_portals!r}'
+                )
                 yield None
                 return
 
+            # log.info(
+            print(
+                f'Found service {service_name!r} -> {maybe_portals}'
+            )
             yield maybe_portals
 
+    # except BaseException as _berr:
+    #     berr = _berr
+    #     log.exception(
+    #         'tractor.find_actor() failed with,\n'
+    #     )
+    #     raise berr
+
 
 async def check_for_service(
     service_name: str,

piker/tsp/__init__.py

@@ -963,7 +963,10 @@ async def tsdb_backfill(
         # concurrently load the provider's most-recent-frame AND any
         # pre-existing tsdb history already saved in `piker` storage.
         dt_eps: list[DateTime, DateTime] = []
-        async with trio.open_nursery() as tn:
+        async with (
+            tractor.trionics.collapse_eg(),
+            trio.open_nursery() as tn
+        ):
             tn.start_soon(
                 push_latest_frame,
                 dt_eps,
@@ -1012,9 +1015,16 @@ async def tsdb_backfill(
                 int,
                 Duration,
             ]|None = config.get('frame_types', None)
+
             if def_frame_durs:
                 def_frame_size: Duration = def_frame_durs[timeframe]
-                assert def_frame_size == calced_frame_size
+
+                if def_frame_size != calced_frame_size:
+                    log.warning(
+                        f'Expected frame size {def_frame_size}\n'
+                        f'Rxed frame {calced_frame_size}\n'
+                    )
+                    # await tractor.pause()
             else:
                 # use what we calced from first frame above.
                 def_frame_size = calced_frame_size
@@ -1043,7 +1053,9 @@ async def tsdb_backfill(
             # if there is a gap to backfill from the first
             # history frame until the last datum loaded from the tsdb
             # continue that now in the background
-            async with trio.open_nursery() as tn:
+            async with trio.open_nursery(
+                strict_exception_groups=False,
+            ) as tn:
 
                 bf_done = await tn.start(
                     partial(
@@ -1308,6 +1320,7 @@ async def manage_history(
         # sampling period) data set since normally differently
         # sampled timeseries can be loaded / process independently
         # ;)
+        tractor.trionics.collapse_eg(),
         trio.open_nursery() as tn,
     ):
         log.info(

piker/ui/_search.py

@@ -15,7 +15,8 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 """
-qompleterz: embeddable search and complete using trio, Qt and rapidfuzz.
+qompleterz: embeddable search and complete using trio, Qt and
+rapidfuzz.
 
 """
 
@@ -46,6 +47,7 @@ import time
 from pprint import pformat
 
 from rapidfuzz import process as fuzzy
+import tractor
 import trio
 from trio_typing import TaskStatus
 
@@ -53,7 +55,7 @@ from piker.ui.qt import (
     size_policy,
     align_flag,
     Qt,
-    QtCore,
+    # QtCore,
     QtWidgets,
     QModelIndex,
     QItemSelectionModel,
@@ -920,7 +922,10 @@ async def fill_results(
 
             # issue multi-provider fan-out search request and place
             # "searching.." statuses on outstanding results providers
-            async with trio.open_nursery() as n:
+            async with (
+                tractor.trionics.collapse_eg(),
+                trio.open_nursery() as tn
+            ):
 
                 for provider, (search, pause) in (
                     _searcher_cache.copy().items()
@@ -944,7 +949,7 @@ async def fill_results(
                                 status_field='-> searchin..',
                             )
 
-                        await n.start(
+                        await tn.start(
                             pack_matches,
                             view,
                             has_results,
@@ -1004,12 +1009,14 @@ async def handle_keyboard_input(
     view.set_font_size(searchbar.dpi_font.px_size)
     send, recv = trio.open_memory_channel(616)
 
-    async with trio.open_nursery() as n:
-
+    async with (
+        tractor.trionics.collapse_eg(),  # needed?
+        trio.open_nursery() as tn
+    ):
         # start a background multi-searcher task which receives
         # patterns relayed from this keyboard input handler and
         # async updates the completer view's results.
-        n.start_soon(
+        tn.start_soon(
             partial(
                 fill_results,
                 searchw,

snippets/qt_screen_info.py

@@ -1,4 +1,22 @@
-"""
+# 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/>.
+
+'''
+A per-display, DPI (scaling) info dumper.
+
 Resource list for mucking with DPIs on multiple screens:
 
 - https://stackoverflow.com/questions/42141354/convert-pixel-size-to-point-size-for-fonts-on-multiple-platforms
@@ -12,89 +30,86 @@ Resource list for mucking with DPIs on multiple screens:
 - https://stackoverflow.com/questions/16561879/what-is-the-difference-between-logicaldpix-and-physicaldpix-in-qt
 - https://doc.qt.io/qt-5/qguiapplication.html#screenAt
 
-"""
+'''
 
 from pyqtgraph import QtGui
-from PyQt5.QtCore import (
-     Qt, QCoreApplication
+from PyQt6 import (
+    QtCore,
+    QtWidgets,
+)
+from PyQt6.QtCore import (
+    Qt,
+    QCoreApplication,
+    QSize,
+    QRect,
 )
 
 # Proper high DPI scaling is available in Qt >= 5.6.0. This attibute
 # must be set before creating the application
 if hasattr(Qt, 'AA_EnableHighDpiScaling'):
-    QCoreApplication.setAttribute(Qt.AA_EnableHighDpiScaling, True)
+    QCoreApplication.setAttribute(
+        Qt.AA_EnableHighDpiScaling,
+        True,
+    )
 
 if hasattr(Qt, 'AA_UseHighDpiPixmaps'):
-    QCoreApplication.setAttribute(Qt.AA_UseHighDpiPixmaps, True)
+    QCoreApplication.setAttribute(
+        Qt.AA_UseHighDpiPixmaps,
+        True,
+    )
 
-
-app = QtGui.QApplication([])
-window = QtGui.QMainWindow()
-main_widget = QtGui.QWidget()
+app = QtWidgets.QApplication([])
+window = QtWidgets.QMainWindow()
+main_widget = QtWidgets.QWidget()
 window.setCentralWidget(main_widget)
 window.show()
 
-pxr = main_widget.devicePixelRatioF()
+pxr: float = main_widget.devicePixelRatioF()
 
-# screen_num = app.desktop().screenNumber()
-# screen = app.screens()[screen_num]
-
-screen = app.screenAt(main_widget.geometry().center())
-
-name = screen.name()
-size = screen.size()
-geo = screen.availableGeometry()
-phydpi = screen.physicalDotsPerInch()
-logdpi = screen.logicalDotsPerInch()
-
-print(
-    # f'screen number: {screen_num}\n',
-    f'screen name: {name}\n'
-    f'screen size: {size}\n'
-    f'screen geometry: {geo}\n\n'
-    f'devicePixelRationF(): {pxr}\n'
-    f'physical dpi: {phydpi}\n'
-    f'logical dpi: {logdpi}\n'
+# explicitly get main widget and primary displays
+current_screen: QtGui.QScreen = app.screenAt(
+    main_widget.geometry().center()
 )
+primary_screen: QtGui.QScreen = app.primaryScreen()
 
-print('-'*50)
+screen: QtGui.QScreen
+for screen in app.screens():
+    name: str = screen.name()
+    model: str = screen.model().rstrip()
+    size: QSize = screen.size()
+    geo: QRect = screen.availableGeometry()
+    phydpi: float = screen.physicalDotsPerInch()
+    logdpi: float = screen.logicalDotsPerInch()
+    is_primary: bool = screen is primary_screen
+    is_current: bool = screen is current_screen
 
-screen = app.primaryScreen()
+    print(
+        f'------ screen name: {name} ------\n'
+        f'|_primary: {is_primary}\n'
+        f' _current: {is_current}\n'
+        f' _model: {model}\n'
+        f' _screen size: {size}\n'
+        f' _screen geometry: {geo}\n'
+        f' _devicePixelRationF(): {pxr}\n'
+        f' _physical dpi: {phydpi}\n'
+        f' _logical dpi: {logdpi}\n'
+    )
 
-name = screen.name()
-size = screen.size()
-geo = screen.availableGeometry()
-phydpi = screen.physicalDotsPerInch()
-logdpi = screen.logicalDotsPerInch()
-
-print(
-    # f'screen number: {screen_num}\n',
-    f'screen name: {name}\n'
-    f'screen size: {size}\n'
-    f'screen geometry: {geo}\n\n'
-    f'devicePixelRationF(): {pxr}\n'
-    f'physical dpi: {phydpi}\n'
-    f'logical dpi: {logdpi}\n'
-)
-
-
-# app-wide font
+# app-wide font info
 font = QtGui.QFont("Hack")
 # use pixel size to be cross-resolution compatible?
 font.setPixelSize(6)
 
-
 fm = QtGui.QFontMetrics(font)
-fontdpi = fm.fontDpi()
-font_h = fm.height()
-
-string = '10000'
-str_br = fm.boundingRect(string)
-str_w = str_br.width()
+fontdpi: float = fm.fontDpi()
+font_h: int = fm.height()
 
+string: str = '10000'
+str_br: QtCore.QRect = fm.boundingRect(string)
+str_w: int = str_br.width()
 
 print(
-    # f'screen number: {screen_num}\n',
+    f'------ global font settings ------\n'
     f'font dpi: {fontdpi}\n'
     f'font height: {font_h}\n'
     f'string bounding rect: {str_br}\n'

tests/conftest.py

@@ -15,6 +15,12 @@ from piker.service import (
 from piker.log import get_console_log
 
 
+# include `tractor`'s built-in fixtures!
+pytest_plugins: tuple[str] = (
+    "tractor._testing.pytest",
+)
+
+
 def pytest_addoption(parser):
     parser.addoption("--ll", action="store", dest='loglevel',
                      default=None, help="logging level to set when testing")

tests/test_questrade.py

@@ -142,7 +142,12 @@ async def test_concurrent_tokens_refresh(us_symbols, loglevel):
         # async with tractor.open_nursery() as n:
         #     await n.run_in_actor('other', intermittently_refresh_tokens)
 
-        async with trio.open_nursery() as n:
+        async with (
+            tractor.trionics.collapse_eg(),
+            trio.open_nursery(
+            # strict_exception_groups=False,
+            ) as n
+        ):
 
             quoter = await qt.stock_quoter(client, us_symbols)
 
@@ -383,7 +388,9 @@ async def test_quote_streaming(tmx_symbols, loglevel, stream_what):
             else:
                 symbols = [tmx_symbols]
 
-            async with trio.open_nursery() as n:
+            async with trio.open_nursery(
+                strict_exception_groups=False,
+            ) as n:
                 for syms, func in zip(symbols, stream_what):
                     n.start_soon(func, feed, syms)