pikers
/
piker
3
0
Fork 0
Code Issues 5 Pull Requests 7 Packages Projects Releases Wiki Activity

WIP: use `userref` field over `reqid`...

kraken_userref_hackzin
Tyler Goodlet 2022-07-29 19:37:02 -04:00
parent 808dbb12e6
commit dc8072c6db
1 changed files with 44 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
piker/brokers/kraken/broker.py
View File

@ -31,6 +31,7 @@ import time
from typing import ( from typing import (
Any, Any,
AsyncIterator, AsyncIterator,
Optional,
Union, Union,
) )
@ -188,10 +189,10 @@ async def handle_order_requests(
'event': ep, 'event': ep,
'token': token, 'token': token,
# XXX: this seems to always get an error response? # XXX: Lol, you can only send one of these..
# 'userref': f"'{reqid}'", 'userref': str(reqid),
# 'reqid': reqid, # remapped-to-int uid from ems
'reqid': reqid, # remapped-to-int uid from ems
'pair': pair, 'pair': pair,
'price': str(order.price), 'price': str(order.price),
'volume': str(order.size), 'volume': str(order.size),
@ -601,7 +602,8 @@ async def handle_order_updates(
# format as tid -> trade event map # format as tid -> trade event map
# eg. received msg format, # eg. received msg format,
# [{'TOKWHY-SMTUB-G5DOI6': {'cost': '95.29047', # [{'TOKWHY-SMTUB-G5DOI6': {
# 'cost': '95.29047',
# 'fee': '0.24776', # 'fee': '0.24776',
# 'margin': '0.00000', # 'margin': '0.00000',
# 'ordertxid': 'OKSUXM-3OLSB-L7TN72', # 'ordertxid': 'OKSUXM-3OLSB-L7TN72',
@ -611,7 +613,8 @@ async def handle_order_updates(
# 'price': '21268.20000', # 'price': '21268.20000',
# 'time': '1657990947.640891', # 'time': '1657990947.640891',
# 'type': 'buy', # 'type': 'buy',
# 'vol': '0.00448042'}}] # 'vol': '0.00448042'
# }}]
trades = { trades = {
tid: trade tid: trade
for entry in trades_msgs for entry in trades_msgs
@ -622,7 +625,9 @@ async def handle_order_updates(
} }
for tid, trade in trades.items(): for tid, trade in trades.items():
txid = trade['ordertxid'] txid = trade['ordertxid']
reqid = trade.get('userref')
if not reqid:
# NOTE: yet again, here we don't have any ref to the # NOTE: yet again, here we don't have any ref to the
# reqid that's generated by us (as the client) and # reqid that's generated by us (as the client) and
# sent in the order request, so we have to look it # sent in the order request, so we have to look it
@ -705,11 +710,11 @@ async def handle_order_updates(
case { case {
'cancel_reason': 'Order replaced', 'cancel_reason': 'Order replaced',
'status': status, 'status': status,
# 'userref': reqid, # XXX: always zero bug XD 'userref': reqid, # XXX: always zero bug XD
# **rest, # **rest,
}: }:
log.info( log.info(
f'Order {txid} was replaced' f'Order {txid}@reqid={reqid} was replaced'
) )
continue continue
@ -768,12 +773,17 @@ async def handle_order_updates(
ourreqid = reqids2txids.inverse.get(txid) ourreqid = reqids2txids.inverse.get(txid)
# XXX: abs necessary in order to enable
# mapping status response messages to the
# reqid-dialog..
reqids2txids[reqid] = txid
if ourreqid != reqid: if ourreqid != reqid:
log.warning( log.warning(
'REQID MISMATCH due to kraken api bugs..\n' 'REQID MISMATCH due to design mess..\n'
f'msg:{reqid}, ours:{ourreqid}' f'msg:{reqid}, ours:{ourreqid}'
) )
reqid = ourreqid # reqid = ourreqid
oid = ids.inverse.get(reqid) oid = ids.inverse.get(reqid)
@ -901,7 +911,7 @@ async def handle_order_updates(
case { case {
'event': etype, 'event': etype,
'status': status, 'status': status,
'reqid': reqid, # 'reqid': reqid,
**rest, **rest,
} as event if ( } as event if (
etype in { etype in {
@ -914,9 +924,14 @@ async def handle_order_updates(
f'{etype}:\n' f'{etype}:\n'
f'{pformat(msg)}' f'{pformat(msg)}'
) )
oid = ids.inverse.get(reqid)
txid = rest.get('txid')
reqid = reqids2txids.inverse.get(txid)
lasttxid = reqids2txids.get(reqid)
# TODO: relay these to EMS once it supports # TODO: relay these to EMS once it supports
# open order loading. # open order loading.
oid = ids.inverse.get(reqid)
if not oid: if not oid:
log.warning( log.warning(
'Unknown order status update?:\n' 'Unknown order status update?:\n'
@ -924,9 +939,7 @@ async def handle_order_updates(
) )
continue continue
lasttxid = reqids2txids.get(reqid) if reqid is not None:
txid = rest.get('txid')
# update the msg chain # update the msg chain
chain = apiflows[reqid] chain = apiflows[reqid]
chain.maps.append(event) chain.maps.append(event)
@ -936,6 +949,7 @@ async def handle_order_updates(
oid, oid,
token, token,
chain, chain,
reqid=reqid,
) )
if resps: if resps:
for resp in resps: for resp in resps:
@ -955,12 +969,13 @@ async def handle_order_updates(
'reqid': reqid or 0, 'reqid': reqid or 0,
'txid': [txid], 'txid': [txid],
}) })
else:
# else:
# XXX: we **must** do this mapping for edit order # XXX: we **must** do this mapping for edit order
# status updates since the `openOrders` sub above # status updates since the `openOrders` sub above
# never relays back the correct client-side `reqid` # never relays back the correct client-side `reqid`
# that is put in the order request.. # that is put in the order request..
reqids2txids[reqid] = txid # reqids2txids[reqid] = txid
case _: case _:
log.warning(f'Unhandled trades update msg: {msg}') log.warning(f'Unhandled trades update msg: {msg}')
@ -971,6 +986,7 @@ def process_status(
oid: str, oid: str,
token: str, token: str,
chain: ChainMap, chain: ChainMap,
reqid: Optional[int] = None,
) -> tuple[list[MsgUnion], bool]: ) -> tuple[list[MsgUnion], bool]:
''' '''
@ -982,7 +998,7 @@ def process_status(
case { case {
'event': etype, 'event': etype,
'status': 'error', 'status': 'error',
'reqid': reqid, # 'reqid': reqid,
'errorMessage': errmsg, 'errorMessage': errmsg,
}: }:
# any of ``{'add', 'edit', 'cancel'}`` # any of ``{'add', 'edit', 'cancel'}``
@ -1006,7 +1022,7 @@ def process_status(
case { case {
'event': 'addOrderStatus', 'event': 'addOrderStatus',
'status': "ok", 'status': "ok",
'reqid': reqid, # oid from ems side # 'reqid': reqid, # oid from ems side
'txid': txid, 'txid': txid,
'descr': descr, # only on success? 'descr': descr, # only on success?
}: }:
@ -1021,7 +1037,7 @@ def process_status(
case { case {
'event': 'editOrderStatus', 'event': 'editOrderStatus',
'status': "ok", 'status': "ok",
'reqid': reqid, # oid from ems side # 'reqid': reqid, # oid from ems side
'descr': descr, 'descr': descr,
# NOTE: for edit request this is a new value # NOTE: for edit request this is a new value
@ -1040,7 +1056,7 @@ def process_status(
case { case {
"event": "cancelOrderStatus", "event": "cancelOrderStatus",
"status": "ok", "status": "ok",
'reqid': reqid, # 'reqid': reqid,
# XXX: sometimes this isn't provided!? # XXX: sometimes this isn't provided!?
# 'txid': txids, # 'txid': txids,