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

Update ledger from api immediately, cruft cleaning

kraken_ws_orders
Tyler Goodlet 2022-07-19 11:02:09 -04:00
parent aa7f24b6db
commit b1419c850d
1 changed files with 23 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -31,7 +31,6 @@ import time
from typing import ( from typing import (
Any, Any,
AsyncIterator, AsyncIterator,
# Optional,
Union, Union,
) )
@ -46,7 +45,6 @@ from piker.pp import (
Position, Position,
PpTable, PpTable,
Transaction, Transaction,
# update_pps_conf,
open_trade_ledger, open_trade_ledger,
open_pps, open_pps,
) )
@ -391,6 +389,7 @@ async def trades_dialogue(
# most recent 50 trades and assume that by ordering we # most recent 50 trades and assume that by ordering we
# already have those records in the ledger. # already have those records in the ledger.
tids2trades = await client.get_trades() tids2trades = await client.get_trades()
ledger_dict.update(tids2trades)
api_trans = norm_trade_records(tids2trades) api_trans = norm_trade_records(tids2trades)
# retrieve kraken reported balances # retrieve kraken reported balances
@ -448,12 +447,13 @@ async def trades_dialogue(
ppmsgs = trades2pps( ppmsgs = trades2pps(
table, table,
acctid, acctid,
# new_trans,
) )
await ctx.started((ppmsgs, [acc_name])) await ctx.started((ppmsgs, [acc_name]))
# XXX: not fucking clue but putting this finally block # XXX: not fucking clue but putting this finally block
# will suppress errors inside the direct await below!?! # will suppress errors inside the direct await below!?!
# likely something to do with the exist stack inside
# the nobsws stuff...
# try: # try:
# Get websocket token for authenticated data stream # Get websocket token for authenticated data stream
@ -494,26 +494,19 @@ async def trades_dialogue(
) )
# enter relay loop # enter relay loop
# try: await handle_order_updates(
try: ws,
await handle_order_updates( stream,
ws, ems_stream,
stream, apiflows,
ems_stream, ids,
apiflows, reqids2txids,
ids, table,
reqids2txids, api_trans,
table, acctid,
api_trans, acc_name,
acctid, token,
acc_name, )
token,
)
# except:
# await tractor.breakpoint()
finally:
# always update ledger on exit
ledger_dict.update(tids2trades)
async def handle_order_updates( async def handle_order_updates(
@ -561,9 +554,13 @@ async def handle_order_updates(
f'ownTrades update_{seq}:\n' f'ownTrades update_{seq}:\n'
f'{pformat(trades_msgs)}' f'{pformat(trades_msgs)}'
) )
# XXX: a fix / todo
# see the comment in the caller about weird error
# suppression around a commented `try:`
# assert 0 # assert 0
# format as tid -> trade event map # format as tid -> trade event map
# eg. msg # 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',
@ -579,15 +576,10 @@ async def handle_order_updates(
tid: trade tid: trade
for entry in trades_msgs for entry in trades_msgs
for (tid, trade) in entry.items() for (tid, trade) in entry.items()
# don't re-process datums we've already seen
if tid not in ledger_trans if tid not in ledger_trans
} }
# if tid in ledger_trans:
# # skip already seen transactions
# log.info(f'Skipping already seen trade {trade}')
# continue
# await tractor.breakpoint()
for tid, trade in trades.items(): for tid, trade in trades.items():
txid = trade['ordertxid'] txid = trade['ordertxid']
@ -642,11 +634,6 @@ async def handle_order_updates(
) )
await ems_stream.send(filled_msg) await ems_stream.send(filled_msg)
# if not trades:
# # skip pp emissions if we have already
# # processed all trades in this msg.
# continue
new_trans = norm_trade_records(trades) new_trans = norm_trade_records(trades)
ppmsgs = trades2pps( ppmsgs = trades2pps(
table, table,
@ -897,9 +884,6 @@ async def handle_order_updates(
chain = apiflows[reqid] chain = apiflows[reqid]
chain.maps.append(event) chain.maps.append(event)
# pretxid = chain['txid']
# print(f'pretxid: {pretxid}')
resps, errored = process_status( resps, errored = process_status(
event, event,
oid, oid,