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

Always load FSPs with the default (fast) sampling period

history_view
Tyler Goodlet 2022-08-30 10:55:11 -04:00
parent 861fe791eb
commit f79c3617d6
1 changed files with 10 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
piker/fsp/_engine.py
View File

@ -37,6 +37,7 @@ from .. import data
from ..data import attach_shm_array from ..data import attach_shm_array
from ..data.feed import Feed from ..data.feed import Feed
from ..data._sharedmem import ShmArray from ..data._sharedmem import ShmArray
from ..data._sampling import _default_delay_s
from ..data._source import Symbol from ..data._source import Symbol
from ._api import ( from ._api import (
Fsp, Fsp,
@ -105,7 +106,7 @@ async def fsp_compute(
filter_quotes_by_sym(fqsn, quote_stream), filter_quotes_by_sym(fqsn, quote_stream),
# XXX: currently the ``ohlcv`` arg # XXX: currently the ``ohlcv`` arg
feed.shm, feed.rt_shm,
) )
# Conduct a single iteration of fsp with historical bars input # Conduct a single iteration of fsp with historical bars input
@ -313,7 +314,7 @@ async def cascade(
profiler(f'{func}: feed up') profiler(f'{func}: feed up')
assert src.token == feed.shm.token assert src.token == feed.rt_shm.token
# last_len = new_len = len(src.array) # last_len = new_len = len(src.array)
func_name = func.__name__ func_name = func.__name__
@ -420,7 +421,11 @@ async def cascade(
# detect sample period step for subscription to increment # detect sample period step for subscription to increment
# signal # signal
times = src.array['time'] times = src.array['time']
if len(times) > 1:
delay_s = times[-1] - times[times != times[-1]][-1] delay_s = times[-1] - times[times != times[-1]][-1]
else:
# our default "HFT" sample rate.
delay_s = _default_delay_s
# Increment the underlying shared memory buffer on every # Increment the underlying shared memory buffer on every
# "increment" msg received from the underlying data feed. # "increment" msg received from the underlying data feed.
@ -431,7 +436,8 @@ async def cascade(
profiler(f'{func_name}: sample stream up') profiler(f'{func_name}: sample stream up')
profiler.finish() profiler.finish()
async for _ in istream: async for i in istream:
# log.runtime(f'FSP incrementing {i}')
# respawn the compute task if the source # respawn the compute task if the source
# array has been updated such that we compute # array has been updated such that we compute