Update fsps and overlays inside main OHLC chart update loop

windows_testing_volume
Tyler Goodlet 2021-09-27 10:25:42 -04:00
parent 6897393715
commit 3f39c2bdfa
1 changed files with 84 additions and 52 deletions

View File

@ -59,9 +59,58 @@ _clear_throttle_rate: int = 58 # Hz
_book_throttle_rate: int = 16 # Hz
def update_fsp_chart(
chart: ChartPlotWidget,
shm: ShmArray,
display_name: str,
array_key: Optional[str],
) -> None:
array = shm.array
# XXX: is this a problem any more after porting to the
# ``tractor.Context`` api or can we remove it?
# TODO: provide a read sync mechanism to avoid this polling. the
# underlying issue is that a backfill (aka prepend) and subsequent
# shm array first/last index update could result in an empty array
# read here since the stream is never torn down on the re-compute
# steps.
# read_tries = 2
# while read_tries > 0:
# try:
# # read last
# array = shm.array
# value = array[-1][array_key]
# break
# except IndexError:
# read_tries -= 1
# continue
# update graphics
chart.update_curve_from_array(
display_name,
array,
array_key=array_key or display_name,
)
last_val_sticky = chart._ysticks.get(display_name)
if last_val_sticky:
# read from last calculated value
# XXX: fsp func names must be unique meaning we don't have
# duplicates of the underlying data even if multiple
# sub-charts reference it under different 'named charts'.
array = shm.array[array_key]
if len(array):
value = array[-1]
last_val_sticky.update_from_data(-1, value)
async def update_chart_from_quotes(
chart: ChartPlotWidget,
linked: LinkedSplits,
stream: tractor.MsgStream,
ohlcv: np.ndarray,
@ -84,6 +133,8 @@ async def update_chart_from_quotes(
# - 1-5 sec bar lookback-autocorrection like tws does?
# (would require a background history checker task)
chart = linked.chart
# update last price sticky
last_price_sticky = chart._ysticks[chart.name]
last_price_sticky.update_from_data(
@ -310,6 +361,33 @@ async def update_chart_from_quotes(
last_mx, last_mn = mx, mn
# run synchronous update on all derived fsp subplots
# print(f'subplots: {linked.subplots.keys()}')
for name, subchart in linked.subplots.items():
update_fsp_chart(
subchart,
subchart._shm,
# XXX: do we really needs seperate names here?
name,
array_key=name,
)
# TODO: all overlays on all subplots..
# run synchronous update on all derived overlays
# print(f'overlays: {chart._overlays}')
for name, shm in chart._overlays.items():
update_fsp_chart(
chart,
shm,
# XXX: do we really needs seperate names here?
name,
array_key=name,
)
def maybe_mk_fsp_shm(
sym: str,
@ -318,7 +396,8 @@ def maybe_mk_fsp_shm(
readonly: bool = True,
) -> (ShmArray, bool):
'''Allocate a single row shm array for an symbol-fsp pair.
'''Allocate a single row shm array for an symbol-fsp pair if none
exists, otherwise load the shm already existing for that token.
'''
uid = tractor.current_actor().uid
@ -436,7 +515,6 @@ async def open_fspd_cluster(
tractor.open_nursery() as n,
trio.open_nursery() as ln,
):
# Currently we spawn an actor per fsp chain but
# likely we'll want to pool them eventually to
# scale horizonatlly once cores are used up.
@ -456,8 +534,6 @@ async def open_fspd_cluster(
# "feeds". assert opened, f"A chart for {key} likely
# already exists?"
# conf['shm'] = shm
portal = await n.start_actor(
enable_modules=['piker.fsp._engine'],
name='fsp.' + display_name,
@ -543,6 +619,8 @@ async def update_chart_from_fsp(
overlay=True,
color='default_light',
)
# specially store ref to shm for lookup in display loop
chart._overlays[display_name] = shm
else:
chart = linkedsplits.add_plot(
@ -636,52 +714,6 @@ async def update_chart_from_fsp(
last = time.time()
def update_fsp_chart(
chart: ChartPlotWidget,
shm: ShmArray,
display_name: str,
array_key: str,
) -> None:
array = shm.array
# XXX: is this a problem any more after porting to the
# ``tractor.Context`` api?
# TODO: provide a read sync mechanism to avoid this polling. the
# underlying issue is that a backfill (aka prepend) and subsequent
# shm array first/last index update could result in an empty array
# read here since the stream is never torn down on the re-compute
# steps.
# read_tries = 2
# while read_tries > 0:
# try:
# # read last
# array = shm.array
# value = array[-1][array_key]
# break
# except IndexError:
# read_tries -= 1
# continue
# update graphics
chart.update_curve_from_array(
display_name,
array,
array_key=array_key,
)
last_val_sticky = chart._ysticks.get(display_name)
if last_val_sticky:
# read from last calculated value
# XXX: fsp func names must be unique meaning we don't have
# duplicates of the underlying data even if multiple
# sub-charts reference it under different 'named charts'.
value = shm.array[array_key][-1]
last_val_sticky.update_from_data(-1, value)
async def check_for_new_bars(feed, ohlcv, linkedsplits):
"""Task which updates from new bars in the shared ohlcv buffer every
``delay_s`` seconds.
@ -1002,7 +1034,7 @@ async def display_symbol_data(
# start graphics update loop(s)after receiving first live quote
ln.start_soon(
update_chart_from_quotes,
chart,
linkedsplits,
feed.stream,
ohlcv,
wap_in_history,