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

Make all `.bsuid`s the normed symbol "altname"s

kraken_ws_orders
Tyler Goodlet 2022-07-10 20:05:31 -04:00
parent 22f9b2552c
commit abb6854e74
1 changed files with 27 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -54,7 +54,6 @@ from .api import (
Client, Client,
BrokerError, BrokerError,
get_client, get_client,
normalize_symbol,
) )
from .feed import ( from .feed import (
get_console_log, get_console_log,
@ -273,7 +272,10 @@ async def trades_dialogue(
log.info( log.info(
f'Loaded {len(trades)} trades from account `{acc_name}`' f'Loaded {len(trades)} trades from account `{acc_name}`'
) )
with open_ledger(acctid, trades) as trans: with open_ledger(
acctid,
trades,
) as trans:
active, closed = pp.update_pps_conf( active, closed = pp.update_pps_conf(
'kraken', 'kraken',
acctid, acctid,
@ -366,7 +368,7 @@ async def handle_order_updates(
emsflow: dict[str, list[MsgUnion]], emsflow: dict[str, list[MsgUnion]],
ids: bidict[str, int], ids: bidict[str, int],
reqids2txids: bidict[int, str], reqids2txids: bidict[int, str],
trans: list[pp.Transaction], trans: set[pp.Transaction],
acctid: str, acctid: str,
acc_name: str, acc_name: str,
token: str, token: str,
@ -381,7 +383,7 @@ async def handle_order_updates(
''' '''
# transaction records which will be updated # transaction records which will be updated
# on new trade clearing events (aka order "fills") # on new trade clearing events (aka order "fills")
trans: list[pp.Transaction] trans: set[pp.Transaction]
async for msg in ws_stream: async for msg in ws_stream:
match msg: match msg:
@ -467,16 +469,27 @@ 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
# update ledger and position tracking # update ledger and position tracking
with open_ledger(acctid, trades) as trans: await tractor.breakpoint()
# TODO: ideally we can pass in an existingn trans: set[pp.Transaction]
with open_ledger(
acctid,
trades,
) as trans:
# TODO: ideally we can pass in an existing
# pps state to this right? such that we # pps state to this right? such that we
# don't have to do a ledger reload all the # don't have to do a ledger reload all the
# time.. # time..
active, closed = pp.update_pps_conf( active, closed = pp.update_pps_conf(
'kraken', 'kraken',
acctid, acctid,
trade_records=trans, trade_records=list(trans),
ledger_reload={}.fromkeys( ledger_reload={}.fromkeys(
t.bsuid for t in trans), t.bsuid for t in trans),
) )
@ -841,8 +854,9 @@ def norm_trade_records(
'buy': 1, 'buy': 1,
'sell': -1, 'sell': -1,
}[record['type']] }[record['type']]
bsuid = record['pair']
norm_sym = normalize_symbol(bsuid) # we normalize to kraken's `altname` always..
bsuid = norm_sym = Client.normalize_symbol(record['pair'])
records.append( records.append(
pp.Transaction( pp.Transaction(
@ -867,7 +881,7 @@ def open_ledger(
acctid: str, acctid: str,
trade_entries: list[dict[str, Any]], trade_entries: list[dict[str, Any]],
) -> list[pp.Transaction]: ) -> set[pp.Transaction]:
''' '''
Write recent session's trades to the user's (local) ledger file. Write recent session's trades to the user's (local) ledger file.
@ -878,8 +892,10 @@ def open_ledger(
) as ledger: ) as ledger:
# normalize to transaction form # normalize to transaction form
# TODO: cawt damn, we should probably delegate to cryptofeed for
# this insteada of re-hacking kraken's total crap?
records = norm_trade_records(trade_entries) records = norm_trade_records(trade_entries)
yield records yield set(records)
# update on exit # update on exit
ledger.update(trade_entries) ledger.update(trade_entries)