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

fix hang when kraken is not in config

kraken_orders
Konstantine Tsafatinos 2022-03-28 18:28:19 -04:00
parent cb8e97a142
commit 2baa1b4605
1 changed files with 74 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

129
piker/brokers/kraken.py
View File

@ -450,11 +450,14 @@ class Client:
async def get_client() -> Client: async def get_client() -> Client:
section = get_config() section = get_config()
client = Client( if section:
name=section['key_descr'], client = Client(
api_key=section['api_key'], name=section['key_descr'],
secret=section['secret'] api_key=section['api_key'],
) secret=section['secret']
)
else:
client = Client()
# at startup, load all symbols locally for fast search # at startup, load all symbols locally for fast search
await client.cache_symbols() await client.cache_symbols()
@ -688,68 +691,80 @@ async def trades_dialogue(
# Authenticated block # Authenticated block
async with get_client() as client: async with get_client() as client:
acc_name = 'kraken.' + client._name if client._api_key:
trades = await client.get_trades() acc_name = 'kraken.' + client._name
trades = await client.get_trades()
position_msgs = pack_positions(acc_name, trades) position_msgs = pack_positions(acc_name, trades)
await ctx.started((position_msgs, (acc_name,))) await ctx.started((position_msgs, (acc_name,)))
# Get websocket token for authenticated data stream # 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', {}) resp = await client.endpoint('GetWebSocketsToken', {})
assert resp['error'] == [] assert resp['error'] == []
token = resp['result']['token'] token = resp['result']['token']
async with ( async with (
ctx.open_stream() as ems_stream, ctx.open_stream() as ems_stream,
trio.open_nursery() as n, trio.open_nursery() as n,
): ):
## TODO: maybe add multiple accounts ## TODO: maybe add multiple accounts
n.start_soon(handle_order_requests, client, ems_stream) n.start_soon(handle_order_requests, client, ems_stream)
# Process trades msg stream of ws # Process trades msg stream of ws
async with open_autorecon_ws( async with open_autorecon_ws(
'wss://ws-auth.kraken.com/', 'wss://ws-auth.kraken.com/',
fixture=subscribe, fixture=subscribe,
token=token, token=token,
) as ws: ) as ws:
async for msg in process_trade_msgs(ws): async for msg in process_trade_msgs(ws):
for trade in msg: for trade in msg:
# check the type of packaged message # check the type of packaged message
assert type(trade) == Trade assert type(trade) == Trade
# prepare and send a status update for line update # prepare and send a status update for line update
trade_msg = BrokerdStatus( trade_msg = BrokerdStatus(
reqid=trade.reqid, reqid=trade.reqid,
time_ns=time.time_ns(), time_ns=time.time_ns(),
account='kraken.spot', account='kraken.spot',
status='executed', status='executed',
filled=float(trade.size), filled=float(trade.size),
reason='Order filled by kraken', reason='Order filled by kraken',
# remaining='' # TODO: not sure what to do here. # remaining='' # TODO: not sure what to do here.
broker_details={ broker_details={
'name': 'kraken', 'name': 'kraken',
'broker_time': trade.broker_time 'broker_time': trade.broker_time
} }
) )
await ems_stream.send(trade_msg.dict()) await ems_stream.send(trade_msg.dict())
# send a fill msg for gui update # send a fill msg for gui update
fill_msg = BrokerdFill( fill_msg = BrokerdFill(
reqid=trade.reqid, reqid=trade.reqid,
time_ns=time.time_ns(), time_ns=time.time_ns(),
action=trade.action, action=trade.action,
size=float(trade.size), size=float(trade.size),
price=float(trade.price), price=float(trade.price),
# TODO: maybe capture more msg data i.e fees? # TODO: maybe capture more msg data i.e fees?
broker_details={'name': 'kraken'}, broker_details={'name': 'kraken'},
broker_time=float(trade.broker_time) broker_time=float(trade.broker_time)
) )
await ems_stream.send(fill_msg.dict()) await ems_stream.send(fill_msg.dict())
else:
log.error('Missing Kraken API key: Trades WS connection failed')
await ctx.started(({}, {'paper',}))
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( async def stream_messages(