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

Attach to order client *after* feed connection to speed up the startup time

syseng_tweaks
Tyler Goodlet 2021-05-24 12:09:03 -04:00
parent 319eacd8c1
commit f51e12819a
1 changed files with 31 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
piker/clearing/_ems.py
View File

@ -242,19 +242,20 @@ async def execute_triggers(
async def exec_loop( async def exec_loop(
ctx: tractor.Context, ctx: tractor.Context,
feed: 'Feed', # noqa feed: 'Feed', # noqa
broker: str, broker: str,
symbol: str, symbol: str,
_exec_mode: str, _exec_mode: str,
task_status: TaskStatus[dict] = trio.TASK_STATUS_IGNORED, task_status: TaskStatus[dict] = trio.TASK_STATUS_IGNORED,
) -> AsyncIterator[dict]: ) -> AsyncIterator[dict]:
"""Main scan loop for order execution conditions and submission """Main scan loop for order execution conditions and submission
to brokers. to brokers.
""" """
# XXX: this should be initial price quote from target provider
# TODO: get initial price quote from target broker
first_quote = await feed.receive() first_quote = await feed.receive()
book = get_dark_book(broker) book = get_dark_book(broker)
@ -351,6 +352,7 @@ async def process_broker_trades(
# TODO: make this a context # TODO: make this a context
# in the paper engine case this is just a mem receive channel # in the paper engine case this is just a mem receive channel
async with feed.receive_trades_data() as trades_stream: async with feed.receive_trades_data() as trades_stream:
first = await trades_stream.__anext__() first = await trades_stream.__anext__()
# startup msg expected as first from broker backend # startup msg expected as first from broker backend
@ -651,9 +653,6 @@ async def _emsd_main(
dark_book = get_dark_book(broker) dark_book = get_dark_book(broker)
# get a portal back to the client
async with tractor.wait_for_actor(client_actor_name) as portal:
# spawn one task per broker feed # spawn one task per broker feed
async with trio.open_nursery() as n: async with trio.open_nursery() as n:
@ -664,6 +663,18 @@ async def _emsd_main(
loglevel='info', loglevel='info',
) as feed: ) as feed:
# get a portal back to the client
async with tractor.wait_for_actor(client_actor_name) as portal:
# connect back to the calling actor (the one that is
# acting as an EMS client and will submit orders) to
# receive requests pushed over a tractor stream
# using (for now) an async generator.
async with portal.open_stream_from(
send_order_cmds,
symbol_key=symbol,
) as order_stream:
# start the condition scan loop # start the condition scan loop
quote, feed, client = await n.start( quote, feed, client = await n.start(
exec_loop, exec_loop,
@ -681,15 +692,6 @@ async def _emsd_main(
dark_book, dark_book,
) )
# connect back to the calling actor (the one that is
# acting as an EMS client and will submit orders) to
# receive requests pushed over a tractor stream
# using (for now) an async generator.
async with portal.open_stream_from(
send_order_cmds,
symbol_key=symbol,
) as order_stream:
# start inbound order request processing # start inbound order request processing
await process_order_cmds( await process_order_cmds(
ctx, ctx,