Cancel any live orders found on connect

More or less just to avoid orders the user wasn't aware of from
persisting until we get "open order relaying" through the ems working.

Some further fixes which required a new `reqids2txids` map which keeps
track of which `kraken` "txid" is mapped to our `reqid: int`; mainly
this was needed for cancel requests which require knowing the underlying
`txid`s (since apparently kraken doesn't keep track of the "reqid"  we
pass it). Pass the ws instance into `handle_order_updates()` to enable
the cancelling orders on startup. Don't key error on unknown `reqid`
values (for eg. when receiving historical trade events on startup).
Handle cancel requests first in the ems side loop.

piker/brokers/kraken/README.rst

@@ -0,0 +1,64 @@
+``kraken`` backend
+------------------
+though they don't have the most liquidity of all the cexes they sure are
+accommodating to those of us who appreciate a little ``xmr``.
+
+status
+******
+current support is *production grade* and both real-time data and order
+management should be correct and fast. this backend is used by core devs
+for live trading.
+
+
+config
+******
+In order to get order mode support your ``brokers.toml``
+needs to have something like the following:
+
+.. code:: toml
+
+   [kraken]
+   accounts.spot = 'spot'
+   key_descr = "spot"
+   api_key = "69696969696969696696969696969696969696969696969696969696"
+   secret = "BOOBSBOOBSBOOBSBOOBSBOOBSSMBZ69696969696969669969696969696"
+
+
+If everything works correctly you should see any current positions
+loaded in the pps pane on chart load and you should also be able to
+check your trade records in the file::
+
+    <pikerk_conf_dir>/ledgers/trades_kraken_spot.toml
+
+
+An example ledger file will have entries written verbatim from the
+trade events schema:
+
+.. code:: toml
+
+    [TFJBKK-SMBZS-VJ4UWS]
+    ordertxid = "SMBZSA-7CNQU-3HWLNJ"
+    postxid = "SMBZSE-M7IF5-CFI7LT"
+    pair = "XXMRZEUR"
+    time = 1655691993.4133966
+    type = "buy"
+    ordertype = "limit"
+    price = "103.97000000"
+    cost = "499.99999977"
+    fee = "0.80000000"
+    vol = "4.80907954"
+    margin = "0.00000000"
+    misc = ""
+
+
+your ``pps.toml`` file will have position entries like,
+
+.. code:: toml
+
+   [kraken.spot."xmreur.kraken"]
+   size = 4.80907954
+   be_price = 103.97000000
+   bsuid = "XXMRZEUR"
+   clears = [
+    { tid = "TFJBKK-SMBZS-VJ4UWS", cost = 0.8, price = 103.97, size = 4.80907954, dt = "2022-05-20T02:26:33.413397+00:00" },
+   ]

piker/brokers/kraken/api.py

@@ -282,6 +282,7 @@ class Client:
                 "volume": str(size),
             }
             return await self.endpoint('AddOrder', data)
+
         else:
             # Edit order data for kraken api
             data["txid"] = reqid

piker/brokers/kraken/broker.py

@@ -18,21 +18,24 @@
 Order api and machinery
 
 '''
-from contextlib import asynccontextmanager as acm
+from contextlib import (
+    asynccontextmanager as acm,
+    contextmanager as cm,
+)
 from functools import partial
-from itertools import chain
+from itertools import chain, count
 from pprint import pformat
 import time
 from typing import (
     Any,
     AsyncIterator,
-    # Callable,
+    Union,
-    # Optional,
-    # Union,
 )
 
+from async_generator import aclosing
+from bidict import bidict
 import pendulum
-from pydantic import BaseModel
+# from pydantic import BaseModel
 import trio
 import tractor
 import wsproto
@@ -61,179 +64,137 @@ from .feed import (
     stream_messages,
 )
 
-
+MsgUnion = Union[
-class Trade(BaseModel):
+    BrokerdCancel,
-    '''
+    BrokerdError,
-    Trade class that helps parse and validate ownTrades stream
+    BrokerdFill,
-
+    BrokerdOrder,
-    '''
+    BrokerdOrderAck,
-    reqid: str  # kraken order transaction id
+    BrokerdPosition,
-    action: str  # buy or sell
+    BrokerdStatus,
-    price: float  # price of asset
+]
-    size: float  # vol of asset
-    broker_time: str  # e.g GTC, GTD
 
 
 async def handle_order_requests(
 
+    ws: NoBsWs,
     client: Client,
     ems_order_stream: tractor.MsgStream,
+    token: str,
+    emsflow: dict[str, list[MsgUnion]],
+    ids: bidict[str, int],
+    reqids2txids: dict[int, str],
 
 ) -> None:
+    '''
+    Process new order submission requests from the EMS
+    and deliver acks or errors.
 
-    request_msg: dict
+    '''
+    # XXX: UGH, let's unify this.. with ``msgspec``.
+    msg: dict[str, Any]
     order: BrokerdOrder
+    counter = count()
 
-    async for request_msg in ems_order_stream:
+    async for msg in ems_order_stream:
-        log.info(
+        log.info(f'Rx order msg:\n{pformat(msg)}')
-            'Received order request:\n'
+        match msg:
-            f'{pformat(request_msg)}'
+            case {
+                'action': 'cancel',
+            }:
+                cancel = BrokerdCancel(**msg)
+                last = emsflow[cancel.oid]
+                reqid = ids[cancel.oid]
+                txid = reqids2txids[reqid]
+
+                # call ws api to cancel:
+                # https://docs.kraken.com/websockets/#message-cancelOrder
+                await ws.send_msg({
+                    'event': 'cancelOrder',
+                    'token': token,
+                    'reqid': reqid,
+                    'txid': [txid],  # should be txid from submission
+                })
+
+            case {
+                'account': 'kraken.spot' as account,
+                'action': action,
+            } if action in {'buy', 'sell'}:
+
+                # validate
+                order = BrokerdOrder(**msg)
+
+                # logic from old `Client.submit_limit()`
+                if order.oid in ids:
+                    ep = 'editOrder'
+                    reqid = ids[order.oid]  # integer not txid
+                    txid = reqids2txids[reqid]
+                    extra = {
+                        'orderid': txid,  # txid
+                    }
+
+                else:
+                    ep = 'addOrder'
+                    reqid = next(counter)
+                    ids[order.oid] = reqid
+                    log.debug(
+                        f"Adding order {reqid}\n"
+                        f'{ids}'
                     )
+                    extra = {
+                        'ordertype': 'limit',
+                        'type': order.action,
+                    }
 
-        action = request_msg['action']
+                    psym = order.symbol.upper()
+                    pair = f'{psym[:3]}/{psym[3:]}'
 
-        if action in {'buy', 'sell'}:
+                # call ws api to submit the order:
+                # https://docs.kraken.com/websockets/#message-addOrder
+                req = {
+                    'event': ep,
+                    'token': token,
 
-            account = request_msg['account']
+                    'reqid': reqid,  # remapped-to-int uid from ems
+                    'pair': pair,
+                    'price': str(order.price),
+                    'volume': str(order.size),
+
+                    # only ensures request is valid, nothing more
+                    # validate: 'true',
+
+                } | extra
+                log.info(f'Submitting WS order request:\n{pformat(req)}')
+                await ws.send_msg(req)
+
+                resp = BrokerdOrderAck(
+                    oid=order.oid,  # ems order request id
+                    reqid=reqid,  # our custom int mapping
+                    account=account,  # piker account
+                )
+                await ems_order_stream.send(resp)
+
+                # placehold for sanity checking in relay loop
+                emsflow.setdefault(order.oid, []).append(order)
+
+            case _:
+                account = msg.get('account')
                 if account != 'kraken.spot':
                     log.error(
                         'This is a kraken account, \
                         only a `kraken.spot` selection is valid'
                     )
-                await ems_order_stream.send(BrokerdError(
-                    oid=request_msg['oid'],
-                    symbol=request_msg['symbol'],
 
-                    # reason=f'Kraken only, No account found: `{account}` ?',
+                await ems_order_stream.send(
+                    BrokerdError(
+                        oid=msg['oid'],
+                        symbol=msg['symbol'],
                         reason=(
-                        'Kraken only, order mode disabled due to '
+                            'Invalid request msg:\n{msg}'
-                        'https://github.com/pikers/piker/issues/299'
                         ),
 
-                ).dict())
-                continue
-
-            # validate
-            order = BrokerdOrder(**request_msg)
-            # call our client api to submit the order
-            resp = await client.submit_limit(
-                symbol=order.symbol,
-                price=order.price,
-                action=order.action,
-                size=order.size,
-                reqid=order.reqid,
-            )
-
-            err = resp['error']
-            if err:
-                oid = order.oid
-                log.error(f'Failed to submit order: {oid}')
-
-                await ems_order_stream.send(
-                    BrokerdError(
-                        oid=order.oid,
-                        reqid=order.reqid,
-                        symbol=order.symbol,
-                        reason="Failed order submission",
-                        broker_details=resp
                     ).dict()
                 )
-            else:
-                # TODO: handle multiple orders (cancels?)
-                #       txid is an array of strings
-                if order.reqid is None:
-                    reqid = resp['result']['txid'][0]
-                else:
-                    # update the internal pairing of oid to krakens
-                    # txid with the new txid that is returned on edit
-                    reqid = resp['result']['txid']
-
-                # deliver ack that order has been submitted to broker routing
-                await ems_order_stream.send(
-                    BrokerdOrderAck(
-
-                        # ems order request id
-                        oid=order.oid,
-
-                        # broker specific request id
-                        reqid=reqid,
-
-                        # account the made the order
-                        account=order.account
-
-                    ).dict()
-                )
-
-        elif action == 'cancel':
-            msg = BrokerdCancel(**request_msg)
-
-            # Send order cancellation to kraken
-            resp = await client.submit_cancel(
-                reqid=msg.reqid
-            )
-
-            # Check to make sure there was no error returned by
-            # the kraken endpoint. Assert one order was cancelled.
-            try:
-                result = resp['result']
-                count = result['count']
-
-            # check for 'error' key if we received no 'result'
-            except KeyError:
-                error = resp.get('error')
-
-                await ems_order_stream.send(
-                    BrokerdError(
-                        oid=msg.oid,
-                        reqid=msg.reqid,
-                        symbol=msg.symbol,
-                        reason="Failed order cancel",
-                        broker_details=resp
-                    ).dict()
-                )
-
-                if not error:
-                    raise BrokerError(f'Unknown order cancel response: {resp}')
-
-            else:
-                if not count:  # no orders were cancelled?
-
-                    # XXX: what exactly is this from and why would we care?
-                    # there doesn't seem to be any docs here?
-                    # https://docs.kraken.com/rest/#operation/cancelOrder
-
-                    # Check to make sure the cancellation is NOT pending,
-                    # then send the confirmation to the ems order stream
-                    pending = result.get('pending')
-                    if pending:
-                        log.error(f'Order {oid} cancel was not yet successful')
-
-                        await ems_order_stream.send(
-                            BrokerdError(
-                                oid=msg.oid,
-                                reqid=msg.reqid,
-                                symbol=msg.symbol,
-                                # TODO: maybe figure out if pending
-                                # cancels will eventually get cancelled
-                                reason="Order cancel is still pending?",
-                                broker_details=resp
-                            ).dict()
-                        )
-
-                else:  # order cancel success case.
-
-                    await ems_order_stream.send(
-                        BrokerdStatus(
-                            reqid=msg.reqid,
-                            account=msg.account,
-                            time_ns=time.time_ns(),
-                            status='cancelled',
-                            reason='Order cancelled',
-                            broker_details={'name': 'kraken'}
-                        ).dict()
-                    )
-    else:
-        log.error(f'Unknown order command: {request_msg}')
 
 
 @acm
@@ -310,7 +271,7 @@ async def trades_dialogue(
         log.info(
             f'Loaded {len(trades)} trades from account `{acc_name}`'
         )
-        trans = await update_ledger(acctid, trades)
+        with open_ledger(acctid, trades) as trans:
             active, closed = pp.update_pps_conf(
                 'kraken',
                 acctid,
@@ -357,27 +318,81 @@ async def trades_dialogue(
                 ),
             ) as ws,
             trio.open_nursery() as n,
+            aclosing(stream_messages(ws)) as stream,
         ):
+            # task local msg dialog tracking
+            emsflow: dict[
+                str,
+                list[MsgUnion],
+            ] = {}
+
+            # 2way map for ems ids to kraken int reqids..
+            ids: bidict[str, int] = bidict()
+            reqids2txids: dict[int, str] = {}
+
             # task for processing inbound requests from ems
-            n.start_soon(handle_order_requests, client, ems_stream)
+            n.start_soon(
+                handle_order_requests,
+                ws,
+                client,
+                ems_stream,
+                token,
+                emsflow,
+                ids,
+                reqids2txids,
+            )
 
-            count: int = 0
+            # enter relay loop
+            await handle_order_updates(
+                ws,
+                stream,
+                ems_stream,
+                emsflow,
+                ids,
+                reqids2txids,
+                trans,
+                acctid,
+                acc_name,
+                token,
+            )
 
-            # process and relay trades events to ems
+
-            # https://docs.kraken.com/websockets/#message-ownTrades
+async def handle_order_updates(
-            async for msg in stream_messages(ws):
+    ws: NoBsWs,
+    ws_stream: AsyncIterator,
+    ems_stream: tractor.MsgStream,
+    emsflow: dict[str, list[MsgUnion]],
+    ids: bidict[str, int],
+    reqids2txids: dict[int, str],
+    trans: list[pp.Transaction],
+    acctid: str,
+    acc_name: str,
+    token: str,
+
+) -> None:
+    '''
+    Main msg handling loop for all things order management.
+
+    This code is broken out to make the context explicit and state variables
+    defined in the signature clear to the reader.
+
+    '''
+    # transaction records which will be updated
+    # on new trade clearing events (aka order "fills")
+    trans: list[pp.Transaction]
+
+    async for msg in ws_stream:
         match msg:
+            # process and relay clearing trade events to ems
+            # https://docs.kraken.com/websockets/#message-ownTrades
             case [
                 trades_msgs,
                 'ownTrades',
+                # won't exist for historical values?
+                # 'userref': reqid,
                 {'sequence': seq},
             ]:
-                        # XXX: do we actually need this orrr?
+                # flatten msgs to an {id -> data} table for processing
-                        # ensure that we are only processing new trades?
-                        assert seq > count
-                        count += 1
-
-                        # flatten msgs for processing
                 trades = {
                     tid: trade
                     for entry in trades_msgs
@@ -388,8 +403,12 @@ async def trades_dialogue(
                 }
                 for tid, trade in trades.items():
 
-                            # parse-cast
+                    # NOTE: try to get the requid sent in the order
-                            reqid = trade['ordertxid']
+                    # request message if posssible; it may not be
+                    # provided since this sub also returns generic
+                    # historical trade events.
+                    reqid = trade.get('userref', trade['ordertxid'])
+
                     action = trade['type']
                     price = float(trade['price'])
                     size = float(trade['vol'])
@@ -398,6 +417,7 @@ async def trades_dialogue(
                     # send a fill msg for gui update
                     fill_msg = BrokerdFill(
                         reqid=reqid,
+
                         time_ns=time.time_ns(),
 
                         action=action,
@@ -435,7 +455,7 @@ async def trades_dialogue(
                     await ems_stream.send(filled_msg.dict())
 
                 # update ledger and position tracking
-                        trans = await update_ledger(acctid, trades)
+                with open_ledger(acctid, trades) as trans:
                     active, closed = pp.update_pps_conf(
                         'kraken',
                         acctid,
@@ -444,7 +464,7 @@ async def trades_dialogue(
                             t.bsuid for t in trans),
                     )
 
-                        # emit pp msgs
+                # emit any new pp msgs to ems
                 for pos in filter(
                     bool,
                     chain(active.values(), closed.values()),
@@ -471,21 +491,268 @@ async def trades_dialogue(
                     )
                     await ems_stream.send(pp_msg.dict())
 
+            # process and relay order state change events
+            # https://docs.kraken.com/websockets/#message-openOrders
             case [
-                        trades_msgs,
+                order_msgs,
                 'openOrders',
                 {'sequence': seq},
             ]:
-                        # TODO: async order update handling which we
+                for order_msg in order_msgs:
-                        # should remove from `handle_order_requests()`
+                    log.info(
-                        # above:
+                        f'Order msg update_{seq}:\n'
-                        # https://github.com/pikers/piker/issues/293
+                        f'{pformat(order_msg)}'
-                        # https://github.com/pikers/piker/issues/310
+                    )
-                        log.info(f'Order update {seq}:{trades_msgs}')
+                    txid, update_msg = list(order_msg.items())[0]
+                    match update_msg:
+
+                        # we ignore internal order updates triggered by
+                        # kraken's "edit" endpoint.
+                        case {
+                            'cancel_reason': 'Order replaced',
+                            'status': status,
+                            'userref': reqid,
+                            **rest,
+                        }:
+                            continue
+
+                        case {
+                            'status': status,
+                            'userref': reqid,
+                            **rest,
+
+                            # XXX: eg. of remaining msg schema:
+                            # 'avg_price': _,
+                            # 'cost': _,
+                            # 'descr': {
+                                # 'close': None,
+                                # 'leverage': None,
+                                # 'order': descr,
+                                # 'ordertype': 'limit',
+                                # 'pair': 'XMR/EUR',
+                                # 'price': '74.94000000',
+                                # 'price2': '0.00000000',
+                                # 'type': 'buy'
+                            # },
+                            # 'expiretm': None,
+                            # 'fee': '0.00000000',
+                            # 'limitprice': '0.00000000',
+                            # 'misc': '',
+                            # 'oflags': 'fciq',
+                            # 'opentm': '1656966131.337344',
+                            # 'refid': None,
+                            # 'starttm': None,
+                            # 'stopprice': '0.00000000',
+                            # 'timeinforce': 'GTC',
+                            # 'vol': submit_vlm,  # '13.34400854',
+                            # 'vol_exec': exec_vlm,  # 0.0000
+                        }:
+                            ems_status = {
+                                'open': 'submitted',
+                                'closed': 'cancelled',
+                                'canceled': 'cancelled',
+                                # do we even need to forward
+                                # this state to the ems?
+                                'pending': 'pending',
+                            }[status]
+
+                            submit_vlm = rest.get('vol', 0)
+                            exec_vlm = rest.get('vol_exec', 0)
+
+                            reqids2txids[reqid] = txid
+
+                            oid = ids.inverse.get(reqid)
+                            if not oid:
+                                # TODO: handle these and relay them
+                                # through the EMS to the client / UI
+                                # side!
+                                log.warning(
+                                    f'Received active order {txid}:\n'
+                                    f'{update_msg}\n'
+                                    'Cancelling order for now!..'
+                                )
+
+                                # call ws api to cancel:
+                                # https://docs.kraken.com/websockets/#message-cancelOrder
+                                await ws.send_msg({
+                                    'event': 'cancelOrder',
+                                    'token': token,
+                                    'reqid': reqid,
+                                    'txid': [txid],
+                                })
+                                continue
+
+                            msgs = emsflow[oid]
+
+                            # send BrokerdStatus messages for all
+                            # order state updates
+                            resp = BrokerdStatus(
+
+                                reqid=reqid,
+                                time_ns=time.time_ns(),  # cuz why not
+                                account=f'kraken.{acctid}',
+
+                                # everyone doin camel case..
+                                status=ems_status,  # force lower case
+
+                                filled=exec_vlm,
+                                reason='',  # why held?
+                                remaining=(
+                                    float(submit_vlm)
+                                    -
+                                    float(exec_vlm)
+                                ),
+
+                                broker_details=dict(
+                                    {'name': 'kraken'}, **update_msg
+                                ),
+                            )
+                            msgs.append(resp)
+                            await ems_stream.send(resp.dict())
 
                         case _:
-                        log.warning(f'Unhandled trades msg: {msg}')
+                            log.warning(
-                        await tractor.breakpoint()
+                                'Unknown orders msg:\n'
+                                f'{txid}:{order_msg}'
+                            )
+
+            case {
+                'event': etype,
+                'status': status,
+                'reqid': reqid,
+                **rest,
+            } as event if (
+                etype in {
+                    'addOrderStatus',
+                    'editOrderStatus',
+                    'cancelOrderStatus',
+                }
+            ):
+                oid = ids.inverse.get(reqid)
+                if not oid:
+                    log.warning(
+                        'Unknown order status update?:\n'
+                        f'{event}'
+                    )
+                    continue
+
+                txid = rest.get('txid')
+                if txid:
+                    reqids2txids[reqid] = txid
+
+                msgs = emsflow[oid]
+                last = msgs[-1]
+                resps, errored = process_status(
+                    event,
+                    oid,
+                    token,
+                    msgs,
+                    last,
+                )
+                if resps:
+                    msgs.extend(resps)
+                    for resp in resps:
+                        await ems_stream.send(resp)
+
+            case _:
+                log.warning(f'Unhandled trades update msg: {msg}')
+
+
+def process_status(
+    event: dict[str, str],
+    oid: str,
+    token: str,
+    msgs: list[MsgUnion],
+    last: MsgUnion,
+
+) -> tuple[list[MsgUnion], bool]:
+    '''
+    Process `'[add/edit/cancel]OrderStatus'` events by translating to
+    and returning the equivalent EMS-msg responses.
+
+    '''
+    match event:
+        case {
+            'event': etype,
+            'status': 'error',
+            'reqid': reqid,
+            'errorMessage': errmsg,
+        }:
+            # any of ``{'add', 'edit', 'cancel'}``
+            action = etype.rstrip('OrderStatus')
+            log.error(
+                f'Failed to {action} order {reqid}:\n'
+                f'{errmsg}'
+            )
+            resp = BrokerdError(
+                oid=oid,
+                # XXX: use old reqid in case it changed?
+                reqid=reqid,
+                symbol=getattr(last, 'symbol', 'N/A'),
+
+                reason=f'Failed {action}:\n{errmsg}',
+                broker_details=event
+            )
+            return [resp], True
+
+        # successful request cases
+        case {
+            'event': 'addOrderStatus',
+            'status': "ok",
+            'reqid': reqid,  # oid from ems side
+            'txid': txid,
+            'descr': descr,  # only on success?
+        }:
+            log.info(
+                f'Submitting order: {descr}\n'
+                f'ems oid: {oid}\n'
+                f're-mapped reqid: {reqid}\n'
+                f'txid: {txid}\n'
+            )
+            return [], False
+
+        case {
+            'event': 'editOrderStatus',
+            'status': "ok",
+            'reqid': reqid,  # oid from ems side
+            'descr': descr,
+
+            # NOTE: for edit request this is a new value
+            'txid': txid,
+            'originaltxid': origtxid,
+        }:
+            log.info(
+                f'Editting order {oid}[requid={reqid}]:\n'
+                f'txid: {origtxid} -> {txid}\n'
+                f'{descr}'
+            )
+            # deliver another ack to update the ems-side `.reqid`.
+            return [], False
+
+        case {
+            "event": "cancelOrderStatus",
+            "status": "ok",
+            'reqid': reqid,
+
+            # XXX: sometimes this isn't provided!?
+            # 'txid': txids,
+            **rest,
+        }:
+            # TODO: should we support "batch" acking of
+            # multiple cancels thus avoiding the below loop?
+            resps: list[MsgUnion] = []
+            for txid in rest.get('txid', [last.reqid]):
+                resp = BrokerdStatus(
+                    reqid=reqid,
+                    account=last.account,
+                    time_ns=time.time_ns(),
+                    status='cancelled',
+                    reason='Cancel success: {oid}@{txid}',
+                    broker_details=event,
+                )
+                resps.append(resp)
+
+            return resps, False
 
 
 def norm_trade_records(
@@ -494,10 +761,9 @@ def norm_trade_records(
 ) -> list[pp.Transaction]:
 
     records: list[pp.Transaction] = []
-
     for tid, record in ledger.items():
 
-        size = record.get('vol') * {
+        size = float(record.get('vol')) * {
             'buy': 1,
             'sell': -1,
         }[record['type']]
@@ -508,7 +774,7 @@ def norm_trade_records(
             pp.Transaction(
                 fqsn=f'{norm_sym}.kraken',
                 tid=tid,
-                size=float(size),
+                size=size,
                 price=float(record['price']),
                 cost=float(record['fee']),
                 dt=pendulum.from_timestamp(float(record['time'])),
@@ -522,19 +788,24 @@ def norm_trade_records(
     return records
 
 
-async def update_ledger(
+@cm
+def open_ledger(
     acctid: str,
     trade_entries: list[dict[str, Any]],
 
 ) -> list[pp.Transaction]:
+    '''
+    Write recent session's trades to the user's (local) ledger file.
 
-    # write recent session's trades to the user's (local) ledger file.
+    '''
     with pp.open_trade_ledger(
         'kraken',
         acctid,
     ) as ledger:
-        ledger.update(trade_entries)
 
         # normalize to transaction form
         records = norm_trade_records(trade_entries)
-    return records
+        yield records
+
+        # update on exit
+        ledger.update(trade_entries)

piker/brokers/kraken/feed.py

@@ -117,9 +117,8 @@ async def stream_messages(
                 too_slow_count = 0
                 continue
 
-        if isinstance(msg, dict):
+        match msg:
-            if msg.get('event') == 'heartbeat':
+            case {'event': 'heartbeat'}:
-
                 now = time.time()
                 delay = now - last_hb
                 last_hb = now
@@ -130,10 +129,19 @@ async def stream_messages(
 
                 continue
 
-            err = msg.get('errorMessage')
+            case {
-            if err:
+                'connectionID': _,
-                raise BrokerError(err)
+                'event': 'systemStatus',
-        else:
+                'status': 'online',
+                'version': _,
+            } as msg:
+                log.info(
+                    'WS connection is up:\n'
+                    f'{msg}'
+                )
+                continue
+
+            case _:
                 yield msg
 
 
@@ -145,16 +153,39 @@ async def process_data_feed_msgs(
 
     '''
     async for msg in stream_messages(ws):
+        match msg:
+            case {
+                'errorMessage': errmsg
+            }:
+                raise BrokerError(errmsg)
 
-        chan_id, *payload_array, chan_name, pair = msg
+            case {
+                'event': 'subscriptionStatus',
+            } as sub:
+                log.info(
+                    'WS subscription is active:\n'
+                    f'{sub}'
+                )
+                continue
 
+            case [
+                chan_id,
+                *payload_array,
+                chan_name,
+                pair
+            ]:
                 if 'ohlc' in chan_name:
-
+                    yield 'ohlc', OHLC(
-            yield 'ohlc', OHLC(chan_id, chan_name, pair, *payload_array[0])
+                        chan_id,
+                        chan_name,
+                        pair,
+                        *payload_array[0]
+                    )
 
                 elif 'spread' in chan_name:
 
-            bid, ask, ts, bsize, asize = map(float, payload_array[0])
+                    bid, ask, ts, bsize, asize = map(
+                        float, payload_array[0])
 
                     # TODO: really makes you think IB has a horrible API...
                     quote = {
@@ -173,9 +204,9 @@ async def process_data_feed_msgs(
                 #     chan_id, *payload_array, chan_name, pair = msg
                 #     print(msg)
 
-        else:
+            case _:
                 print(f'UNHANDLED MSG: {msg}')
-            yield msg
+                # yield msg
 
 
 def normalize(
@@ -385,7 +416,7 @@ async def stream_quotes(
             msg_gen = process_data_feed_msgs(ws)
 
             # TODO: use ``anext()`` when it lands in 3.10!
-            typ, ohlc_last = await msg_gen.__anext__()
+            typ, ohlc_last = await anext(msg_gen)
 
             topic, quote = normalize(ohlc_last)
 

piker/clearing/_ems.py

@@ -88,7 +88,8 @@ def mk_check(
 
 @dataclass
 class _DarkBook:
-    '''EMS-trigger execution book.
+    '''
+    EMS-trigger execution book.
 
     Contains conditions for executions (aka "orders" or "triggers")
     which are not exposed to brokers and thus the market; i.e. these are
@@ -652,6 +653,13 @@ async def translate_and_relay_brokerd_events(
         else:
             # check for existing live flow entry
             entry = book._ems_entries.get(oid)
+            old_reqid = entry.reqid
+
+            if old_reqid and old_reqid != reqid:
+                log.warning(
+                    f'Brokerd order id change for {oid}:\n'
+                    f'{old_reqid} -> {reqid}'
+                )
 
             # initial response to brokerd order request
             if name == 'ack':
@@ -662,6 +670,10 @@ async def translate_and_relay_brokerd_events(
                 # a ``BrokerdOrderAck`` **must** be sent after an order
                 # request in order to establish this id mapping.
                 book._ems2brokerd_ids[oid] = reqid
+                log.info(
+                    'Rx ACK for order\n'
+                    f'oid: {oid} -> reqid: {reqid}'
+                )
 
                 # new order which has not yet be registered into the
                 # local ems book, insert it now and handle 2 cases: