Make `ib` backend multi-client capable

This adds full support for a single `brokerd` managing multiple API
endpoint clients in tandem. Get the client scan loop correct and load
accounts from all discovered clients as specified in a user's
`broker.toml`. We now just always re-scan for all clients and if there's
a cache hit just skip a creation/connection logic.

Route orders with an account name to the correct client in the
`handle_order_requests()` endpoint and spawn an event relay task per
client for transmitting trade events back to `emsd`.

piker/brokers/ib.py

@@ -219,6 +219,8 @@ class Client:
     Note: this client requires running inside an ``asyncio`` loop.
 
     """
+    _contracts: dict[str, Contract] = {}
+
     def __init__(
         self,
 
@@ -229,7 +231,6 @@ class Client:
         self.ib.RaiseRequestErrors = True
 
         # contract cache
-        self._contracts: dict[str, Contract] = {}
         self._feeds: dict[str, trio.abc.SendChannel] = {}
 
         # NOTE: the ib.client here is "throttled" to 45 rps by default
@@ -505,7 +506,7 @@ class Client:
         return contract, ticker, details
 
     # async to be consistent for the client proxy, and cuz why not.
-    async def submit_limit(
+    def submit_limit(
         self,
         # ignored since ib doesn't support defining your
         # own order id
@@ -554,7 +555,7 @@ class Client:
         # their own weird client int counting ids..
         return trade.order.orderId
 
-    async def submit_cancel(
+    def submit_cancel(
         self,
         reqid: str,
     ) -> None:
@@ -571,6 +572,7 @@ class Client:
     async def recv_trade_updates(
         self,
         to_trio: trio.abc.SendChannel,
+
     ) -> None:
         """Stream a ticker using the std L1 api.
         """
@@ -720,30 +722,9 @@ async def load_aio_clients(
     global _client_cache
 
     conf = get_config()
+    ib = None
     client = None
 
-    # first check cache for existing client
-    if port:
-        log.info(f'Loading requested client on port: {port}')
-        client = _client_cache.get((host, port))
-
-    if client and client.ib.isConnected():
-        yield client, _client_cache, _accounts2clients
-        return
-
-    # allocate new and/or reload disconnected but cached clients
-    try:
-
-        # TODO: in case the arbiter has no record
-        # of existing brokerd we need to broadcast for one.
-
-        if client_id is None:
-            # if this is a persistent brokerd, try to allocate a new id for
-            # each client
-            client_id = next(_client_ids)
-
-        ib = NonShittyIB()
-
     # attempt to get connection info from config; if no .toml entry
     # exists, we try to load from a default localhost connection.
     host = conf.get('host', '127.0.0.1')
@@ -764,6 +745,9 @@ async def load_aio_clients(
     try_ports = [ports[key] for key in order]
     ports = try_ports if port is None else [port]
 
+    we_connected = []
+    # allocate new and/or reload disconnected but cached clients
+    try:
         # TODO: support multiple clients allowing for execution on
         # multiple accounts (including a paper instance running on the
         # same machine) and switching between accounts in the EMs
@@ -774,9 +758,15 @@ async def load_aio_clients(
         # from connection details in ``brokers.toml``.
         for port in ports:
             client = _client_cache.get((host, port))
-
+            accounts_found: dict[str, Client] = {}
             if not client or not client.ib.isConnected():
                 try:
+                    ib = NonShittyIB()
+
+                    # if this is a persistent brokerd, try to allocate
+                    # a new id for each client
+                    client_id = next(_client_ids)
+
                     log.info(f"Connecting to the EYEBEE on port {port}!")
                     await ib.connectAsync(host, port, clientId=client_id)
 
@@ -789,7 +779,7 @@ async def load_aio_clients(
                     pps = ib.positions()
                     if pps:
                         for pp in pps:
-                            _accounts2clients[
+                            accounts_found[
                                 accounts_def.inverse[pp.account]
                             ] = client
 
@@ -798,18 +788,21 @@ async def load_aio_clients(
                     # them for this client
                     for value in ib.accountValues():
                         acct = value.account
-                        if acct not in _accounts2clients:
+                        if acct not in accounts_found:
-                            _accounts2clients[
+                            accounts_found[
                                 accounts_def.inverse[acct]
                             ] = client
 
                     log.info(
-                        f'Loaded accounts: {_accounts2clients} for {client} '
+                        f'Loaded accounts for client @ {host}:{port}\n'
-                        f'@ {host}:{port}'
+                        f'{pformat(accounts_found)}'
                     )
 
+                    # update all actor-global caches
                     log.info(f"Caching client for {(host, port)}")
                     _client_cache[(host, port)] = client
+                    we_connected.append(client)
+                    _accounts2clients.update(accounts_found)
 
                 except ConnectionRefusedError as ce:
                     _err = ce
@@ -826,7 +819,8 @@ async def load_aio_clients(
         yield client, _client_cache, _accounts2clients
 
     except BaseException:
-        ib.disconnect()
+        for client in we_connected:
+            client.ib.disconnect()
         raise
 
 
@@ -834,14 +828,16 @@ async def _aio_run_client_method(
     meth: str,
     to_trio=None,
     from_trio=None,
+    client=None,
     **kwargs,
 ) -> None:
 
     async with load_aio_clients() as (
-        client,
+        _client,
         clients,
         accts2clients,
     ):
+        client = client or _client
         async_meth = getattr(client, meth)
 
         # handle streaming methods
@@ -855,7 +851,9 @@ async def _aio_run_client_method(
 
 async def _trio_run_client_method(
     method: str,
+    client: Optional[Client] = None,
     **kwargs,
+
 ) -> None:
     """Asyncio entry point to run tasks against the ``ib_insync`` api.
 
@@ -875,12 +873,12 @@ async def _trio_run_client_method(
     ):
         kwargs['_treat_as_stream'] = True
 
-    result = await tractor.to_asyncio.run_task(
+    return await tractor.to_asyncio.run_task(
         _aio_run_client_method,
         meth=method,
+        client=client,
         **kwargs
     )
-    return result
 
 
 class _MethodProxy:
@@ -1371,11 +1369,11 @@ def pack_position(pos: Position) -> dict[str, Any]:
 async def handle_order_requests(
 
     ems_order_stream: tractor.MsgStream,
+    accounts_def: dict[str, str],
 
 ) -> None:
 
     global _accounts2clients
-    accounts_def = config.load_accounts('ib')
 
     # request_msg: dict
     async for request_msg in ems_order_stream:
@@ -1415,15 +1413,13 @@ async def handle_order_requests(
             order = BrokerdOrder(**request_msg)
 
             # call our client api to submit the order
-            reqid = await _trio_run_client_method(
+            reqid = client.submit_limit(
-
-                method='submit_limit',
                 oid=order.oid,
                 symbol=order.symbol,
                 price=order.price,
                 action=order.action,
                 size=order.size,
-                account=order.account,
+                account=acct_number,
 
                 # XXX: by default 0 tells ``ib_insync`` methods that
                 # there is no existing order so ask the client to create
@@ -1446,11 +1442,7 @@ async def handle_order_requests(
 
         elif action == 'cancel':
             msg = BrokerdCancel(**request_msg)
-
+            client.submit_cancel(reqid=msg.reqid)
-            await _trio_run_client_method(
-                method='submit_cancel',
-                reqid=msg.reqid
-            )
 
         else:
             log.error(f'Unknown order command: {request_msg}')
@@ -1467,15 +1459,24 @@ async def trades_dialogue(
     # XXX: required to propagate ``tractor`` loglevel to piker logging
     get_console_log(loglevel or tractor.current_actor().loglevel)
 
-    ib_trade_events_stream = await _trio_run_client_method(
+    accounts_def = config.load_accounts('ib')
-        method='recv_trade_updates',
+
-    )
     global _accounts2clients
     global _client_cache
 
     # deliver positions to subscriber before anything else
     all_positions = {}
 
+    clients: list[tuple[Client, trio.MemoryReceiveChannel]] = []
+    for account, client in _accounts2clients.items():
+
+        # each client to an api endpoint will have it's own event stream
+        trade_event_stream = await _trio_run_client_method(
+            method='recv_trade_updates',
+            client=client,
+        )
+        clients.append((client, trade_event_stream))
+
         for client in _client_cache.values():
             for pos in client.positions():
                 msg = pack_position(pos)
@@ -1483,19 +1484,42 @@ async def trades_dialogue(
 
     await ctx.started(all_positions)
 
-    action_map = {'BOT': 'buy', 'SLD': 'sell'}
-
     async with (
         ctx.open_stream() as ems_stream,
         trio.open_nursery() as n,
     ):
         # start order request handler **before** local trades event loop
-        n.start_soon(handle_order_requests, ems_stream)
+        n.start_soon(handle_order_requests, ems_stream, accounts_def)
+
+        # allocate event relay tasks for each client connection
+        for client, stream in clients:
+            n.start_soon(
+                deliver_trade_events,
+                stream,
+                ems_stream,
+                accounts_def
+            )
+
+        # block until cancelled
+        await trio.sleep_forever()
+
+
+async def deliver_trade_events(
+
+    trade_event_stream: trio.MemoryReceiveChannel,
+    ems_stream: tractor.MsgStream,
+    accounts_def: dict[str, str],
+
+) -> None:
+    '''Format and relay all trade events for a given client to the EMS.
+
+    '''
+    action_map = {'BOT': 'buy', 'SLD': 'sell'}
 
     # 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...
-        async for event_name, item in ib_trade_events_stream:
+    async for event_name, item in trade_event_stream:
 
         log.info(f'ib sending {event_name}:\n{pformat(item)}')
 
@@ -1537,7 +1561,7 @@ async def trades_dialogue(
 
                 reqid=trade.order.orderId,
                 time_ns=time.time_ns(),  # cuz why not
-                    # account=client.
+                account=accounts_def.inverse[trade.order.account],
 
                 # everyone doin camel case..
                 status=status.status.lower(),  # force lower case
@@ -1605,7 +1629,8 @@ async def trades_dialogue(
             if err['reqid'] == -1:
                 log.error(f'TWS external order error:\n{pformat(err)}')
 
-                # don't forward for now, it's unecessary.. but if we wanted to,
+            # TODO: what schema for this msg if we're going to make it
+            # portable across all backends?
             # msg = BrokerdError(**err)
             continue