Spurious first-draft of EG collapsing

Topically, throughout various (seemingly) console-UX-affecting or benign
spots in the code base; nothing that required more intervention beyond
things superficial. A few spots also include `trio.Nursery` ref renames
(always to something with a `tn` in it) and log-level reductions to
quiet (benign) console noise oriented around issues meant to be solved
long..

Note there's still a couple spots i left with the loose-ify flag because
i haven't fully tested them without using the latest version of
`tractor.trionics.collapse_eg()`, but more then likely they should flip
over fine.

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

@@ -22,7 +22,9 @@ you know when you're losing money (if possible) XD
 from __future__ import annotations
 from collections.abc import ValuesView
 from contextlib import contextmanager as cm
+from functools import partial
 from math import copysign
+from pprint import pformat
 from typing import (
     Any,
     Callable,
@@ -37,12 +39,16 @@ from pendulum import (
     parse,
 )
 
+from ..log import get_logger
+
 if TYPE_CHECKING:
     from ._ledger import (
         Transaction,
         TransactionLedger,
     )
 
+log = get_logger(__name__)
+
 
 def ppu(
     clears: Iterator[Transaction],
@@ -238,6 +244,9 @@ def iter_by_dt(
 
     def dyn_parse_to_dt(
         tx: tuple[str, dict[str, Any]] | Transaction,
+
+        debug: bool = False,
+        _invalid: list|None = None,
     ) -> DateTime:
 
         # handle `.items()` inputs
@@ -250,52 +259,81 @@ def iter_by_dt(
         # get best parser for this record..
         for k in parsers:
             if (
-                isdict and k in tx
+                (v := getattr(tx, k, None))
                 or
-                getattr(tx, k, None)
+                (
+                    isdict
+                    and
+                    (v := tx.get(k))
+                )
             ):
-                v = (
-                    tx[k] if isdict
-                    else tx.dt
-                )
-                assert v is not None, (
-                    f'No valid value for `{k}`!?'
-                )
-
                 # only call parser on the value if not None from
                 # the `parsers` table above (when NOT using
                 # `.get()`), otherwise pass through the value and
                 # sort on it directly
                 if (
                     not isinstance(v, DateTime)
-                    and (parser := parsers.get(k))
+                    and
+                    (parser := parsers.get(k))
                 ):
-                    return parser(v)
+                    ret = parser(v)
                 else:
-                    return v
+                    ret = v
+
+                return ret
 
             else:
-            # TODO: move to top?
+                log.debug(
-            from piker.log import get_logger
+                    f'Parser-field not found in txn\n'
-            log = get_logger(__name__)
+                    f'\n'
+                    f'parser-field: {k!r}\n'
+                    f'txn: {tx!r}\n'
+                    f'\n'
+                    f'Trying next..\n'
+                )
+                continue
 
+        # XXX: should never get here..
+        else:
             # XXX: we should really never get here..
             # only if a ledger record has no expected sort(able)
             # field will we likely hit this.. like with ze IB.
             # if no sortable field just deliver epoch?
             log.warning(
                 'No (time) sortable field for TXN:\n'
-                f'{tx}\n'
+                f'{tx!r}\n'
+            )
+            if debug:
+                import tractor
+                with tractor.devx.maybe_open_crash_handler():
+                    raise ValueError(
+                        f'No supported time-field found in txn !?\n'
+                        f'\n'
+                        f'supported-time-fields: {parsers!r}\n'
+                        f'\n'
+                        f'txn: {tx!r}\n'
                     )
-            return from_timestamp(0)
-            # breakpoint()
 
+            if _invalid is not None:
+                _invalid.append(tx)
+                return from_timestamp(0.)
 
-    entry: tuple[str, dict] | Transaction
+    entry: tuple[str, dict]|Transaction
+    invalid: list = []
     for entry in sorted(
         records,
-        key=key or dyn_parse_to_dt,
+        key=key or partial(
+            dyn_parse_to_dt,
+            _invalid=invalid,
+        ),
     ):
+        if entry in invalid:
+            log.warning(
+                f'Ignoring txn w invalid timestamp ??\n'
+                f'{pformat(entry)}\n'
+            )
+            continue
+
         # NOTE the type sig above; either pairs or txns B)
         yield entry
 

piker/brokers/__init__.py

@@ -98,13 +98,14 @@ async def open_cached_client(
     If one has not been setup do it and cache it.
 
     '''
-    brokermod = get_brokermod(brokername)
+    brokermod: ModuleType = get_brokermod(brokername)
+
+    # TODO: make abstract or `typing.Protocol`
+    # client: Client
     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/_daemon.py

@@ -96,7 +96,10 @@ async def _setup_persistent_brokerd(
     # - `open_symbol_search()`
     # NOTE: see ep invocation details inside `.data.feed`.
     try:
-        async with trio.open_nursery() as service_nursery:
+        async with (
+            tractor.trionics.collapse_eg(),
+            trio.open_nursery() as service_nursery
+        ):
             bus: _FeedsBus = feed.get_feed_bus(
                 brokername,
                 service_nursery,

piker/brokers/binance/api.py

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

@@ -97,6 +97,13 @@ 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,
@@ -142,7 +149,11 @@ class SpotPair(Pair, frozen=True):
     defaultSelfTradePreventionMode: str
     allowedSelfTradePreventionModes: list[str]
     permissions: list[str]
-    permissionSets: list[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
 
     # NOTE: see `.data._symcache.SymbologyCache.load()` for why
     ns_path: str = 'piker.brokers.binance:SpotPair'

piker/brokers/cli.py

@@ -471,11 +471,15 @@ def search(
 
     '''
     # global opts
-    brokermods = list(config['brokermods'].values())
+    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
 
     # define tractor entrypoint
     async def main(func):
-
         async with maybe_open_pikerd(
             loglevel=config['loglevel'],
             debug_mode=pdb,

piker/brokers/core.py

@@ -22,7 +22,9 @@ routines should be primitive data types where possible.
 """
 import inspect
 from types import ModuleType
-from typing import List, Dict, Any, Optional
+from typing import (
+    Any,
+)
 
 import trio
 
@@ -34,8 +36,10 @@ 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)
@@ -62,10 +66,14 @@ async def api(brokername: str, methname: str, **kwargs) -> dict:
 
 async def stocks_quote(
     brokermod: ModuleType,
-    tickers: List[str]
+    tickers: list[str]
-) -> Dict[str, Dict[str, Any]]:
+
-    """Return quotes dict for ``tickers``.
+) -> dict[str, dict[str, Any]]:
-    """
+    '''
+    Return a `dict` of snapshot quotes for the provided input
+    `tickers`: a `list` of fqmes.
+
+    '''
     async with brokermod.get_client() as client:
         return await client.quote(tickers)
 
@@ -74,13 +82,15 @@ async def stocks_quote(
 async def option_chain(
     brokermod: ModuleType,
     symbol: str,
-    date: Optional[str] = None,
+    date: str|None = None,
-) -> Dict[str, Dict[str, Dict[str, Any]]]:
+) -> dict[str, dict[str, dict[str, Any]]]:
-    """Return option chain for ``symbol`` for ``date``.
+    '''
+    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])
@@ -98,7 +108,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:
@@ -110,15 +120,24 @@ async def bars(
     brokermod: ModuleType,
     symbol: str,
     **kwargs,
-) -> Dict[str, Dict[str, Dict[str, Any]]]:
+) -> dict[str, dict[str, dict[str, Any]]]:
-    """Return option contracts (all expiries) for ``symbol``.
+    '''
-    """
+    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.
@@ -130,12 +149,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 = []
+    results: list[str] = []
 
     async def search_backend(
         brokermod: ModuleType
@@ -143,6 +162,13 @@ 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(
@@ -162,7 +188,6 @@ async def symbol_search(
             ))
 
     async with trio.open_nursery() as n:
-
         for mod in brokermods:
             n.start_soon(search_backend, mod.name)
 
@@ -172,11 +197,13 @@ async def symbol_search(
 async def mkt_info(
     brokermod: ModuleType,
     fqme: str,
+
     **kwargs,
 
 ) -> MktPair:
     '''
-    Return MktPair info from broker including src and dst assets.
+    Return the `piker.accounting.MktPair` info struct from a given
+    backend broker tradable src/dst asset pair.
 
     '''
     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,6 +34,7 @@ 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')
 
@@ -48,6 +49,37 @@ _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,
@@ -90,15 +122,9 @@ 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
+            no_setup_msg.format(vnc_sockaddr)
             +
             'REQUIRES A `vnc_addrs: array` ENTRY'
         )
@@ -119,27 +145,38 @@ async def data_reset_hack(
                         port=vnc_port,
                     )
                 )
-            except OSError:
+            except (
-                if vnc_host != 'localhost':
+                OSError,  # no VNC server avail..
-                    log.warning(no_setup_msg)
+                PermissionError,  # asyncvnc pw fail..
-                    return False
+            ):
-
                 try:
                     import i3ipc  # noqa  (since a deps dynamic check)
                 except ModuleNotFoundError:
-                    log.warning(no_setup_msg)
+                    log.warning(
+                        no_setup_msg.format(vnc_sockaddr)
+                    )
                     return False
 
-                try:
+                if vnc_host not in {
-                    i3ipc_xdotool_manual_click_hack()
+                    'localhost',
-                    _reset_tech = 'i3ipc_xdotool'
+                    '127.0.0.1',
-                    return True
+                }:
-                except OSError:
+                    focussed, matches = i3ipc_fin_wins_titled()
-                    log.exception(no_setup_msg)
+                    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)
 
         case 'i3ipc_xdotool':
-            i3ipc_xdotool_manual_click_hack()
+            try_xdo_manual(vnc_sockaddr)
+            # i3ipc_xdotool_manual_click_hack()
 
         case _ as tech:
             raise RuntimeError(f'{tech} is not supported for reset tech!?')
@@ -178,9 +215,9 @@ async def vnc_click_hack(
         host,
         port=port,
 
-        # TODO: doesn't work see:
+        # TODO: doesn't work?
-        # https://github.com/barneygale/asyncvnc/issues/7
+        # see, https://github.com/barneygale/asyncvnc/issues/7
-        # password='ibcansmbz',
+        password='doggy',
 
     ) as client:
 
@@ -194,34 +231,67 @@ 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`.
 
     '''
-    import i3ipc
+    focussed, matches = i3ipc_fin_wins_titled()
-    i3 = i3ipc.Connection()
+    orig_win_id = focussed.window
-
-    # 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 in win_names:
+        for name, con in matches:
-            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

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
+    This is a bootstrap entrypoint to call from
-    a ``tractor.to_asyncio.open_channel_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,32 +1243,47 @@ 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']
-                if code in {
+                reason: str = err['reason']
-                    200,  # uhh
+                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
 
                     # 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{pformat(err)}')
+                    log.error(
+                        f'TWS external order error ??\n'
+                        f'{pformat(err)}\n'
+                    )
 
                 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 nurse.start(
+                    data_cs, reset_done = await tn.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 nurse:
+    async with trio.open_nursery() as tn:
 
         # start history request that we allow
         # to run indefinitely until a result is acquired
-        nurse.start_soon(query)
+        tn.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 nurse.start(
+            data_cs, reset_done = await tn.start(
                 partial(
                     wait_on_data_reset,
                     proxy,
@@ -705,7 +705,9 @@ 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
@@ -761,7 +763,10 @@ async def open_aio_quote_stream(
     symbol: str,
     contract: Contract | None = None,
 
-) -> trio.abc.ReceiveStream:
+) -> (
+    trio.abc.Channel|  # iface
+    tractor.to_asyncio.LinkedTaskChannel  # actually
+):
 
     from tractor.trionics import broadcast_receiver
     global _quote_streams
@@ -778,6 +783,7 @@ 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,
@@ -983,17 +989,18 @@ 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 nurse,
+                    trio.open_nursery() as tn,
                     open_aio_quote_stream(
                         symbol=sym,
                         contract=con,
-                    ) as stream,
+                    ) as iter_quotes,
                 ):
                     # ugh, clear ticks since we've consumed them
                     # (ahem, ib_insync is stateful trash)
@@ -1021,9 +1028,9 @@ async def stream_quotes(
                         await rt_ev.wait()
                         cs.cancel()  # cancel called should now be set
 
-                    nurse.start_soon(reset_on_feed)
+                    tn.start_soon(reset_on_feed)
 
-                    async with aclosing(stream):
+                    async with aclosing(iter_quotes):
                         # if syminfo.get('no_vlm', False):
                         if not init_msg.shm_write_opts['has_vlm']:
 
@@ -1038,19 +1045,21 @@ async def stream_quotes(
                             # wait for real volume on feed (trading might be
                             # closed)
                             while True:
-                                ticker = await stream.receive()
+                                ticker = await iter_quotes.receive()
 
                                 # for a real volume contract we rait for
                                 # the first "real" trade to take place
                                 if (
                                     # not calc_price
                                     # and not ticker.rtTime
-                                    not ticker.rtTime
+                                    False
+                                    # 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
@@ -1066,13 +1075,18 @@ async def stream_quotes(
                             log.debug(f"First ticker received {quote}")
 
                         # tell data-layer spawner-caller that live
-                        # quotes are now streaming.
+                        # quotes are now active desptie not having
+                        # necessarily received a first vlm/clearing
+                        # tick.
+                        ticker = await iter_quotes.receive()
                         feed_is_live.set()
+                        fqme: str = quote['fqme']
+                        await send_chan.send({fqme: quote})
 
                         # last = time.time()
-                        async for ticker in stream:
+                        async for ticker in iter_quotes:
                             quote = normalize(ticker)
-                            fqme = quote['fqme']
+                            fqme: str = quote['fqme']
                             await send_chan.send({fqme: quote})
 
                             # ugh, clear ticks since we've consumed them

piker/brokers/kraken/api.py

@@ -34,6 +34,7 @@ import urllib.parse
 import hashlib
 import hmac
 import base64
+import tractor
 import trio
 
 from piker import config
@@ -372,8 +373,7 @@ class Client:
         #     1658347714, 'status': 'Success'}]}
 
         if xfers:
-            import tractor
+            await tractor.pause()
-            await tractor.pp()
 
         trans: dict[str, Transaction] = {}
         for entry in xfers:
@@ -501,6 +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)
 
                 # register the above `Pair` structs for all

piker/brokers/kraken/broker.py

@@ -175,9 +175,8 @@ async def handle_order_requests(
 
             case {
                 'account': 'kraken.spot' as account,
-                'action': action,
+                'action': 'buy'|'sell',
-            } if action in {'buy', 'sell'}:
+            }:
-
                 # validate
                 order = BrokerdOrder(**msg)
 
@@ -262,6 +261,12 @@ 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
@@ -544,7 +549,7 @@ async def open_trade_dialog(
             # to be reloaded.
             balances: dict[str, float] = await client.get_balances()
 
-            verify_balances(
+            await verify_balances(
                 acnt,
                 src_fiat,
                 balances,
@@ -1085,6 +1090,8 @@ 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,7 +21,6 @@ Symbology defs and search.
 from decimal import Decimal
 
 import tractor
-from rapidfuzz import process as fuzzy
 
 from piker._cacheables import (
     async_lifo_cache,
@@ -41,8 +40,13 @@ 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)
@@ -53,7 +57,6 @@ 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
 
@@ -79,6 +82,7 @@ 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,6 +37,12 @@ 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/_client.py

@@ -25,7 +25,10 @@ from typing import TYPE_CHECKING
 
 import trio
 import tractor
-from tractor.trionics import broadcast_receiver
+from tractor.trionics import (
+    broadcast_receiver,
+    collapse_eg,
+)
 
 from ._util import (
     log,  # sub-sys logger
@@ -285,8 +288,11 @@ async def open_ems(
             client._ems_stream = trades_stream
 
             # start sync code order msg delivery task
-            async with trio.open_nursery() as n:
+            async with (
-                n.start_soon(
+                collapse_eg(),
+                trio.open_nursery() as tn,
+            ):
+                tn.start_soon(
                     relay_orders_from_sync_code,
                     client,
                     fqme,
@@ -302,4 +308,4 @@ async def open_ems(
                 )
 
                 # stop the sync-msg-relay task on exit.
-                n.cancel_scope.cancel()
+                tn.cancel_scope.cancel()

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
@@ -76,7 +77,6 @@ if TYPE_CHECKING:
 
 # TODO: numba all of this
 def mk_check(
-
     trigger_price: float,
     known_last: float,
     action: str,
@@ -162,7 +162,7 @@ async def clear_dark_triggers(
 
     router: Router,
     brokerd_orders_stream: tractor.MsgStream,
-    quote_stream: tractor.ReceiveMsgStream,  # noqa
+    quote_stream: tractor.MsgStream,
     broker: str,
     fqme: str,
 
@@ -178,6 +178,7 @@ async def clear_dark_triggers(
     '''
     # XXX: optimize this for speed!
     # TODO:
+    # - port to the new ringbuf stuff in `tractor.ipc`!
     # - numba all this!
     # - this stream may eventually contain multiple symbols
     quote_stream._raise_on_lag = False
@@ -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
+    # open a root "service task-nursery" for the `emsd`-actor
-    async with trio.open_nursery() as service_nursery:
+    async with (
-
+        trionics.collapse_eg(),
-        _router = Router(nursery=service_nursery)
+        trio.open_nursery() as tn
+    ):
+        _router = Router(_tn=tn)
 
         # TODO: send back the full set of persistent
         # orders/execs?
@@ -1190,12 +1193,16 @@ async def process_client_order_cmds(
     submitting live orders immediately if requested by the client.
 
     '''
-    # cmd: dict
+    # TODO, only allow `msgspec.Struct` form!
+    cmd: dict
     async for cmd in client_order_stream:
-        log.info(f'Received order cmd:\n{pformat(cmd)}')
+        log.info(
+            f'Received order cmd:\n'
+            f'{pformat(cmd)}\n'
+        )
 
         # CAWT DAMN we need struct support!
-        oid = str(cmd['oid'])
+        oid: str = str(cmd['oid'])
 
         # register this stream as an active order dialog (msg flow) for
         # this order id such that translated message from the brokerd
@@ -1301,7 +1308,7 @@ async def process_client_order_cmds(
             case {
                 'oid': oid,
                 'symbol': fqme,
-                'price': trigger_price,
+                'price': price,
                 'size': size,
                 'action': ('buy' | 'sell') as action,
                 'exec_mode': ('live' | 'paper'),
@@ -1333,7 +1340,7 @@ async def process_client_order_cmds(
 
                     symbol=sym,
                     action=action,
-                    price=trigger_price,
+                    price=price,
                     size=size,
                     account=req.account,
                 )
@@ -1355,7 +1362,11 @@ async def process_client_order_cmds(
                 # (``translate_and_relay_brokerd_events()`` above) will
                 # handle relaying the ems side responses back to
                 # the client/cmd sender from this request
-                log.info(f'Sending live order to {broker}:\n{pformat(msg)}')
+                log.info(
+                    f'Sending live order to {broker}:\n'
+                    f'{pformat(msg)}'
+                )
+
                 await brokerd_order_stream.send(msg)
 
                 # an immediate response should be ``BrokerdOrderAck``
@@ -1371,7 +1382,7 @@ async def process_client_order_cmds(
             case {
                 'oid': oid,
                 'symbol': fqme,
-                'price': trigger_price,
+                'price': price,
                 'size': size,
                 'exec_mode': exec_mode,
                 'action': action,
@@ -1399,7 +1410,12 @@ async def process_client_order_cmds(
                 if isnan(last):
                     last = flume.rt_shm.array[-1]['close']
 
-                pred = mk_check(trigger_price, last, action)
+                trigger_price: float = float(price)
+                pred = mk_check(
+                    trigger_price,
+                    last,
+                    action,
+                )
 
                 # NOTE: for dark orders currently we submit
                 # the triggered live order at a price 5 ticks
@@ -1506,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,
@@ -1539,7 +1555,7 @@ async def _emsd_main(
     ctx: tractor.Context,
     fqme: str,
     exec_mode: str,  # ('paper', 'live')
-    loglevel: str | None = None,
+    loglevel: str|None = None,
 
 ) -> tuple[
     dict[

piker/clearing/_messages.py

@@ -19,6 +19,7 @@ Clearing sub-system message and protocols.
 
 """
 from __future__ import annotations
+from decimal import Decimal
 from typing import (
     Literal,
 )
@@ -71,7 +72,15 @@ class Order(Struct):
     symbol: str  # | MktPair
     account: str  # should we set a default as '' ?
 
-    price: float
+    # https://docs.python.org/3/library/decimal.html#decimal-objects
+    #
+    # ?TODO? decimal usage throughout?
+    # -[ ] possibly leverage the `Encoder(decimal_format='number')`
+    #  bit?
+    # |_https://jcristharif.com/msgspec/supported-types.html#decimal
+    # -[ ] should we also use it for .size?
+    #
+    price: Decimal
     size: float  # -ve is "sell", +ve is "buy"
 
     brokers: list[str] = []
@@ -178,7 +187,7 @@ class BrokerdOrder(Struct):
     time_ns: int
 
     symbol: str  # fqme
-    price: float
+    price: Decimal
     size: float
 
     # TODO: if we instead rely on a +ve/-ve size to determine

piker/clearing/_paper_engine.py

@@ -510,7 +510,7 @@ async def handle_order_requests(
                 reqid = await client.submit_limit(
                     oid=order.oid,
                     symbol=f'{order.symbol}.{client.broker}',
-                    price=order.price,
+                    price=float(order.price),
                     action=order.action,
                     size=order.size,
                     # XXX: by default 0 tells ``ib_insync`` methods that

piker/cli/__init__.py

@@ -184,33 +184,12 @@ def pikerd(
                     registry_addrs=regaddrs,
                     loglevel=loglevel,
                     debug_mode=pdb,
-
+                    enable_transports=['uds'],
-                ) as service_mngr,  # normally delivers a ``Services`` handle
+                ) as service_mngr,
-
-                # AsyncExitStack() as stack,
             ):
-                # TODO: spawn all other sub-actor daemons according to
-                # multiaddress endpoint spec defined by user config
                 assert service_mngr
-
+                # ?TODO? spawn all other sub-actor daemons according to
-                # if tsdb:
+                # multiaddress endpoint spec defined by user config
-                #     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)
@@ -335,7 +314,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

@@ -740,7 +740,7 @@ async def sample_and_broadcast(
 
                                 log.warning(
                                     f'Feed OVERRUN {sub_key}'
-                                    '@{bus.brokername} -> \n'
+                                    f'@{bus.brokername} -> \n'
                                     f'feed @ {chan.uid}\n'
                                     f'throttle = {throttle} Hz'
                                 )

piker/data/feed.py

@@ -39,6 +39,7 @@ from typing import (
     AsyncContextManager,
     Awaitable,
     Sequence,
+    TYPE_CHECKING,
 )
 
 import trio
@@ -75,6 +76,10 @@ from ._sampling import (
     uniform_rate_send,
 )
 
+if TYPE_CHECKING:
+    from tractor._addr import Address
+    from tractor.msg.types import Aid
+
 
 class Sub(Struct, frozen=True):
     '''
@@ -723,7 +728,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:
@@ -786,7 +794,6 @@ async def install_brokerd_search(
 
 @acm
 async def maybe_open_feed(
-
     fqmes: list[str],
     loglevel: str | None = None,
 
@@ -840,13 +847,12 @@ async def maybe_open_feed(
 
 @acm
 async def open_feed(
-
     fqmes: list[str],
 
-    loglevel: str | None = None,
+    loglevel: str|None = None,
     allow_overruns: bool = True,
     start_stream: bool = True,
-    tick_throttle: float | None = None,  # Hz
+    tick_throttle: float|None = None,  # Hz
 
     allow_remote_ctl_ui: bool = False,
 
@@ -899,19 +905,19 @@ async def open_feed(
             feed.portals[brokermod] = portal
 
             # fill out "status info" that the UI can show
-            host, port = portal.channel.raddr
+            chan: tractor.Channel = portal.chan
-            if host == '127.0.0.1':
+            raddr: Address = chan.raddr
-                host = 'localhost'
+            aid: Aid = chan.aid
-
+            # TAG_feed_status_update
             feed.status.update({
-                'actor_name': portal.channel.uid[0],
+                'actor_id': aid,
-                'host': host,
+                'actor_short_id': f'{aid.name}@{aid.pid}',
-                'port': port,
+                'ipc': chan.raddr.proto_key,
+                'ipc_addr': raddr,
                 'hist_shm': 'NA',
                 'rt_shm': 'NA',
-                'throttle_rate': tick_throttle,
+                'throttle_hz': tick_throttle,
             })
-            # feed.status.update(init_msg.pop('status', {}))
 
             # (allocate and) connect to any feed bus for this broker
             bus_ctxs.append(

piker/data/flows.py

@@ -36,10 +36,10 @@ from ._sharedmem import (
     ShmArray,
     _Token,
 )
+from piker.accounting import MktPair
 
 if TYPE_CHECKING:
-    from ..accounting import MktPair
+    from piker.data.feed import Feed
-    from .feed import Feed
 
 
 class Flume(Struct):
@@ -82,7 +82,7 @@ class Flume(Struct):
 
     # TODO: do we need this really if we can pull the `Portal` from
     # ``tractor``'s internals?
-    feed: Feed | None = None
+    feed: Feed|None = None
 
     @property
     def rt_shm(self) -> ShmArray:

piker/data/validate.py

@@ -113,9 +113,9 @@ def validate_backend(
             )
             if ep is None:
                 log.warning(
-                    f'Provider backend {mod.name} is missing '
+                    f'Provider backend {mod.name!r} is missing '
-                    f'{daemon_name} support :(\n'
+                    f'{daemon_name!r} support?\n'
-                    f'The following endpoint is missing: {name}'
+                    f'|_module endpoint-func missing: {name!r}\n'
                 )
 
     inits: list[

piker/fsp/_engine.py

@@ -498,6 +498,7 @@ async def cascade(
 
         func_name: str = func.__name__
         async with (
+            tractor.trionics.collapse_eg(),  # avoid multi-taskc tb in console
             trio.open_nursery() as tn,
         ):
             # TODO: might be better to just make a "restart" method where

piker/service/_actor_runtime.py

@@ -200,7 +200,8 @@ async def open_pikerd(
             reg_addrs,
         ),
         tractor.open_nursery() as actor_nursery,
-        trio.open_nursery() as service_nursery,
+        tractor.trionics.collapse_eg(),
+        trio.open_nursery() as service_tn,
     ):
         for addr in reg_addrs:
             if addr not in root_actor.accept_addrs:
@@ -211,7 +212,7 @@ async def open_pikerd(
 
         # assign globally for future daemon/task creation
         Services.actor_n = actor_nursery
-        Services.service_n = service_nursery
+        Services.service_n = service_tn
         Services.debug_mode = debug_mode
 
         try:
@@ -221,7 +222,7 @@ async def open_pikerd(
             # TODO: is this more clever/efficient?
             # if 'samplerd' in Services.service_tasks:
             #     await Services.cancel_service('samplerd')
-            service_nursery.cancel_scope.cancel()
+            service_tn.cancel_scope.cancel()
 
 
 # TODO: do we even need this?

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/tsp/_anal.py

@@ -517,7 +517,7 @@ def with_dts(
 
     '''
     return df.with_columns([
-        pl.col(time_col).shift(1).suffix('_prev'),
+        pl.col(time_col).shift(1).name.suffix('_prev'),
         pl.col(time_col).diff().alias('s_diff'),
         pl.from_epoch(pl.col(time_col)).alias('dt'),
     ]).with_columns([
@@ -623,7 +623,7 @@ def detect_vlm_gaps(
 
 ) -> pl.DataFrame:
 
-    vnull: pl.DataFrame = w_dts.filter(
+    vnull: pl.DataFrame = df.filter(
         pl.col(col) == 0
     )
     return vnull

piker/ui/_app.py

@@ -21,6 +21,7 @@ Main app startup and run.
 from functools import partial
 from types import ModuleType
 
+import tractor
 import trio
 
 from piker.ui.qt import (
@@ -116,6 +117,7 @@ async def _async_main(
         needed_brokermods[brokername] = brokers[brokername]
 
     async with (
+        tractor.trionics.collapse_eg(),
         trio.open_nursery() as root_n,
     ):
         # set root nursery and task stack for spawning other charts/feeds

piker/ui/_chart.py

@@ -33,7 +33,6 @@ import trio
 
 from piker.ui.qt import (
     QtCore,
-    QtWidgets,
     Qt,
     QLineF,
     QFrame,

piker/ui/_display.py

@@ -1445,7 +1445,10 @@ async def display_symbol_data(
         # for pause/resume on mouse interaction
         rt_chart.feed = feed
 
-        async with trio.open_nursery() as ln:
+        async with (
+            tractor.trionics.collapse_eg(),
+            trio.open_nursery() as ln,
+        ):
             # if available load volume related built-in display(s)
             vlm_charts: dict[
                 str,

piker/ui/_event.py

@@ -22,7 +22,10 @@ from contextlib import asynccontextmanager as acm
 from typing import Callable
 
 import trio
-from tractor.trionics import gather_contexts
+from tractor.trionics import (
+    gather_contexts,
+    collapse_eg,
+)
 
 from piker.ui.qt import (
     QtCore,
@@ -207,7 +210,10 @@ async def open_signal_handler(
         async for args in recv:
             await async_handler(*args)
 
-    async with trio.open_nursery() as tn:
+    async with (
+        collapse_eg(),
+        trio.open_nursery() as tn
+    ):
         tn.start_soon(proxy_to_handler)
         async with send:
             yield
@@ -242,6 +248,7 @@ async def open_handlers(
     widget: QWidget
     streams: list[trio.abc.ReceiveChannel]
     async with (
+        collapse_eg(),
         trio.open_nursery() as tn,
         gather_contexts([
             open_event_stream(

piker/ui/_feedstatus.py

@@ -18,10 +18,11 @@
 Feed status and controls widget(s) for embedding in a UI-pane.
 
 """
-
 from __future__ import annotations
-from textwrap import dedent
+from typing import (
-from typing import TYPE_CHECKING
+    Any,
+    TYPE_CHECKING,
+)
 
 # from PyQt5.QtCore import Qt
 
@@ -49,35 +50,55 @@ def mk_feed_label(
     a feed control protocol.
 
     '''
-    status = feed.status
+    status: dict[str, Any] = feed.status
     assert status
 
-    msg = dedent("""
+    # SO tips on ws/nls,
-        actor: **{actor_name}**\n
+    # https://stackoverflow.com/a/15721400
-        |_ @**{host}:{port}**\n
+    ws: str = ' '
-    """)
+    # nl: str = '<br>'  # dun work?
+    actor_info_repr: str = (
+        f')> **{status["actor_short_id"]}**\n'
+        '\n'  # bc md?
+    )
 
-    for key, val in status.items():
+    # fields to select *IN* for display
-        if key in ('host', 'port', 'actor_name'):
+    # (see `.data.feed.open_feed()` status
-            continue
+    #  update -> TAG_feed_status_update)
-        msg += f'\n|_ {key}: **{{{key}}}**\n'
+    for key in [
+        'ipc',
+        'hist_shm',
+        'rt_shm',
+        'throttle_hz',
+    ]:
+        # NOTE, the 2nd key is filled via `.format()` updates.
+        actor_info_repr += (
+            f'\n'  # bc md?
+            f'{ws}|_{key}: **{{{key}}}**\n'
+        )
+        # ^TODO? formatting and content..
+        # -[ ] showing which fqme is "forward" on the
+        #    chart/fsp/order-mode?
+        #  '|_ flows: **{symbols}**\n'
+        #
+        # -[x] why isn't the indent working?
+        #  => markdown, now solved..
 
     feed_label = FormatLabel(
-        fmt_str=msg,
+        fmt_str=actor_info_repr,
-        # |_ streams: **{symbols}**\n
         font=_font.font,
         font_size=_font_small.px_size,
         font_color='default_lightest',
     )
 
+    # ?TODO, remove this?
     # form.vbox.setAlignment(feed_label, Qt.AlignBottom)
     # form.vbox.setAlignment(Qt.AlignBottom)
-    _ = chart.height() - (
+    # _ = chart.height() - (
-        form.height() +
+    #     form.height() +
-        form.fill_bar.height()
+    #     form.fill_bar.height()
-        # feed_label.height()
+    #     # feed_label.height()
-    )
+    # )
 
     feed_label.format(**feed.status)
-
     return feed_label

piker/ui/_fsp.py

@@ -600,6 +600,7 @@ async def open_fsp_admin(
             kwargs=kwargs,
         ) as (cache_hit, cluster_map),
 
+        tractor.trionics.collapse_eg(),
         trio.open_nursery() as tn,
     ):
         if cache_hit:
@@ -613,6 +614,8 @@ async def open_fsp_admin(
         )
         try:
             yield admin
+
+        # ??TODO, does this *need* to be inside a finally?
         finally:
             # terminate all tasks via signals
             for key, entry in admin._registry.items():

piker/ui/_label.py

@@ -285,18 +285,20 @@ class FormatLabel(QLabel):
         font_size: int,
         font_color: str,
 
+        use_md: bool = True,
+
         parent=None,
 
     ) -> None:
 
         super().__init__(parent)
 
-        # by default set the format string verbatim and expect user to
+        # by default set the format string verbatim and expect user
-        # call ``.format()`` later (presumably they'll notice the
+        # to call ``.format()`` later (presumably they'll notice the
         # unformatted content if ``fmt_str`` isn't meant to be
         # unformatted).
         self.fmt_str = fmt_str
-        self.setText(fmt_str)
+        # self.setText(fmt_str)  # ?TODO, why here?
 
         self.setStyleSheet(
             f"""QLabel {{
@@ -306,6 +308,7 @@ class FormatLabel(QLabel):
             """
         )
         self.setFont(_font.font)
+        if use_md:
             self.setTextFormat(
                 Qt.TextFormat.MarkdownText
             )
@@ -316,7 +319,10 @@ class FormatLabel(QLabel):
             size_policy.Expanding,
         )
         self.setAlignment(
-            Qt.AlignVCenter | Qt.AlignLeft
+            Qt.AlignLeft
+            |
+            Qt.AlignBottom
+            # Qt.AlignVCenter
         )
         self.setText(self.fmt_str)
 

piker/ui/_style.py

@@ -269,6 +269,8 @@ def hcolor(name: str) -> str:
 
         # default ohlc-bars/curve gray
         'bracket': '#666666',  # like the logo
+        'pikers': '#616161',  # a trader shade of..
+        'beast': '#161616',  # in the dark alone.
 
         # bluish
         'charcoal': '#36454F',

piker/ui/order_mode.py

@@ -21,6 +21,7 @@ Chart trading, the only way to scalp.
 from __future__ import annotations
 from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
+from decimal import Decimal
 from functools import partial
 from pprint import pformat
 import time
@@ -41,7 +42,6 @@ from piker.accounting import (
     Position,
     mk_allocator,
     MktPair,
-    Symbol,
 )
 from piker.clearing import (
     open_ems,
@@ -143,6 +143,15 @@ class OrderMode:
     }
     _staged_order: Order | None = None
 
+    @property
+    def curr_mkt(self) -> MktPair:
+        '''
+        Deliver the currently selected `MktPair` according
+        chart state.
+
+        '''
+        return self.chart.linked.mkt
+
     def on_level_change_update_next_order_info(
         self,
         level: float,
@@ -172,7 +181,11 @@ class OrderMode:
         line.update_labels(order_info)
 
         # update bound-in staged order
-        order.price = level
+        mkt: MktPair = self.curr_mkt
+        order.price: Decimal = mkt.quantize(
+            size=level,
+            quantity_type='price',
+        )
         order.size = order_info['size']
 
         # when an order is changed we flip the settings side-pane to
@@ -187,7 +200,9 @@ class OrderMode:
 
     ) -> LevelLine:
 
-        level = order.price
+        # TODO, if we instead just always decimalize at the ems layer
+        # we can avoid this back-n-forth casting?
+        level = float(order.price)
 
         line = order_line(
             chart or self.chart,
@@ -224,7 +239,11 @@ class OrderMode:
             # the order mode allocator but we still need to update the
             # "staged" order message we'll send to the ems
             def update_order_price(y: float) -> None:
-                order.price = y
+                mkt: MktPair = self.curr_mkt
+                order.price: Decimal = mkt.quantize(
+                    size=y,
+                    quantity_type='price',
+                )
 
             line._on_level_change = update_order_price
 
@@ -275,34 +294,31 @@ class OrderMode:
         chart = cursor.linked.chart
         if (
             not chart
-            and cursor
+            and
-            and cursor.active_plot
+            cursor
+            and
+            cursor.active_plot
         ):
             return
 
         chart = cursor.active_plot
-        price = cursor._datum_xy[1]
+        price: float = cursor._datum_xy[1]
         if not price:
             # zero prices are not supported by any means
             # since that's illogical / a no-op.
             return
 
-        mkt: MktPair = self.chart.linked.mkt
-
-        # NOTE : we could also use instead,
-        # mkt.quantize(price, quantity_type='price')
-        # but it returns a Decimal and it's probably gonna
-        # be slower?
         # TODO: should we be enforcing this precision
-        # at a different layer in the stack? right now
+        # at a different layer in the stack?
-        # any precision error will literally be relayed
+        # |_ might require `MktPair` tracking in the EMS?
-        # all the way back from the backend.
+        # |_ right now any precision error will be relayed
-
+        #    all the way back from the backend and vice-versa..
-        price = round(
+        #
-            price,
+        mkt: MktPair = self.curr_mkt
-            ndigits=mkt.price_tick_digits,
+        price: Decimal = mkt.quantize(
+            size=price,
+            quantity_type='price',
         )
-
         order = self._staged_order = Order(
             action=action,
             price=price,
@@ -378,7 +394,7 @@ class OrderMode:
                 'oid': oid,
             })
 
-        if order.price <= 0:
+        if float(order.price) <= 0:
             log.error(
                 '*!? Invalid `Order.price <= 0` ?!*\n'
                 # TODO: make this present multi-line in object form
@@ -515,14 +531,15 @@ class OrderMode:
             # if an order msg is provided update the line
             # **from** that msg.
             if order:
-                if order.price <= 0:
+                price: float = float(order.price)
+                if price <= 0:
                     log.error(f'Order has 0 price, cancelling..\n{order}')
                     self.cancel_orders([order.oid])
                     return None
 
-                line.set_level(order.price)
+                line.set_level(price)
                 self.on_level_change_update_next_order_info(
-                    level=order.price,
+                    level=price,
                     line=line,
                     order=order,
                     # use the corresponding position tracker for the
@@ -681,9 +698,9 @@ class OrderMode:
     ) -> Dialog | None:
         # NOTE: the `.order` attr **must** be set with the
         # equivalent order msg in order to be loaded.
-        order = msg.req
+        order: Order = msg.req
         oid = str(msg.oid)
-        symbol = order.symbol
+        symbol: str = order.symbol
 
         # TODO: MEGA UGGG ZONEEEE!
         src = msg.src
@@ -702,13 +719,22 @@ class OrderMode:
         order.oid = str(order.oid)
         order.brokers = [brokername]
 
-        # TODO: change this over to `MktPair`, but it's
+        # ?TODO? change this over to `MktPair`, but it's gonna be
-        # gonna be tough since we don't have any such data
+        # tough since we don't have any such data really in our
-        # really in our clearing msg schema..
+        # clearing msg schema..
-        order.symbol = Symbol.from_fqme(
+        # BUT WAIT! WHY do we even want/need this!?
-            fqsn=fqme,
+        #
-            info={},
+        # order.symbol = self.curr_mkt
-        )
+        #
+        # XXX, the old approach.. which i don't quire member why..
+        # -[ ] verify we for sure don't require this any more!
+        #  |_https://github.com/pikers/piker/issues/517
+        #
+        # order.symbol = Symbol.from_fqme(
+        #     fqsn=fqme,
+        #     info={},
+        # )
+
         maybe_dialog: Dialog | None = self.submit_order(
             send_msg=False,
             order=order,
@@ -766,6 +792,7 @@ async def open_order_mode(
             brokerd_accounts,
             ems_dialog_msgs,
         ),
+        tractor.trionics.collapse_eg(),
         trio.open_nursery() as tn,
 
     ):
@@ -1101,7 +1128,7 @@ async def process_trade_msg(
                         )
                     )
                 ):
-                    msg.req = order
+                    msg.req: Order = order
                     dialog: (
                         Dialog
                         # NOTE: on an invalid order submission (eg.
@@ -1166,7 +1193,7 @@ async def process_trade_msg(
             tm = time.time()
             mode.on_fill(
                 oid,
-                price=req.price,
+                price=float(req.price),
                 time_s=tm,
             )
             mode.lines.remove_line(uuid=oid)
@@ -1221,7 +1248,7 @@ async def process_trade_msg(
             tm = details['broker_time']
             mode.on_fill(
                 oid,
-                price=details['price'],
+                price=float(details['price']),
                 time_s=tm,
                 pointing='up' if action == 'buy' else 'down',
             )

tags

@@ -0,0 +1 @@
+TAG_feed_status_update	./piker/data/feed.py	/TAG_feed_status_update/

tests/test_ems.py

@@ -179,7 +179,7 @@ def test_ems_err_on_bad_broker(
         # NOTE: emsd should error on the actor's enabled modules
         # import phase, when looking for a backend named `doggy`.
         except tractor.RemoteActorError as re:
-            assert re.type == ModuleNotFoundError
+            assert re.type is ModuleNotFoundError
 
     run_and_tollerate_cancels(load_bad_fqme)
 

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)