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

Flip to `open_order_mode()` as ctx mngr 

We need a subtask to compute the current pp PnL in real-time but really
only if a pp exists - a spawnable subtask would be ideal for this. Stage
a tick streaming task using a stream bcaster; no actual pnl calc yet.

Since we're going to need subtasks anyway might as well stick the order
mode UI processing loop in a task as well and then just give the whole
thing a ctx mngr api. This'll probably be handy for when we have
auto-strats that need to dynamically use the mode's api as well.

Oh, and move the time -> index mapper to a chart method for now.
fsp_feeds
Tyler Goodlet 2021-08-27 14:53:30 -04:00
parent ee377e6d6b
commit dfe4ca948a
2 changed files with 161 additions and 124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
piker/ui/_chart.py
View File

@ -71,7 +71,7 @@ from .. import brokers
from ..log import get_logger from ..log import get_logger
from ._exec import run_qtractor from ._exec import run_qtractor
from ._interaction import ChartView from ._interaction import ChartView
from .order_mode import run_order_mode from .order_mode import open_order_mode
from .. import fsp from .. import fsp
from ._forms import ( from ._forms import (
FieldsForm, FieldsForm,
@ -1084,6 +1084,22 @@ class ChartPlotWidget(pg.PlotWidget):
self.sig_mouse_leave.emit(self) self.sig_mouse_leave.emit(self)
self.scene().leaveEvent(ev) self.scene().leaveEvent(ev)
def get_index(self, time: float) -> int:
# TODO: this should go onto some sort of
# data-view strimg thinger..right?
ohlc = self._shm.array
# ohlc = chart._shm.array
# XXX: not sure why the time is so off here
# looks like we're gonna have to do some fixing..
indexes = ohlc['time'] >= time
if any(indexes):
return ohlc['index'][indexes][-1]
else:
return ohlc['index'][-1]
_clear_throttle_rate: int = 60 # Hz _clear_throttle_rate: int = 60 # Hz
_book_throttle_rate: int = 16 # Hz _book_throttle_rate: int = 16 # Hz
@ -1695,19 +1711,6 @@ async def display_symbol_data(
) as feed, ) as feed,
trio.open_nursery() as n, trio.open_nursery() as n,
): ):
# async def print_quotes():
# async with feed.stream.subscribe() as bstream:
# last_tick = time.time()
# async for quotes in bstream:
# now = time.time()
# period = now - last_tick
# for sym, quote in quotes.items():
# ticks = quote.get('ticks', ())
# if ticks:
# # print(f'{1/period} Hz')
# last_tick = time.time()
# n.start_soon(print_quotes)
ohlcv: ShmArray = feed.shm ohlcv: ShmArray = feed.shm
bars = ohlcv.array bars = ohlcv.array
@ -1833,12 +1836,37 @@ async def display_symbol_data(
linkedsplits linkedsplits
) )
await run_order_mode( async with (
chart,
symbol, open_order_mode(
provider, chart,
order_mode_started symbol,
) provider,
order_mode_started
) as order_mode,
):
pp = order_mode.pp
live = pp.live_pp
# real-time update pnl on the order mode
async with feed.stream.subscribe() as bstream:
last_tick = time.time()
async for quotes in bstream:
now = time.time()
period = now - last_tick
for sym, quote in quotes.items():
ticks = quote.get('ticks', ())
if ticks:
print(f'{1/period} Hz')
if live.size:
# compute and display pnl
pass
last_tick = time.time()
async def load_provider_search( async def load_provider_search(

217
piker/ui/order_mode.py
View File

@ -18,6 +18,7 @@
Chart trading, the only way to scalp. Chart trading, the only way to scalp.
""" """
from contextlib import asynccontextmanager
from dataclasses import dataclass, field from dataclasses import dataclass, field
from functools import partial from functools import partial
from pprint import pformat from pprint import pformat
@ -36,7 +37,7 @@ from ..data._source import Symbol
from ..log import get_logger from ..log import get_logger
from ._editors import LineEditor, ArrowEditor from ._editors import LineEditor, ArrowEditor
from ._lines import order_line, LevelLine from ._lines import order_line, LevelLine
from ._position import PositionTracker, OrderModePane, Allocator, _size_units from ._position import PositionTracker, SettingsPane, Allocator, _size_units
from ._window import MultiStatus from ._window import MultiStatus
from ..clearing._messages import Order from ..clearing._messages import Order
from ._forms import open_form_input_handling from ._forms import open_form_input_handling
@ -91,7 +92,7 @@ class OrderMode:
multistatus: MultiStatus multistatus: MultiStatus
pp: PositionTracker pp: PositionTracker
allocator: 'Allocator' # noqa allocator: 'Allocator' # noqa
pane: OrderModePane pane: SettingsPane
active: bool = False active: bool = False
@ -462,7 +463,8 @@ class OrderMode:
return ids return ids
async def run_order_mode( @asynccontextmanager
async def open_order_mode(
chart: 'ChartPlotWidget', # noqa chart: 'ChartPlotWidget', # noqa
symbol: Symbol, symbol: Symbol,
@ -493,6 +495,7 @@ async def run_order_mode(
trades_stream, trades_stream,
positions positions
), ),
trio.open_nursery() as n,
): ):
log.info(f'Opening order mode for {brokername}.{symbol.key}') log.info(f'Opening order mode for {brokername}.{symbol.key}')
@ -521,7 +524,7 @@ async def run_order_mode(
pp_tracker.hide() pp_tracker.hide()
# order pane widgets and allocation model # order pane widgets and allocation model
order_pane = OrderModePane( order_pane = SettingsPane(
tracker=pp_tracker, tracker=pp_tracker,
form=form, form=form,
alloc=alloc, alloc=alloc,
@ -574,21 +577,6 @@ async def run_order_mode(
# make fill bar and positioning snapshot # make fill bar and positioning snapshot
order_pane.init_status_ui() order_pane.init_status_ui()
# TODO: this should go onto some sort of
# data-view strimg thinger..right?
def get_index(time: float):
# XXX: not sure why the time is so off here
# looks like we're gonna have to do some fixing..
ohlc = chart._shm.array
indexes = ohlc['time'] >= time
if any(indexes):
return ohlc['index'][indexes][-1]
else:
return ohlc['index'][-1]
# Begin order-response streaming # Begin order-response streaming
done() done()
@ -611,110 +599,131 @@ async def run_order_mode(
# to handle input since the ems connection is ready # to handle input since the ems connection is ready
started.set() started.set()
# this is where we receive **back** messages n.start_soon(
# about executions **from** the EMS actor process_trades_and_update_ui,
async for msg in trades_stream: mode,
trades_stream,
book,
)
yield mode
# await trio.sleep_forever()
fmsg = pformat(msg)
log.info(f'Received order msg:\n{fmsg}')
name = msg['name'] async def process_trades_and_update_ui(
if name in (
'position',
):
# show line label once order is live
sym = mode.chart.linked.symbol mode: OrderMode,
if msg['symbol'].lower() in sym.key: trades_stream: tractor.MsgStream,
pp_tracker.update(msg) book: OrderBook,
# update order pane widgets ) -> None:
mode.pane.update_status_ui()
# short circuit to next msg to avoid get_index = mode.chart.get_index
# uncessary msg content lookups tracker = mode.pp
continue
resp = msg['resp'] # this is where we receive **back** messages
oid = msg['oid'] # about executions **from** the EMS actor
async for msg in trades_stream:
dialog = mode.dialogs.get(oid) fmsg = pformat(msg)
if dialog is None: log.info(f'Received order msg:\n{fmsg}')
log.warning(f'received msg for untracked dialog:\n{fmsg}')
# TODO: enable pure tracking / mirroring of dialogs name = msg['name']
# is desired. if name in (
continue 'position',
):
# show line label once order is live
# record message to dialog tracking sym = mode.chart.linked.symbol
dialog.msgs[oid] = msg if msg['symbol'].lower() in sym.key:
tracker.update(msg)
# response to 'action' request (buy/sell) # update order pane widgets
if resp in ( mode.pane.update_status_ui()
'dark_submitted',
'broker_submitted'
):
# show line label once order is live # short circuit to next msg to avoid
mode.on_submit(oid) # uncessary msg content lookups
continue
# resp to 'cancel' request or error condition resp = msg['resp']
# for action request oid = msg['oid']
elif resp in (
'broker_cancelled',
'broker_inactive',
'dark_cancelled'
):
# delete level line from view
mode.on_cancel(oid)
elif resp in ( dialog = mode.dialogs.get(oid)
'dark_triggered' if dialog is None:
): log.warning(f'received msg for untracked dialog:\n{fmsg}')
log.info(f'Dark order triggered for {fmsg}')
elif resp in ( # TODO: enable pure tracking / mirroring of dialogs
'alert_triggered' # is desired.
): continue
# should only be one "fill" for an alert
# add a triangle and remove the level line
mode.on_fill(
oid,
price=msg['trigger_price'],
arrow_index=get_index(time.time()),
)
mode.lines.remove_line(uuid=oid)
await mode.on_exec(oid, msg)
# response to completed 'action' request for buy/sell # record message to dialog tracking
elif resp in ( dialog.msgs[oid] = msg
'broker_executed',
):
# right now this is just triggering a system alert
await mode.on_exec(oid, msg)
if msg['brokerd_msg']['remaining'] == 0: # response to 'action' request (buy/sell)
mode.lines.remove_line(uuid=oid) if resp in (
'dark_submitted',
'broker_submitted'
):
# each clearing tick is responded individually # show line label once order is live
elif resp in ('broker_filled',): mode.on_submit(oid)
known_order = book._sent_orders.get(oid) # resp to 'cancel' request or error condition
if not known_order: # for action request
log.warning(f'order {oid} is unknown') elif resp in (
continue 'broker_cancelled',
'broker_inactive',
'dark_cancelled'
):
# delete level line from view
mode.on_cancel(oid)
action = known_order.action elif resp in (
details = msg['brokerd_msg'] 'dark_triggered'
):
log.info(f'Dark order triggered for {fmsg}')
# TODO: some kinda progress system elif resp in (
mode.on_fill( 'alert_triggered'
oid, ):
price=details['price'], # should only be one "fill" for an alert
pointing='up' if action == 'buy' else 'down', # add a triangle and remove the level line
mode.on_fill(
oid,
price=msg['trigger_price'],
arrow_index=get_index(time.time()),
)
mode.lines.remove_line(uuid=oid)
await mode.on_exec(oid, msg)
# TODO: put the actual exchange timestamp # response to completed 'action' request for buy/sell
arrow_index=get_index(details['broker_time']), elif resp in (
) 'broker_executed',
):
# right now this is just triggering a system alert
await mode.on_exec(oid, msg)
pp_tracker.live_pp.fills.append(msg) if msg['brokerd_msg']['remaining'] == 0:
mode.lines.remove_line(uuid=oid)
# each clearing tick is responded individually
elif resp in ('broker_filled',):
known_order = book._sent_orders.get(oid)
if not known_order:
log.warning(f'order {oid} is unknown')
continue
action = known_order.action
details = msg['brokerd_msg']
# TODO: some kinda progress system
mode.on_fill(
oid,
price=details['price'],
pointing='up' if action == 'buy' else 'down',
# TODO: put the actual exchange timestamp
arrow_index=get_index(details['broker_time']),
)
tracker.live_pp.fills.append(msg)