Drop `Flume.feed`, it's unused yet causes import cycles..

Replace the debug breakpoint with a warning-log when a single-bar
null-segment is detected in `get_null_segs()`. This lets the gap
analysis continue while still alerting about the anomaly.

Deats,
- extract the 3-bar window (before, null, after) and calculate
  a `gap: pendulum.Interval` for the warning msg.
- comment-out the old breakpoint block for optional debugging as needed.

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

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_insync` project.
+and we interface to their APIs using the `ib_async` 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_insync``.
+  wrapping around `ib_async`.
 
 """
 from .api import (

piker/brokers/ib/_flex_reports.py

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

piker/brokers/ib/api.py

@@ -15,7 +15,8 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 '''
-Core API client machinery; mostly sane/useful wrapping around `ib_insync`..
+Core API client machinery; mostly sane/useful wrapping around
+`ib_async`..
 
 '''
 from __future__ import annotations
@@ -57,7 +58,7 @@ from pendulum import (
     Interval,
 )
 from eventkit import Event
-from ib_insync import (
+from ib_async import (
     client as ib_client,
     IB,
     Contract,
@@ -143,7 +144,7 @@ _bar_sizes = {
 _show_wap_in_history: bool = False
 
 # overrides to sidestep pretty questionable design decisions in
-# ``ib_insync``:
+# ``ib_async``:
 class NonShittyWrapper(Wrapper):
     def tcpDataArrived(self):
         """Override time stamps to be floats for now.
@@ -183,7 +184,7 @@ class NonShittyIB(IB):
     '''
     def __init__(self):
 
-        # override `ib_insync` internal loggers so we can see wtf
+        # override `ib_async` internal loggers so we can see wtf
         # it's doing..
         self._logger = get_logger(
             name=__name__,
@@ -194,7 +195,7 @@ class NonShittyIB(IB):
         self.wrapper = NonShittyWrapper(self)
         self.client = ib_client.Client(self.wrapper)
         self.client._logger = get_logger(
-            name='ib_insync.client',
+            name='ib_async.client',
         )
 
         # self.errorEvent += self._onError
@@ -779,13 +780,14 @@ class Client:
                 ContFuture(symbol, exchange=exchange)
             ))[0]
         else:
-            con = (await self.ib.qualifyContractsAsync(
+            cons = (await self.ib.qualifyContractsAsync(
                 Future(
                     symbol,
                     exchange=exchange,
                     lastTradeDateOrContractMonth=expiry,
                 )
-            ))[0]
+            ))
+            con = cons[0]
 
         return con
 
@@ -878,7 +880,7 @@ class Client:
                 currency='USD',
                 exchange='PAXOS',
             )
-            # XXX, on `ib_insync` when first tried this,
+            # XXX, on `ib_async` 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',
@@ -992,7 +994,6 @@ class Client:
     async def get_sym_details(
         self,
         fqme: str,
-
     ) -> tuple[
         Contract,
         ContractDetails,
@@ -1092,7 +1093,7 @@ class Client:
         size: int,
         account: str,  # if blank the "default" tws account is used
 
-        # XXX: by default 0 tells ``ib_insync`` methods that there is no
+        # XXX: by default 0 tells ``ib_async`` 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,
@@ -1281,7 +1282,7 @@ async def load_aio_clients(
     port: int = None,
     client_id: int = 6116,
 
-    # the API TCP in `ib_insync` connection can be flaky af so instead
+    # the API TCP in `ib_async` 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
@@ -1289,7 +1290,7 @@ async def load_aio_clients(
 
 ) -> dict[str, Client]:
     '''
-    Return an ``ib_insync.IB`` instance wrapped in our client API.
+    Return an ``ib_async.IB`` instance wrapped in our client API.
 
     Client instances are cached for later use.
 
@@ -1631,6 +1632,7 @@ 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)
 
@@ -1640,7 +1642,11 @@ 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: tuple[str, dict]|dict|None = await chan.get()
+        msg: (
+            None
+            |tuple[str, dict]
+            |dict
+        ) = await chan.get()
         match msg:
             case None:  # termination sentinel
                 log.info('asyncio `Client` method-proxy SHUTDOWN!')
@@ -1742,7 +1748,7 @@ async def get_client(
 
 ) -> Client:
     '''
-    Init the ``ib_insync`` client in another actor and return
+    Init the ``ib_async`` 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_insync.contract import (
+from ib_async.contract import (
     Contract,
 )
-from ib_insync.order import (
+from ib_async.order import (
     Trade,
     OrderStatus,
 )
-from ib_insync.objects import (
+from ib_async.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_insync`` methods that
+            # XXX: by default 0 tells ``ib_async`` 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_insync`, pack as tuples and push over mem-chan to our
+    from `ib_async`, 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_insync internal
+        # XXX: these all seem to be weird ib_async 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_insync
+    Proxy wrapper for starting trade event stream from ib_async
     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

piker/brokers/ib/feed.py

@@ -36,7 +36,7 @@ from typing import (
 )
 
 from async_generator import aclosing
-import ib_insync as ibis
+import ib_async as ibis
 import numpy as np
 from pendulum import (
     now,
@@ -100,7 +100,7 @@ tick_types = {
     5: 'size',
     8: 'volume',
 
-    # ``ib_insync`` already packs these into
+    # `ib_async` already packs these into
     # quotes under the following fields.
     55: 'trades_per_min',  # `'tradeRate'`
     56: 'vlm_per_min',  # `'volumeRate'`
@@ -304,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_insync` machinery, for eg. maybe
+        # a lookthrough of the `ib_async` machinery, for eg. maybe
         # we have to do the batch queries on the `asyncio` side?
         yield (
             get_hist,
@@ -1051,6 +1051,21 @@ 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
 
 
@@ -1227,7 +1242,7 @@ async def stream_quotes(
                 ):
                     # ?TODO? can we rm this - particularly for `ib_async`?
                     # ugh, clear ticks since we've consumed them
-                    # (ahem, ib_insync is stateful trash)
+                    # (ahem, ib_async 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_insync import (
+from ib_async import (
     Contract,
     Commodity,
     Fill,

piker/brokers/ib/symbols.py

@@ -30,7 +30,7 @@ from typing import (
 )
 
 from rapidfuzz import process as fuzzy
-import ib_insync as ibis
+import ib_async as ibis
 import tractor
 import trio
 
@@ -522,7 +522,11 @@ async def get_mkt_info(
     if atype == 'commodity':
         venue: str = 'cmdty'
     else:
-        venue = con.primaryExchange or con.exchange
+        venue: str = (
+            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,7 +41,7 @@ from pendulum import (
 )
 
 if TYPE_CHECKING:
-    from ib_insync import (
+    from ib_async import (
         TradingSession,
         ContractDetails,
     )
@@ -236,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_insync.contract.ContractDetails._parseSessions()` and
+# `ib_async.contract.ContractDetails._parseSessions()` and
 # it's downstream meths..
 #
 # This is still likely useful to keep for now to parse the

piker/data/feed.py

@@ -973,9 +973,6 @@ async def open_feed(
                     # assert flume.mkt.fqme == fqme
                     feed.flumes[fqme] = flume
 
-                    # TODO: do we need this?
-                    flume.feed = feed
-
                     # attach and cache shm handles
                     rt_shm = flume.rt_shm
                     assert rt_shm

piker/data/flows.py

@@ -22,9 +22,6 @@ real-time data processing data-structures.
 
 """
 from __future__ import annotations
-from typing import (
-    TYPE_CHECKING,
-)
 
 import tractor
 import pendulum
@@ -38,9 +35,6 @@ from ._sharedmem import (
 )
 from piker.accounting import MktPair
 
-if TYPE_CHECKING:
-    from piker.data.feed import Feed
-
 
 class Flume(Struct):
     '''
@@ -80,10 +74,6 @@ class Flume(Struct):
     izero_rt: int = 0
     throttle_rate: int | None = None
 
-    # TODO: do we need this really if we can pull the `Portal` from
-    # ``tractor``'s internals?
-    feed: Feed|None = None
-
     @property
     def rt_shm(self) -> ShmArray:
 
@@ -156,7 +146,6 @@ class Flume(Struct):
         # will get instead some kind of msg-compat version
         # that it can load.
         msg.pop('stream')
-        msg.pop('feed')
         msg.pop('_rt_shm')
         msg.pop('_hist_shm')
 

piker/tsp/_anal.py

@@ -276,14 +276,41 @@ def get_null_segs(
         absi_zdiff: np.ndarray = np.diff(absi_zeros)
 
         if zero_t.size < 2:
-            try:
+            idx: int = zero_t['index'][0]
-                breakpoint()
+            idx_before: int = idx - 1
-            except RuntimeError:
+            idx_after: int = idx + 1
-                # XXX, if greenback not active from
+            index = frame['index']
-                # piker store ldshm cmd..
+            before_cond = idx_before <= index
-                log.exception(
+            after_cond = index <= idx_after
-                    "Can't debug single-sample null!\n"
+            bars: np.ndarray = frame[
+                before_cond
+                &
+                after_cond
+            ]
+            time: np.ndarray = bars['time']
+            from pendulum import (
+                from_timestamp,
+                Interval,
             )
+            gap: Interval = (
+                from_timestamp(time[-1])
+                -
+                from_timestamp(time[0])
+            )
+            log.warning(
+                f'Single OHLCV-bar null-segment detected??\n'
+                f'gap -> {gap}\n'
+            )
+
+            # ^^XXX, if you want to debug the above bar-gap^^
+            # try:
+            #     breakpoint()
+            # except RuntimeError:
+            #     # XXX, if greenback not active from
+            #     # piker store ldshm cmd..
+            #     log.exception(
+            #         "Can't debug single-sample null!\n"
+            #     )
 
             return None
 

pyproject.toml

@@ -52,7 +52,6 @@ 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",
@@ -76,6 +75,7 @@ dependencies = [
     "numba>=0.61.0",
     "pyvnc",
     "exchange-calendars>=4.13.1",
+    "ib-async>=2.1.0",
 ]
 # ------ dependencies ------
 # NOTE, by default we ship only a "headless" deps set bc

uv.lock

@@ -10,6 +10,18 @@ resolution-markers = [
     "python_full_version < '3.14' and 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"
@@ -396,18 +408,6 @@ 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"
@@ -630,16 +630,17 @@ wheels = [
 ]
 
 [[package]]
-name = "ib-insync"
+name = "ib-async"
-version = "0.9.86"
+version = "2.1.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
-    { name = "eventkit" },
+    { name = "aeventkit" },
     { name = "nest-asyncio" },
+    { name = "tzdata" },
 ]
-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" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]
 
 [[package]]
@@ -1107,7 +1108,7 @@ dependencies = [
     { name = "cryptofeed" },
     { name = "exchange-calendars" },
     { name = "httpx" },
-    { name = "ib-insync" },
+    { name = "ib-async" },
     { name = "msgspec" },
     { name = "numba" },
     { name = "numpy" },
@@ -1183,7 +1184,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-insync", specifier = ">=0.9.86,<0.10.0" },
+    { name = "ib-async", specifier = ">=2.1.0" },
     { name = "msgspec", specifier = ">=0.19.0,<0.20" },
     { name = "numba", specifier = ">=0.61.0" },
     { name = "numpy", specifier = ">=2.0" },