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

Split existing live-open case into its own block

open_order_loading
Tyler Goodlet 2022-08-11 22:48:01 -04:00
parent e2cd8c4aef
commit f6ba95a6c7
1 changed files with 96 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -741,6 +741,57 @@ async def handle_order_updates(
txid, update_msg = list(order_msg.items())[0] txid, update_msg = list(order_msg.items())[0]
match update_msg: match update_msg:
# EMS-unknown live order that needs to be
# delivered and loaded on the client-side.
case {
'userref': reqid,
# during a fill this field is **not**
# provided! but, it is always avail on
# actual status updates.. see case above.
'status': status,
**rest,
} if (
ids.inverse.get(reqid) is None
):
# parse out existing live order
descr = rest['descr']
fqsn = descr['pair'].replace(
'/', '').lower()
price = float(descr['price'])
size = float(rest['vol'])
action = descr['type']
# register the userref value from
# kraken (usually an `int` staring
# at 1?) as our reqid.
reqids2txids[reqid] = txid
oid = str(reqid)
ids[oid] = reqid # NOTE!: str -> int
# fill out ``Status`` + boxed ``Order``
status_msg = Status(
time_ns=time.time_ns(),
resp='open',
oid=oid,
reqid=reqid,
# embedded order info
req=Order(
action=action,
exec_mode='live',
oid=oid,
symbol=fqsn,
account=acc_name,
price=price,
size=size,
),
src='kraken',
)
apiflows[reqid].maps.append(status_msg)
await ems_stream.send(status_msg)
continue
# XXX: eg. of full msg schema: # XXX: eg. of full msg schema:
# {'avg_price': _, # {'avg_price': _,
# 'cost': _, # 'cost': _,
@ -819,10 +870,6 @@ async def handle_order_updates(
) )
oid = ids.inverse.get(reqid) oid = ids.inverse.get(reqid)
if (
status == 'open'
and (
# XXX: too fast edit handled by the # XXX: too fast edit handled by the
# request handler task: this # request handler task: this
# scenario occurs when ems side # scenario occurs when ems side
@ -848,17 +895,13 @@ async def handle_order_updates(
# coming in too quickly on the other # coming in too quickly on the other
# side of the ems, aka the client # side of the ems, aka the client
# <-> ems dialog. # <-> ems dialog.
(toofast := isinstance( if (
status == 'open'
and isinstance(
reqids2txids.get(reqid), reqids2txids.get(reqid),
TooFastEdit TooFastEdit
))
# pre-existing open order NOT from
# this EMS session.
or (noid := oid is None)
) )
): ):
if toofast:
# TODO: don't even allow this case # TODO: don't even allow this case
# by not moving the client side line # by not moving the client side line
# until an edit confirmation # until an edit confirmation
@ -878,47 +921,6 @@ async def handle_order_updates(
}) })
continue continue
# a non-ems-active order, emit live
# order embedded in status msg.
elif noid:
# parse out existing live order
descr = rest['descr']
fqsn = descr['pair'].replace(
'/', '').lower()
price = float(descr['price'])
size = float(rest['vol'])
action = descr['type']
# register the userref value from
# kraken (usually an `int` staring
# at 1?) as our reqid.
reqids2txids[reqid] = txid
oid = str(reqid)
ids[oid] = reqid # NOTE!: str -> int
# fill out ``Status`` + boxed ``Order``
status_msg = Status(
time_ns=time.time_ns(),
resp='open',
oid=oid,
reqid=reqid,
# embedded order info
req=Order(
action=action,
exec_mode='live',
oid=oid,
symbol=fqsn,
account=acc_name,
price=price,
size=size,
),
src='kraken',
)
apiflows[reqid].maps.append(status_msg)
await ems_stream.send(status_msg)
continue
# send BrokerdStatus messages for all # send BrokerdStatus messages for all
# order state updates # order state updates
resp = BrokerdStatus( resp = BrokerdStatus(