Strip `None` values from symcache dict before TOML write

(this commit msg was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code

Note that (again) this patch was orig by @dnks and broken out from
a larger commit which added unnecessary/out-of-scope changes we didn't
end up requiring.

piker/brokers/deribit/api.py

@@ -586,7 +586,7 @@ async def open_price_feed(
                 fh,
                 instrument
             )
-        ) as (chan, first):
+        ) as (first, chan):
             yield chan
 
 
@@ -653,7 +653,7 @@ async def open_order_feed(
                 fh,
                 instrument
             )
-        ) as (chan, first):
+        ) as (first, chan):
             yield chan
 
 

piker/brokers/deribit/feed.py

@@ -32,7 +32,7 @@ import tractor
 
 from piker.brokers import open_cached_client
 from piker.log import get_logger, get_console_log
-from tractor.ipc._shm import ShmArray
+from piker.data import ShmArray
 from piker.brokers._util import (
     BrokerError,
     DataUnavailable,

piker/brokers/ib/README.rst

@@ -2,7 +2,7 @@
 --------------
 more or less the "everything broker" for traditional and international
 markets. they are the "go to" provider for automatic retail trading
-and we interface to their APIs using the `ib_async` project.
+and we interface to their APIs using the `ib_insync` project.
 
 status
 ******

piker/brokers/ib/__init__.py

@@ -22,7 +22,7 @@ Sub-modules within break into the core functionalities:
 - ``broker.py`` part for orders / trading endpoints
 - ``feed.py`` for real-time data feed endpoints
 - ``api.py`` for the core API machinery which is ``trio``-ized
-  wrapping around `ib_async`.
+  wrapping around ``ib_insync``.
 
 """
 from .api import (

piker/brokers/ib/_flex_reports.py

@@ -111,7 +111,7 @@ def load_flex_trades(
 
 ) -> dict[str, Any]:
 
-    from ib_async import flexreport, util
+    from ib_insync import flexreport, util
 
     conf = get_config()
 
@@ -154,7 +154,8 @@ def load_flex_trades(
         trade_entries,
     )
 
-    ledger_dict: dict|None
+    ledger_dict: dict | None = None
+
     for acctid in trades_by_account:
         trades_by_id = trades_by_account[acctid]
 

piker/brokers/ib/_util.py

@@ -20,7 +20,6 @@ runnable script-programs.
 
 '''
 from __future__ import annotations
-import asyncio
 from datetime import (  # noqa
     datetime,
     date,
@@ -141,8 +140,7 @@ async def data_reset_hack(
             except (
                 OSError,  # no VNC server avail..
                 PermissionError,  # asyncvnc pw fail..
-            ) as _vnc_err:
-                vnc_err = _vnc_err
+            ):
                 try:
                     import i3ipc  # noqa  (since a deps dynamic check)
                 except ModuleNotFoundError:
@@ -168,22 +166,14 @@ async def data_reset_hack(
 
                 # localhost but no vnc-client or it borked..
                 else:
-                    log.error(
-                        'VNC CLICK HACK FAILE with,\n'
-                        f'{vnc_err!r}\n'
-                    )
-
-                    # breakpoint()
-                    # try_xdo_manual(client)
+                    try_xdo_manual(client)
 
         case 'i3ipc_xdotool':
             try_xdo_manual(client)
             # i3ipc_xdotool_manual_click_hack()
 
         case _ as tech:
-            raise RuntimeError(
-                f'{tech!r} is not supported for reset tech!?'
-            )
+            raise RuntimeError(f'{tech} is not supported for reset tech!?')
 
     # we don't really need the ``xdotool`` approach any more B)
     return True
@@ -275,39 +265,14 @@ async def vnc_click_hack(
             # 640x1800
             await client.move(
                 Point(
-                    500,  # x from left
-                    400,  # y from top
+                    500,
+                    500,
                 )
             )
-            # in case a prior dialog win is open/active.
-            await client.press('ISO_Enter')
-
             # ensure the ib-gw window is active
             await client.click(MOUSE_BUTTON_LEFT)
-
             # send the hotkeys combo B)
-            await client.press(
-                'Ctrl',
-                'Alt',
-                key,
-            )  # NOTE, keys are stacked
-
-            # XXX, sometimes a dialog asking if you want to "simulate
-            # a reset" will show, in which case we want to select
-            # "Yes" (by tabbing) and then hit enter.
-            iters: int = 1
-            delay: float = 0.3
-            await asyncio.sleep(delay)
-
-            for i in range(iters):
-                log.info(f'Sending TAB {i}')
-                await client.press('Tab')
-                await asyncio.sleep(delay)
-
-            for i in range(iters):
-                log.info(f'Sending ENTER {i}')
-                await client.press('KP_Enter')
-                await asyncio.sleep(delay)
+            await client.press('Ctrl', 'Alt', key)  # keys are stacked
 
 
 def i3ipc_fin_wins_titled(

piker/brokers/ib/api.py

@@ -15,8 +15,7 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 '''
-Core API client machinery; mostly sane/useful wrapping around
-`ib_async`..
+Core API client machinery; mostly sane/useful wrapping around `ib_insync`..
 
 '''
 from __future__ import annotations
@@ -58,7 +57,7 @@ from pendulum import (
     Interval,
 )
 from eventkit import Event
-from ib_async import (
+from ib_insync import (
     client as ib_client,
     IB,
     Contract,
@@ -144,7 +143,7 @@ _bar_sizes = {
 _show_wap_in_history: bool = False
 
 # overrides to sidestep pretty questionable design decisions in
-# ``ib_async``:
+# ``ib_insync``:
 class NonShittyWrapper(Wrapper):
     def tcpDataArrived(self):
         """Override time stamps to be floats for now.
@@ -184,7 +183,7 @@ class NonShittyIB(IB):
     '''
     def __init__(self):
 
-        # override `ib_async` internal loggers so we can see wtf
+        # override `ib_insync` internal loggers so we can see wtf
         # it's doing..
         self._logger = get_logger(
             name=__name__,
@@ -195,7 +194,7 @@ class NonShittyIB(IB):
         self.wrapper = NonShittyWrapper(self)
         self.client = ib_client.Client(self.wrapper)
         self.client._logger = get_logger(
-            name='ib_async.client',
+            name='ib_insync.client',
         )
 
         # self.errorEvent += self._onError
@@ -561,7 +560,7 @@ class Client:
                     # f'Recursing for more bars:\n'
                 )
                 # XXX, debug!
-                # breakpoint()
+                breakpoint()
                 # XXX ? TODO? recursively try to re-request?
                 # => i think *NO* right?
                 #
@@ -768,48 +767,25 @@ class Client:
         expiry: str = '',
         front: bool = False,
 
-    ) -> Contract|list[Contract]:
+    ) -> Contract:
         '''
         Get an unqualifed contract for the current "continous"
         future.
 
-        When input params result in a so called "ambiguous contract"
-        situation, we return the list of all matches provided by,
-
-        `IB.qualifyContractsAsync(..., returnAll=True)`
-
         '''
         # it's the "front" contract returned here
         if front:
-            cons = (
-                await self.ib.qualifyContractsAsync(
-                    ContFuture(symbol, exchange=exchange),
-                    returnAll=True,
-                )
-            )
+            con = (await self.ib.qualifyContractsAsync(
+                ContFuture(symbol, exchange=exchange)
+            ))[0]
         else:
-            cons = (
-                await self.ib.qualifyContractsAsync(
-                    Future(
-                        symbol,
-                        exchange=exchange,
-                        lastTradeDateOrContractMonth=expiry,
-                    ),
-                    returnAll=True,
+            con = (await self.ib.qualifyContractsAsync(
+                Future(
+                    symbol,
+                    exchange=exchange,
+                    lastTradeDateOrContractMonth=expiry,
                 )
-            )
-
-        con = cons[0]
-        if isinstance(con, list):
-            log.warning(
-                f'{len(con)!r} futes cons matched for input params,\n'
-                f'symbol={symbol!r}\n'
-                f'exchange={exchange!r}\n'
-                f'expiry={expiry!r}\n'
-                f'\n'
-                f'cons:\n'
-                f'{con!r}\n'
-            )
+            ))[0]
 
         return con
 
@@ -902,7 +878,7 @@ class Client:
                 currency='USD',
                 exchange='PAXOS',
             )
-            # XXX, on `ib_async` when first tried this,
+            # XXX, on `ib_insync` when first tried this,
             # > Error 10299, reqId 141: Expected what to show is
             # > AGGTRADES, please use that instead of TRADES.,
             # > contract: Crypto(conId=479624278, symbol='BTC',
@@ -934,17 +910,11 @@ class Client:
             )
             exch = 'SMART' if not exch else exch
 
-        if isinstance(con, list):
-            contracts: list[Contract] = con
-        else:
-            contracts: list[Contract] = [con]
-
+        contracts: list[Contract] = [con]
         if qualify:
             try:
                 contracts: list[Contract] = (
-                    await self.ib.qualifyContractsAsync(
-                        *contracts
-                    )
+                    await self.ib.qualifyContractsAsync(con)
                 )
             except RequestError as err:
                 msg = err.message
@@ -1022,6 +992,7 @@ class Client:
     async def get_sym_details(
         self,
         fqme: str,
+
     ) -> tuple[
         Contract,
         ContractDetails,
@@ -1121,7 +1092,7 @@ class Client:
         size: int,
         account: str,  # if blank the "default" tws account is used
 
-        # XXX: by default 0 tells ``ib_async`` methods that there is no
+        # XXX: by default 0 tells ``ib_insync`` methods that there is no
         # existing order so ask the client to create a new one (which it
         # seems to do by allocating an int counter - collision prone..)
         reqid: int = None,
@@ -1310,7 +1281,7 @@ async def load_aio_clients(
     port: int = None,
     client_id: int = 6116,
 
-    # the API TCP in `ib_async` connection can be flaky af so instead
+    # the API TCP in `ib_insync` connection can be flaky af so instead
     # retry a few times to get the client going..
     connect_retries: int = 3,
     connect_timeout: float = 30,  # in case a remote-host
@@ -1318,7 +1289,7 @@ async def load_aio_clients(
 
 ) -> dict[str, Client]:
     '''
-    Return an ``ib_async.IB`` instance wrapped in our client API.
+    Return an ``ib_insync.IB`` instance wrapped in our client API.
 
     Client instances are cached for later use.
 
@@ -1529,7 +1500,7 @@ async def open_client_proxies() -> tuple[
             # TODO: maybe this should be the default in tractor?
             key=tractor.current_actor().uid,
 
-        ) as (cache_hit, (_, clients)),
+        ) as (cache_hit, (clients, _)),
 
         AsyncExitStack() as stack
     ):
@@ -1660,7 +1631,6 @@ async def open_aio_client_method_relay(
 
 ) -> None:
 
-    # with tractor.devx.maybe_open_crash_handler() as _bxerr:
     # sync with `open_client_proxy()` caller
     chan.started_nowait(client)
 
@@ -1670,11 +1640,7 @@ async def open_aio_client_method_relay(
     # relay all method requests to ``asyncio``-side client and deliver
     # back results
     while not chan._to_trio._closed:  # <- TODO, better check like `._web_bs`?
-        msg: (
-            None
-            |tuple[str, dict]
-            |dict
-        ) = await chan.get()
+        msg: tuple[str, dict]|dict|None = await chan.get()
         match msg:
             case None:  # termination sentinel
                 log.info('asyncio `Client` method-proxy SHUTDOWN!')
@@ -1718,7 +1684,7 @@ async def open_client_proxy(
             open_aio_client_method_relay,
             client=client,
             event_consumers=event_table,
-        ) as (chan, first),
+        ) as (first, chan),
 
         trionics.collapse_eg(),  # loose-ify
         trio.open_nursery() as relay_tn,
@@ -1776,7 +1742,7 @@ async def get_client(
 
 ) -> Client:
     '''
-    Init the ``ib_async`` client in another actor and return
+    Init the ``ib_insync`` client in another actor and return
     a method proxy to it.
 
     '''

piker/brokers/ib/broker.py

@@ -35,14 +35,14 @@ from trio_typing import TaskStatus
 import tractor
 from tractor.to_asyncio import LinkedTaskChannel
 from tractor import trionics
-from ib_async.contract import (
+from ib_insync.contract import (
     Contract,
 )
-from ib_async.order import (
+from ib_insync.order import (
     Trade,
     OrderStatus,
 )
-from ib_async.objects import (
+from ib_insync.objects import (
     Fill,
     Execution,
     CommissionReport,
@@ -181,7 +181,7 @@ async def handle_order_requests(
             # validate
             order = BrokerdOrder(**request_msg)
 
-            # XXX: by default 0 tells ``ib_async`` methods that
+            # XXX: by default 0 tells ``ib_insync`` methods that
             # there is no existing order so ask the client to create
             # a new one (which it seems to do by allocating an int
             # counter - collision prone..)
@@ -237,7 +237,7 @@ async def recv_trade_updates(
 ) -> None:
     '''
     Receive and relay order control and positioning related events
-    from `ib_async`, pack as tuples and push over mem-chan to our
+    from `ib_insync`, pack as tuples and push over mem-chan to our
     trio relay task for processing and relay to EMS.
 
     '''
@@ -303,7 +303,7 @@ async def recv_trade_updates(
         # much more then a few more pnl fields..
         # 'updatePortfolioEvent',
 
-        # XXX: these all seem to be weird ib_async internal
+        # XXX: these all seem to be weird ib_insync internal
         # events that we probably don't care that much about
         # given the internal design is wonky af..
         # 'newOrderEvent',
@@ -499,7 +499,7 @@ async def open_trade_event_stream(
     ] = trio.TASK_STATUS_IGNORED,
 ):
     '''
-    Proxy wrapper for starting trade event stream from ib_async
+    Proxy wrapper for starting trade event stream from ib_insync
     which spawns an asyncio task that registers an internal closure
     (`push_tradies()`) which in turn relays trading events through
     a `tractor.to_asyncio.LinkedTaskChannel` which the parent
@@ -514,8 +514,8 @@ async def open_trade_event_stream(
         recv_trade_updates,
         client=client,
     ) as (
-        trade_event_stream,
         _,  # first pushed val
+        trade_event_stream,
     ):
         task_status.started(trade_event_stream)
         # block forever to keep session trio-asyncio session
@@ -991,9 +991,6 @@ _statuses: dict[str, str] = {
     # TODO: see a current ``ib_insync`` issue around this:
     # https://github.com/erdewit/ib_insync/issues/363
     'Inactive': 'pending',
-
-    # XXX, uhh wut the heck is this?
-    'ValidationError': 'error',
 }
 
 _action_map = {
@@ -1066,19 +1063,8 @@ async def deliver_trade_events(
     # TODO: for some reason we can receive a ``None`` here when the
     # ib-gw goes down? Not sure exactly how that's happening looking
     # at the eventkit code above but we should probably handle it...
-    event_name: str
-    item: (
-        Trade
-        |tuple[Trade, Fill]
-        |CommissionReport
-        |IbPosition
-        |dict
-    )
     async for event_name, item in trade_event_stream:
-        log.info(
-            f'Relaying {event_name!r}:\n'
-            f'{pformat(item)}\n'
-        )
+        log.info(f'Relaying `{event_name}`:\n{pformat(item)}')
         match event_name:
             case 'orderStatusEvent':
 
@@ -1089,12 +1075,11 @@ async def deliver_trade_events(
                 trade: Trade = item
                 reqid: str = str(trade.order.orderId)
                 status: OrderStatus = trade.orderStatus
-                status_str: str = _statuses.get(
-                    status.status,
-                    'error',
-                )
+                status_str: str = _statuses[status.status]
                 remaining: float = status.remaining
-                if status_str == 'filled':
+                if (
+                    status_str == 'filled'
+                ):
                     fill: Fill = trade.fills[-1]
                     execu: Execution = fill.execution
 
@@ -1125,12 +1110,6 @@ async def deliver_trade_events(
                         # all units were cleared.
                         status_str = 'closed'
 
-                elif status_str == 'error':
-                    log.error(
-                        f'IB reported error status for order ??\n'
-                        f'{status.status!r}\n'
-                    )
-
                 # skip duplicate filled updates - we get the deats
                 # from the execution details event
                 msg = BrokerdStatus(
@@ -1291,23 +1270,13 @@ async def deliver_trade_events(
             case 'error':
                 # NOTE: see impl deats in
                 # `Client.inline_errors()::push_err()`
-                err: dict|str = item
+                err: dict = item
 
-                # std case, never relay errors for non-order-control
-                # related issues.
+                # never relay errors for non-broker related issues
                 # https://interactivebrokers.github.io/tws-api/message_codes.html
-                if isinstance(err, dict):
-                    code: int = err['error_code']
-                    reason: str = err['reason']
-                    reqid: str = str(err['reqid'])
-
-                # XXX, sometimes you'll get just a `str` of the form,
-                # '[code 104] connection failed' or something..
-                elif isinstance(err, str):
-                    code_part, _, reason = err.rpartition(']')
-                    if code_part:
-                        _, _, code = code_part.partition('[code')
-                    reqid: str = '<unknown>'
+                code: int = err['error_code']
+                reason: str = err['reason']
+                reqid: str = str(err['reqid'])
 
                 # "Warning:" msg codes,
                 # https://interactivebrokers.github.io/tws-api/message_codes.html#warning_codes

piker/brokers/ib/feed.py

@@ -36,7 +36,7 @@ from typing import (
 )
 
 from async_generator import aclosing
-import ib_async as ibis
+import ib_insync as ibis
 import numpy as np
 from pendulum import (
     now,
@@ -100,7 +100,7 @@ tick_types = {
     5: 'size',
     8: 'volume',
 
-    # `ib_async` already packs these into
+    # ``ib_insync`` already packs these into
     # quotes under the following fields.
     55: 'trades_per_min',  # `'tradeRate'`
     56: 'vlm_per_min',  # `'volumeRate'`
@@ -201,15 +201,6 @@ async def open_history_client(
                 fqme,
                 timeframe,
                 end_dt=end_dt,
-
-                # XXX WARNING, we don't actually use this inside
-                # `Client.bars()` since it isn't really supported,
-                # the API instead supports a "duration" of time style
-                # from the `end_dt` (or at least that was the best
-                # way to get it working sanely)..
-                # 
-                # SO, with that in mind be aware that any downstream
-                # logic based on this may be mostly futile Xp
                 start_dt=start_dt,
             )
             latency = time.time() - query_start
@@ -287,27 +278,19 @@ async def open_history_client(
                 trimmed_bars = bars_array[
                     bars_array['time'] >= start_dt.timestamp()
                 ]
-                # XXX, should NEVER get HERE!
-                if trimmed_bars.size:
-                    trimmed_first_dt: datetime = from_timestamp(trimmed_bars['time'][0])
-                    if (
-                        trimmed_first_dt
-                        >=
-                        start_dt
-                    ):
-                        msg: str = (
-                            f'OHLC-bars array start is gt `start_dt` limit !!\n'
-                            f'start_dt: {start_dt}\n'
-                            f'first_dt: {first_dt}\n'
-                            f'trimmed_first_dt: {trimmed_first_dt}\n'
-                            f'\n'
-                            f'Delivering shorted frame of {trimmed_bars.size!r}\n'
-                        )
-                        log.warning(msg)
-                        # TODO! rm this once we're more confident it
-                        # never breaks anything (in the caller)!
-                        # breakpoint()
-                        # raise RuntimeError(msg)
+                if (
+                    trimmed_first_dt := from_timestamp(trimmed_bars['time'][0])
+                    !=
+                    start_dt
+                ):
+                    # TODO! rm this once we're more confident it never hits!
+                    # breakpoint()
+                    raise RuntimeError(
+                        f'OHLC-bars array start is gt `start_dt` limit !!\n'
+                        f'start_dt: {start_dt}\n'
+                        f'first_dt: {first_dt}\n'
+                        f'trimmed_first_dt: {trimmed_first_dt}\n'
+                    )
 
                 # XXX, overwrite with start_dt-limited frame
                 bars_array = trimmed_bars
@@ -321,7 +304,7 @@ async def open_history_client(
         # TODO: it seems like we can do async queries for ohlc
         # but getting the order right still isn't working and I'm not
         # quite sure why.. needs some tinkering and probably
-        # a lookthrough of the `ib_async` machinery, for eg. maybe
+        # a lookthrough of the `ib_insync` machinery, for eg. maybe
         # we have to do the batch queries on the `asyncio` side?
         yield (
             get_hist,
@@ -989,7 +972,7 @@ async def open_aio_quote_stream(
         symbol=symbol,
         contract=contract,
 
-    ) as (from_aio, contract):
+    ) as (contract, from_aio):
 
         assert contract
 
@@ -1068,21 +1051,6 @@ def normalize(
     #         ticker.rtTime.timestamp) / 1000.
     data.pop('rtTime')
 
-    # XXX, `ib_async` seems to set a
-    # `'timezone': datetime.timezone.utc` in this `dict`
-    # which is NOT IPC serializeable sin codec!
-    #
-    # pretty sure we don't need any of this field for now anyway?
-    data.pop('defaults')
-
-    if lts := data.get('lastTimeStamp'):
-        lts.replace(tzinfo=None)
-        log.warning(
-            f'Stripping `.tzinfo` from datetime\n'
-            f'{lts}\n'
-        )
-        # breakpoint()
-
     return data
 
 
@@ -1259,7 +1227,7 @@ async def stream_quotes(
                 ):
                     # ?TODO? can we rm this - particularly for `ib_async`?
                     # ugh, clear ticks since we've consumed them
-                    # (ahem, ib_async is stateful trash)
+                    # (ahem, ib_insync is stateful trash)
                     # first_ticker.ticks = []
 
                     # only on first entry at feed boot up

piker/brokers/ib/ledger.py

@@ -36,7 +36,7 @@ from pendulum import (
     parse,
     from_timestamp,
 )
-from ib_async import (
+from ib_insync import (
     Contract,
     Commodity,
     Fill,

piker/brokers/ib/symbols.py

@@ -23,7 +23,6 @@ from contextlib import (
     nullcontext,
 )
 from decimal import Decimal
-from functools import partial
 import time
 from typing import (
     Awaitable,
@@ -31,9 +30,8 @@ from typing import (
 )
 
 from rapidfuzz import process as fuzzy
-import ib_async as ibis
+import ib_insync as ibis
 import tractor
-from tractor.devx.pformat import ppfmt
 import trio
 
 from piker.accounting import (
@@ -217,19 +215,18 @@ async def open_symbol_search(ctx: tractor.Context) -> None:
                 f'{ib_client}\n'
             )
 
-            last: float = time.time()
+            last = time.time()
             async for pattern in stream:
                 log.info(f'received {pattern}')
                 now: float = time.time()
 
-                # TODO? check this is no longer true?
                 # this causes tractor hang...
                 # assert 0
 
                 assert pattern, 'IB can not accept blank search pattern'
 
                 # throttle search requests to no faster then 1Hz
-                diff: float = now - last
+                diff = now - last
                 if diff < 1.0:
                     log.debug('throttle sleeping')
                     await trio.sleep(diff)
@@ -240,12 +237,11 @@ async def open_symbol_search(ctx: tractor.Context) -> None:
 
                 if (
                     not pattern
-                    or
-                    pattern.isspace()
-                    or
+                    or pattern.isspace()
+
                     # XXX: not sure if this is a bad assumption but it
                     # seems to make search snappier?
-                    len(pattern) < 1
+                    or len(pattern) < 1
                 ):
                     log.warning('empty pattern received, skipping..')
 
@@ -258,58 +254,36 @@ async def open_symbol_search(ctx: tractor.Context) -> None:
                     # XXX: this unblocks the far end search task which may
                     # hold up a multi-search nursery block
                     await stream.send({})
+
                     continue
 
-                log.info(
-                    f'Searching for FQME with,\n'
-                    f'pattern: {pattern!r}\n'
-                )
+                log.info(f'searching for {pattern}')
 
-                last: float = time.time()
+                last = time.time()
 
-                # async batch search using api stocks endpoint and
-                # module defined adhoc symbol set.
-                stock_results: list[dict] = []
+                # async batch search using api stocks endpoint and module
+                # defined adhoc symbol set.
+                stock_results = []
 
                 async def extend_results(
-                    # ?TODO, how to type async-fn!?
-                    target: Awaitable[list],
-                    pattern: str,
-                    **kwargs,
+                    target: Awaitable[list]
                 ) -> None:
                     try:
-                        results = await target(
-                            pattern=pattern,
-                            **kwargs,
-                        )
-                        client_repr: str = proxy._aio_ns.ib.client.__class__.__name__
-                        meth_repr: str = target.keywords["meth"]
-                        log.info(
-                            f'Search query,\n'
-                            f'{client_repr}.{meth_repr}(\n'
-                            f'    pattern={pattern!r}\n'
-                            f'    **kwargs={kwargs!r},\n'
-                            f') = {ppfmt(list(results))}'
-                            # XXX ^ just the keys since that's what
-                            # shows in UI results table.
-                        )
+                        results = await target
                     except tractor.trionics.Lagged:
-                        log.exception(
-                            'IB SYM-SEARCH OVERRUN?!?\n'
-                        )
+                        print("IB SYM-SEARCH OVERRUN?!?")
                         return
 
                     stock_results.extend(results)
 
                 for _ in range(10):
                     with trio.move_on_after(3) as cs:
-                        async with trio.open_nursery() as tn:
-                            tn.start_soon(
-                                partial(
-                                    extend_results,
+                        async with trio.open_nursery() as sn:
+                            sn.start_soon(
+                                extend_results,
+                                proxy.search_symbols(
                                     pattern=pattern,
-                                    target=proxy.search_symbols,
-                                    upto=10,
+                                    upto=5,
                                 ),
                             )
 
@@ -339,9 +313,7 @@ async def open_symbol_search(ctx: tractor.Context) -> None:
                     #     adhoc_match_results = {i[0]: {} for i in
                     #     adhoc_matches}
 
-                log.debug(
-                    f'fuzzy matching stocks {ppfmt(stock_results)}'
-                )
+                log.debug(f'fuzzy matching stocks {stock_results}')
                 stock_matches = fuzzy.extract(
                     pattern,
                     stock_results,
@@ -355,10 +327,7 @@ async def open_symbol_search(ctx: tractor.Context) -> None:
                 # TODO: we used to deliver contract details
                 # {item[2]: item[0] for item in stock_matches}
 
-                log.debug(
-                    f'Sending final matches\n'
-                    f'{matches.keys()}'
-                )
+                log.debug(f"sending matches: {matches.keys()}")
                 await stream.send(matches)
 
 
@@ -553,11 +522,7 @@ async def get_mkt_info(
     if atype == 'commodity':
         venue: str = 'cmdty'
     else:
-        venue: str = (
-            con.primaryExchange
-            or
-            con.exchange
-        )
+        venue = con.primaryExchange or con.exchange
 
     price_tick: Decimal = Decimal(str(details.minTick))
     ib_min_tick_gt_2: Decimal = Decimal('0.01')

piker/brokers/ib/venues.py

@@ -41,9 +41,8 @@ from pendulum import (
 )
 
 if TYPE_CHECKING:
-    from ib_async import (
+    from ib_insync import (
         TradingSession,
-        Contract,
         ContractDetails,
     )
     from exchange_calendars.exchange_calendars import (
@@ -83,20 +82,8 @@ def has_holiday(
 
     '''
     tz: str = con_deats.timeZoneId
-    con: Contract = con_deats.contract
-    exch: str = (
-        con.primaryExchange
-        or
-        con.exchange
-    )
-
-    # XXX, ad-hoc handle any IB exchange which are non-std
-    # via lookup table..
-    std_exch: dict = {
-        'ARCA': 'ARCX',
-    }.get(exch, exch)
-
-    cal: ExchangeCalendar = xcals.get_calendar(std_exch)
+    exch: str = con_deats.contract.primaryExchange
+    cal: ExchangeCalendar = xcals.get_calendar(exch)
     end: datetime = period.end
     # _start: datetime = period.start
     # ?TODO, can rm ya?
@@ -249,7 +236,7 @@ def is_venue_closure(
 #
 # NOTE, this was generated by @guille from a gpt5 prompt
 # and was originally thot to be needed before learning about
-# `ib_async.contract.ContractDetails._parseSessions()` and
+# `ib_insync.contract.ContractDetails._parseSessions()` and
 # it's downstream meths..
 #
 # This is still likely useful to keep for now to parse the

piker/config.py

@@ -19,6 +19,7 @@ Platform configuration (files) mgmt.
 
 """
 import platform
+import sys
 import os
 import shutil
 from typing import (
@@ -28,7 +29,6 @@ from typing import (
 from pathlib import Path
 
 from bidict import bidict
-import platformdirs
 import tomlkit
 try:
     import tomllib
@@ -41,7 +41,7 @@ from .log import get_logger
 log = get_logger('broker-config')
 
 
-# XXX NOTE: orig impl was taken from `click`
+# XXX NOTE: taken from `click`
 # |_https://github.com/pallets/click/blob/main/src/click/utils.py#L449
 #
 # (since apparently they have some super weirdness with SIGINT and
@@ -54,21 +54,44 @@ def get_app_dir(
     force_posix: bool = False,
 
 ) -> str:
-    '''
-    Returns the config folder for the application.  The default behavior
+    r"""Returns the config folder for the application.  The default behavior
     is to return whatever is most appropriate for the operating system.
 
-    ----
-    NOTE, below is originally from `click` impl fn, we can prolly remove?
-    ----
+    To give you an idea, for an app called ``"Foo Bar"``, something like
+    the following folders could be returned:
 
+    Mac OS X:
+      ``~/Library/Application Support/Foo Bar``
+    Mac OS X (POSIX):
+      ``~/.foo-bar``
+    Unix:
+      ``~/.config/foo-bar``
+    Unix (POSIX):
+      ``~/.foo-bar``
+    Win XP (roaming):
+      ``C:\Documents and Settings\<user>\Local Settings\Application Data\Foo``
+    Win XP (not roaming):
+      ``C:\Documents and Settings\<user>\Application Data\Foo Bar``
+    Win 7 (roaming):
+      ``C:\Users\<user>\AppData\Roaming\Foo Bar``
+    Win 7 (not roaming):
+      ``C:\Users\<user>\AppData\Local\Foo Bar``
+
+    .. versionadded:: 2.0
+
+    :param app_name: the application name.  This should be properly capitalized
+                     and can contain whitespace.
     :param roaming: controls if the folder should be roaming or not on Windows.
                     Has no affect otherwise.
     :param force_posix: if this is set to `True` then on any POSIX system the
                         folder will be stored in the home folder with a leading
                         dot instead of the XDG config home or darwin's
                         application support folder.
-    '''
+    """
+
+    def _posixify(name):
+        return "-".join(name.split()).lower()
+
     # NOTE: for testing with `pytest` we leverage the `tmp_dir`
     # fixture to generate (and clean up) a test-request-specific
     # directory for isolated configuration files such that,
@@ -94,30 +117,23 @@ def get_app_dir(
     #         assert testdirpath.exists(), 'piker test harness might be borked!?'
     #         app_name = str(testdirpath)
 
-    os_name: str = platform.system()
-    conf_dir: Path = platformdirs.user_config_path()
-    app_dir: Path = conf_dir / app_name
-
-    # ?TODO, from `click`; can remove?
+    if platform.system() == 'Windows':
+        key = "APPDATA" if roaming else "LOCALAPPDATA"
+        folder = os.environ.get(key)
+        if folder is None:
+            folder = os.path.expanduser("~")
+        return os.path.join(folder, app_name)
     if force_posix:
-        def _posixify(name):
-            return "-".join(name.split()).lower()
-
         return os.path.join(
-            os.path.expanduser(
-                "~/.{}".format(
-                    _posixify(app_name)
-                )
-            )
+            os.path.expanduser("~/.{}".format(_posixify(app_name))))
+    if sys.platform == "darwin":
+        return os.path.join(
+            os.path.expanduser("~/Library/Application Support"), app_name
         )
-
-    log.info(
-        f'Using user config directory,\n'
-        f'platform.system(): {os_name!r}\n'
-        f'conf_dir: {conf_dir!r}\n'
-        f'app_dir: {conf_dir!r}\n'
+    return os.path.join(
+        os.environ.get("XDG_CONFIG_HOME", os.path.expanduser("~/.config")),
+        _posixify(app_name),
     )
-    return app_dir
 
 
 _click_config_dir: Path = Path(get_app_dir('piker'))
@@ -234,9 +250,7 @@ def repodir() -> Path:
         repodir: Path = Path(os.environ.get('GITHUB_WORKSPACE'))
         confdir: Path = repodir / 'config'
 
-    assert confdir.is_dir(), (
-        f'{confdir} DNE, {repodir} is likely incorrect!'
-    )
+    assert confdir.is_dir(), f'{confdir} DNE, {repodir} is likely incorrect!'
     return repodir
 
 

piker/data/__init__.py

@@ -23,13 +23,13 @@ sharing live streams over a network.
 
 """
 from .ticktools import iterticks
-from tractor.ipc._shm import (
-    ShmArray,
+from ._sharedmem import (
+    maybe_open_shm_array,
+    attach_shm_array,
+    open_shm_array,
     get_shm_token,
-    open_shm_ndarray as open_shm_array,
-    attach_shm_ndarray as attach_shm_array,
+    ShmArray,
 )
-from ._sharedmem import maybe_open_shm_array
 from ._source import (
     def_iohlcv_fields,
     def_ohlcv_fields,

piker/data/_formatters.py

@@ -28,7 +28,9 @@ from msgspec import field
 import numpy as np
 from numpy.lib import recfunctions as rfn
 
-from tractor.ipc._shm import ShmArray
+from ._sharedmem import (
+    ShmArray,
+)
 from ._pathops import (
     path_arrays_from_ohlc,
 )

piker/data/_sampling.py

@@ -55,7 +55,9 @@ from ._util import (
 from ..service import maybe_spawn_daemon
 
 if TYPE_CHECKING:
-    from tractor.ipc._shm import ShmArray
+    from ._sharedmem import (
+        ShmArray,
+    )
     from .feed import (
         _FeedsBus,
         Sub,
@@ -376,16 +378,16 @@ async def register_with_sampler(
                 # feed_is_live.is_set()
                 # ^TODO? pass it in instead?
             ):
-                from tractor.ipc._shm import (
-                    attach_shm_ndarray,
-                    NDToken,
+                from ._sharedmem import (
+                    attach_shm_array,
+                    _Token,
                 )
                 for period in shms_by_period:
 
                     # load and register shm handles
                     shm_token_msg = shms_by_period[period]
-                    shm = attach_shm_ndarray(
-                        NDToken.from_msg(shm_token_msg),
+                    shm = attach_shm_array(
+                        _Token.from_msg(shm_token_msg),
                         readonly=False,
                     )
                     shms_by_period[period] = shm

piker/data/_sharedmem.py

@@ -1,106 +1,661 @@
 # piker: trading gear for hackers
 # Copyright (C) Tyler Goodlet (in stewardship for pikers)
 
-# This program is free software: you can redistribute it
-# and/or modify it under the terms of the GNU Affero General
-# Public License as published by the Free Software
-# Foundation, either version 3 of the License, or (at your
-# option) any later version.
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
 
-# This program is distributed in the hope that it will be
-# useful, but WITHOUT ANY WARRANTY; without even the implied
-# warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
-# PURPOSE.  See the GNU Affero General Public License for
-# more details.
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
 
-# You should have received a copy of the GNU Affero General
-# Public License along with this program.  If not, see
-# <https://www.gnu.org/licenses/>.
+# 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/>.
 
-'''
-Piker-specific shared memory helpers.
+"""
+NumPy compatible shared memory buffers for real-time IPC streaming.
 
-Thin shim providing piker-only wrappers around
-``tractor.ipc._shm``; all core types and functions
-are now imported directly from tractor throughout
-the codebase.
+"""
+from __future__ import annotations
+from sys import byteorder
+import time
+from typing import Optional
+from multiprocessing.shared_memory import SharedMemory, _USE_POSIX
 
-'''
+if _USE_POSIX:
+    from _posixshmem import shm_unlink
+
+# import msgspec
 import numpy as np
-
-from tractor.ipc._shm import (
-    NDToken,
-    ShmArray,
-    _known_tokens,
-    _make_token as _tractor_make_token,
-    open_shm_ndarray,
-    attach_shm_ndarray,
-)
+from numpy.lib import recfunctions as rfn
+import tractor
 
 from ._util import log
+from ._source import def_iohlcv_fields
+from piker.types import Struct
+
+
+def cuckoff_mantracker():
+    '''
+    Disable all ``multiprocessing``` "resource tracking" machinery since
+    it's an absolute multi-threaded mess of non-SC madness.
+
+    '''
+    from multiprocessing import resource_tracker as mantracker
+
+    # Tell the "resource tracker" thing to fuck off.
+    class ManTracker(mantracker.ResourceTracker):
+        def register(self, name, rtype):
+            pass
+
+        def unregister(self, name, rtype):
+            pass
+
+        def ensure_running(self):
+            pass
+
+    # "know your land and know your prey"
+    # https://www.dailymotion.com/video/x6ozzco
+    mantracker._resource_tracker = ManTracker()
+    mantracker.register = mantracker._resource_tracker.register
+    mantracker.ensure_running = mantracker._resource_tracker.ensure_running
+    mantracker.unregister = mantracker._resource_tracker.unregister
+    mantracker.getfd = mantracker._resource_tracker.getfd
+
+
+cuckoff_mantracker()
+
+
+class SharedInt:
+    """Wrapper around a single entry shared memory array which
+    holds an ``int`` value used as an index counter.
+
+    """
+    def __init__(
+        self,
+        shm: SharedMemory,
+    ) -> None:
+        self._shm = shm
+
+    @property
+    def value(self) -> int:
+        return int.from_bytes(self._shm.buf, byteorder)
+
+    @value.setter
+    def value(self, value) -> None:
+        self._shm.buf[:] = value.to_bytes(self._shm.size, byteorder)
+
+    def destroy(self) -> None:
+        if _USE_POSIX:
+            # We manually unlink to bypass all the "resource tracker"
+            # nonsense meant for non-SC systems.
+            name = self._shm.name
+            try:
+                shm_unlink(name)
+            except FileNotFoundError:
+                # might be a teardown race here?
+                log.warning(f'Shm for {name} already unlinked?')
+
+
+class _Token(Struct, frozen=True):
+    '''
+    Internal represenation of a shared memory "token"
+    which can be used to key a system wide post shm entry.
+
+    '''
+    shm_name: str  # this servers as a "key" value
+    shm_first_index_name: str
+    shm_last_index_name: str
+    dtype_descr: tuple
+    size: int  # in struct-array index / row terms
+
+    @property
+    def dtype(self) -> np.dtype:
+        return np.dtype(list(map(tuple, self.dtype_descr))).descr
+
+    def as_msg(self):
+        return self.to_dict()
+
+    @classmethod
+    def from_msg(cls, msg: dict) -> _Token:
+        if isinstance(msg, _Token):
+            return msg
+
+        # TODO: native struct decoding
+        # return _token_dec.decode(msg)
+
+        msg['dtype_descr'] = tuple(map(tuple, msg['dtype_descr']))
+        return _Token(**msg)
+
+
+# _token_dec = msgspec.msgpack.Decoder(_Token)
+
+# TODO: this api?
+# _known_tokens = tractor.ActorVar('_shm_tokens', {})
+# _known_tokens = tractor.ContextStack('_known_tokens', )
+# _known_tokens = trio.RunVar('shms', {})
+
+# process-local store of keys to tokens
+_known_tokens = {}
+
+
+def get_shm_token(key: str) -> _Token:
+    """Convenience func to check if a token
+    for the provided key is known by this process.
+    """
+    return _known_tokens.get(key)
 
 
 def _make_token(
     key: str,
     size: int,
-    dtype: np.dtype|None = None,
-) -> NDToken:
+    dtype: Optional[np.dtype] = None,
+) -> _Token:
     '''
-    Wrap tractor's ``_make_token()`` with piker's
-    default dtype fallback to ``def_iohlcv_fields``.
+    Create a serializable token that can be used
+    to access a shared array.
 
     '''
-    from ._source import def_iohlcv_fields
-    dtype = (
-        def_iohlcv_fields
-        if dtype is None
-        else dtype
+    dtype = def_iohlcv_fields if dtype is None else dtype
+    return _Token(
+        shm_name=key,
+        shm_first_index_name=key + "_first",
+        shm_last_index_name=key + "_last",
+        dtype_descr=tuple(np.dtype(dtype).descr),
+        size=size,
     )
-    return _tractor_make_token(
+
+
+class ShmArray:
+    '''
+    A shared memory ``numpy`` (compatible) array API.
+
+    An underlying shared memory buffer is allocated based on
+    a user specified ``numpy.ndarray``. This fixed size array
+    can be read and written to by pushing data both onto the "front"
+    or "back" of a set index range. The indexes for the "first" and
+    "last" index are themselves stored in shared memory (accessed via
+    ``SharedInt`` interfaces) values such that multiple processes can
+    interact with the same array using a synchronized-index.
+
+    '''
+    def __init__(
+        self,
+        shmarr: np.ndarray,
+        first: SharedInt,
+        last: SharedInt,
+        shm: SharedMemory,
+        # readonly: bool = True,
+    ) -> None:
+        self._array = shmarr
+
+        # indexes for first and last indices corresponding
+        # to fille data
+        self._first = first
+        self._last = last
+
+        self._len = len(shmarr)
+        self._shm = shm
+        self._post_init: bool = False
+
+        # pushing data does not write the index (aka primary key)
+        dtype = shmarr.dtype
+        if dtype.fields:
+            self._write_fields = list(shmarr.dtype.fields.keys())[1:]
+        else:
+            self._write_fields = None
+
+    # TODO: ringbuf api?
+
+    @property
+    def _token(self) -> _Token:
+        return _Token(
+            shm_name=self._shm.name,
+            shm_first_index_name=self._first._shm.name,
+            shm_last_index_name=self._last._shm.name,
+            dtype_descr=tuple(self._array.dtype.descr),
+            size=self._len,
+        )
+
+    @property
+    def token(self) -> dict:
+        """Shared memory token that can be serialized and used by
+        another process to attach to this array.
+        """
+        return self._token.as_msg()
+
+    @property
+    def index(self) -> int:
+        return self._last.value % self._len
+
+    @property
+    def array(self) -> np.ndarray:
+        '''
+        Return an up-to-date ``np.ndarray`` view of the
+        so-far-written data to the underlying shm buffer.
+
+        '''
+        a = self._array[self._first.value:self._last.value]
+
+        # first, last = self._first.value, self._last.value
+        # a = self._array[first:last]
+
+        # TODO: eventually comment this once we've not seen it in the
+        # wild in a long time..
+        # XXX: race where first/last indexes cause a reader
+        # to load an empty array..
+        if len(a) == 0 and self._post_init:
+            raise RuntimeError('Empty array race condition hit!?')
+
+        return a
+
+    def ustruct(
+        self,
+        fields: Optional[list[str]] = None,
+
+        # type that all field values will be cast to
+        # in the returned view.
+        common_dtype: np.dtype = float,
+
+    ) -> np.ndarray:
+
+        array = self._array
+
+        if fields:
+            selection = array[fields]
+            # fcount = len(fields)
+        else:
+            selection = array
+            # fcount = len(array.dtype.fields)
+
+        # XXX: manual ``.view()`` attempt that also doesn't work.
+        # uview = selection.view(
+        #     dtype='<f16',
+        # ).reshape(-1, 4, order='A')
+
+        # assert len(selection) == len(uview)
+
+        u = rfn.structured_to_unstructured(
+            selection,
+            # dtype=float,
+            copy=True,
+        )
+
+        # unstruct = np.ndarray(u.shape, dtype=a.dtype, buffer=shm.buf)
+        # array[:] = a[:]
+        return u
+        # return ShmArray(
+        #     shmarr=u,
+        #     first=self._first,
+        #     last=self._last,
+        #     shm=self._shm
+        # )
+
+    def last(
+        self,
+        length: int = 1,
+
+    ) -> np.ndarray:
+        '''
+        Return the last ``length``'s worth of ("row") entries from the
+        array.
+
+        '''
+        return self.array[-length:]
+
+    def push(
+        self,
+        data: np.ndarray,
+
+        field_map: Optional[dict[str, str]] = None,
+        prepend: bool = False,
+        update_first: bool = True,
+        start: int | None = None,
+
+    ) -> int:
+        '''
+        Ring buffer like "push" to append data
+        into the buffer and return updated "last" index.
+
+        NB: no actual ring logic yet to give a "loop around" on overflow
+        condition, lel.
+
+        '''
+        length = len(data)
+
+        if prepend:
+            index = (start or self._first.value) - length
+
+            if index < 0:
+                raise ValueError(
+                    f'Array size of {self._len} was overrun during prepend.\n'
+                    f'You have passed {abs(index)} too many datums.'
+                )
+
+        else:
+            index = start if start is not None else self._last.value
+
+        end = index + length
+
+        if field_map:
+            src_names, dst_names = zip(*field_map.items())
+        else:
+            dst_names = src_names = self._write_fields
+
+        try:
+            self._array[
+                list(dst_names)
+            ][index:end] = data[list(src_names)][:]
+
+            # NOTE: there was a race here between updating
+            # the first and last indices and when the next reader
+            # tries to access ``.array`` (which due to the index
+            # overlap will be empty). Pretty sure we've fixed it now
+            # but leaving this here as a reminder.
+            if (
+                prepend
+                and update_first
+                and length
+            ):
+                assert index < self._first.value
+
+            if (
+                index < self._first.value
+                and update_first
+            ):
+                assert prepend, 'prepend=True not passed but index decreased?'
+                self._first.value = index
+
+            elif not prepend:
+                self._last.value = end
+
+            self._post_init = True
+            return end
+
+        except ValueError as err:
+            if field_map:
+                raise
+
+            # should raise if diff detected
+            self.diff_err_fields(data)
+            raise err
+
+    def diff_err_fields(
+        self,
+        data: np.ndarray,
+    ) -> None:
+        # reraise with any field discrepancy
+        our_fields, their_fields = (
+            set(self._array.dtype.fields),
+            set(data.dtype.fields),
+        )
+
+        only_in_ours = our_fields - their_fields
+        only_in_theirs = their_fields - our_fields
+
+        if only_in_ours:
+            raise TypeError(
+                f"Input array is missing field(s): {only_in_ours}"
+            )
+        elif only_in_theirs:
+            raise TypeError(
+                f"Input array has unknown field(s): {only_in_theirs}"
+            )
+
+    # TODO: support "silent" prepends that don't update ._first.value?
+    def prepend(
+        self,
+        data: np.ndarray,
+    ) -> int:
+        end = self.push(data, prepend=True)
+        assert end
+
+    def close(self) -> None:
+        self._first._shm.close()
+        self._last._shm.close()
+        self._shm.close()
+
+    def destroy(self) -> None:
+        if _USE_POSIX:
+            # We manually unlink to bypass all the "resource tracker"
+            # nonsense meant for non-SC systems.
+            shm_unlink(self._shm.name)
+
+        self._first.destroy()
+        self._last.destroy()
+
+    def flush(self) -> None:
+        # TODO: flush to storage backend like markestore?
+        ...
+
+
+def open_shm_array(
+    size: int,
+    key: str | None = None,
+    dtype: np.dtype | None = None,
+    append_start_index: int | None = None,
+    readonly: bool = False,
+
+) -> ShmArray:
+    '''Open a memory shared ``numpy`` using the standard library.
+
+    This call unlinks (aka permanently destroys) the buffer on teardown
+    and thus should be used from the parent-most accessor (process).
+
+    '''
+    # create new shared mem segment for which we
+    # have write permission
+    a = np.zeros(size, dtype=dtype)
+    a['index'] = np.arange(len(a))
+
+    shm = SharedMemory(
+        name=key,
+        create=True,
+        size=a.nbytes
+    )
+    array = np.ndarray(
+        a.shape,
+        dtype=a.dtype,
+        buffer=shm.buf
+    )
+    array[:] = a[:]
+    array.setflags(write=int(not readonly))
+
+    token = _make_token(
         key=key,
         size=size,
         dtype=dtype,
     )
 
+    # create single entry arrays for storing an first and last indices
+    first = SharedInt(
+        shm=SharedMemory(
+            name=token.shm_first_index_name,
+            create=True,
+            size=4,  # std int
+        )
+    )
+
+    last = SharedInt(
+        shm=SharedMemory(
+            name=token.shm_last_index_name,
+            create=True,
+            size=4,  # std int
+        )
+    )
+
+    # start the "real-time" updated section after 3-days worth of 1s
+    # sampled OHLC. this allows appending up to a days worth from
+    # tick/quote feeds before having to flush to a (tsdb) storage
+    # backend, and looks something like,
+    # -------------------------
+    # |              |        i
+    # _________________________
+    # <-------------> <------->
+    #  history         real-time
+    #
+    # Once fully "prepended", the history section will leave the
+    # ``ShmArray._start.value: int = 0`` and the yet-to-be written
+    # real-time section will start at ``ShmArray.index: int``.
+
+    # this sets the index to nearly 2/3rds into the the length of
+    # the buffer leaving at least a "days worth of second samples"
+    # for the real-time section.
+    if append_start_index is None:
+        append_start_index = round(size * 0.616)
+
+    last.value = first.value = append_start_index
+
+    shmarr = ShmArray(
+        array,
+        first,
+        last,
+        shm,
+    )
+
+    assert shmarr._token == token
+    _known_tokens[key] = shmarr.token
+
+    # "unlink" created shm on process teardown by
+    # pushing teardown calls onto actor context stack
+    stack = tractor.current_actor(
+        err_on_no_runtime=False,
+    ).lifetime_stack
+    if stack:
+        stack.callback(shmarr.close)
+        stack.callback(shmarr.destroy)
+
+    return shmarr
+
+
+def attach_shm_array(
+    token: tuple[str, str, tuple[str, str]],
+    readonly: bool = True,
+
+) -> ShmArray:
+    '''
+    Attach to an existing shared memory array previously
+    created by another process using ``open_shared_array``.
+
+    No new shared mem is allocated but wrapper types for read/write
+    access are constructed.
+
+    '''
+    token = _Token.from_msg(token)
+    key = token.shm_name
+
+    if key in _known_tokens:
+        assert _Token.from_msg(_known_tokens[key]) == token, "WTF"
+
+    # XXX: ugh, looks like due to the ``shm_open()`` C api we can't
+    # actually place files in a subdir, see discussion here:
+    # https://stackoverflow.com/a/11103289
+
+    # attach to array buffer and view as per dtype
+    _err: Optional[Exception] = None
+    for _ in range(3):
+        try:
+            shm = SharedMemory(
+                name=key,
+                create=False,
+            )
+            break
+        except OSError as oserr:
+            _err = oserr
+            time.sleep(0.1)
+    else:
+        if _err:
+            raise _err
+
+    shmarr = np.ndarray(
+        (token.size,),
+        dtype=token.dtype,
+        buffer=shm.buf
+    )
+    shmarr.setflags(write=int(not readonly))
+
+    first = SharedInt(
+        shm=SharedMemory(
+            name=token.shm_first_index_name,
+            create=False,
+            size=4,  # std int
+        ),
+    )
+    last = SharedInt(
+        shm=SharedMemory(
+            name=token.shm_last_index_name,
+            create=False,
+            size=4,  # std int
+        ),
+    )
+
+    # make sure we can read
+    first.value
+
+    sha = ShmArray(
+        shmarr,
+        first,
+        last,
+        shm,
+    )
+    # read test
+    sha.array
+
+    # Stash key -> token knowledge for future queries
+    # via `maybe_opepn_shm_array()` but only after we know
+    # we can attach.
+    if key not in _known_tokens:
+        _known_tokens[key] = token
+
+    # "close" attached shm on actor teardown
+    if (actor := tractor.current_actor(
+        err_on_no_runtime=False,
+    )):
+        actor.lifetime_stack.callback(sha.close)
+
+    return sha
+
 
 def maybe_open_shm_array(
     key: str,
     size: int,
-    dtype: np.dtype|None = None,
-    append_start_index: int|None = None,
+    dtype: np.dtype | None = None,
+    append_start_index: int | None = None,
     readonly: bool = False,
     **kwargs,
+
 ) -> tuple[ShmArray, bool]:
     '''
-    Attempt to attach to a shared memory block
-    using a "key" lookup to registered blocks in
-    the user's overall "system" registry (presumes
-    you don't have the block's explicit token).
+    Attempt to attach to a shared memory block using a "key" lookup
+    to registered blocks in the users overall "system" registry
+    (presumes you don't have the block's explicit token).
 
-    This is a thin wrapper around tractor's
-    ``maybe_open_shm_ndarray()`` preserving piker's
-    historical defaults (``readonly=False``,
-    ``append_start_index=None``).
+    This function is meant to solve the problem of discovering whether
+    a shared array token has been allocated or discovered by the actor
+    running in **this** process. Systems where multiple actors may seek
+    to access a common block can use this function to attempt to acquire
+    a token as discovered by the actors who have previously stored
+    a "key" -> ``_Token`` map in an actor local (aka python global)
+    variable.
 
-    If you know the explicit ``NDToken`` for your
-    memory segment instead use
-    ``tractor.ipc._shm.attach_shm_ndarray()``.
+    If you know the explicit ``_Token`` for your memory segment instead
+    use ``attach_shm_array``.
 
     '''
     try:
         # see if we already know this key
         token = _known_tokens[key]
         return (
-            attach_shm_ndarray(
+            attach_shm_array(
                 token=token,
                 readonly=readonly,
             ),
             False,
         )
     except KeyError:
-        log.debug(
-            f'Could not find {key} in shms cache'
-        )
+        log.debug(f"Could not find {key} in shms cache")
         if dtype:
             token = _make_token(
                 key,
@@ -108,18 +663,9 @@ def maybe_open_shm_array(
                 dtype=dtype,
             )
             try:
-                return (
-                    attach_shm_ndarray(
-                        token=token,
-                        **kwargs,
-                    ),
-                    False,
-                )
+                return attach_shm_array(token=token, **kwargs), False
             except FileNotFoundError:
-                log.debug(
-                    f'Could not attach to shm'
-                    f' with token {token}'
-                )
+                log.debug(f"Could not attach to shm with token {token}")
 
         # This actor does not know about memory
         # associated with the provided "key".
@@ -127,7 +673,7 @@ def maybe_open_shm_array(
         # to fail if a block has been allocated
         # on the OS by someone else.
         return (
-            open_shm_ndarray(
+            open_shm_array(
                 key=key,
                 size=size,
                 dtype=dtype,
@@ -137,20 +683,18 @@ def maybe_open_shm_array(
             True,
         )
 
-
 def try_read(
-    array: np.ndarray,
-) -> np.ndarray|None:
-    '''
-    Try to read the last row from a shared mem
-    array or ``None`` if the array read returns
-    a zero-length array result.
+    array: np.ndarray
 
-    Can be used to check for backfilling race
-    conditions where an array is currently being
-    (re-)written by a writer actor but the reader
-    is unaware and reads during the window where
-    the first and last indexes are being updated.
+) -> Optional[np.ndarray]:
+    '''
+    Try to read the last row from a shared mem array or ``None``
+    if the array read returns a zero-length array result.
+
+    Can be used to check for backfilling race conditions where an array
+    is currently being (re-)written by a writer actor but the reader is
+    unaware and reads during the window where the first and last indexes
+    are being updated.
 
     '''
     try:
@@ -158,13 +702,14 @@ def try_read(
     except IndexError:
         # XXX: race condition with backfilling shm.
         #
-        # the underlying issue is that a backfill
-        # (aka prepend) and subsequent shm array
-        # first/last index update could result in an
-        # empty array read here since the indices may
-        # be updated in such a way that a read delivers
-        # an empty array (though it seems like we
-        # *should* be able to prevent that?).
+        # the underlying issue is that a backfill (aka prepend) and subsequent
+        # shm array first/last index update could result in an empty array
+        # read here since the indices may be updated in such a way that
+        # a read delivers an empty array (though it seems like we
+        # *should* be able to prevent that?). also, as and alt and
+        # something we need anyway, maybe there should be some kind of
+        # signal that a prepend is taking place and this consumer can
+        # respond (eg. redrawing graphics) accordingly.
 
-        # the array read was empty
+        # the array read was emtpy
         return None

piker/data/_symcache.py

@@ -105,6 +105,15 @@ class SymbologyCache(Struct):
 
     def write_config(self) -> None:
 
+        def clean_dict_for_toml(d):
+            '''Remove None values from dict recursively for TOML serialization'''
+            if isinstance(d, dict):
+                return {k: clean_dict_for_toml(v) for k, v in d.items() if v is not None}
+            elif isinstance(d, list):
+                return [clean_dict_for_toml(item) for item in d if item is not None]
+            else:
+                return d
+
         # put the backend's pair-struct type ref at the top
         # of file if possible.
         cachedict: dict[str, Any] = {
@@ -125,7 +134,9 @@ class SymbologyCache(Struct):
 
             dct = cachedict[key] = {}
             for key, struct in table.items():
-                dct[key] = struct.to_dict(include_non_members=False)
+                raw_dict = struct.to_dict(include_non_members=False)
+                # Clean None values for TOML compatibility
+                dct[key] = clean_dict_for_toml(raw_dict)
 
         try:
             with self.fp.open(mode='wb') as fp:

piker/data/flows.py

@@ -31,10 +31,10 @@ import pendulum
 import numpy as np
 
 from piker.types import Struct
-from tractor.ipc._shm import (
+from ._sharedmem import (
+    attach_shm_array,
     ShmArray,
-    NDToken,
-    attach_shm_ndarray,
+    _Token,
 )
 from piker.accounting import MktPair
 
@@ -64,11 +64,11 @@ class Flume(Struct):
     '''
     mkt: MktPair
     first_quote: dict
-    _rt_shm_token: NDToken
+    _rt_shm_token: _Token
 
     # optional since some data flows won't have a "downsampled" history
     # buffer/stream (eg. FSPs).
-    _hist_shm_token: NDToken|None = None
+    _hist_shm_token: _Token | None = None
 
     # private shm refs loaded dynamically from tokens
     _hist_shm: ShmArray | None = None
@@ -88,7 +88,7 @@ class Flume(Struct):
     def rt_shm(self) -> ShmArray:
 
         if self._rt_shm is None:
-            self._rt_shm = attach_shm_ndarray(
+            self._rt_shm = attach_shm_array(
                 token=self._rt_shm_token,
                 readonly=self._readonly,
             )
@@ -104,7 +104,7 @@ class Flume(Struct):
             )
 
         if self._hist_shm is None:
-            self._hist_shm = attach_shm_ndarray(
+            self._hist_shm = attach_shm_array(
                 token=self._hist_shm_token,
                 readonly=self._readonly,
             )

piker/fsp/_api.py

@@ -37,12 +37,12 @@ import numpy as np
 import tractor
 from tractor.msg import NamespacePath
 
-from tractor.ipc._shm import (
+from ..data._sharedmem import (
     ShmArray,
-    NDToken,
-    attach_shm_ndarray,
+    maybe_open_shm_array,
+    attach_shm_array,
+    _Token,
 )
-from ..data._sharedmem import maybe_open_shm_array
 from ..log import get_logger
 
 log = get_logger(__name__)
@@ -78,8 +78,8 @@ class Fsp:
     # + the consuming fsp *to* the consumers output
     # shm flow.
     _flow_registry: dict[
-        tuple[NDToken, str],
-        tuple[NDToken, Optional[ShmArray]],
+        tuple[_Token, str],
+        tuple[_Token, Optional[ShmArray]],
     ] = {}
 
     def __init__(
@@ -148,7 +148,7 @@ class Fsp:
                 # times as possible as per:
                 # - https://github.com/pikers/piker/issues/359
                 # - https://github.com/pikers/piker/issues/332
-                maybe_array := attach_shm_ndarray(dst_token)
+                maybe_array := attach_shm_array(dst_token)
             )
 
         return maybe_array

piker/fsp/_engine.py

@@ -40,7 +40,7 @@ from ..log import (
 )
 from .. import data
 from ..data.flows import Flume
-from tractor.ipc._shm import ShmArray
+from ..data._sharedmem import ShmArray
 from ..data._sampling import (
     _default_delay_s,
     open_sample_stream,
@@ -49,7 +49,7 @@ from ..accounting import MktPair
 from ._api import (
     Fsp,
     _load_builtins,
-    NDToken,
+    _Token,
 )
 from ..toolz import Profiler
 
@@ -414,7 +414,7 @@ async def cascade(
     dst_flume_addr: dict,
     ns_path: NamespacePath,
 
-    shm_registry: dict[str, NDToken],
+    shm_registry: dict[str, _Token],
 
     zero_on_step: bool = False,
     loglevel: str|None = None,
@@ -465,9 +465,9 @@ async def cascade(
     # not sure how else to do it.
     for (token, fsp_name, dst_token) in shm_registry:
         Fsp._flow_registry[(
-            NDToken.from_msg(token),
+            _Token.from_msg(token),
             fsp_name,
-        )] = NDToken.from_msg(dst_token), None
+        )] = _Token.from_msg(dst_token), None
 
     fsp: Fsp = reg.get(
         NamespacePath(ns_path)

piker/fsp/_momo.py

@@ -25,7 +25,7 @@ from numba import jit, float64, optional, int64
 
 from ._api import fsp
 from ..data import iterticks
-from tractor.ipc._shm import ShmArray
+from ..data._sharedmem import ShmArray
 
 
 @jit(

piker/fsp/_volume.py

@@ -21,7 +21,7 @@ from tractor.trionics._broadcast import AsyncReceiver
 
 from ._api import fsp
 from ..data import iterticks
-from tractor.ipc._shm import ShmArray
+from ..data._sharedmem import ShmArray
 from ._momo import _wma
 from ..log import get_logger
 

piker/storage/cli.py

@@ -37,7 +37,9 @@ import typer
 
 from piker.service import open_piker_runtime
 from piker.cli import cli
-from tractor.ipc._shm import ShmArray
+from piker.data import (
+    ShmArray,
+)
 from piker import tsp
 from . import log
 from . import (

piker/storage/nativedb.py

@@ -64,8 +64,10 @@ from pendulum import (
 
 from piker import config
 from piker import tsp
-from tractor.ipc._shm import ShmArray
-from piker.data import def_iohlcv_fields
+from piker.data import (
+    def_iohlcv_fields,
+    ShmArray,
+)
 from piker.log import get_logger
 from . import TimeseriesNotFound
 

piker/tsp/_history.py

@@ -59,12 +59,11 @@ from piker.brokers import NoData
 from piker.accounting import (
     MktPair,
 )
-from piker.log import (
-    get_logger,
-    get_console_log,
+from piker.log import get_logger
+from ..data._sharedmem import (
+    maybe_open_shm_array,
+    ShmArray,
 )
-from tractor.ipc._shm import ShmArray
-from ..data._sharedmem import maybe_open_shm_array
 from piker.data._source import (
     def_iohlcv_fields,
 )
@@ -250,20 +249,10 @@ async def maybe_fill_null_segments(
             end_dt=end_dt,
         )
 
-        if array.size == 0:
-            log.warning(
-                f'Valid gap from backend ??\n'
-                f'{end_dt} -> {start_dt}\n'
-            )
-            # ?TODO? do we want to remove the nulls and push
-            # the close price here for the gap duration?
-            await tractor.pause()
-            break
-
         if (
-            frame_start_dt := (from_timestamp(array['time'][0]))
-            <
-            backfill_until_dt
+            frame_start_dt := (
+                from_timestamp(array['time'][0])
+            ) < backfill_until_dt
         ):
             log.error(
                 f'Invalid frame_start !?\n'
@@ -625,17 +614,10 @@ async def start_backfill(
 
             else:
                 log.warning(
-                    f'0 BARS TO PUSH after diff!?\n'
+                    '0 BARS TO PUSH after diff!?\n'
                     f'{next_start_dt} -> {last_start_dt}'
-                    f'\n'
-                    f'This might mean we rxed a gap frame which starts BEFORE,\n'
-                    f'backfill_until_dt: {backfill_until_dt}\n'
-                    f'end_dt_param: {end_dt_param}\n'
-
                 )
-                # XXX, to debug it and be sure.
-                # await tractor.pause()
-                break
+                await tractor.pause()
 
             # Check if we're about to exceed buffer capacity BEFORE
             # attempting the push
@@ -1387,10 +1369,6 @@ async def manage_history(
         engages.
 
     '''
-    get_console_log(
-        name=__name__,
-        level=loglevel,
-    )
     # TODO: is there a way to make each shm file key
     # actor-tree-discovery-addr unique so we avoid collisions
     # when doing tests which also allocate shms for certain instruments

piker/ui/_chart.py

@@ -49,7 +49,7 @@ from ._cursor import (
     Cursor,
     ContentsLabel,
 )
-from tractor.ipc._shm import ShmArray
+from ..data._sharedmem import ShmArray
 from ._ohlc import BarItems
 from ._curve import (
     Curve,

piker/ui/_dataviz.py

@@ -42,7 +42,9 @@ from numpy import (
 import pyqtgraph as pg
 
 from piker.ui.qt import QLineF
-from tractor.ipc._shm import ShmArray
+from ..data._sharedmem import (
+    ShmArray,
+)
 from ..data.flows import Flume
 from ..data._formatters import (
     IncrementalFormatter,

piker/ui/_fsp.py

@@ -44,12 +44,14 @@ from piker.fsp import (
     dolla_vlm,
     flow_rates,
 )
-from tractor.ipc._shm import (
+from piker.data import (
+    Flume,
     ShmArray,
-    NDToken,
 )
-from piker.data import Flume
-from piker.data._sharedmem import try_read
+from piker.data._sharedmem import (
+    _Token,
+    try_read,
+)
 from piker.log import get_logger
 from piker.toolz import Profiler
 from piker.types import Struct
@@ -380,7 +382,7 @@ class FspAdmin:
             tuple,
             tuple[tractor.MsgStream, ShmArray]
         ] = {}
-        self._flow_registry: dict[NDToken, str] = {}
+        self._flow_registry: dict[_Token, str] = {}
 
         # TODO: make this a `.src_flume` and add
         # a `dst_flume`?

piker/ui/order_mode.py

@@ -34,7 +34,6 @@ import uuid
 
 from bidict import bidict
 import tractor
-from tractor.devx.pformat import ppfmt
 import trio
 
 from piker import config
@@ -1208,10 +1207,11 @@ async def process_trade_msg(
                 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)}'
+                # f'{ppfmt(broker_msg)}'
             )
             # do all the things for a cancel:
             # - drop order-msg dialog from client table

pyproject.toml

@@ -52,6 +52,7 @@ dependencies = [
     "bidict >=0.23.1",
     "colorama >=0.4.6, <0.5.0",
     "colorlog >=6.7.0, <7.0.0",
+    "ib-insync >=0.9.86, <0.10.0",
     "numpy>=2.0",
     "polars >=0.20.6",
     "polars-fuzzy-match>=0.1.5",
@@ -75,8 +76,6 @@ dependencies = [
     "numba>=0.61.0",
     "pyvnc",
     "exchange-calendars>=4.13.1",
-    "ib-async>=2.1.0",
-    "aeventkit>=2.1.0",  # XXX, imports as eventkit?
 ]
 # ------ dependencies ------
 # NOTE, by default we ship only a "headless" deps set bc
@@ -203,8 +202,9 @@ pyvnc = { git = "https://github.com/regulad/pyvnc.git" }
 # xonsh = { git = 'https://github.com/xonsh/xonsh.git', branch = 'main' }
 
 # XXX since, we're like, always hacking new shite all-the-time. Bp
-tractor = { git = "https://github.com/goodboy/tractor.git", branch ="main" }
+tractor = { git = "https://github.com/goodboy/tractor.git", branch ="piker_pin" }
 # tractor = { git = "https://pikers.dev/goodboy/tractor", branch = "piker_pin" }
+# tractor = { git = "https://pikers.dev/goodboy/tractor", branch = "main" }
 # ------ goodboy ------
 # hackin dev-envs, usually there's something new he's hackin in..
 # tractor = { path = "../tractor", editable = true }

uv.lock

@@ -7,18 +7,6 @@ resolution-markers = [
     "sys_platform != 'emscripten' and sys_platform != 'win32'",
 ]
 
-[[package]]
-name = "aeventkit"
-version = "2.1.0"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
-    { name = "numpy" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/5c/8c/c08db1a1910f8d04ec6a524de522edd0bac181bdf94dbb01183f7685cd77/aeventkit-2.1.0.tar.gz", hash = "sha256:4e7d81bb0a67227121da50a23e19e5bbf13eded541a9f4857eeb6b7b857b738a", size = 24703, upload-time = "2025-06-22T15:54:03.961Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/8d/8c/2a4b912b1afa201b25bdd0f5bccf96d5a8b5dccb6131316a8dd2d9cabcc1/aeventkit-2.1.0-py3-none-any.whl", hash = "sha256:962d43f79e731ac43527f2d0defeed118e6dbaa85f1487f5667540ebb8f00729", size = 26678, upload-time = "2025-06-22T15:54:02.141Z" },
-]
-
 [[package]]
 name = "aiodns"
 version = "3.6.0"
@@ -363,6 +351,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/56/01/6f77d042b83260ef9ed73ea9647dfa0ef8414eba0a3fc57a509a088ad39b/elasticsearch-8.19.2-py3-none-any.whl", hash = "sha256:c16ba20c4c76cf6952e836dae7f4e724e00ba7bf31b94b79472b873683accdd4", size = 949706, upload-time = "2025-10-28T16:36:41.003Z" },
 ]
 
+[[package]]
+name = "eventkit"
+version = "1.0.3"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "numpy" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/16/1e/0fac4e45d71ace143a2673ec642701c3cd16f833a0e77a57fa6a40472696/eventkit-1.0.3.tar.gz", hash = "sha256:99497f6f3c638a50ff7616f2f8cd887b18bbff3765dc1bd8681554db1467c933", size = 28320, upload-time = "2023-12-11T11:41:35.339Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/93/d9/7497d650b69b420e1a913329a843e16c715dac883750679240ef00a921e2/eventkit-1.0.3-py3-none-any.whl", hash = "sha256:0e199527a89aff9d195b9671ad45d2cc9f79ecda0900de8ecfb4c864d67ad6a2", size = 31837, upload-time = "2023-12-11T11:41:33.358Z" },
+]
+
 [[package]]
 name = "exceptiongroup"
 version = "1.3.1"
@@ -538,17 +538,16 @@ wheels = [
 ]
 
 [[package]]
-name = "ib-async"
-version = "2.1.0"
+name = "ib-insync"
+version = "0.9.86"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
-    { name = "aeventkit" },
+    { name = "eventkit" },
     { name = "nest-asyncio" },
-    { name = "tzdata" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/30/4d/dfc1da8224c3ffcdcd668da7283c4e5f14239a07f83ea66af99700296fc3/ib_async-2.1.0.tar.gz", hash = "sha256:6a03a87d6c06acacb0217a5bea60a8a168ecd5b5a7e86e1c73678d5b48cbc796", size = 87678, upload-time = "2025-12-08T01:42:32.004Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/55/bb/733d5c81c8c2f54e90898afc7ff3a99f4d53619e6917c848833f9cc1ab56/ib_insync-0.9.86.tar.gz", hash = "sha256:73af602ca2463f260999970c5bd937b1c4325e383686eff301743a4de08d381e", size = 69859, upload-time = "2023-07-02T12:43:31.968Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/80/e7/8f33801788c66f15e9250957ff7f53a8000843f79af1a3ed7a96def0e96b/ib_async-2.1.0-py3-none-any.whl", hash = "sha256:f6d8b991bdbd6dd38e700c61b3dced06ebe0f14be4e5263e2ef10ba10b88d434", size = 88876, upload-time = "2025-12-08T01:42:30.883Z" },
+    { url = "https://files.pythonhosted.org/packages/8f/f3/28ea87be30570f4d6b8fd24380d12fa74e59467ee003755e76aeb29082b8/ib_insync-0.9.86-py3-none-any.whl", hash = "sha256:a61fbe56ff405d93d211dad8238d7300de76dd6399eafc04c320470edec9a4a4", size = 72980, upload-time = "2023-07-02T12:43:29.928Z" },
 ]
 
 [[package]]
@@ -934,7 +933,6 @@ name = "piker"
 version = "0.1.0a0.dev0"
 source = { editable = "." }
 dependencies = [
-    { name = "aeventkit" },
     { name = "async-generator" },
     { name = "attrs" },
     { name = "bidict" },
@@ -943,7 +941,7 @@ dependencies = [
     { name = "cryptofeed" },
     { name = "exchange-calendars" },
     { name = "httpx" },
-    { name = "ib-async" },
+    { name = "ib-insync" },
     { name = "msgspec" },
     { name = "numba" },
     { name = "numpy" },
@@ -1011,7 +1009,6 @@ uis = [
 
 [package.metadata]
 requires-dist = [
-    { name = "aeventkit", specifier = ">=2.1.0" },
     { name = "async-generator", specifier = ">=1.10,<2.0.0" },
     { name = "attrs", specifier = ">=23.1.0,<24.0.0" },
     { name = "bidict", specifier = ">=0.23.1" },
@@ -1020,7 +1017,7 @@ requires-dist = [
     { name = "cryptofeed", specifier = ">=2.4.0,<3.0.0" },
     { name = "exchange-calendars", specifier = ">=4.13.1" },
     { name = "httpx", specifier = ">=0.27.0,<0.28.0" },
-    { name = "ib-async", specifier = ">=2.1.0" },
+    { name = "ib-insync", specifier = ">=0.9.86,<0.10.0" },
     { name = "msgspec", specifier = ">=0.19.0,<0.20" },
     { name = "numba", specifier = ">=0.61.0" },
     { name = "numpy", specifier = ">=2.0" },
@@ -1034,7 +1031,7 @@ requires-dist = [
     { name = "tomli", specifier = ">=2.0.1,<3.0.0" },
     { name = "tomli-w", specifier = ">=1.0.0,<2.0.0" },
     { name = "tomlkit", git = "https://github.com/pikers/tomlkit.git?branch=piker_pin" },
-    { name = "tractor", git = "https://github.com/goodboy/tractor.git?branch=main" },
+    { name = "tractor", git = "https://github.com/goodboy/tractor.git?branch=piker_pin" },
     { name = "trio", specifier = ">=0.27" },
     { name = "trio-typing", specifier = ">=0.10.0" },
     { name = "trio-util", specifier = ">=0.7.0,<0.8.0" },
@@ -1676,7 +1673,7 @@ wheels = [
 [[package]]
 name = "tractor"
 version = "0.1.0a6.dev0"
-source = { git = "https://github.com/goodboy/tractor.git?branch=main#e77198bb64f0467a50e251ed140daee439752354" }
+source = { git = "https://github.com/goodboy/tractor.git?branch=piker_pin#566a17ba92469000a01fd18c709ea3e336e72796" }
 dependencies = [
     { name = "bidict" },
     { name = "cffi" },