ib: explicit ib_insync import in prep for passing client to reset h…

…acker

piker/brokers/binance/api.py

@@ -661,10 +661,14 @@ async def submit_limit(
             signed=True,
             action='post'
         )
-        reqid: str = resp['orderId']
+        # import tractor
+        # await tractor.breakpoint()
+
+        # ensure our id is tracked by them
         if oid:
-            assert oid == reqid
+            assert oid == resp['clientOrderId']
 
+        reqid: str = resp['orderId']
         return reqid
 
     async def submit_cancel(

piker/brokers/ib/_util.py

@@ -21,13 +21,20 @@
 '''
 from __future__ import annotations
 from functools import partial
-from typing import Literal
+from typing import (
+    Literal,
+    TYPE_CHECKING,
+)
 import subprocess
 
 import tractor
 
 from .._util import get_logger
 
+if TYPE_CHECKING:
+    from .api import Client
+    from ib_insync import IB
+
 log = get_logger('piker.brokers.ib')
 
 _reset_tech: Literal[
@@ -42,7 +49,8 @@
 
 
 async def data_reset_hack(
-    vnc_host: str,
+    # vnc_host: str,
+    client: Client,
     reset_type: Literal['data', 'connection'],
 
 ) -> None:

piker/brokers/ib/broker.py

@@ -512,9 +512,9 @@ async def open_trade_event_stream(
     async with tractor.to_asyncio.open_channel_from(
         recv_trade_updates,
         client=client,
-    ) as (ibclient, trade_event_stream):
+    ) as (_, trade_event_stream):
 
-        assert ibclient is client.ib
+        # assert ibclient is client.ib
         task_status.started(trade_event_stream)
         await trio.sleep_forever()
 

piker/brokers/ib/feed.py

@@ -37,6 +37,7 @@
 
 from async_generator import aclosing
 from fuzzywuzzy import process as fuzzy
+import ib_insync as ibis
 import numpy as np
 import pendulum
 import tractor
@@ -50,10 +51,10 @@
 )
 from .api import (
     # _adhoc_futes_set,
+    Client,
     con2fqme,
     log,
     load_aio_clients,
-    ibis,
     MethodProxy,
     open_client_proxies,
     get_preferred_data_client,
@@ -276,7 +277,8 @@ async def wait_on_data_reset(
     # )
     # try to wait on the reset event(s) to arrive, a timeout
     # will trigger a retry up to 6 times (for now).
-    client = proxy._aio_ns.ib.client
+    client: Client = proxy._aio_ns
+    ib_client: ibis.IB = client.ib
 
     done = trio.Event()
     with trio.move_on_after(timeout) as cs:
@@ -285,10 +287,11 @@ async def wait_on_data_reset(
 
         log.warning(
             'Sending DATA RESET request:\n'
-            f'{client}'
+            f'{ib_client.client}'
         )
         res = await data_reset_hack(
-            vnc_host=client.host,
+            # vnc_host=client.host,
+            ib_client=ib_client,
             reset_type=reset_type,
         )