piker/brokers/kraken.py

@@ -33,6 +33,7 @@ import tractor
 from pydantic.dataclasses import dataclass
 from pydantic import BaseModel
 import wsproto
+import itertools
 import urllib.parse
 import hashlib
 import hmac
@@ -160,6 +161,13 @@ 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 {}
@@ -300,38 +308,34 @@ class Client:
 
     async def submit_limit(
         self,
+        oid: str,
         symbol: str,
         price: float,
         action: str,
         size: float,
-        reqid: str = None,
+        reqid: int = None,
-        validate: bool = False # set True test call without a real submission
+    ) -> int:
-    ) -> dict:
         '''
         Place an order and return integer request id provided by client.
 
         '''
-        # Build common data dict for common keys from both endpoints
+        # Build order data for kraken api
         data = {
+            "userref": reqid,
+            "ordertype": "limit",
+            "type": action,
+            "volume": str(size),
             "pair": symbol,
             "price": str(price),
-            "validate": validate
+            # set to True test AddOrder call without a real submission
+            "validate": False
         }
-        if reqid is None:
+        return await self.endpoint('AddOrder', data)
-            # Build order data for kraken api
-            data["ordertype"] = "limit"
-            data["type"] = action
-            data["volume"] = str(size)
-            return await self.endpoint('AddOrder', data)
-        else:
-            # Edit order data for kraken api
-            data["txid"] = reqid
-            return await self.endpoint('EditOrder', data)
 
     async def submit_cancel(
         self,
         reqid: str,
-    ) -> dict:
+    ) -> None:
         '''
         Send cancel request for order id ``reqid``.
 
@@ -542,6 +546,7 @@ async def handle_order_requests(
 
     request_msg: dict
     order: BrokerdOrder
+    userref_counter = itertools.count()
 
     async for request_msg in ems_order_stream:
         log.info(f'Received order request {request_msg}')
@@ -570,14 +575,17 @@ async def handle_order_requests(
                 continue
 
             # validate
+            temp_id = next(userref_counter)
             order = BrokerdOrder(**request_msg)
+
             # call our client api to submit the order
             resp = await client.submit_limit(
+                oid=order.oid,
                 symbol=order.symbol,
                 price=order.price,
                 action=order.action,
                 size=order.size,
-                reqid=order.reqid,
+                reqid=temp_id,
             )
 
             err = resp['error']
@@ -588,21 +596,16 @@ async def handle_order_requests(
                 await ems_order_stream.send(
                     BrokerdError(
                         oid=order.oid,
-                        reqid=order.reqid,
+                        reqid=temp_id,
                         symbol=order.symbol,
                         reason="Failed order submission",
                         broker_details=resp
                     ).dict()
                 )
             else:
-                # TODO: handle multiple orders (cancels?)
+                # TODO: handle multiple cancels
                 #       txid is an array of strings
-                if order.reqid is None:
+                reqid = resp['result']['txid'][0]
-                    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(
@@ -639,9 +642,9 @@ async def handle_order_requests(
 
                 await ems_order_stream.send(
                     BrokerdError(
-                        oid=msg.oid,
+                        oid=order.oid,
-                        reqid=msg.reqid,
+                        reqid=temp_id,
-                        symbol=msg.symbol,
+                        symbol=order.symbol,
                         reason="Failed order cancel",
                         broker_details=resp
                     ).dict()
@@ -665,11 +668,9 @@ async def handle_order_requests(
 
                         await ems_order_stream.send(
                             BrokerdError(
-                                oid=msg.oid,
+                                oid=order.oid,
-                                reqid=msg.reqid,
+                                reqid=temp_id,
-                                symbol=msg.symbol,
+                                symbol=order.symbol,
-                                # TODO: maybe figure out if pending cancels will
-                                # eventually get cancelled
                                 reason="Order cancel is still pending?",
                                 broker_details=resp
                             ).dict()