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

Async enter/open the symcache in paper engine 

Since we don't want to be doing a `trio.run()` from async code (being
already in the `tractor` runtime and all); for now just put a top level
block wrapping async enter until we figure out to embed it (likely)
inside `open_account()` and pass the ref to `open_trade_ledger()`.
account_tests
Tyler Goodlet 2023-07-10 17:52:23 -04:00
parent 108e8c7082
commit 06c581bfab
1 changed files with 123 additions and 117 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

240
piker/clearing/_paper_engine.py
View File

@ -49,9 +49,10 @@ from ..accounting import (
Transaction, Transaction,
TransactionLedger, TransactionLedger,
open_trade_ledger, open_trade_ledger,
open_pps, open_account,
) )
from ..data import iterticks from ..data import iterticks
from ..data._symcache import open_symcache
from ..accounting import unpack_fqme from ..accounting import unpack_fqme
from ._util import ( from ._util import (
log, # sub-sys logger log, # sub-sys logger
@ -541,141 +542,146 @@ async def open_trade_dialog(
# enable piker.clearing console log for *this* subactor # enable piker.clearing console log for *this* subactor
get_console_log(loglevel) get_console_log(loglevel)
acnt: Account async with open_symcache(get_brokermod(broker)) as symcache:
ledger: TransactionLedger acnt: Account
with ( ledger: TransactionLedger
open_pps( with (
broker,
'paper',
write_on_exit=True,
) as acnt,
open_trade_ledger( # TODO: probably do the symcache and ledger loading
broker, # implicitly behind this? Deliver an account, and ledger
'paper', # pair or make the ledger an attr of the account?
) as ledger open_account(
): broker,
# NOTE: retreive market(pair) info from the backend broker 'paper',
# since ledger entries (in their backend native format) often write_on_exit=True,
# don't contain necessary market info per trade record entry.. ) as acnt,
# - if no fqme was passed in, we presume we're running in
# "ledger-sync-only mode" and thus we load mkt info for
# each symbol found in the ledger to a acnt table manually.
# TODO: how to process ledger info from backends? open_trade_ledger(
# - should we be rolling our own actor-cached version of these broker,
# client API refs or using portal IPC to send requests to the 'paper',
# existing brokerd daemon? symcache=symcache,
# - alternatively we can possibly expect and use ) as ledger
# a `.broker.norm_trade_records()` ep? ):
brokermod = get_brokermod(broker) # NOTE: retreive market(pair) info from the backend broker
gmi = getattr(brokermod, 'get_mkt_info', None) # since ledger entries (in their backend native format) often
# don't contain necessary market info per trade record entry..
# - if no fqme was passed in, we presume we're running in
# "ledger-sync-only mode" and thus we load mkt info for
# each symbol found in the ledger to a acnt table manually.
# update all transactions with mkt info before # TODO: how to process ledger info from backends?
# loading any pps # - should we be rolling our own actor-cached version of these
mkt_by_fqme: dict[str, MktPair] = {} # client API refs or using portal IPC to send requests to the
if fqme: # existing brokerd daemon?
bs_fqme, _, broker = fqme.rpartition('.') # - alternatively we can possibly expect and use
mkt, pair = await brokermod.get_mkt_info(bs_fqme) # a `.broker.norm_trade_records()` ep?
mkt_by_fqme[mkt.fqme] = mkt brokermod = get_brokermod(broker)
gmi = getattr(brokermod, 'get_mkt_info', None)
# for each sym in the ledger load it's `MktPair` info # update all transactions with mkt info before
for tid, txdict in ledger.data.items(): # loading any pps
l_fqme: str = txdict.get('fqme') or txdict['fqsn'] mkt_by_fqme: dict[str, MktPair] = {}
if fqme:
bs_fqme, _, broker = fqme.rpartition('.')
mkt, pair = await brokermod.get_mkt_info(bs_fqme)
mkt_by_fqme[mkt.fqme] = mkt
if ( # for each sym in the ledger load it's `MktPair` info
gmi for tid, txdict in ledger.data.items():
and l_fqme not in mkt_by_fqme l_fqme: str = txdict.get('fqme') or txdict['fqsn']
):
mkt, pair = await gmi(
l_fqme.rstrip(f'.{broker}'),
)
mkt_by_fqme[l_fqme] = mkt
# if an ``fqme: str`` input was provided we only if (
# need a ``MktPair`` for that one market, since we're gmi
# running in real simulated-clearing mode, not just ledger and l_fqme not in mkt_by_fqme
# syncing. ):
if ( mkt, pair = await gmi(
fqme is not None l_fqme.rstrip(f'.{broker}'),
and fqme in mkt_by_fqme )
): mkt_by_fqme[l_fqme] = mkt
break
# update pos table from ledger history and provide a ``MktPair`` # if an ``fqme: str`` input was provided we only
# lookup for internal position accounting calcs. # need a ``MktPair`` for that one market, since we're
acnt.update_from_ledger(ledger) # running in real simulated-clearing mode, not just ledger
# syncing.
if (
fqme is not None
and fqme in mkt_by_fqme
):
break
pp_msgs: list[BrokerdPosition] = [] # update pos table from ledger history and provide a ``MktPair``
pos: Position # lookup for internal position accounting calcs.
token: str # f'{symbol}.{self.broker}' acnt.update_from_ledger(ledger)
for token, pos in acnt.pps.items():
pp_msgs.append(BrokerdPosition( pp_msgs: list[BrokerdPosition] = []
broker=broker, pos: Position
account='paper', token: str # f'{symbol}.{self.broker}'
symbol=pos.mkt.fqme, for token, pos in acnt.pps.items():
size=pos.size, pp_msgs.append(BrokerdPosition(
avg_price=pos.ppu, broker=broker,
account='paper',
symbol=pos.mkt.fqme,
size=pos.size,
avg_price=pos.ppu,
))
await ctx.started((
pp_msgs,
['paper'],
)) ))
await ctx.started(( # write new positions state in case ledger was
pp_msgs, # newer then that tracked in pps.toml
['paper'], acnt.write_config()
))
# write new positions state in case ledger was # exit early since no fqme was passed,
# newer then that tracked in pps.toml # normally this case is just to load
acnt.write_config() # positions "offline".
if fqme is None:
# exit early since no fqme was passed, log.warning(
# normally this case is just to load 'Paper engine only running in position delivery mode!\n'
# positions "offline". 'NO SIMULATED CLEARING LOOP IS ACTIVE!'
if fqme is None: )
log.warning( await trio.sleep_forever()
'Paper engine only running in position delivery mode!\n' return
'NO SIMULATED CLEARING LOOP IS ACTIVE!'
)
await trio.sleep_forever()
return
async with (
data.open_feed(
[fqme],
loglevel=loglevel,
) as feed,
):
# sanity check all the mkt infos
for fqme, flume in feed.flumes.items():
mkt = mkt_by_fqme[fqme]
print(mkt - flume.mkt)
assert mkt == flume.mkt
async with ( async with (
ctx.open_stream() as ems_stream, data.open_feed(
trio.open_nursery() as n, [fqme],
loglevel=loglevel,
) as feed,
): ):
client = PaperBoi( # sanity check all the mkt infos
broker=broker, for fqme, flume in feed.flumes.items():
ems_trades_stream=ems_stream, mkt = mkt_by_fqme[fqme]
acnt=acnt, assert mkt == flume.mkt
ledger=ledger,
_buys=_buys, async with (
_sells=_sells, ctx.open_stream() as ems_stream,
_reqids=_reqids, trio.open_nursery() as n,
):
client = PaperBoi(
broker=broker,
ems_trades_stream=ems_stream,
acnt=acnt,
ledger=ledger,
_mkts=mkt_by_fqme, _buys=_buys,
_sells=_sells,
_reqids=_reqids,
) _mkts=mkt_by_fqme,
n.start_soon( )
handle_order_requests,
client,
ems_stream,
)
# paper engine simulator clearing task n.start_soon(
await simulate_fills(feed.streams[broker], client) handle_order_requests,
client,
ems_stream,
)
# paper engine simulator clearing task
await simulate_fills(feed.streams[broker], client)
@acm @acm