pikers
/
piker
3
0
Fork 0
Code Issues 4 Pull Requests 6 Packages Projects Releases Wiki Activity

Run dark-clear-loop in daemon task 

This enables "headless" dark order matching and clearing where an `emsd`
daemon subactor can be left running with active dark (or other
algorithmic) orders which will still trigger despite to attached-controlling
ems-client.

Impl details:
- rename/add `Router.maybe_open_trade_relays()` which now does all work
  of starting up ems-side long living clearing and relay tasks and the
  associated data feed; make is a `Nursery.start()`-able task instead of
  an `@acm`.
- drop `open_brokerd_trades_dialog()` and move/factor contents into the
  above method.
- add support for a `router.client_broadcast('all', msg)` to wholesale
  fan out a msg to all clients.
offline_dark_clearing
Tyler Goodlet 2022-10-04 21:52:24 -04:00
parent 31b0d8cee8
commit 49433ea87d
1 changed files with 234 additions and 271 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

293
piker/clearing/_ems.py
View File

@ -23,7 +23,6 @@ from collections import (
defaultdict, defaultdict,
# ChainMap, # ChainMap,
) )
from contextlib import asynccontextmanager
from math import isnan from math import isnan
from pprint import pformat from pprint import pformat
import time import time
@ -41,9 +40,12 @@ import tractor
from ..log import get_logger from ..log import get_logger
from ..data._normalize import iterticks from ..data._normalize import iterticks
from ..data.feed import Feed, maybe_open_feed from ..data.feed import (
Feed,
maybe_open_feed,
)
from ..data.types import Struct from ..data.types import Struct
from .._daemon import maybe_spawn_brokerd # from .._daemon import maybe_spawn_brokerd
from . import _paper_engine as paper from . import _paper_engine as paper
from ._messages import ( from ._messages import (
Order, Order,
@ -135,7 +137,6 @@ class _DarkBook(Struct):
float, float,
] = {} ] = {}
# _ems_entries: dict[str, str] = {}
_active: dict = {} _active: dict = {}
_ems2brokerd_ids: dict[str, str] = bidict() _ems2brokerd_ids: dict[str, str] = bidict()
@ -247,7 +248,6 @@ async def clear_dark_triggers(
await brokerd_orders_stream.send(brokerd_msg) await brokerd_orders_stream.send(brokerd_msg)
# book._ems_entries[oid] = live_req
# book._msgflows[oid].maps.insert(0, live_req) # book._msgflows[oid].maps.insert(0, live_req)
case _: case _:
@ -383,123 +383,55 @@ class Router(Struct):
if not stream._closed if not stream._closed
) )
@asynccontextmanager async def maybe_open_trade_relays(
async def maybe_open_brokerd_trades_dialogue(
self, self,
feed: Feed, fqsn: str,
symbol: str,
dark_book: _DarkBook,
exec_mode: str, exec_mode: str,
loglevel: str, loglevel: str,
) -> tuple[dict, tractor.MsgStream]: task_status: TaskStatus[
tuple[TradesRelay, Feed]
] = trio.TASK_STATUS_IGNORED,
) -> tuple[TradesRelay, Feed]:
''' '''
Open and yield ``brokerd`` trades dialogue context-stream if Open and yield ``brokerd`` trades dialogue context-stream if
none already exists. none already exists.
''' '''
broker = feed.mod.name from ..data._source import unpack_fqsn
relay: TradesRelay = self.relays.get(broker) broker, symbol, suffix = unpack_fqsn(fqsn)
async with (
maybe_open_feed(
[fqsn],
loglevel=loglevel,
) as (feed, quote_stream),
):
brokermod = feed.mod
broker = brokermod.name
# XXX: this should be initial price quote from target provider
first_quote: dict = feed.first_quotes[fqsn]
book: _DarkBook = self.get_dark_book(broker)
book.lasts[fqsn]: float = first_quote['last']
relay: TradesRelay = self.relays.get(broker)
if ( if (
relay is None relay
# We always want to spawn a new relay for the paper engine # We always want to spawn a new relay for the paper engine
# per symbol since we need a new tractor context to be # per symbol since we need a new tractor context to be
# opened for every every symbol such that a new data feed # opened for every every symbol such that a new data feed
# and ``PaperBoi`` client will be created and then used to # and ``PaperBoi`` client will be created and then used to
# simulate clearing events. # simulate clearing events.
or exec_mode == 'paper' and exec_mode != 'paper'
): ):
task_status.started((relay, feed))
await trio.sleep_forever()
return
relay = await self.nursery.start( trades_endpoint = getattr(brokermod, 'trades_dialogue', None)
open_brokerd_trades_dialog,
self,
feed,
symbol,
exec_mode,
loglevel,
)
self.nursery.start_soon(
translate_and_relay_brokerd_events,
broker,
relay.brokerd_stream,
self,
)
relay.consumers += 1
# TODO: get updated positions here?
assert relay.brokerd_stream
try:
yield relay
finally:
# TODO: what exactly needs to be torn down here or
# are we just consumer tracking?
relay.consumers -= 1
async def client_broadcast(
self,
sub_key: str,
msg: dict,
) -> None:
to_remove: set[tractor.MsgStream] = set()
subs = self.subscribers[sub_key]
for client_stream in subs:
try:
await client_stream.send(msg)
except (
trio.ClosedResourceError,
trio.BrokenResourceError,
):
to_remove.add(client_stream)
self.clients.remove(client_stream)
log.warning(
f'client for {client_stream} was already closed?')
if to_remove:
subs.difference_update(to_remove)
_router: Router = None
async def open_brokerd_trades_dialog(
router: Router,
feed: Feed,
symbol: str,
exec_mode: str,
loglevel: str,
task_status: TaskStatus[TradesRelay] = trio.TASK_STATUS_IGNORED,
) -> tuple[dict, tractor.MsgStream]:
'''
Open and yield ``brokerd`` trades dialogue context-stream if none
already exists.
'''
trades_endpoint = getattr(feed.mod, 'trades_dialogue', None)
broker = feed.mod.name
# TODO: make a `tractor` bug/test for this!
# if only i could member what the problem was..
# probably some GC of the portal thing?
# portal = feed.portal
# XXX: we must have our own portal + channel otherwise
# when the data feed closes it may result in a half-closed
# channel that the brokerd side thinks is still open somehow!?
async with maybe_spawn_brokerd(
broker,
loglevel=loglevel,
) as portal:
if ( if (
trades_endpoint is None trades_endpoint is None
or exec_mode == 'paper' or exec_mode == 'paper'
@ -522,33 +454,40 @@ async def open_brokerd_trades_dialog(
else: else:
# open live brokerd trades endpoint # open live brokerd trades endpoint
open_trades_endpoint = portal.open_context( open_trades_endpoint = feed.portal.open_context(
trades_endpoint, trades_endpoint,
loglevel=loglevel, loglevel=loglevel,
) )
# open trades-dialog endpoint with backend broker
try: try:
positions: list[BrokerdPosition] positions: list[BrokerdPosition]
accounts: tuple[str] accounts: tuple[str]
async with ( async with (
open_trades_endpoint as (brokerd_ctx, (positions, accounts,)), open_trades_endpoint as (
brokerd_ctx,
(positions, accounts,),
),
brokerd_ctx.open_stream() as brokerd_trades_stream, brokerd_ctx.open_stream() as brokerd_trades_stream,
): ):
# XXX: really we only want one stream per `emsd` actor # XXX: really we only want one stream per `emsd`
# to relay global `brokerd` order events unless we're # actor to relay global `brokerd` order events
# going to expect each backend to relay only orders # unless we're going to expect each backend to
# affiliated with a particular ``trades_dialogue()`` # relay only orders affiliated with a particular
# session (seems annoying for implementers). So, here # ``trades_dialogue()`` session (seems annoying
# we cache the relay task and instead of running multiple # for implementers). So, here we cache the relay
# tasks (which will result in multiples of the same msg being # task and instead of running multiple tasks
# relayed for each EMS client) we just register each client # (which will result in multiples of the same
# stream to this single relay loop in the dialog table. # msg being relayed for each EMS client) we just
# register each client stream to this single
# relay loop in the dialog table.
# begin processing order events from the target brokerd backend # begin processing order events from the target
# by receiving order submission response messages, # brokerd backend by receiving order submission
# normalizing them to EMS messages and relaying back to # response messages, normalizing them to EMS
# the piker order client set. # messages and relaying back to the piker order
# client set.
# locally cache and track positions per account with # locally cache and track positions per account with
# a table of (brokername, acctid) -> `BrokerdPosition` # a table of (brokername, acctid) -> `BrokerdPosition`
@ -576,12 +515,31 @@ async def open_brokerd_trades_dialog(
consumers=1, consumers=1,
) )
router.relays[broker] = relay self.relays[broker] = relay
# the ems scan loop may be cancelled by the client but we # spawn a ``brokerd`` order control dialog stream
# want to keep the ``brokerd`` dialogue up regardless # that syncs lifetime with the parent `emsd` daemon.
self.nursery.start_soon(
translate_and_relay_brokerd_events,
broker,
relay.brokerd_stream,
self,
)
task_status.started(relay) # dark book clearing loop, also lives with parent
# daemon to allow dark order clearing while no
# client is connected.
self.nursery.start_soon(
clear_dark_triggers,
self,
relay.brokerd_stream,
quote_stream,
broker,
fqsn, # form: <name>.<venue>.<suffix>.<broker>
book
)
task_status.started((relay, feed))
# this context should block here indefinitely until # this context should block here indefinitely until
# the ``brokerd`` task either dies or is cancelled # the ``brokerd`` task either dies or is cancelled
@ -590,10 +548,43 @@ async def open_brokerd_trades_dialog(
finally: finally:
# parent context must have been closed remove from cache so # parent context must have been closed remove from cache so
# next client will respawn if needed # next client will respawn if needed
relay = router.relays.pop(broker, None) relay = self.relays.pop(broker, None)
if not relay: if not relay:
log.warning(f'Relay for {broker} was already removed!?') log.warning(f'Relay for {broker} was already removed!?')
async def client_broadcast(
self,
sub_key: str,
msg: dict,
) -> None:
to_remove: set[tractor.MsgStream] = set()
if sub_key == 'all':
subs = set()
for s in self.subscribers.values():
subs |= s
else:
subs = self.subscribers[sub_key]
for client_stream in subs:
try:
await client_stream.send(msg)
except (
trio.ClosedResourceError,
trio.BrokenResourceError,
):
to_remove.add(client_stream)
self.clients.remove(client_stream)
log.warning(
f'client for {client_stream} was already closed?')
if to_remove:
subs.difference_update(to_remove)
_router: Router = None
@tractor.context @tractor.context
async def _setup_persistent_emsd( async def _setup_persistent_emsd(
@ -677,7 +668,7 @@ async def translate_and_relay_brokerd_events(
# fan-out-relay position msgs immediately by # fan-out-relay position msgs immediately by
# broadcasting updates on all client streams # broadcasting updates on all client streams
await router.client_broadcast(sym, pos_msg) await router.client_broadcast('all', pos_msg)
continue continue
# BrokerdOrderAck # BrokerdOrderAck
@ -827,7 +818,7 @@ async def translate_and_relay_brokerd_events(
# ``.oid`` in the msg since we're planning to # ``.oid`` in the msg since we're planning to
# maybe-kinda offer that via using ``Status`` # maybe-kinda offer that via using ``Status``
# in the longer run anyway? # in the longer run anyway?
log.warning(f'Unkown fill for {fmsg}') log.warning(f'Unknown fill for {fmsg}')
continue continue
# proxy through the "fill" result(s) # proxy through the "fill" result(s)
@ -1026,7 +1017,6 @@ async def process_client_order_cmds(
# acked yet by a brokerd, so register a cancel for when # acked yet by a brokerd, so register a cancel for when
# the order ack does show up later such that the brokerd # the order ack does show up later such that the brokerd
# order request can be cancelled at that time. # order request can be cancelled at that time.
# dark_book._ems_entries[oid] = msg
# special case for now.. # special case for now..
status.req = to_brokerd_msg status.req = to_brokerd_msg
@ -1286,7 +1276,6 @@ async def _emsd_main(
from ..data._source import unpack_fqsn from ..data._source import unpack_fqsn
broker, symbol, suffix = unpack_fqsn(fqsn) broker, symbol, suffix = unpack_fqsn(fqsn)
dark_book = _router.get_dark_book(broker)
# TODO: would be nice if in tractor we can require either a ctx arg, # TODO: would be nice if in tractor we can require either a ctx arg,
# or a named arg with ctx in it and a type annotation of # or a named arg with ctx in it and a type annotation of
@ -1294,44 +1283,29 @@ async def _emsd_main(
ems_ctx = ctx ems_ctx = ctx
# spawn one task per broker feed # spawn one task per broker feed
relay: TradesRelay
feed: Feed feed: Feed
async with (
maybe_open_feed(
[fqsn],
loglevel=loglevel,
) as (feed, quote_stream),
):
# XXX: this should be initial price quote from target provider # open a stream with the brokerd backend for order flow dialogue
first_quote: dict = feed.first_quotes[fqsn] # only open if one isn't already up: we try to keep as few duplicate
book: _DarkBook = _router.get_dark_book(broker) # streams as necessary.
book.lasts[fqsn]: float = first_quote['last'] # TODO: should we try using `tractor.trionics.maybe_open_context()`
# here?
# open a stream with the brokerd backend for order relay, feed = await _router.nursery.start(
# flow dialogue _router.maybe_open_trade_relays,
async with ( fqsn,
# only open if one isn't already up: we try to keep
# as few duplicate streams as necessary
_router.maybe_open_brokerd_trades_dialogue(
feed,
symbol,
dark_book,
exec_mode, exec_mode,
loglevel, loglevel,
)
) as relay,
trio.open_nursery() as n,
):
brokerd_stream = relay.brokerd_stream brokerd_stream = relay.brokerd_stream
dark_book = _router.get_dark_book(broker)
# signal to client that we're started and deliver # signal to client that we're started and deliver
# all known pps and accounts for this ``brokerd``. # all known pps and accounts for this ``brokerd``.
await ems_ctx.started(( await ems_ctx.started((
relay.positions, relay.positions,
list(relay.accounts), list(relay.accounts),
book._active, dark_book._active,
)) ))
# establish 2-way stream with requesting order-client and # establish 2-way stream with requesting order-client and
@ -1349,17 +1323,6 @@ async def _emsd_main(
# updates per dialog. # updates per dialog.
_router.subscribers[fqsn].add(client_stream) _router.subscribers[fqsn].add(client_stream)
# trigger scan and exec loop
n.start_soon(
clear_dark_triggers,
_router,
brokerd_stream,
quote_stream,
broker,
fqsn, # form: <name>.<venue>.<suffix>.<broker>
book
)
# start inbound (from attached client) order request processing # start inbound (from attached client) order request processing
# main entrypoint, run here until cancelled. # main entrypoint, run here until cancelled.
try: try: