Merge pull request #298 from pikers/kraken_cleaning

Kraken cleaning, disable order support due to #299!

piker/brokers/kraken.py

@@ -33,7 +33,7 @@ import tractor
 from pydantic.dataclasses import dataclass
 from pydantic import BaseModel
 import wsproto
-from itertools import count
+import itertools
 import urllib.parse
 import hashlib
 import hmac
@@ -159,9 +159,15 @@ class OHLC:
 def get_config() -> dict[str, Any]:
 
     conf, path = config.load()
-
     section = conf.get('kraken')
 
+    if section:
+        log.warning(
+            'Kraken order mode is currently disabled due to bug!\n'
+            'See https://github.com/pikers/piker/issues/299'
+        )
+        return {}
+
     if section is None:
         log.warning(f'No config section found for kraken in {path}')
         return {}
@@ -446,7 +452,6 @@ class Client:
             raise SymbolNotFound(json['error'][0] + f': {symbol}')
 
 
-
 @asynccontextmanager
 async def get_client() -> Client:
 
@@ -541,7 +546,8 @@ async def handle_order_requests(
 
     request_msg: dict
     order: BrokerdOrder
-    userref_counter = count()
+    userref_counter = itertools.count()
+
     async for request_msg in ems_order_stream:
         log.info(f'Received order request {request_msg}')
 
@@ -558,7 +564,13 @@ async def handle_order_requests(
                 await ems_order_stream.send(BrokerdError(
                     oid=request_msg['oid'],
                     symbol=request_msg['symbol'],
-                    reason=f'Kraken only, No account found: `{account}` ?',
+
+                    # reason=f'Kraken only, No account found: `{account}` ?',
+                    reason=(
+                        'Kraken only, order mode disabled due to '
+                        'https://github.com/pikers/piker/issues/299'
+                    ),
+
                 ).dict())
                 continue
 
@@ -578,7 +590,9 @@ async def handle_order_requests(
 
             err = resp['error']
             if err:
-                log.error(f'Failed to submit order')
+                oid = order.oid
+                log.error(f'Failed to submit order: {oid}')
+
                 await ems_order_stream.send(
                     BrokerdError(
                         oid=order.oid,
@@ -616,30 +630,16 @@ async def handle_order_requests(
                 reqid=msg.reqid
             )
 
-            try:
             # Check to make sure there was no error returned by
-                # the kraken endpoint. Assert one order was cancelled
+            # the kraken endpoint. Assert one order was cancelled.
-                assert resp['error'] == []
-                assert resp['result']['count'] == 1
-
-                # TODO: Change this code using .get
             try:
-                    pending = resp['result']['pending']
+                result = resp['result']
-                # Check to make sure the cancellation is NOT pending,
+                count = result['count']
-                # then send the confirmation to the ems order stream
+
+            # check for 'error' key if we received no 'result'
             except KeyError:
-                    await ems_order_stream.send(
+                error = resp.get('error')
-                        BrokerdStatus(
+
-                            reqid=msg.reqid,
-                            account=msg.account,
-                            time_ns=time.time_ns(),
-                            status='cancelled',
-                            reason='Order cancelled',
-                            broker_details={'name': 'kraken'}
-                        ).dict()
-                    )
-            except AssertionError:
-                log.error(f'Order cancel was not successful')
                 await ems_order_stream.send(
                     BrokerdError(
                         oid=order.oid,
@@ -650,6 +650,44 @@ async def handle_order_requests(
                     ).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=order.oid,
+                                reqid=temp_id,
+                                symbol=order.symbol,
+                                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}')
 
@@ -694,7 +732,7 @@ async def trades_dialogue(
     async with get_client() as client:
         if not client._api_key:
             log.error('Missing Kraken API key: Trades WS connection failed')
-            await ctx.started(({}, {'paper',}))
+            await ctx.started(({}, ['paper']))
 
             async with (
                 ctx.open_stream() as ems_stream,
@@ -713,7 +751,7 @@ async def trades_dialogue(
                     _positions={},
                 )
 
-                ## TODO: maybe add multiple accounts
+                # TODO: maybe add multiple accounts
                 n.start_soon(handle_order_requests, client, ems_stream)
 
         acc_name = 'kraken.' + client._name
@@ -724,7 +762,7 @@ async def trades_dialogue(
         await ctx.started((position_msgs, (acc_name,)))
 
         # Get websocket token for authenticated data stream
-        # Assert that a token was actually received 
+        # Assert that a token was actually received.
         resp = await client.endpoint('GetWebSocketsToken', {})
         assert resp['error'] == []
         token = resp['result']['token']
@@ -733,7 +771,7 @@ async def trades_dialogue(
             ctx.open_stream() as ems_stream,
             trio.open_nursery() as n,
         ):
-            ## TODO: maybe add multiple accounts
+            # TODO: maybe add multiple accounts
             n.start_soon(handle_order_requests, client, ems_stream)
 
             # Process trades msg stream of ws
@@ -746,24 +784,8 @@ async def trades_dialogue(
                     for trade in msg:
                         # check the type of packaged message
                         assert type(trade) == Trade
-                        # prepare and send a status update for line update
-                        trade_msg = BrokerdStatus(
-                            reqid=trade.reqid,
-                            time_ns=time.time_ns(),
-
-                            account='kraken.spot',
-                            status='executed',
-                            filled=float(trade.size),
-                            reason='Order filled by kraken',
-                            # remaining='' # TODO: not sure what to do here.
-                            broker_details={
-                                'name': 'kraken',
-                                'broker_time': trade.broker_time
-                            }
-                        )
-                        
-                        await ems_stream.send(trade_msg.dict())
 
+                        # prepare and send a filled status update
                         filled_msg = BrokerdStatus(
                             reqid=trade.reqid,
                             time_ns=time.time_ns(),
@@ -772,11 +794,19 @@ async def trades_dialogue(
                             status='filled',
                             filled=float(trade.size),
                             reason='Order filled by kraken',
-                            # remaining='' # TODO: not sure what to do here.
                             broker_details={
                                 'name': 'kraken',
                                 'broker_time': trade.broker_time
-                            }
+                            },
+
+                            # TODO: figure out if kraken gives a count
+                            # of how many units of underlying were
+                            # filled. Alternatively we can decrement
+                            # this value ourselves by associating and
+                            # calcing from the diff with the original
+                            # client-side request, see:
+                            # https://github.com/pikers/piker/issues/296
+                            remaining=0,
                         )
 
                         await ems_stream.send(filled_msg.dict())
@@ -899,9 +929,9 @@ async def process_trade_msgs(
     async for msg in stream_messages(ws):
 
         try:
-            # check that we are on the ownTrades stream and that msgs are 
+            # check that we are on the ownTrades stream and that msgs
-            # arriving in sequence with kraken
+            # are arriving in sequence with kraken For clarification the
-            # For clarification the kraken ws api docs for this stream: 
+            # kraken ws api docs for this stream:
             # https://docs.kraken.com/websockets/#message-ownTrades
             assert msg[1] == 'ownTrades'
             assert msg[2]['sequence'] > sequence_counter

piker/clearing/_messages.py

@@ -184,7 +184,7 @@ class BrokerdStatus(BaseModel):
     # {
     #   'submitted',
     #   'cancelled',
-    #   'executed',
+    #   'filled',
     # }
     status: str