pikers
/
piker
3
0
Fork 0
Code Issues 4 Pull Requests 6 Packages Projects Releases Wiki Activity

get basic order request loop receiving msgs

kraken_gb
Konstantine Tsafatinos 2022-02-07 19:18:00 -05:00 committed by Tyler Goodlet
parent a3a6718ba5
commit 7627fb5141
1 changed files with 53 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

102
piker/brokers/kraken.py
View File

@ -438,7 +438,7 @@ def normalize_symbol(
async def handle_order_requests(
client: #kraken,
client: 'test',#kraken,
ems_order_stream: tractor.MsgStream,
) -> None:
@ -447,61 +447,61 @@ async def handle_order_requests(
async for request_msg in ems_order_stream:
log.info(f'Received order request {request_msg}')
action = request_msg['action']
# action = request_msg['action']
if action in {'buy', 'sell'}:
# if action in {'buy', 'sell'}:
account = request_msg['account']
if account != 'kraken.spot':
log.error(
'This is a kraken account, \
only a `kraken.spot` selection is valid'
)
await ems_order_stream.send(BrokerError(
oid=request_msg['oid']
symbol=request_msg['symbol']
reason=f'Kraken only, No account found: `{account}` ?',
).dict())
continue
# account = request_msg['account']
# if account != 'kraken.spot':
# log.error(
# 'This is a kraken account, \
# only a `kraken.spot` selection is valid'
# )
# await ems_order_stream.send(BrokerError(
# oid=request_msg['oid'],
# symbol=request_msg['symbol'],
# reason=f'Kraken only, No account found: `{account}` ?',
# ).dict())
# continue
# validate
order = BrokerdOrder(**request_msg)
# # validate
# order = BrokerdOrder(**request_msg)
# call our client api to submit the order
## TODO: look into the submit_limit method, do it write my own?
reqid = await client.submit_limit(
# # call our client api to submit the order
# ## TODO: look into the submit_limit method, do it write my own?
# reqid = await client.submit_limit(
oid=order.oid,
symbol=order.symbol,
price=order.price,
action=order.action,
size=order.size,
## XXX: how do I handle new orders
reqid=order.reqid,
)
# oid=order.oid,
# symbol=order.symbol,
# price=order.price,
# action=order.action,
# size=order.size,
# ## XXX: how do I handle new orders
# reqid=order.reqid,
# )
# deliver ack that order has been submitted to broker routing
await ems_order_stream.send(
BrokerdOrderAck(
# # deliver ack that order has been submitted to broker routing
# await ems_order_stream.send(
# BrokerdOrderAck(
# ems order request id
oid=order.oid,
# # ems order request id
# oid=order.oid,
# broker specific request id
reqid=reqid,
# # broker specific request id
# reqid=reqid,
).dict()
)
# ).dict()
# )
elif action == 'cancel':
msg = BrokerdCancel(**request_msg)
# elif action == 'cancel':
# msg = BrokerdCancel(**request_msg)
await client.submit_cancel(
reqid=msg.reqid
)
# await client.submit_cancel(
# reqid=msg.reqid
# )
else:
log.error(f'Unknown order command: {request_msg}')
# else:
# log.error(f'Unknown order command: {request_msg}')
@tractor.context
@ -527,16 +527,18 @@ async def trades_dialogue(
all_positions.append(msg.dict())
open_orders = await client.get_user_data('OpenOrders', {})
await tractor.breakpoint()
#await tractor.breakpoint()
await ctx.started((all_positions, (acc_name,)))
await trio.sleep_forever()
#await trio.sleep_forever()
# async with (
# ctx.open_stream() as ems_stream,
# trio.open_nursery as n,
# ):
async with (
ctx.open_stream() as ems_stream,
trio.open_nursery() as n,
):
## TODO: maybe add multiple accounts
n.start_soon(handle_order_requests, client, ems_stream)
async def stream_messages(ws):