config/brokers.toml

@@ -1,5 +1,6 @@
+################
 # ---- CEXY ----
-
+################
 [binance]
 accounts.paper = 'paper'
 
@@ -12,41 +13,28 @@ accounts.spot = 'spot'
 spot.use_testnet = false
 spot.api_key = ''
 spot.api_secret = ''
-# ------ binance ------
 
 
 [deribit]
-# std assets
 key_id = ''
 key_secret = ''
-# options
-accounts.option = 'option'
-option.use_testnet = false
-option.key_id = ''
-option.key_secret = ''
-# aux logging from `cryptofeed`
-option.log.filename = 'cryptofeed.log'
-option.log.level = 'DEBUG'
-option.log.disabled = true
-# ------ deribit ------
 
 
 [kraken]
 key_descr = ''
 api_key = ''
 secret = ''
-# ------ kraken ------
 
 
 [kucoin]
 key_id = ''
 key_secret = ''
 key_passphrase = ''
-# ------ kucoin ------
 
 
+################
 # -- BROKERZ ---
-
+################
 [questrade]
 refresh_token = ''
 access_token = ''
@@ -54,55 +42,44 @@ api_server = 'https://api06.iq.questrade.com/'
 expires_in = 1800
 token_type = 'Bearer'
 expires_at = 1616095326.355846
-# ------ questrade ------
 
 
 [ib]
-# define the (set of) host-port socketaddrs that
-# brokerd.ib will scan to connect to an API endpoint
-# (ib-gw or ib-tws listening instances)
 hosts = [
     '127.0.0.1',
 ]
+# XXX: the order in which ports will be scanned
+# (by the `brokerd` daemon-actor)
+# is determined # by the line order here.
+# TODO: when we eventually spawn gateways in our
+# container, we can just dynamically allocate these
+# using IBC.
 ports = [
     4002,  # gw
     7497,  # tws
 ]
 
-# When API endpoints are being scanned durin startup, the order
+# XXX: for a paper account the flex web query service
-# of user-defined-account "names" (as defined below) here
+# is not supported so you have to manually download
-# determines which py-client connection is given priority to be
+# and XML report and put it in a location that can be
-# used for data-feed-requests by according to whichever client
+# accessed by the ``brokerd.ib`` backend code for parsing.
-# connected to an API endpoing which reported the equivalent
+flex_token = ''
-# account number for that name.
+flex_trades_query_id = ''  # live account
+
+# when clients are being scanned this determines
+# which clients are preferred to be used for data
+# feeds based on the order of account names, if
+# detected as active on an API client.
 prefer_data_account = [
     'paper',
     'margin',
     'ira',
 ]
 
-# For long-term trades txn (transaction) history
-# processing (i.e your txn ledger with IB) you can
-# (automatically for live accounts) query the FLEX
-# report system for past history.
-#
-# (For paper accounts the web query service
-# is not supported so you have to manually download
-# an XML report and put it in a location that can be
-# accessed by our `brokerd.ib` backend code for parsing).
-#
-flex_token = ''
-flex_trades_query_id = ''  # live account
-
-# define "aliases" (names) for each account number
-# such that the names can be reffed and logged throughout
-# `piker.accounting` subsys and more easily
-# referred to  by the user.
-#
-# These keys will be the set exposed through the order-mode
-# account-selection UI so that numbers are never shown.
 [ib.accounts]
-paper = 'DU0000000'  # <- literal account #
+# the order in which accounts will be selectable
-margin = 'U0000000'
+# in the order mode UI (if found via clients during
-ira = 'U0000000'
+# API-app scanning)when a new symbol is loaded.
-# ------ ib ------
+paper = 'XX0000000'
+margin = 'X0000000'
+ira = 'X0000000'

dockering/ib/README.rst

@@ -1,138 +1,30 @@
 running ``ib`` gateway in ``docker``
 ------------------------------------
-We have a config based on a well maintained community
+We have a config based on the (now defunct)
-image from `@gnzsnz`:
+image from "waytrade":
 
-https://github.com/gnzsnz/ib-gateway-docker
+https://github.com/waytrade/ib-gateway-docker
 
-
+To startup this image with our custom settings
-To startup this image simply run the command::
+simply run the command::
 
     docker compose up
 
-(For further usage^ see the official `docker-compose`_ docs)
+And you should have the following socket-available services:
 
-
+- ``x11vnc1@127.0.0.1:3003``
-And you should have the following socket-available services by
-default:
-
-- ``x11vnc1 @ 127.0.0.1:5900``
 - ``ib-gw@127.0.0.1:4002``
 
-You can now attach to the container via a VNC client with password-auth;
+You can attach to the container via a VNC client
-here is an example using ``vncclient`` on ``linux``::
+without password auth.
 
-    vncviewer localhost:5900
+SECURITY STUFF!?!?!
-
+-------------------
-now enter the pw (password) you set via an (see second code blob)
+Though "``ib``" claims they host filter connections outside
-`.env file`_ or pw-file according to the `credentials section`_.
+localhost (aka ``127.0.0.1``) it's probably better if you filter
-
+the socket at the OS level using a stateless firewall rule::
-If you want to change away from their default config see the example
-`docker-compose.yml`-config issue and config-section of the readme,
-
-  - https://github.com/gnzsnz/ib-gateway-docker?tab=readme-ov-file#configuration
-  - https://github.com/gnzsnz/ib-gateway-docker/discussions/103
-
-.. _.env file: https://github.com/gnzsnz/ib-gateway-docker?tab=readme-ov-file#how-to-use-it
-.. _docker-compose: https://docs.docker.com/compose/
-.. _credentials section: https://github.com/gnzsnz/ib-gateway-docker?tab=readme-ov-file#credentials
-
-
-Connecting to the API from `piker`
----------------------------------
-In order to expose the container's API endpoint to the
-`brokerd/datad/ib` actor, we need to add a section to the user's
-`brokers.toml` config (note the below is similar to the repo-shipped
-template file),
-
-.. code:: toml
-
-    [ib]
-    # define the (set of) host-port socketaddrs that
-    # brokerd.ib will scan to connect to an API endpoint
-    # (ib-gw or ib-tws listening instances)
-    hosts = [
-        '127.0.0.1',
-    ]
-    ports = [
-        4002,  # gw
-        7497,  # tws
-    ]
-
-    # When API endpoints are being scanned durin startup, the order
-    # of user-defined-account "names" (as defined below) here
-    # determines which py-client connection is given priority to be
-    # used for data-feed-requests by according to whichever client
-    # connected to an API endpoing which reported the equivalent
-    # account number for that name.
-    prefer_data_account = [
-        'paper',
-        'margin',
-        'ira',
-    ]
-
-    # define "aliases" (names) for each account number
-    # such that the names can be reffed and logged throughout
-    # `piker.accounting` subsys and more easily
-    # referred to  by the user.
-    #
-    # These keys will be the set exposed through the order-mode
-    # account-selection UI so that numbers are never shown.
-    [ib.accounts]
-    paper = 'XX0000000'
-    margin = 'X0000000'
-    ira = 'X0000000'
-
-
-the broker daemon can also connect to the container's VNC server for
-added functionalies including,
-
-- viewing the API endpoint program's GUI for manual interventions,
-- workarounds for historical data throttling using hotkey hacks,
-
-Add a further section to `brokers.toml` which maps each API-ep's
-port to a table of VNC server connection info like,
-
-.. code:: toml
-
-    [ib.vnc_addrs]
-    4002 = {host = 'localhost', port = 5900, pw = 'doggy'}
-
-The `pw = 'doggy'` here ^ should the same value as the particular
-container instances `.env` file setting (when it was run),
-
-.. code:: ini
-
-    VNC_SERVER_PASSWORD='doggy'
-
-
-IF you also want to run ``TWS``
--------------------------------
-You can also run it containerized,
-
-https://github.com/gnzsnz/ib-gateway-docker?tab=readme-ov-file#using-tws
-
-
-SECURITY stuff (advanced, only if you're paranoid)
---------------------------------------------------
-First and foremost if doing a "distributed" container setup where you
-run the ``ib-gw`` docker container and your connecting API client
-(likely ``ib_async`` from python) on **different hosts** be sure to
-read the `security considerations`_ section!
-
-And for a further (somewhat paranoid) perspective from
-a long-time-ago serious devops eng..
-
-Though "``ib``" claims they filter remote host connections outside
-``localhost`` (aka ``127.0.0.1`` on ipv4) it's prolly justified if
-you'd like to filter the socket at the *OS level* using a stateless
-firewall rule::
 
     ip rule add not unicast iif lo to 0.0.0.0/0 dport 4002
 
-
+We will soon have this baked into our own custom image but for
-We will soon have this either baked into our own custom derivative
+now you'll have to do it urself dawgy.
-image (or patched into the current upstream one after further testin)
-but for now you'll have to do it urself, diggity dawg.
-
-.. _security considerations: https://github.com/gnzsnz/ib-gateway-docker?tab=readme-ov-file#security-considerations

dockering/ib/docker-compose.yml

@@ -1,15 +1,10 @@
-# a community maintained IB API container!
+# rework from the original @
-#
+# https://github.com/waytrade/ib-gateway-docker/blob/master/docker-compose.yml
-# https://github.com/gnzsnz/ib-gateway-docker
+version: "3.5"
-#
+
-# For piker we (currently) include some minor deviations
-# for some config files in the `volumes` section.
-#
-# See full configuration settings @
-# - https://github.com/gnzsnz/ib-gateway-docker?tab=readme-ov-file#configuration
-# - https://github.com/gnzsnz/ib-gateway-docker/discussions/103
 
 services:
+
   ib_gw_paper:
 
     # apparently java is a mega cukc:
@@ -55,22 +50,16 @@ services:
         target: /root/scripts/run_x11_vnc.sh
         read_only: true
 
-    # NOTE: an alt method to fill these out is to
+    # NOTE:to fill these out, define an `.env` file in the same dir as
-    # define an `.env` file in the same dir as
+    # this compose file which looks something like:
-    # this compose file.
+    # TWS_USERID='myuser'
+    # TWS_PASSWORD='guest'
     environment:
       TWS_USERID: ${TWS_USERID}
-      # TWS_USERID: 'myuser'
       TWS_PASSWORD: ${TWS_PASSWORD}
-      # TWS_PASSWORD: 'guest'
+      TRADING_MODE: 'paper'
-      TRADING_MODE: ${TRADING_MODE}
+      VNC_SERVER_PASSWORD: 'doggy'
-      # TRADING_MODE: 'paper'
+      VNC_SERVER_PORT: '3003'
-      VNC_SERVER_PASSWORD: ${VNC_SERVER_PASSWORD}
-      # VNC_SERVER_PASSWORD: 'doggy'
-
-      # TODO, see if we can get this supported like it
-      # was on the old `waytrade` image?
-      # VNC_SERVER_PORT: '3003'
 
     # ports:
     #   - target: 4002
@@ -87,9 +76,6 @@ services:
       # - "127.0.0.1:4002:4002"
       # - "127.0.0.1:5900:5900"
 
-  # TODO, a masked but working example of dual paper + live
-  # ib-gw instances running in a single app run!
-  #
   # ib_gw_live:
   #   image: waytrade/ib-gateway:1012.2i
   #   restart: no

piker/accounting/__init__.py

@@ -33,6 +33,7 @@ from ._pos import (
     Account,
     load_account,
     load_account_from_ledger,
+    open_pps,
     open_account,
     Position,
 )
@@ -67,6 +68,7 @@ __all__ = [
     'load_account_from_ledger',
     'mk_allocator',
     'open_account',
+    'open_pps',
     'open_trade_ledger',
     'unpack_fqme',
     'DerivTypes',

piker/accounting/_pos.py

@@ -356,12 +356,13 @@ class Position(Struct):
     ) -> bool:
         '''
         Update clearing table by calculating the rolling ppu and
-        (accumulative) size in both the clears entry and local attrs
+        (accumulative) size in both the clears entry and local
-        state.
+        attrs state.
 
         Inserts are always done in datetime sorted order.
 
         '''
+        # added: bool = False
         tid: str = t.tid
         if tid in self._events:
             log.debug(
@@ -369,7 +370,7 @@ class Position(Struct):
                 f'\n'
                 f'{t}\n'
             )
-            return False
+            # return added
 
         # TODO: apparently this IS possible with a dict but not
         # common and probably not that beneficial unless we're also
@@ -450,12 +451,6 @@ class Position(Struct):
     # def suggest_split(self) -> float:
     #     ...
 
-    # ?TODO, for sending rendered state over the wire?
-    # def summary(self) -> PositionSummary:
-        # do minimal conversion to a subset of fields
-        # currently defined in `.clearing._messages.BrokerdPosition`
-
-
 
 class Account(Struct):
     '''
@@ -754,7 +749,7 @@ class Account(Struct):
         # XXX WTF: if we use a tomlkit.Integer here we get this
         # super weird --1 thing going on for cumsize!?1!
         # NOTE: the fix was to always float() the size value loaded
-        # in open_account() below!
+        # in open_pps() below!
         config.write(
             config=self.conf,
             path=self.conf_path,
@@ -938,6 +933,7 @@ def open_account(
             clears_table['dt'] = dt
             trans.append(Transaction(
                 fqme=bs_mktid,
+                # sym=mkt,
                 bs_mktid=bs_mktid,
                 tid=tid,
                 # XXX: not sure why sometimes these are loaded as
@@ -960,18 +956,7 @@ def open_account(
         ):
             expiry: pendulum.DateTime = pendulum.parse(expiry)
 
-        # !XXX, should never be duplicates over
+        pp = pp_objs[bs_mktid] = Position(
-        # a backend-(broker)-system's unique market-IDs!
-        if pos := pp_objs.get(bs_mktid):
-            if mkt != pos.mkt:
-                log.warning(
-                    f'Duplicated position but diff `MktPair.fqme` ??\n'
-                    f'bs_mktid: {bs_mktid!r}\n'
-                    f'pos.mkt: {pos.mkt}\n'
-                    f'mkt: {mkt}\n'
-                )
-        else:
-            pos = pp_objs[bs_mktid] = Position(
             mkt,
             split_ratio=split_ratio,
             bs_mktid=bs_mktid,
@@ -983,13 +968,8 @@ def open_account(
         # state, since today's records may have already been
         # processed!
         for t in trans:
-            added: bool = pos.add_clear(t)
+            pp.add_clear(t)
-            if not added:
+
-                log.warning(
-                    f'Txn already recorded in pp ??\n'
-                    f'\n'
-                    f'{t}\n'
-                )
     try:
         yield acnt
     finally:
@@ -997,6 +977,20 @@ def open_account(
             acnt.write_config()
 
 
+# TODO: drop the old name and THIS!
+@cm
+def open_pps(
+    *args,
+    **kwargs,
+) -> Generator[Account, None, None]:
+    log.warning(
+        '`open_pps()` is now deprecated!\n'
+        'Please use `with open_account() as cnt:`'
+    )
+    with open_account(*args, **kwargs) as acnt:
+        yield acnt
+
+
 def load_account_from_ledger(
 
     brokername: str,

piker/accounting/calc.py

@@ -22,9 +22,7 @@ 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,
@@ -32,7 +30,6 @@ from typing import (
     TYPE_CHECKING,
 )
 
-from tractor.devx import maybe_open_crash_handler
 import polars as pl
 from pendulum import (
     DateTime,
@@ -40,16 +37,12 @@ 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],
@@ -245,9 +238,6 @@ 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
@@ -260,90 +250,52 @@ def iter_by_dt(
         # get best parser for this record..
         for k in parsers:
             if (
-                (v := getattr(tx, k, None))
+                isdict and k in tx
                 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
+                    and (parser := parsers.get(k))
-                    (parser := parsers.get(k))
                 ):
-                    ret = parser(v)
+                    return parser(v)
                 else:
-                    ret = v
+                    return v
-
-                return ret
 
         else:
-                log.debug(
+            # TODO: move to top?
-                    f'Parser-field not found in txn\n'
+            from piker.log import get_logger
-                    f'\n'
+            log = get_logger(__name__)
-                    f'parser-field: {k!r}\n'
-                    f'txn: {tx!r}\n'
-                    f'\n'
-                    f'Trying next..\n'
-                )
-                continue
 
-        # XXX: we should never really get here bc it means some kinda
-        # bad txn-record (field) data..
-        #
-        # -> set the `debug_mode = True` if you want to trace such
-        # cases from REPL ;)
-        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!r}\n'
+                f'{tx}\n'
             )
-            report: str = (
+            return from_timestamp(0)
-                f'No supported time-field found in txn !?\n'
+            # breakpoint()
-                f'\n'
-                f'supported-time-fields: {parsers!r}\n'
-                f'\n'
-                f'txn: {tx!r}\n'
-            )
-            if debug:
-                with maybe_open_crash_handler(
-                    pdb=debug,
-                    raise_on_exit=False,
-                ):
-                    raise ValueError(report)
-            else:
-                log.error(report)
 
-            if _invalid is not None:
-                _invalid.append(tx)
-                return from_timestamp(0.)
 
     entry: tuple[str, dict] | Transaction
-    invalid: list = []
     for entry in sorted(
         records,
-        key=key or partial(
+        key=key or dyn_parse_to_dt,
-            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
 
@@ -406,7 +358,6 @@ def open_ledger_dfs(
     acctname: str,
 
     ledger: TransactionLedger | None = None,
-    debug_mode: bool = False,
 
     **kwargs,
 
@@ -421,10 +372,8 @@ def open_ledger_dfs(
     can update the ledger on exit.
 
     '''
-    with maybe_open_crash_handler(
+    from piker.toolz import open_crash_handler
-        pdb=debug_mode,
+    with open_crash_handler():
-        # raise_on_exit=False,
-    ):
         if not ledger:
             import time
             from ._ledger import open_trade_ledger

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 (
+from typing import List, Dict, Any, Optional
-    Any,
-)
 
 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]
+    tickers: List[str]
-
+) -> Dict[str, Dict[str, Any]]:
-) -> dict[str, dict[str, Any]]:
+    """Return quotes dict for ``tickers``.
-    '''
+    """
-    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)
 
@@ -82,15 +74,13 @@ async def stocks_quote(
 async def option_chain(
     brokermod: ModuleType,
     symbol: str,
-    date: str|None = None,
+    date: Optional[str] = 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])
@@ -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]]]:
+) -> 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(
+async def search_w_brokerd(name: str, pattern: str) -> dict:
-    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
+    Return MktPair info from broker including src and dst assets.
-    backend broker tradable src/dst asset pair.
 
     '''
     async with open_cached_client(brokermod.name) as client:

piker/brokers/ib/_util.py

@@ -20,11 +20,6 @@ runnable script-programs.
 
 '''
 from __future__ import annotations
-from datetime import (  # noqa
-    datetime,
-    date,
-    tzinfo as TzInfo,
-)
 from functools import partial
 from typing import (
     Literal,
@@ -38,7 +33,7 @@ from piker.brokers._util import get_logger
 
 if TYPE_CHECKING:
     from .api import Client
-    import i3ipc
+    from ib_insync import IB
 
 log = get_logger('piker.brokers.ib')
 
@@ -53,39 +48,8 @@ _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(
-    client: Client,
-):
-    '''
-    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:
-        vnc_sockaddr: str = client.conf.vnc_addrs
-        log.exception(
-            no_setup_msg.format(vnc_sockaddr=vnc_sockaddr)
-        )
-        return False
-
-
 async def data_reset_hack(
+    # vnc_host: str,
     client: Client,
     reset_type: Literal['data', 'connection'],
 
@@ -117,60 +81,65 @@ async def data_reset_hack(
           that need to be wrangle.
 
     '''
+    ib_client: IB = client.ib
+
     # look up any user defined vnc socket address mapped from
     # a particular API socket port.
-    vnc_addrs: tuple[str]|None = client.conf.get('vnc_addrs')
+    api_port: str = str(ib_client.client.port)
-    if not vnc_addrs:
+    vnc_host: str
+    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=client.conf)
+            no_setup_msg
             +
             'REQUIRES A `vnc_addrs: array` ENTRY'
         )
 
+    vnc_host, vnc_port = vnc_sockaddr.get(
+        api_port,
+        ('localhost', 3003)
+    )
     global _reset_tech
+
     match _reset_tech:
         case 'vnc':
             try:
                 await tractor.to_asyncio.run_task(
                     partial(
                         vnc_click_hack,
-                        client=client,
+                        host=vnc_host,
+                        port=vnc_port,
                     )
                 )
-            except (
+            except OSError:
-                OSError,  # no VNC server avail..
+                if vnc_host != 'localhost':
-                PermissionError,  # asyncvnc pw fail..
+                    log.warning(no_setup_msg)
-            ):
+                    return False
+
                 try:
                     import i3ipc  # noqa  (since a deps dynamic check)
                 except ModuleNotFoundError:
-                    log.warning(
+                    log.warning(no_setup_msg)
-                        no_setup_msg.format(vnc_sockaddr=client.conf)
-                    )
                     return False
 
-                # XXX, Xorg only workaround..
+                try:
-                # TODO? remove now that we have `pyvnc`?
+                    i3ipc_xdotool_manual_click_hack()
-                # if vnc_host not in {
+                    _reset_tech = 'i3ipc_xdotool'
-                #     'localhost',
+                    return True
-                #     '127.0.0.1',
+                except OSError:
-                # }:
+                    log.exception(no_setup_msg)
-                #     focussed, matches = i3ipc_fin_wins_titled()
+                    return False
-                #     if not matches:
-                #         log.warning(
-                #             no_setup_msg.format(vnc_sockaddr=vnc_sockaddr)
-                #         )
-                #         return False
-                #     else:
-                #         try_xdo_manual(vnc_sockaddr)
-
-                # localhost but no vnc-client or it borked..
-                else:
-                    try_xdo_manual(client)
 
         case 'i3ipc_xdotool':
-            try_xdo_manual(client)
+            i3ipc_xdotool_manual_click_hack()
-            # i3ipc_xdotool_manual_click_hack()
 
         case _ as tech:
             raise RuntimeError(f'{tech} is not supported for reset tech!?')
@@ -180,66 +149,21 @@ async def data_reset_hack(
 
 
 async def vnc_click_hack(
-    client: Client,
+    host: str,
-    reset_type: str = 'data',
+    port: int,
-    pw: str|None = None,
+    reset_type: str = 'data'
-
 ) -> None:
     '''
     Reset the data or network connection for the VNC attached
-    ib-gateway using a (magic) keybinding combo.
+    ib gateway using magic combos.
-
-    A vnc-server password can be set either by an input `pw` param or
-    set in the client's config with the latter loaded from the user's
-    `brokers.toml` in a vnc-addrs-port-mapping section,
-
-    .. code:: toml
-
-        [ib.vnc_addrs]
-        4002 = {host = 'localhost', port = 5900, pw = 'doggy'}
 
     '''
-    api_port: str = str(client.ib.client.port)
-    conf: dict = client.conf
-    vnc_addrs: dict[int, tuple] = conf.get('vnc_addrs')
-    if not vnc_addrs:
-        return None
-
-    addr_entry: dict|tuple = vnc_addrs.get(
-        api_port,
-        ('localhost', 5900)  # a typical default
-    )
-    if pw is None:
-        match addr_entry:
-            case (
-                host,
-                port,
-            ):
-                pass
-
-            case {
-                'host': host,
-                'port': port,
-                'pw': pw
-            }:
-                pass
-
-            case _:
-                raise ValueError(
-                    f'Invalid `ib.vnc_addrs` entry ?\n'
-                    f'{addr_entry!r}\n'
-                )
     try:
-        from pyvnc import (
+        import asyncvnc
-            AsyncVNCClient,
-            VNCConfig,
-            Point,
-            MOUSE_BUTTON_LEFT,
-        )
     except ModuleNotFoundError:
         log.warning(
             "In order to leverage `piker`'s built-in data reset hacks, install "
-            "the `pyvnc` project: https://github.com/regulad/pyvnc.git"
+            "the `asyncvnc` project: https://github.com/barneygale/asyncvnc"
         )
         return
 
@@ -250,79 +174,24 @@ async def vnc_click_hack(
         'connection': 'r'
     }[reset_type]
 
-    with tractor.devx.open_crash_handler():
+    async with asyncvnc.connect(
-        client = await AsyncVNCClient.connect(
+        host,
-            VNCConfig(
-                host=host,
         port=port,
-                password=pw,
+
-            )
+        # TODO: doesn't work see:
-        )
+        # https://github.com/barneygale/asyncvnc/issues/7
-        async with client:
+        # password='ibcansmbz',
+
+    ) as client:
+
         # move to middle of screen
         # 640x1800
-            await client.move(
+        client.mouse.move(
-                Point(
+            x=500,
-                    500,
+            y=500,
-                    500,
         )
-            )
+        client.mouse.click()
-            # ensure the ib-gw window is active
+        client.keyboard.press('Ctrl', 'Alt', key)  # keys are stacked
-            await client.click(MOUSE_BUTTON_LEFT)
-            # send the hotkeys combo B)
-            await client.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:
@@ -330,10 +199,29 @@ 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()
+    import i3ipc
-    orig_win_id = focussed.window
+    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:
+        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
@@ -379,99 +267,3 @@ def i3ipc_xdotool_manual_click_hack() -> None:
         ])
     except subprocess.TimeoutExpired:
         log.exception('xdotool timed out?')
-
-
-
-def is_current_time_in_range(
-    start_dt: datetime,
-    end_dt: datetime,
-) -> bool:
-    '''
-    Check if current time is within the datetime range.
-
-    Use any/the-same timezone as provided by `start_dt.tzinfo` value
-    in the range.
-
-    '''
-    now: datetime = datetime.now(start_dt.tzinfo)
-    return start_dt <= now <= end_dt
-
-
-# TODO, put this into `._util` and call it from here!
-#
-# NOTE, this was generated by @guille from a gpt5 prompt
-# and was originally thot to be needed before learning about
-# `ib_insync.contract.ContractDetails._parseSessions()` and
-# it's downstream meths..
-#
-# This is still likely useful to keep for now to parse the
-# `.tradingHours: str` value manually if we ever decide
-# to move off `ib_async` and implement our own `trio`/`anyio`
-# based version Bp
-#
-# >attempt to parse the retarted ib "time stampy thing" they
-# >do for "venue hours" with this.. written by
-# >gpt5-"thinking",
-#
-
-
-def parse_trading_hours(
-    spec: str,
-    tz: TzInfo|None = None
-) -> dict[
-    date,
-    tuple[datetime, datetime]
-]|None:
-    '''
-    Parse venue hours like:
-    'YYYYMMDD:HHMM-YYYYMMDD:HHMM;YYYYMMDD:CLOSED;...'
-
-    Returns `dict[date] = (open_dt, close_dt)` or `None` if
-    closed.
-
-    '''
-    if (
-        not isinstance(spec, str)
-        or
-        not spec
-    ):
-        raise ValueError('spec must be a non-empty string')
-
-    out: dict[
-        date,
-        tuple[datetime, datetime]
-    ]|None = {}
-
-    for part in (p.strip() for p in spec.split(';') if p.strip()):
-        if part.endswith(':CLOSED'):
-            day_s, _ = part.split(':', 1)
-            d = datetime.strptime(day_s, '%Y%m%d').date()
-            out[d] = None
-            continue
-
-        try:
-            start_s, end_s = part.split('-', 1)
-            start_dt = datetime.strptime(start_s, '%Y%m%d:%H%M')
-            end_dt = datetime.strptime(end_s, '%Y%m%d:%H%M')
-        except ValueError as exc:
-            raise ValueError(f'invalid segment: {part}') from exc
-
-        if tz is not None:
-            start_dt = start_dt.replace(tzinfo=tz)
-            end_dt = end_dt.replace(tzinfo=tz)
-
-        out[start_dt.date()] = (start_dt, end_dt)
-
-    return out
-
-
-# ORIG desired usage,
-#
-# TODO, for non-drunk tomorrow,
-# - call above fn and check that `output[today] is not None`
-# trading_hrs: dict = parse_trading_hours(
-#     details.tradingHours
-# )
-# liq_hrs: dict = parse_trading_hours(
-#     details.liquidHours
-    # )

piker/brokers/ib/api.py

@@ -944,7 +944,6 @@ class Client:
                     )
                     if tkr:
                         break
-
                 except TimeoutError as err:
                     timeouterr = err
                     await asyncio.sleep(0.01)
@@ -953,9 +952,7 @@ class Client:
                 else:
                     if not warnset:
                         log.warning(
-                            f'Quote req timed out..\n'
+                            f'Quote req timed out..maybe venue is closed?\n'
-                            f'Maybe the venue is closed?\n'
-                            f'\n'
                             f'{asdict(contract)}'
                         )
                         warnset = True
@@ -967,11 +964,9 @@ class Client:
                 )
                 break
         else:
-            if (
+            if timeouterr and raise_on_timeout:
-                timeouterr
+                import pdbp
-                and
+                pdbp.set_trace()
-                raise_on_timeout
-            ):
                 raise timeouterr
 
             if not warnset:
@@ -1368,7 +1363,9 @@ async def load_aio_clients(
 
 
 async def load_clients_for_trio(
-    chan: tractor.to_asyncio.LinkedTaskChannel,
+    from_trio: asyncio.Queue,
+    to_trio: trio.abc.SendChannel,
+
 ) -> None:
     '''
     Pure async mngr proxy to ``load_aio_clients()``.
@@ -1381,7 +1378,8 @@ async def load_clients_for_trio(
         disconnect_on_exit=False,
     ) as accts2clients:
 
-        chan.started_nowait(accts2clients)
+        to_trio.send_nowait(accts2clients)
+
         # TODO: maybe a sync event to wait on instead?
         await asyncio.sleep(float('inf'))
 
@@ -1528,22 +1526,23 @@ class MethodProxy:
 
 
 async def open_aio_client_method_relay(
-    chan: tractor.to_asyncio.LinkedTaskChannel,
+    from_trio: asyncio.Queue,
+    to_trio: trio.abc.SendChannel,
     client: Client,
     event_consumers: dict[str, trio.Event],
 
 ) -> None:
 
     # sync with `open_client_proxy()` caller
-    chan.started_nowait(client)
+    to_trio.send_nowait(client)
 
     # TODO: separate channel for error handling?
-    client.inline_errors(chan)
+    client.inline_errors(to_trio)
 
     # relay all method requests to ``asyncio``-side client and deliver
     # back results
-    while not chan._to_trio._closed:  # <- TODO, better check like `._web_bs`?
+    while not to_trio._closed:
-        msg: tuple[str, dict]|dict|None = await chan.get()
+        msg: tuple[str, dict] | dict | None = await from_trio.get()
         match msg:
             case None:  # termination sentinel
                 log.info('asyncio `Client` method-proxy SHUTDOWN!')
@@ -1556,7 +1555,7 @@ async def open_aio_client_method_relay(
                 try:
                     resp = await meth(**kwargs)
                     # echo the msg back
-                    chan.send_nowait({'result': resp})
+                    to_trio.send_nowait({'result': resp})
 
                 except (
                     RequestError,
@@ -1564,10 +1563,10 @@ async def open_aio_client_method_relay(
                     # TODO: relay all errors to trio?
                     # BaseException,
                 ) as err:
-                    chan.send_nowait({'exception': err})
+                    to_trio.send_nowait({'exception': err})
 
             case {'error': content}:
-                chan.send_nowait({'exception': content})
+                to_trio.send_nowait({'exception': content})
 
             case _:
                 raise ValueError(f'Unhandled msg {msg}')

piker/brokers/ib/broker.py

@@ -117,11 +117,7 @@ def pack_position(
             symbol=fqme,
             currency=con.currency,
             size=float(pos.position),
-            avg_price=(
+            avg_price=float(pos.avgCost) / float(con.multiplier or 1.0),
-                float(pos.avgCost)
-                /
-                float(con.multiplier or 1.0)
-            ),
         ),
     )
 
@@ -362,10 +358,6 @@ async def update_and_audit_pos_msg(
         size=ibpos.position,
 
         avg_price=pikerpos.ppu,
-
-        # XXX ensures matching even if multiple venue-names
-        # in `.bs_fqme`, likely from txn records..
-        bs_mktid=mkt.bs_mktid,
     )
 
     ibfmtmsg: str = pformat(ibpos._asdict())
@@ -434,8 +426,7 @@ async def aggr_open_orders(
 
 ) -> None:
     '''
-    Collect all open orders from client and fill in `order_msgs:
+    Collect all open orders from client and fill in `order_msgs: list`.
-    list`.
 
     '''
     trades: list[Trade] = client.ib.openTrades()
@@ -556,10 +547,7 @@ async def open_trade_dialog(
         ),
 
         # TODO: do this as part of `open_account()`!?
-        open_symcache(
+        open_symcache('ib', only_from_memcache=True) as symcache,
-            'ib',
-            only_from_memcache=True,
-        ) as symcache,
     ):
         # Open a trade ledgers stack for appending trade records over
         # multiple accounts.
@@ -567,10 +555,8 @@ async def open_trade_dialog(
         ledgers: dict[str, TransactionLedger] = {}
         tables: dict[str, Account] = {}
         order_msgs: list[Status] = []
-        conf: dict = get_config()
+        conf = get_config()
-        accounts_def_inv: bidict[str, str] = bidict(
+        accounts_def_inv: bidict[str, str] = bidict(conf['accounts']).inverse
-            conf['accounts']
-        ).inverse
 
         with (
             ExitStack() as lstack,
@@ -720,11 +706,7 @@ async def open_trade_dialog(
             # client-account and build out position msgs to deliver to
             # EMS.
             for acctid, acnt in tables.items():
-                active_pps: dict[str, Position]
+                active_pps, closed_pps = acnt.dump_active()
-                (
-                    active_pps,
-                    closed_pps,
-                ) = acnt.dump_active()
 
                 for pps in [active_pps, closed_pps]:
                     piker_pps: list[Position] = list(pps.values())
@@ -740,7 +722,6 @@ async def open_trade_dialog(
                         )
                         if ibpos:
                             bs_mktid: str = str(ibpos.contract.conId)
-
                             msg = await update_and_audit_pos_msg(
                                 acctid,
                                 pikerpos,
@@ -1262,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']
+                if code in {
-                reqid: str = str(err['reqid'])
+                    200,  # uhh
-
-                # "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(
+                    log.error(f'TWS external order error:\n{pformat(err)}')
-                        f'TWS external order error ??\n'
-                        f'{pformat(err)}\n'
-                    )
 
                 flow: dict = dict(
                     flows.get(reqid)

piker/brokers/ib/feed.py

@@ -1,5 +1,5 @@
 # piker: trading gear for hackers
-# Copyright (C) 2018-forever Tyler Goodlet (in stewardship for pikers)
+# 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
@@ -13,12 +13,10 @@
 
 # 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/>.
+"""
+Data feed endpoints pre-wrapped and ready for use with ``tractor``/``trio``.
 
-'''
+"""
-Data feed endpoints pre-wrapped and ready for use with `tractor`/`trio`
-via "infected-asyncio-mode".
-
-'''
 from __future__ import annotations
 import asyncio
 from contextlib import (
@@ -28,6 +26,7 @@ from dataclasses import asdict
 from datetime import datetime
 from functools import partial
 from pprint import pformat
+from math import isnan
 import time
 from typing import (
     Any,
@@ -41,6 +40,7 @@ import numpy as np
 from pendulum import (
     now,
     from_timestamp,
+    # DateTime,
     Duration,
     duration as mk_duration,
 )
@@ -69,10 +69,7 @@ from .api import (
     Contract,
     RequestError,
 )
-from ._util import (
+from ._util import data_reset_hack
-    data_reset_hack,
-    is_current_time_in_range,
-)
 from .symbols import get_mkt_info
 
 if TYPE_CHECKING:
@@ -187,8 +184,7 @@ async def open_history_client(
 
             if (
                 start_dt
-                and
+                and start_dt.timestamp() == 0
-                start_dt.timestamp() == 0
             ):
                 await tractor.pause()
 
@@ -207,16 +203,14 @@ async def open_history_client(
             ):
                 count += 1
                 mean += latency / count
-                log.debug(
+                print(
                     f'HISTORY FRAME QUERY LATENCY: {latency}\n'
                     f'mean: {mean}'
                 )
 
             # could be trying to retreive bars over weekend
             if out is None:
-                log.error(
+                log.error(f"Can't grab bars starting at {end_dt}!?!?")
-                    f"No bars starting at {end_dt!r} !?!?"
-                )
                 if (
                     end_dt
                     and head_dt
@@ -291,9 +285,8 @@ _pacing: str = (
 
 async def wait_on_data_reset(
     proxy: MethodProxy,
-
     reset_type: str = 'data',
-    timeout: float = 16,
+    timeout: float = 16, # float('inf'),
 
     task_status: TaskStatus[
         tuple[
@@ -302,47 +295,29 @@ async def wait_on_data_reset(
         ]
     ] = trio.TASK_STATUS_IGNORED,
 ) -> bool:
-    '''
-    Wait on a (global-ish) "data-farm" event to be emitted
-    by the IB api server.
 
-    Allows syncing to reconnect event-messages emitted on the API
+    # TODO: we might have to put a task lock around this
-    console, such as:
+    # method..
-
+    hist_ev = proxy.status_event(
-     - 'HMDS data farm connection is OK:ushmds'
-     - 'Market data farm is connecting:usfuture'
-     - 'Market data farm connection is OK:usfuture'
-
-     Deliver a `(cs, done: Event)` pair to the caller to support it
-     waiting or cancelling the associated "data-reset-request";
-     normally a manual data-reset-req is expected to be the cause and
-     thus trigger such events (such as our click-hack-magic from
-     `.ib._util`).
-
-    '''
-    # ?TODO, do we need a task-lock around this method?
-    #
-    # register for an API "status event" wrapped for `trio`-sync.
-    hist_ev: trio.Event = proxy.status_event(
         'HMDS data farm connection is OK:ushmds'
     )
-    #
+
-    # ^TODO: other event-messages we might want to support waiting-for
+    # TODO: other event messages we might want to try and
-    # but i wasn't able to get reliable..
+    # wait for but i wasn't able to get any of this
-    #
+    # reliable..
     # reconnect_start = proxy.status_event(
     #     'Market data farm is connecting:usfuture'
     # )
     # live_ev = proxy.status_event(
     #     'Market data farm connection is OK:usfuture'
     # )
-
     # try to wait on the reset event(s) to arrive, a timeout
     # will trigger a retry up to 6 times (for now).
     client: Client = proxy._aio_ns
 
     done = trio.Event()
     with trio.move_on_after(timeout) as cs:
+
         task_status.started((cs, done))
 
         log.warning(
@@ -421,9 +396,8 @@ async def get_bars(
     bool,  # timed out hint
 ]:
     '''
-    Request-n-retrieve historical data frames from a `trio.Task`
+    Retrieve historical data from a ``trio``-side task using
-    using a `MethoProxy` to query the `asyncio`-side's
+    a ``MethoProxy``.
-    `.ib.api.Client` methods.
 
     '''
     global _data_resetter_task, _failed_resets
@@ -633,10 +607,7 @@ async def get_bars(
     # such that simultaneous symbol queries don't try data resettingn
     # too fast..
     unset_resetter: bool = False
-    async with (
+    async with trio.open_nursery() as tn:
-        tractor.trionics.collapse_eg(),
-        trio.open_nursery() as tn
-    ):
 
         # start history request that we allow
         # to run indefinitely until a result is acquired
@@ -682,12 +653,14 @@ async def get_bars(
     )
 
 
-# per-actor cache of inter-eventloop-chans
 _quote_streams: dict[str, trio.abc.ReceiveStream] = {}
 
 
 async def _setup_quote_stream(
-    chan: tractor.to_asyncio.LinkedTaskChannel,
+
+    from_trio: asyncio.Queue,
+    to_trio: trio.abc.SendChannel,
+
     symbol: str,
     opts: tuple[int] = (
         '375',  # RT trade volume (excludes utrades)
@@ -705,13 +678,10 @@ async def _setup_quote_stream(
 
 ) -> trio.abc.ReceiveChannel:
     '''
-    Stream L1 quotes via the `Ticker.updateEvent.connect(push)`
+    Stream a ticker using the std L1 api.
-    callback API by registering a `push` callback which simply
-    `chan.send_nowait()`s quote msgs back to the calling
-    parent-`trio.Task`-side.
 
-    NOTE, that this task-fn is run on the `asyncio.Task`-side ONLY
+    This task is ``asyncio``-side and must be called from
-    and is thus run via `tractor.to_asyncio.open_channel_from()`.
+    ``tractor.to_asyncio.open_channel_from()``.
 
     '''
     global _quote_streams
@@ -719,79 +689,39 @@ async def _setup_quote_stream(
     async with load_aio_clients(
         disconnect_on_exit=False,
     ) as accts2clients:
-
-        # XXX since this is an `asyncio.Task`, we must use
-        # tractor.pause_from_sync()
-
         caccount_name, client = get_preferred_data_client(accts2clients)
-        contract = (
+        contract = contract or (await client.find_contract(symbol))
-            contract
+        to_trio.send_nowait(contract)  # cuz why not
-            or
+        ticker: Ticker = client.ib.reqMktData(contract, ','.join(opts))
-            (await client.find_contract(symbol))
-        )
-        chan.started_nowait(contract)  # cuz why not
-        ticker: Ticker = client.ib.reqMktData(
-            contract,
-            ','.join(opts),
-        )
-        maybe_exc: BaseException|None = None
-        handler_tries: int = 0
-        aio_task: asyncio.Task = asyncio.current_task()
 
-        # ?TODO? this API is batch-wise and quite slow-af but,
+        # NOTE: it's batch-wise and slow af but I guess could
-        # - seems to be 5s updates?
+        # be good for backchecking? Seems to be every 5s maybe?
-        # - maybe we could use it for backchecking?
-        #
         # ticker: Ticker = client.ib.reqTickByTickData(
         #     contract, 'Last',
         # )
 
-        # define a very naive queue-pushing callback that relays
+        # # define a simple queue push routine that streams quote packets
-        # quote-packets directly the calling (parent) `trio.Task`.
+        # # to trio over the ``to_trio`` memory channel.
-        # Ensure on teardown we cancel the feed via their cancel API.
+        # to_trio, from_aio = trio.open_memory_channel(2**8)  # type: ignore
-        #
         def teardown():
-            '''
-            Disconnect our `push`-er callback and cancel the data-feed
-            for `contract`.
-
-            '''
-            nonlocal maybe_exc
             ticker.updateEvent.disconnect(push)
-            report: str = f'Disconnected mkt-data for {symbol!r} due to '
+            log.error(
-            if maybe_exc is not None:
+                f'Disconnected stream for `{symbol}`'
-                report += (
-                    'error,\n'
-                    f'{maybe_exc!r}\n'
             )
-                log.error(report)
-            else:
-                report += (
-                    'cancellation.\n'
-                )
-                log.cancel(report)
-
             client.ib.cancelMktData(contract)
 
             # decouple broadcast mem chan
             _quote_streams.pop(symbol, None)
 
-        def push(
+        def push(t: Ticker) -> None:
-            t: Ticker,
+            """
-            tries_before_raise: int = 6,
+            Push quotes to trio task.
-        ) -> None:
-            '''
-            Push quotes verbatim to parent-side `trio.Task`.
 
-            '''
+            """
-            nonlocal maybe_exc, handler_tries
+            # log.debug(t)
-            # log.debug(f'new IB quote: {t}\n')
             try:
-                chan.send_nowait(t)
+                to_trio.send_nowait(t)
 
-                # XXX TODO XXX replicate in `tractor` tests
-                # as per `CancelledError`-handler notes below!
-                # assert 0
             except (
                 trio.BrokenResourceError,
 
@@ -806,93 +736,30 @@ async def _setup_quote_stream(
                 # resulting in tracebacks spammed to console..
                 # Manually do the dereg ourselves.
                 teardown()
-
-            # for slow debugging purposes to avoid clobbering prompt
-            # with log msgs
             except trio.WouldBlock:
-                log.exception(
+                # log.warning(
-                    f'Asyncio->Trio `chan.send_nowait()` blocked !?\n'
+                #     f'channel is blocking symbol feed for {symbol}?'
-                    f'\n'
+                #     f'\n{to_trio.statistics}'
-                    f'{chan._to_trio.statistics()}\n'
+                # )
-                )
+                pass
-
-            # ?TODO, handle re-connection attempts?
-            except BaseException as _berr:
-                berr = _berr
-                if handler_tries >= tries_before_raise:
-                    # breakpoint()
-                    maybe_exc = _berr
-                    # task.set_exception(berr)
-                    aio_task.cancel(msg=berr.args)
-                    raise berr
-                else:
-                    handler_tries += 1
-
-                log.exception(
-                    f'Failed to push ticker quote !?\n'
-                    f'handler_tries={handler_tries!r}\n'
-                    f'ticker: {t!r}\n'
-                    f'\n'
-                    f'{chan._to_trio.statistics()}\n'
-                    f'\n'
-                    f'CAUSE: {berr}\n'
-                )
 
+            # except trio.WouldBlock:
+            #     # for slow debugging purposes to avoid clobbering prompt
+            #     # with log msgs
+            #     pass
 
         ticker.updateEvent.connect(push)
         try:
             await asyncio.sleep(float('inf'))
-
-            # XXX, for debug.. TODO? can we rm again?
-            #
-            # tractor.pause_from_sync()
-            # while True:
-            #     await asyncio.sleep(1.6)
-            #     if ticker.ticks:
-            #         log.debug(
-            #             f'ticker.ticks = \n'
-            #             f'{ticker.ticks}\n'
-            #         )
-            #     else:
-            #         log.warning(
-            #             'UHH no ticker.ticks ??'
-            #         )
-
-        # XXX TODO XXX !?!?
-        # apparently **without this handler** and the subsequent
-        # re-raising of `maybe_exc from _taskc` cancelling the
-        # `aio_task` from the `push()`-callback will cause a very
-        # strange chain of exc raising that breaks alll sorts of
-        # downstream callers, tasks and remote-actor tasks!?
-        #
-        # -[ ] we need some lowlevel reproducting tests to replicate
-        #      those worst-case scenarios in `tractor` core!!
-        # -[ ] likely we should factor-out the `tractor.to_asyncio`
-        #      attempts at workarounds in `.translate_aio_errors()`
-        #      for failed `asyncio.Task.set_exception()` to either
-        #      call `aio_task.cancel()` and/or
-        #      `aio_task._fut_waiter.set_exception()` to a re-useable
-        #      toolset in something like a `.to_asyncio._utils`??
-        #
-        except asyncio.CancelledError as _taskc:
-            if maybe_exc is not None:
-                raise maybe_exc from _taskc
-
-            raise _taskc
-
-        except BaseException as _berr:
-            # stash any crash cause for reporting in `teardown()`
-            maybe_exc = _berr
-            raise _berr
-
         finally:
-            # always disconnect our `push()` and cancel the
-            # ib-"mkt-data-feed".
             teardown()
 
+        # return from_aio
+
 
 @acm
 async def open_aio_quote_stream(
+
     symbol: str,
     contract: Contract | None = None,
 
@@ -900,13 +767,7 @@ async def open_aio_quote_stream(
     trio.abc.Channel|  # iface
     tractor.to_asyncio.LinkedTaskChannel  # actually
 ):
-    '''
-    Open a real-time `Ticker` quote stream from an `asyncio.Task`
-    spawned via `tractor.to_asyncio.open_channel_from()`, deliver the
-    inter-event-loop channel to the `trio.Task` caller and cache it
-    globally for re-use.
 
-    '''
     from tractor.trionics import broadcast_receiver
     global _quote_streams
 
@@ -932,10 +793,6 @@ async def open_aio_quote_stream(
 
         assert contract
 
-        # TODO? de-reg on teardown of last consumer task?
-        # -> why aren't we using `.trionics.maybe_open_context()`
-        # here again?? (we are in `open_client_proxies()` tho?)
-        #
         # cache feed for later consumers
         _quote_streams[symbol] = from_aio
 
@@ -950,12 +807,7 @@ def normalize(
     calc_price: bool = False
 
 ) -> dict:
-    '''
-    Translate `ib_async`'s `Ticker.ticks` values to a `piker`
-    normalized `dict` form for transmit to downstream `.data` layer
-    consumers.
 
-    '''
     # check for special contract types
     con = ticker.contract
     fqme, calc_price = con2fqme(con)
@@ -974,7 +826,7 @@ def normalize(
 
             tbt = ticker.tickByTicks
             if tbt:
-                log.info(f'tickbyticks:\n {ticker.tickByTicks}')
+                print(f'tickbyticks:\n {ticker.tickByTicks}')
 
     ticker.ticks = new_ticks
 
@@ -1010,39 +862,27 @@ def normalize(
     return data
 
 
-# ?TODO? feels like this task-fn could be factored to reduce some
-# indentation levels?
-# -[ ] the reconnect while loop on ib-gw "data farm connection.."s
-# -[ ] everything embedded under the `async with aclosing(stream):`
-#     as the "meat" of the quote delivery once the connection is
-#     stable.
-#
 async def stream_quotes(
+
     send_chan: trio.abc.SendChannel,
     symbols: list[str],
     feed_is_live: trio.Event,
-
+    loglevel: str = None,
-    # TODO? we need to hook into the `ib_async` logger like
-    # we can with i3ipc from modden!
-    # loglevel: str|None = None,
 
     # startup sync
     task_status: TaskStatus[tuple[dict, dict]] = trio.TASK_STATUS_IGNORED,
 
 ) -> None:
     '''
-    Stream `symbols[0]` quotes back via `send_chan`.
+    Stream symbol quotes.
 
-    The `feed_is_live: Event` is set to signal the caller that it can
+    This is a ``trio`` callable routine meant to be invoked
-    begin processing msgs from the mem-chan.
+    once the brokerd is up.
 
     '''
     # TODO: support multiple subscriptions
-    sym: str = symbols[0]
+    sym = symbols[0]
-    log.info(
+    log.info(f'request for real-time quotes: {sym}')
-        f'request for real-time quotes\n'
-        f'sym: {sym!r}\n'
-    )
 
     init_msgs: list[FeedInit] = []
 
@@ -1051,30 +891,21 @@ async def stream_quotes(
     details: ibis.ContractDetails
     async with (
         open_data_client() as proxy,
+        # trio.open_nursery() as tn,
     ):
         mkt, details = await get_mkt_info(
             sym,
             proxy=proxy,  # passed to avoid implicit client load
         )
 
-        # is venue active rn?
-        venue_is_open: bool = any(
-            is_current_time_in_range(
-                start_dt=sesh.start,
-                end_dt=sesh.end,
-            )
-            for sesh in details.tradingSessions()
-        )
-
         init_msg = FeedInit(mkt_info=mkt)
 
-        # NOTE, tell sampler (via config) to skip vlm summing for dst
-        # assets which provide no vlm data..
         if mkt.dst.atype in {
             'fiat',
             'index',
             'commodity',
         }:
+            # tell sampler config that it shouldn't do vlm summing.
             init_msg.shm_write_opts['sum_tick_vlm'] = False
             init_msg.shm_write_opts['has_vlm'] = False
 
@@ -1082,21 +913,12 @@ async def stream_quotes(
 
         con: Contract = details.contract
         first_ticker: Ticker | None = None
-
+        with trio.move_on_after(1):
-        timeout: float = 1.6
-        with trio.move_on_after(timeout) as quote_cs:
             first_ticker: Ticker = await proxy.get_quote(
                 contract=con,
                 raise_on_timeout=False,
             )
 
-        # XXX should never happen with this ep right?
-        # but if so then, more then likely mkt is closed?
-        if quote_cs.cancelled_caught:
-            log.warning(
-                f'First quote req timed out after {timeout!r}s'
-            )
-
         if first_ticker:
             first_quote: dict = normalize(first_ticker)
 
@@ -1108,27 +930,28 @@ async def stream_quotes(
                 f'{pformat(first_quote)}\n'
             )
 
-        # XXX NOTE: whenever we're "outside regular trading hours"
+        # NOTE: it might be outside regular trading hours for
-        # (only relevant for assets coming from the "legacy markets"
+        # assets with "standard venue operating hours" so we
-        # space) so we basically (from an API/runtime-operational
+        # only "pretend the feed is live" when the dst asset
-        # perspective) "pretend the feed is live" even if it's
+        # type is NOT within the NON-NORMAL-venue set: aka not
-        # actually closed.
+        # commodities, forex or crypto currencies which CAN
-        #
+        # always return a NaN on a snap quote request during
-        # IOW, we signal to the effective caller (task) that the live
+        # normal venue hours. In the case of a closed venue
-        # feed is "already up" but really we're just indicating that
+        # (equitiies, futes, bonds etc.) we at least try to
-        # the OHLCV history can start being loaded immediately by the
+        # grab the OHLC history.
-        # `piker.data`/`.tsp` layers.
+        if (
-        #
+            first_ticker
-        # XXX, deats: the "pretend we're live" is just done by
+            and
-        # a `feed_is_live.set()` even though nothing is actually live
+            isnan(first_ticker.last)
-        # Bp
+            # SO, if the last quote price value is NaN we ONLY
-        if not venue_is_open:
+            # "pretend to do" `feed_is_live.set()` if it's a known
-            log.warning(
+            # dst asset venue with a lot of closed operating hours.
-                f'Venue is closed, unable to establish real-time feed.\n'
+            and mkt.dst.atype not in {
-                f'mkt: {mkt!r}\n'
+                'commodity',
-                f'\n'
+                'fiat',
-                f'first_ticker: {first_ticker}\n'
+                'crypto',
-            )
+            }
+        ):
             task_status.started((
                 init_msgs,
                 first_quote,
@@ -1139,12 +962,10 @@ async def stream_quotes(
             feed_is_live.set()
 
             # block and let data history backfill code run.
-            # XXX obvi given the venue is closed, we never expect feed
-            # to come up; a taskc should be the only way to
-            # terminate this task.
             await trio.sleep_forever()
+            return  # we never expect feed to come up?
 
-        # ?TODO, we could instead spawn a task that waits on a feed
+        # TODO: we should instead spawn a task that waits on a feed
         # to start and let it wait indefinitely..instead of this
         # hard coded stuff.
         # async def wait_for_first_quote():
@@ -1171,22 +992,19 @@ async def stream_quotes(
         iter_quotes: trio.abc.Channel
         while (
             startup
-            or
+            or cs.cancel_called
-            cs.cancel_called
         ):
             with trio.CancelScope() as cs:
                 async with (
-                    tractor.trionics.collapse_eg(),
                     trio.open_nursery() as tn,
                     open_aio_quote_stream(
                         symbol=sym,
                         contract=con,
                     ) as iter_quotes,
                 ):
-                    # ?TODO? can we rm this - particularly for `ib_async`?
                     # ugh, clear ticks since we've consumed them
                     # (ahem, ib_insync is stateful trash)
-                    # first_ticker.ticks = []
+                    first_ticker.ticks = []
 
                     # only on first entry at feed boot up
                     if startup:
@@ -1200,8 +1018,8 @@ async def stream_quotes(
                     # data feed event.
                     async def reset_on_feed():
 
-                        # ??TODO? this seems to be surpressed from the
+                        # TODO: this seems to be surpressed from the
-                        # traceback in `tractor`?
+                        # traceback in ``tractor``?
                         # assert 0
 
                         rt_ev = proxy.status_event(
@@ -1247,7 +1065,7 @@ async def stream_quotes(
                                     # ugh, clear ticks since we've
                                     # consumed them (ahem, ib_insync is
                                     # truly stateful trash)
-                                    # ticker.ticks = []
+                                    ticker.ticks = []
 
                                     # XXX: this works because we don't use
                                     # ``aclosing()`` above?
@@ -1269,12 +1087,8 @@ async def stream_quotes(
                         async for ticker in iter_quotes:
                             quote = normalize(ticker)
                             fqme: str = quote['fqme']
-                            log.debug(
-                                f'Sending quote\n'
-                                f'{quote}'
-                            )
                             await send_chan.send({fqme: quote})
 
                             # ugh, clear ticks since we've consumed them
-                            # ticker.ticks = []
+                            ticker.ticks = []
                             # last = time.time()

piker/clearing/_ems.py

@@ -388,7 +388,6 @@ async def open_brokerd_dialog(
     for ep_name in [
         'open_trade_dialog',  # probably final name?
         'trades_dialogue',  # legacy
-        # ^!TODO, rm this since all backends ported no ?!?
     ]:
         trades_endpoint = getattr(
             brokermod,
@@ -1028,18 +1027,8 @@ async def translate_and_relay_brokerd_events(
                     )
 
                 if status == 'closed':
-                    log.info(
+                    log.info(f'Execution for {oid} is complete!')
-                        f'Execution is complete!\n'
+                    status_msg = book._active.pop(oid)
-                        f'oid: {oid!r}\n'
-                    )
-                    status_msg = book._active.pop(oid, None)
-                    if status_msg is None:
-                        log.warning(
-                            f'Order was already cleared from book ??\n'
-                            f'oid: {oid!r}\n'
-                            f'\n'
-                            f'Maybe the order cancelled before submitted ??\n'
-                        )
 
                 elif status == 'canceled':
                     log.cancel(f'Cancellation for {oid} is complete!')
@@ -1563,18 +1552,19 @@ async def maybe_open_trade_relays(
 
 @tractor.context
 async def _emsd_main(
-    ctx: tractor.Context,   # becomes `ems_ctx` below
+    ctx: tractor.Context,
     fqme: str,
     exec_mode: str,  # ('paper', 'live')
     loglevel: str|None = None,
 
-) -> tuple[  # `ctx.started()` value!
+) -> tuple[
-    dict[  # positions
+    dict[
-        tuple[str, str],  # brokername, acctid
+        # brokername, acctid
+        tuple[str, str],
         list[BrokerdPosition],
     ],
-    list[str],  # accounts
+    list[str],
-    dict[str, Status],  # dialogs
+    dict[str, Status],
 ]:
     '''
     EMS (sub)actor entrypoint providing the execution management

piker/clearing/_messages.py

@@ -301,9 +301,6 @@ class BrokerdError(Struct):
 
 # TODO: yeah, so we REALLY need to completely deprecate
 # this and use the `.accounting.Position` msg-type instead..
-# -[ ] an alternative might be to add a `Position.summary() ->
-#   `PositionSummary`-msg that we generate since `Position` has a lot
-#   of fields by default we likely don't want to send over the wire?
 class BrokerdPosition(Struct):
     '''
     Position update event from brokerd.
@@ -316,4 +313,3 @@ class BrokerdPosition(Struct):
     avg_price: float
     currency: str = ''
     name: str = 'position'
-    bs_mktid: str|int|None = None

piker/ui/_remote_ctl.py

@@ -15,8 +15,8 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 '''
-Remote control tasks for sending annotations (and maybe more cmds) to
+Remote control tasks for sending annotations (and maybe more cmds)
-a chart from some other actor.
+to a chart from some other actor.
 
 '''
 from __future__ import annotations
@@ -32,7 +32,6 @@ from typing import (
 )
 
 import tractor
-import trio
 from tractor import trionics
 from tractor import (
     Portal,
@@ -317,8 +316,6 @@ class AnnotCtl(Struct):
             )
             yield aid
         finally:
-            # async ipc send op
-            with trio.CancelScope(shield=True):
             await self.remove(aid)
 
     async def redraw(

piker/ui/order_mode.py

@@ -555,13 +555,14 @@ class OrderMode:
 
     def on_fill(
         self,
+
         uuid: str,
         price: float,
         time_s: float,
 
         pointing: str | None = None,
 
-    ) -> bool:
+    ) -> None:
         '''
         Fill msg handler.
 
@@ -574,33 +575,13 @@ class OrderMode:
         - update fill bar size
 
         '''
-        # XXX WARNING XXX
+        dialog = self.dialogs[uuid]
-        # if a `Status(resp='error')` arrives *before* this
-        # fill-status, the `.dialogs` entry may have already been
-        # popped and thus the below will skipped.
-        #
-        # NOTE, to avoid this confusing scenario ensure that any
-        # errors delivered thru from the broker-backend are not just
-        # "noisy reporting" (like is very common from IB..) and are
-        # instead ONLY errors-causing-order-dialog-cancellation!
-        if not (dialog := self.dialogs.get(uuid)):
-            log.warning(
-                f'Order was already cleared from `.dialogs` ??\n'
-                f'uuid: {uuid!r}\n'
-            )
-            return False
-
         lines = dialog.lines
         chart = self.chart
 
-        if not lines:
+        # XXX: seems to fail on certain types of races?
-            log.warn("No line(s) for order {uuid}!?")
-            return False
-
-        # update line state(s)
-        #
-        # ?XXX this fails on certain types of races?
         # assert len(lines) == 2
+        if lines:
             flume: Flume = self.feed.flumes[chart.linked.mkt.fqme]
             _, _, ratio = flume.get_ds_info()
 
@@ -626,31 +607,28 @@ class OrderMode:
                     pointing=pointing,
                     color=lines[0].color
                 )
+        else:
+            log.warn("No line(s) for order {uuid}!?")
 
     def on_cancel(
         self,
-        uuid: str,
+        uuid: str
 
-    ) -> bool:
+    ) -> None:
 
-        msg: Order|None = self.client._sent_orders.pop(uuid, None)
+        msg: Order = self.client._sent_orders.pop(uuid, None)
-        if msg is None:
-            log.warning(
-                f'Received cancel for unsubmitted order {pformat(msg)}'
-            )
-            return False
 
-        # remove GUI line, show cursor.
+        if msg is not None:
             self.lines.remove_line(uuid=uuid)
             self.chart.linked.cursor.show_xhair()
 
-        # remove msg dialog (history)
+            dialog = self.dialogs.pop(uuid, None)
-        dialog: Dialog|None = self.dialogs.pop(uuid, None)
             if dialog:
                 dialog.last_status_close()
-
+        else:
-        return True
+            log.warning(
-
+                f'Received cancel for unsubmitted order {pformat(msg)}'
+            )
 
     def cancel_orders_under_cursor(self) -> list[str]:
         return self.cancel_orders(
@@ -1079,23 +1057,13 @@ async def process_trade_msg(
     if name in (
         'position',
     ):
-        mkt: MktPair = mode.chart.linked.mkt
+        sym: MktPair = mode.chart.linked.mkt
         pp_msg_symbol = msg['symbol'].lower()
-        pp_msg_bsmktid = msg['bs_mktid']
+        fqme = sym.fqme
-        fqme = mkt.fqme
+        broker = sym.broker
-        broker = mkt.broker
         if (
-            # match on any backed-specific(-unique)-ID first!
-            (
-                pp_msg_bsmktid
-                and
-                mkt.bs_mktid == pp_msg_bsmktid
-            )
-            or
-            # OW try against what's provided as an FQME..
             pp_msg_symbol == fqme
-            or
+            or pp_msg_symbol == fqme.removesuffix(f'.{broker}')
-            pp_msg_symbol == fqme.removesuffix(f'.{broker}')
         ):
             log.info(
                 f'Loading position for `{fqme}`:\n'
@@ -1118,7 +1086,7 @@ async def process_trade_msg(
         return
 
     msg = Status(**msg)
-    # resp: str = msg.resp
+    resp = msg.resp
     oid = msg.oid
     dialog: Dialog = mode.dialogs.get(oid)
 
@@ -1182,33 +1150,20 @@ async def process_trade_msg(
                         mode.on_submit(oid)
 
         case Status(resp='error'):
-            # TODO: parse into broker-side msg, or should we
-            # expect it to just be **that** msg verbatim (since
-            # we'd presumably have only 1 `Error` msg-struct)
-            broker_msg: dict = msg.brokerd_msg
 
-            # XXX NOTE, this presumes the rxed "error" is
-            # order-dialog-cancel-causing, THUS backends much ONLY
-            # relay errors of this "severity"!!
-            log.error(
-                f'Order errored ??\n'
-                f'oid: {oid!r}\n'
-                f'\n'
-                f'{pformat(broker_msg)}\n'
-                f'\n'
-                f'=> CANCELLING ORDER DIALOG <=\n'
-
-                # from tractor.devx.pformat import ppfmt
-                # !TODO LOL, wtf the msg is causing
-                # a recursion bug!
-                # -[ ] get this shit on msgspec stat!
-                # f'{ppfmt(broker_msg)}'
-            )
             # do all the things for a cancel:
             # - drop order-msg dialog from client table
             # - delete level line from view
             mode.on_cancel(oid)
 
+            # TODO: parse into broker-side msg, or should we
+            # expect it to just be **that** msg verbatim (since
+            # we'd presumably have only 1 `Error` msg-struct)
+            broker_msg: dict = msg.brokerd_msg
+            log.error(
+                f'Order {oid}->{resp} with:\n{pformat(broker_msg)}'
+            )
+
         case Status(resp='canceled'):
             # delete level line from view
             mode.on_cancel(oid)
@@ -1223,10 +1178,10 @@ async def process_trade_msg(
             # TODO: UX for a "pending" clear/live order
             log.info(f'Dark order triggered for {fmtmsg}')
 
-        # TODO: do the struct-msg version, blah blah..
-        # req=Order(exec_mode='live', action='alert') as req,
         case Status(
             resp='triggered',
+            # TODO: do the struct-msg version, blah blah..
+            # req=Order(exec_mode='live', action='alert') as req,
             req={
                 'exec_mode': 'live',
                 'action': 'alert',

tests/test_accounting.py

@@ -12,14 +12,12 @@ from piker import config
 from piker.accounting import (
     Account,
     calc,
-    open_account,
-    load_account,
-    load_account_from_ledger,
-    open_trade_ledger,
     Position,
     TransactionLedger,
+    open_trade_ledger,
+    load_account,
+    load_account_from_ledger,
 )
-import tractor
 
 
 def test_root_conf_networking_section(
@@ -55,17 +53,12 @@ def test_account_file_default_empty(
 )
 def test_paper_ledger_position_calcs(
     fq_acnt: tuple[str, str],
-    debug_mode: bool,
 ):
     broker: str
     acnt_name: str
     broker, acnt_name = fq_acnt
 
-    accounts_path: Path = (
+    accounts_path: Path = config.repodir() / 'tests' / '_inputs'
-        config.repodir()
-        / 'tests'
-        / '_inputs'  # tests-local-subdir
-    )
 
     ldr: TransactionLedger
     with (
@@ -84,7 +77,6 @@ def test_paper_ledger_position_calcs(
             ledger=ldr,
 
             _fp=accounts_path,
-            debug_mode=debug_mode,
 
         ) as (dfs, ledger),
 
@@ -110,87 +102,3 @@ def test_paper_ledger_position_calcs(
         df = dfs[xrp]
         assert df['cumsize'][-1] == 0
         assert pos.cumsize == 0
-
-
-
-@pytest.mark.parametrize(
-    'fq_acnt',
-    [
-        ('ib', 'algopaper'),
-    ],
-)
-def test_ib_account_with_duplicated_mktids(
-    fq_acnt: tuple[str, str],
-    debug_mode: bool,
-):
-    # ?TODO, once we start symcache-incremental-update-support?
-    # from piker.data import (
-    #     open_symcache,
-    # )
-    #
-    # async def main():
-    #     async with (
-    #         # TODO: do this as part of `open_account()`!?
-    #         open_symcache(
-    #             'ib',
-    #             only_from_memcache=True,
-    #         ) as symcache,
-    #     ):
-
-
-    from piker.brokers.ib.ledger import (
-        tx_sort,
-
-        # ?TODO, once we want to pull lowlevel txns and process them?
-        # norm_trade_records,
-        # update_ledger_from_api_trades,
-    )
-
-    broker: str
-    acnt_id: str = 'algopaper'
-    broker, acnt_id = fq_acnt
-    accounts_def = config.load_accounts([broker])
-    assert accounts_def[f'{broker}.{acnt_id}']
-
-    ledger: TransactionLedger
-    acnt: Account
-    with (
-        tractor.devx.maybe_open_crash_handler(pdb=debug_mode),
-
-        open_trade_ledger(
-            'ib',
-            acnt_id,
-            tx_sort=tx_sort,
-
-            # TODO, eventually incrementally updated for IB..
-            # symcache=symcache,
-            symcache=None,
-            allow_from_sync_code=True,
-
-        ) as ledger,
-
-        open_account(
-            'ib',
-            acnt_id,
-            write_on_exit=True,
-        ) as acnt,
-    ):
-        # per input params
-        symcache = ledger.symcache
-        assert not (
-            symcache.pairs
-            or
-            symcache.pairs
-            or
-            symcache.mktmaps
-        )
-        # re-compute all positions that have changed state.
-        # TODO: likely we should change the API to return the
-        # position updates from `.update_from_ledger()`?
-        active, closed = acnt.dump_active()
-
-        # breakpoint()
-
-        # TODO, (see above imports as well) incremental update from
-        #      (updated) ledger?
-        # -[ ] pull some code from `.ib.broker` content.

tests/test_ems.py

@@ -42,7 +42,7 @@ from piker.accounting import (
     unpack_fqme,
 )
 from piker.accounting import (
-    open_account,
+    open_pps,
     Position,
 )
 
@@ -136,7 +136,7 @@ def load_and_check_pos(
 
 ) -> None:
 
-    with open_account(ppmsg.broker, ppmsg.account) as table:
+    with open_pps(ppmsg.broker, ppmsg.account) as table:
 
         if ppmsg.size == 0:
             assert ppmsg.symbol not in table.pps