diff --git a/piker/data/flows.py b/piker/data/flows.py index 075b9953..343904ce 100644 --- a/piker/data/flows.py +++ b/piker/data/flows.py @@ -24,7 +24,6 @@ real-time data processing data-structures. from __future__ import annotations import tractor -import pendulum import numpy as np from piker.types import Struct @@ -73,6 +72,8 @@ class Flume(Struct): izero_hist: int = 0 izero_rt: int = 0 throttle_rate: int | None = None + rt_sample_period_s: float = 1. + hist_sample_period_s: float = 60. @property def rt_shm(self) -> ShmArray: @@ -115,17 +116,9 @@ class Flume(Struct): period and ratio between them. ''' - times: np.ndarray = self.hist_shm.array['time'] - end: float | int = pendulum.from_timestamp(times[-1]) - start: float | int = pendulum.from_timestamp(times[times != times[-1]][-1]) - hist_step_size_s: float = (end - start).seconds - - times = self.rt_shm.array['time'] - end = pendulum.from_timestamp(times[-1]) - start = pendulum.from_timestamp(times[times != times[-1]][-1]) - rt_step_size_s = (end - start).seconds - - ratio = hist_step_size_s / rt_step_size_s + rt_step_size_s: float = self.rt_sample_period_s + hist_step_size_s: float = self.hist_sample_period_s + ratio: float = hist_step_size_s / rt_step_size_s return ( rt_step_size_s, hist_step_size_s, diff --git a/piker/fsp/_engine.py b/piker/fsp/_engine.py index 8795a1fb..3218fce3 100644 --- a/piker/fsp/_engine.py +++ b/piker/fsp/_engine.py @@ -22,7 +22,10 @@ from __future__ import annotations from contextlib import asynccontextmanager as acm from functools import partial from typing import ( + Any, AsyncIterator, + Awaitable, + cast, Callable, TYPE_CHECKING, ) @@ -39,15 +42,20 @@ from ..log import ( get_console_log, ) from .. import data +from ..data.ticktools import FeedQuote from ..data.flows import Flume +from ..data._sharedmem import NDTokenMsg from tractor.ipc._shm import ShmArray from ..data._sampling import ( - _default_delay_s, open_sample_stream, ) from ..accounting import MktPair from ._api import ( Fsp, + FspFunc, + FspStream, + FspYield, + FspHistory, _load_builtins, NDToken, ) @@ -58,13 +66,70 @@ if TYPE_CHECKING: log = get_logger(__name__) +type QuoteFrame = dict[str, FeedQuote] +type BackfillNotice = tuple[str, float]|list[str|float] +type SampleValue = int|float|BackfillNotice +type SampleMsg = dict[str, SampleValue] +type EdgeFunc = FspFunc + + +def _is_relevant_sample_msg( + msg: SampleMsg, + fqme: str, + period_s: float, + +) -> bool: + backfill: SampleValue|None = msg.get('backfilling') + if backfill is None: + return True + + if ( + not isinstance(backfill, (list, tuple)) + or + len(backfill) != 2 + ): + return False + + backfill_fqme: str | float + timeframe: str | float + backfill_fqme, timeframe = backfill + return ( + backfill_fqme == fqme + and + timeframe == period_s + ) + + +def _should_advance_dst( + msg: SampleMsg, + step_diff: int, + +) -> bool: + return ( + 'backfilling' not in msg + and + step_diff == 1 + ) + + +def _needs_history_resync( + msg: SampleMsg, + synced: bool, + +) -> bool: + return ( + 'backfilling' in msg + or + not synced + ) + async def filter_quotes_by_sym( sym: str, - quote_stream: tractor.MsgStream, + quote_stream: AsyncIterator[QuoteFrame], -) -> AsyncIterator[dict]: +) -> AsyncIterator[FeedQuote]: ''' Filter quote stream by target symbol. @@ -75,10 +140,91 @@ async def filter_quotes_by_sym( yield {} async for quotes in quote_stream: - quote = quotes.get(sym) + quote: FeedQuote|None = quotes.get(sym) if quote: yield quote + +class _HistoryShm: + ''' + Present an immutable history view before switching to live SHM. + + ''' + def __init__(self, shm: ShmArray) -> None: + self._shm: ShmArray = shm + self.first: int = shm._first.value + self.last_index: int = shm._last.value + self.history: np.ndarray = shm._array[ + self.first:self.last_index + ].copy() + self.live: bool = False + + @property + def array(self) -> np.ndarray: + if self.live: + return self._shm.array + return self.history + + @property + def index(self) -> int: + if self.live: + return self._shm.index + return self.last_index % self._shm._len + + def last(self, length: int = 1) -> np.ndarray: + return self.array[-length:] + + def __getattr__(self, name: str) -> Any: + return getattr(self._shm, name) + + +async def _open_history_snapshot( + edge_func: EdgeFunc, + quote_stream: AsyncIterator[QuoteFrame], + fqme: str, + src_shm: ShmArray, + +) -> tuple[ + FspStream, + FspHistory, + tuple[int, int], + np.ndarray, +]: + ''' + Start an FSP against an immutable source-history snapshot. + + ''' + history_shm: _HistoryShm = _HistoryShm(src_shm) + out_stream: FspStream = edge_func( + filter_quotes_by_sym(fqme, quote_stream), + cast(ShmArray, history_shm), + ) + try: + first_yield: FspYield = await anext(out_stream) + except BaseException: + with trio.CancelScope(shield=True): + await out_stream.aclose() + raise + + if isinstance(first_yield, tuple): + await out_stream.aclose() + raise TypeError( + f'FSP `{edge_func.__name__}` yielded a realtime update ' + f'before its historical output' + ) + + history_output: FspHistory = first_yield + history_shm.live = True + return ( + out_stream, + history_output, + ( + history_shm.first, + history_shm.last_index, + ), + history_shm.history, + ) + # TODO: unifying the abstractions in this FSP subsys/layer: # -[ ] move the `.data.flows.Flume` type into this # module/subsys/pkg? @@ -124,24 +270,33 @@ class Cascade(Struct): fsp: Fsp # UI-side middleware ctl API # filled during cascade/.bind_func() (fsp_compute) init phases - bind_func: Callable | None = None - complete: trio.Event | None = None - cs: trio.CancelScope | None = None - client_stream: tractor.MsgStream | None = None + bind_func: Callable[..., Awaitable[None]]|None = None + complete: trio.Event|None = None + cs: trio.CancelScope|None = None + client_stream: tractor.MsgStream|None = None async def resync(self) -> int: # TODO: adopt an incremental update engine/approach # where possible here eventually! log.info(f're-syncing fsp {self.fsp.name} to source') - self.cs.cancel() - await self.complete.wait() - index: int = await self.tn.start(self.bind_func) + cs: trio.CancelScope|None = self.cs + complete: trio.Event|None = self.complete + bind_func = self.bind_func + client_stream = self.client_stream + assert cs is not None + assert complete is not None + assert bind_func is not None + assert client_stream is not None + + cs.cancel() + await complete.wait() + index: int = await self.tn.start(bind_func) # always trigger UI refresh after history update, # see ``piker.ui._fsp.FspAdmin.open_chain()`` and # ``piker.ui._display.trigger_update()``. dst_shm: ShmArray = self.dst.rt_shm - await self.client_stream.send({ + await client_stream.send({ 'fsp_update': { 'key': dst_shm.token, 'first': dst_shm._first.value, @@ -158,28 +313,49 @@ class Cascade(Struct): ''' src_shm: ShmArray = self.src.rt_shm dst_shm: ShmArray = self.dst.rt_shm - step_diff = src_shm.index - dst_shm.index - len_diff = abs(len(src_shm.array) - len(dst_shm.array)) - synced: bool = not ( - # the source is likely backfilling and we must - # sync history calculations - len_diff > 2 + src_first = src_shm._first.value + src_last = src_shm._last.value + dst_first = dst_shm._first.value + dst_last = dst_shm._last.value - # we aren't step synced to the source and may be - # leading/lagging by a step - or step_diff > 1 - or step_diff < 0 + # Compare absolute bounds instead of ``ShmArray.index``, which + # wraps at the allocation size. A one-step source lead is the + # normal state before the destination sample is appended. + step_diff = src_last - dst_last + len_diff = abs( + (src_last - src_first) + - (dst_last - dst_first) + ) + synced: bool = ( + src_first == dst_first + and + step_diff in (0, 1) ) if not synced: fsp: Fsp = self.fsp - log.warning( - f'***DESYNCED fsp***\n' - f'------------------\n' + report = ( f'ns-path: {fsp.ns_path!r}\n' f'shm-token: {src_shm.token}\n' + f'src-bounds: {(src_first, src_last)}\n' + f'dst-bounds: {(dst_first, dst_last)}\n' f'step_diff: {step_diff}\n' f'len_diff: {len_diff}\n' ) + if ( + src_first < dst_first + and + src_last == dst_last + ): + log.info( + f'FSP source history grew:\n' + f'{report}' + ) + else: + log.warning( + f'***DESYNCED fsp***\n' + f'------------------\n' + f'{report}' + ) return ( synced, step_diff, @@ -187,17 +363,20 @@ class Cascade(Struct): ) async def poll_and_sync_to_step(self) -> int: - synced, step_diff, _ = self.is_synced() - while not synced: + while True: await self.resync() - synced, step_diff, _ = self.is_synced() - - return step_diff + ( + synced, + step_diff, + _, + ) = self.is_synced() + if synced: + return step_diff @acm async def open_edge( self, - bind_func: Callable, + bind_func: Callable[..., Awaitable[None]], ) -> int: self.bind_func = bind_func index = await self.tn.start(bind_func) @@ -206,14 +385,123 @@ class Cascade(Struct): # -[ ] dynamic reconnection after update? +async def _close_fsp_stream(out_stream: FspStream) -> None: + with trio.CancelScope(shield=True): + await out_stream.aclose() + + +def _write_history_output( + dst_shm: ShmArray, + history_output: FspHistory, + src_bounds: tuple[int, int], + src_history: np.ndarray, + func_name: str, + profiler: Profiler, + +) -> int: + src_first, src_last = src_bounds + fields = dst_shm.array.dtype.fields + if fields is None: + raise TypeError('FSP destination SHM must be a structured array') + fields = fields.copy() + fields.pop('index') + fields.pop('time') + history_by_field: np.ndarray|None = None + + if len(fields) > 1: + if not isinstance(history_output, dict): + raise ValueError( + f'`{func_name}` is a multi-output FSP and should ' + f'yield a `dict[str, np.ndarray]` for history' + ) + + output_fields: list[str] = [ + key + for key in fields + if key in history_output + ] + if not output_fields: + raise ValueError( + f'`{func_name}` produced no declared history fields' + ) + + for key in output_fields: + output: np.ndarray|None = history_output[key] + if history_by_field is None: + length: int = ( + len(src_history) + if output is None + else len(output) + ) + + # using the first output, determine the length of the + # struct-array that will be pushed to shm. + history_by_field = np.zeros( + length, + dtype=dst_shm.array.dtype, + ) + + if output is not None: + history_by_field[key] = output + + else: + if not isinstance(history_output, np.ndarray): + raise ValueError( + f'`{func_name}` is a single output FSP and should ' + f'yield an `np.ndarray` for history' + ) + history_by_field = np.zeros( + len(history_output), + dtype=dst_shm.array.dtype, + ) + history_by_field[func_name] = history_output + + if history_by_field is None: + raise ValueError(f'`{func_name}` produced no history fields') + if len(history_by_field) != len(src_history): + raise ValueError( + f'`{func_name}` history length ' + f'{len(history_by_field)} does not match source length ' + f'{len(src_history)}' + ) + + history_by_field['time'] = src_history['time'] + + # TODO: XXX: + # THERE'S A BIG BUG HERE WITH THE `index` field since we're + # prepending a copy of the first value a few times to make + # sub-curves align with the parent bar chart. + # This likely needs to be fixed either by, + # - manually assigning the index and historical data + # seperately to the shm array (i.e. not using .push()) + # - developing some system on top of the shared mem array that + # is `index` aware such that historical data can be indexed + # relative to the true first datum? Not sure if this is sane + # for incremental compuations. + first: int = src_first + dst_shm._first.value = first + + # TODO: can we use this `start` flag instead of the manual + # setting above? + index: int = dst_shm.push( + history_by_field, + start=first, + ) + + profiler(f'{func_name} pushed history') + profiler.finish() + assert index == src_last + return index + + async def connect_streams( casc: Cascade, mkt: MktPair, - quote_stream: trio.abc.ReceiveChannel, + quote_stream: AsyncIterator[QuoteFrame], src: Flume, dst: Flume, - edge_func: Callable, + edge_func: EdgeFunc, # attach_stream: bool = False, task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED, @@ -239,119 +527,38 @@ async def connect_streams( # fqme: str = mkt.fqme fqme: str = src.mkt.fqme - # TODO: dynamic introspection of what the underlying (vertex) - # function actually requires from input node (flumes) then - # deliver those inputs as part of a graph "compilation" step? - out_stream = edge_func( - - # TODO: do we even need this if we do the feed api right? - # shouldn't a local stream do this before we get a handle - # to the async iterable? it's that or we do some kinda - # async itertools style? - filter_quotes_by_sym(fqme, quote_stream), - - # XXX: currently the ``ohlcv`` arg, but we should allow - # (dynamic) requests for src flume (node) streams? - src.rt_shm, - ) - # HISTORY COMPUTE PHASE # conduct a single iteration of fsp with historical bars input # and get historical output. - history_output: ( - dict[str, np.ndarray] # multi-output case - | np.ndarray, # single output case + history_output: FspHistory + src_shm: ShmArray = src.rt_shm + ( + out_stream, + history_output, + src_bounds, + src_history, + ) = await _open_history_snapshot( + edge_func, + quote_stream, + fqme, + src_shm, ) - history_output = await anext(out_stream) - - func_name = edge_func.__name__ + func_name: str = edge_func.__name__ profiler(f'{func_name} generated history') - # build struct array with an 'index' field to push as history - - # TODO: push using a[['f0', 'f1', .., 'fn']] = .. syntax no? - # if the output array is multi-field then push - # each respective field. dst_shm: ShmArray = dst.rt_shm - fields = getattr(dst_shm.array.dtype, 'fields', None).copy() - fields.pop('index') - history_by_field: np.ndarray | None = None - src_shm: ShmArray = src.rt_shm - src_time = src_shm.array['time'] - - if ( - fields and - len(fields) > 1 - ): - if not isinstance(history_output, dict): - raise ValueError( - f'`{func_name}` is a multi-output FSP and should yield a ' - '`dict[str, np.ndarray]` for history' - ) - - for key in fields.keys(): - if key in history_output: - output = history_output[key] - - if history_by_field is None: - - if output is None: - length = len(src_shm.array) - else: - length = len(output) - - # using the first output, determine - # the length of the struct-array that - # will be pushed to shm. - history_by_field = np.zeros( - length, - dtype=dst_shm.array.dtype - ) - - if output is None: - continue - - history_by_field[key] = output - - # single-key output stream - else: - if not isinstance(history_output, np.ndarray): - raise ValueError( - f'`{func_name}` is a single output FSP and should yield an ' - '`np.ndarray` for history' - ) - history_by_field = np.zeros( - len(history_output), - dtype=dst_shm.array.dtype + try: + index: int = _write_history_output( + dst_shm, + history_output, + src_bounds, + src_history, + func_name, + profiler, ) - history_by_field[func_name] = history_output - - history_by_field['time'] = src_time[-len(history_by_field):] - - history_output['time'] = src_shm.array['time'] - - # TODO: XXX: - # THERE'S A BIG BUG HERE WITH THE `index` field since we're - # prepending a copy of the first value a few times to make - # sub-curves align with the parent bar chart. - # This likely needs to be fixed either by, - # - manually assigning the index and historical data - # seperately to the shm array (i.e. not using .push()) - # - developing some system on top of the shared mem array that - # is `index` aware such that historical data can be indexed - # relative to the true first datum? Not sure if this is sane - # for incremental compuations. - first = dst_shm._first.value = src_shm._first.value - - # TODO: can we use this `start` flag instead of the manual - # setting above? - index = dst_shm.push( - history_by_field, - start=first, - ) - - profiler(f'{func_name} pushed history') - profiler.finish() + except BaseException: + await _close_fsp_stream(out_stream) + raise # setup a respawn handle with trio.CancelScope() as cs: @@ -399,7 +606,10 @@ async def connect_streams( # log.info(f'FSP quote too fast: {hz}') # last = time.time() finally: - casc.complete.set() + try: + await _close_fsp_stream(out_stream) + finally: + casc.complete.set() @tractor.context @@ -410,11 +620,13 @@ async def cascade( fqme: str, # flume pair cascaded using an "edge function" - src_flume_addr: dict, - dst_flume_addr: dict, + src_flume_addr: dict[str, Any], + dst_flume_addr: dict[str, Any], ns_path: NamespacePath, - shm_registry: dict[str, NDToken], + shm_registry: list[ + tuple[NDTokenMsg, str, NDTokenMsg] + ], zero_on_step: bool = False, loglevel: str|None = None, @@ -452,7 +664,7 @@ async def cascade( # src: ShmArray = attach_shm_array(token=src_shm_token) # dst: ShmArray = attach_shm_array(readonly=False, token=dst_shm_token) - reg = _load_builtins() + reg: dict[NamespacePath, Fsp] = _load_builtins() lines = '\n'.join([f'{key.rpartition(":")[2]} => {key}' for key in reg]) log.info( f'Registered FSP set:\n{lines}' @@ -465,18 +677,21 @@ async def cascade( # not sure how else to do it. for (token, fsp_name, dst_token) in shm_registry: Fsp._flow_registry[( - NDToken.from_msg(token), + NDToken.from_msg(dict(token)), fsp_name, - )] = NDToken.from_msg(dst_token), None + )] = NDToken.from_msg(dict(dst_token)), None - fsp: Fsp = reg.get( + fsp: Fsp|None = reg.get( NamespacePath(ns_path) ) - func: Callable = fsp.func - - if not func: + if ( + fsp is None + or + not fsp.func + ): # TODO: assume it's a func target path raise ValueError(f'Unknown fsp target: {ns_path}') + func: FspFunc = fsp.func _fqme: str = src.mkt.fqme assert _fqme == fqme @@ -521,7 +736,7 @@ async def cascade( # the target task is spawned implicitly and then the event is # set via some higher level api? At that poing we might as well # be writing a one-cancels-one nursery though right? - casc = Cascade( + casc: Cascade = Cascade( src, dst, tn, @@ -529,7 +744,7 @@ async def cascade( ) # TODO: this seems like it should be wrapped somewhere? - fsp_target = partial( + fsp_target: Callable[..., Awaitable[None]] = partial( connect_streams, casc=casc, mkt=mkt, @@ -543,9 +758,20 @@ async def cascade( # and renders dst flume output(s) edge_func=func ) - async with casc.open_edge( - bind_func=fsp_target, - ) as index: + + # Subscribe before history bootstrap so in-place repairs + # with unchanged SHM bounds can not lose invalidations. + period_s: float = src.rt_sample_period_s + + async with ( + open_sample_stream( + period_s=period_s, + loglevel=loglevel, + ) as istream, + casc.open_edge( + bind_func=fsp_target, + ) as index, + ): # casc.bind_func = fsp_target # index = await tn.start(fsp_target) dst_shm: ShmArray = dst.rt_shm @@ -567,72 +793,82 @@ async def cascade( async with ctx.open_stream() as client_stream: casc.client_stream: tractor.MsgStream = client_stream - s, step, ld = casc.is_synced() + profiler(f'{func_name}: sample stream up') + profiler.finish() - # detect sample period step for subscription to increment - # signal - times = src.rt_shm.array['time'] - if len(times) > 1: - last_ts = times[-1] - delay_s: float = float(last_ts - times[times != last_ts][-1]) - else: - # our default "HFT" sample rate. - delay_s: float = _default_delay_s + async for sample_msg in istream: + # print(f'FSP incrementing {sample_msg}') + msg: SampleMsg = cast( + SampleMsg, + sample_msg, + ) - # sub and increment the underlying shared memory buffer - # on every step msg received from the global `samplerd` - # service. - async with open_sample_stream( - period_s=float(delay_s), - loglevel=loglevel, - ) as istream: + # ``samplerd`` broadcasts history events for + # every feed and period. Only this cascade's + # source can invalidate its destination. + if not _is_relevant_sample_msg( + msg, + fqme, + period_s, + ): + continue - profiler(f'{func_name}: sample stream up') - profiler.finish() + # Respawn the compute task after source history + # revisions or an actual bounds desync. + ( + synced, + step_diff, + _, + ) = casc.is_synced() + if _needs_history_resync( + msg, + synced, + ): + step_diff = ( + await casc.poll_and_sync_to_step() + ) - async for i in istream: - # print(f'FSP incrementing {i}') + # Backfill broadcasts only announce source + # history revisions. They never represent a + # new sample row. Duplicate sample wakeups are + # likewise non-advancing when already aligned. + if not _should_advance_dst( + msg, + step_diff, + ): + continue - # respawn the compute task if the source - # array has been updated such that we compute - # new history from the (prepended) source. - synced, step_diff, _ = casc.is_synced() - if not synced: - step_diff: int = await casc.poll_and_sync_to_step() + array = dst_shm.array - # skip adding a last bar since we should already - # be step alinged - if step_diff == 0: - continue + # some metrics like vlm should be reset + # to zero every step. + if zero_on_step: + last = zeroed + else: + last = array[-1:].copy() - # read out last shm row, copy and write new row - array = dst_shm.array + dst.rt_shm.push(last) - # some metrics like vlm should be reset - # to zero every step. - if zero_on_step: - last = zeroed - else: - last = array[-1:].copy() + # sync with source buffer's time step + src_l2 = src_shm.array[-2:] + src_li, src_lt = src_l2[-1][ + ['index', 'time'] + ] + src_2li, src_2lt = src_l2[-2][ + ['index', 'time'] + ] + dst_shm._array['time'][src_li] = src_lt + dst_shm._array['time'][src_2li] = src_2lt - dst.rt_shm.push(last) - - # sync with source buffer's time step - src_l2 = src_shm.array[-2:] - src_li, src_lt = src_l2[-1][['index', 'time']] - src_2li, src_2lt = src_l2[-2][['index', 'time']] - dst_shm._array['time'][src_li] = src_lt - dst_shm._array['time'][src_2li] = src_2lt - - # last2 = dst.array[-2:] - # if ( - # last2[-1]['index'] != src_li - # or last2[-2]['index'] != src_2li - # ): - # dstl2 = list(last2) - # srcl2 = list(src_l2) - # print( - # # f'{dst.token}\n' - # f'src: {srcl2}\n' - # f'dst: {dstl2}\n' - # ) + # last2 = dst.array[-2:] + # if ( + # last2[-1]['index'] != src_li + # or last2[-2]['index'] != src_2li + # ): + # dstl2 = list(last2) + # srcl2 = list(src_l2) + # print( + # # f'{dst.token}\n' + # f'src: {srcl2}\n' + # f'dst: {dstl2}\n' + # ) diff --git a/tests/test_fsp_sync.py b/tests/test_fsp_sync.py index df151e83..cfdedc1b 100644 --- a/tests/test_fsp_sync.py +++ b/tests/test_fsp_sync.py @@ -3,12 +3,22 @@ FSP history synchronization regressions. ''' from collections.abc import AsyncIterator +from types import SimpleNamespace from typing import cast import numpy as np import pytest import trio +from piker.fsp._engine import ( + _is_relevant_sample_msg, + _needs_history_resync, + _open_history_snapshot, + _should_advance_dst, + Cascade, + EdgeFunc, + QuoteFrame, +) from piker.fsp._momo import ( rsi, wma, @@ -16,14 +26,56 @@ from piker.fsp._momo import ( from piker.fsp._volume import ( tina_vwap, ) +from piker.accounting import MktPair +from piker.data.flows import Flume from piker.data._sharedmem import NDTokenMsg from piker.data.ticktools import FeedQuote from piker.fsp._api import Fsp +from piker.ui._chart import LinkedSplits +from piker.ui._fsp import update_fsp_vizs from tractor.ipc._shm import ( NDToken, ShmArray, ) +type SampleMsg = dict[ + str, + int|float|tuple[str, float], +] + + +def test_sample_period_ignores_market_closure_gap() -> None: + ''' + Keep FSP cascades subscribed to their regular sampler period. + + Starting a cascade immediately after a market closure left the + source SHM tail with a large gap between its last two distinct + timestamps. Using only that pair subscribed the cascade to the gap + duration instead of the regular one-second stream. Realtime FSP + writes then repeatedly replaced one row without advancing its SHM + bound. Give both source SHMs closure-gap tail deltas and prove + `Flume.get_ds_info()` returns its declared one- and 60-second + periods instead of deriving cadence from those rows. + + ''' + rt_shm = OhlcvShm(length=4) + hist_shm = OhlcvShm(length=4) + rt_shm._array['time'][:4] = [100, 101, 102, 3600] + hist_shm._array['time'][:4] = [100, 160, 220, 3600] + flume = Flume( + mkt=cast(MktPair, SimpleNamespace()), + first_quote={}, + _rt_shm_token=rt_shm._token, + _hist_shm_token=hist_shm._token, + ) + flume._rt_shm = cast(ShmArray, rt_shm) + flume._hist_shm = cast(ShmArray, hist_shm) + + assert np.diff(rt_shm.array['time'])[-1] == 3498 + assert np.diff(hist_shm.array['time'])[-1] == 3380 + assert flume.get_ds_info() == (1, 60, 60) + + class Value: def __init__(self, value: int) -> None: self.value: int = value @@ -96,6 +148,236 @@ class OhlcvShm(Shm): ) +class FspShm(Shm): + def __init__( + self, + first: int = 0, + last: int = 4, + token: str = 'fsp', + + ) -> None: + dtype = np.dtype([ + ('index', ' Cascade: + fsp: Fsp = cast( + Fsp, + SimpleNamespace( + name='test_fsp', + ns_path='tests:test_fsp', + ), + ) + src: Flume = cast( + Flume, + SimpleNamespace( + rt_shm=Shm(*src_bounds, token='src'), + ), + ) + dst: Flume = cast( + Flume, + SimpleNamespace( + rt_shm=Shm(*dst_bounds, token='dst'), + ), + ) + nursery: trio.Nursery = cast(trio.Nursery, None) + return Cascade(src, dst, nursery, fsp) + + +@pytest.mark.parametrize( + 'src_bounds,dst_bounds,expected', + [ + ((10, 20), (10, 20), (True, 0, 0)), + ((10, 21), (10, 20), (True, 1, 1)), + ((0, 3000), (2000, 3000), (False, 0, 2000)), + ((9, 20), (10, 20), (False, 0, 1)), + ((8, 20), (10, 20), (False, 0, 2)), + ((10, 20), (9, 20), (False, 0, 1)), + ((11, 21), (10, 20), (False, 1, 0)), + ((10, 22), (10, 20), (False, 2, 2)), + ((10, 19), (10, 20), (False, -1, 1)), + ], +) +def test_cascade_sync_uses_absolute_bounds( + src_bounds: tuple[int, int], + dst_bounds: tuple[int, int], + expected: tuple[bool, int, int], + +) -> None: + ''' + Detect every history-bound mismatch without modulo-index skew. + + Source prepends move only its first bound, while a normal realtime + step moves only its last bound. The former requires a historical + recompute even for one or two rows; the latter permits exactly one + destination append. This matrix also shifts equal-length bounds and + puts the destination ahead to prove neither state is accepted. + + ''' + cascade = mk_cascade(src_bounds, dst_bounds) + + assert cascade.is_synced() == expected + + +@pytest.mark.parametrize( + 'msg,expected', + [ + ({'index': 1}, True), + ({'backfilling': ('tsla.nasdaq.ib', 1)}, True), + ({'backfilling': ('btcusdt.binance', 1)}, False), + ({'backfilling': ('tsla.nasdaq.ib', 60)}, False), + ], +) +def test_cascade_filters_foreign_backfill_events( + msg: SampleMsg, + expected: bool, + +) -> None: + ''' + Ignore backfill wakeups from overlay markets and other periods. + + Samplerd broadcasts every backfill marker to every period and FSP + subscriber. With multiple markets overlaid, an unrelated provider + frame used to wake each primary-market cascade and race its own SHM + inspection. Feed regular sample steps and matching history events + through, but reject both a foreign FQME and the 60-second period. + + ''' + assert _is_relevant_sample_msg( + msg, + fqme='tsla.nasdaq.ib', + period_s=1, + ) is expected + + +@pytest.mark.parametrize( + 'msg,step_diff,expected', + [ + ({'index': 1}, 1, True), + ({'index': 1}, 0, False), + ({'backfilling': ('tsla.nasdaq.ib', 1)}, 1, False), + ({'backfilling': ('tsla.nasdaq.ib', 1)}, 0, False), + ], +) +def test_only_new_sample_steps_advance_destination( + msg: SampleMsg, + step_diff: int, + expected: bool, + +) -> None: + ''' + Keep queued history and duplicate wakeups from adding FSP rows. + + A recomputation can consume several prepends before their sampler + markers are delivered. Those stale markers then observe aligned SHM + bounds and previously appended duplicate destination rows. Model the + queued marker and regular duplicate cases, and prove only a genuine + one-step source lead authorizes destination advancement. + + ''' + assert _should_advance_dst(msg, step_diff) is expected + + +@pytest.mark.parametrize( + 'msg,synced,expected', + [ + ({'index': 1}, True, False), + ({'index': 1}, False, True), + ({'backfilling': ('tsla.nasdaq.ib', 1)}, True, True), + ({'backfilling': ('tsla.nasdaq.ib', 1)}, False, True), + ], +) +def test_backfill_always_invalidates_fsp_history( + msg: SampleMsg, + synced: bool, + expected: bool, + +) -> None: + ''' + Recompute in-place history repairs with unchanged SHM bounds. + + Gap repair can replace null source rows without moving either SHM + bound. A bounds-only predicate therefore reports synchronization + even though derived values are stale. Prove every relevant backfill + marker forces history replay while an aligned sample event does not. + + ''' + assert _needs_history_resync(msg, synced) is expected + + +def test_history_compute_uses_immutable_source_snapshot() -> None: + ''' + Do not publish output and timestamps from different source ranges. + + A source append or prepend can occur while the FSP's first yield is + computing. The old path read timestamps only afterward and could pair + an N-row result with a shifted N-row timestamp tail. Mutate the live + last bound during the first yield, then prove the accepted output and + timestamps retain the original range while later reads switch to live + SHM. + + ''' + shm = Shm(10, 20, token='source') + live_lengths: list[int] = [] + + async def edge( + _source: AsyncIterator[FeedQuote], + src_shm: ShmArray, + + ) -> AsyncIterator[np.ndarray]: + shm._last.value += 1 + yield np.ones(len(src_shm.array)) + live_lengths.append(len(src_shm.array)) + + async def main() -> None: + ( + out_stream, + history, + bounds, + src_history, + ) = await _open_history_snapshot( + cast(EdgeFunc, edge), + quote_stream=cast( + AsyncIterator[QuoteFrame], + object(), + ), + fqme='tsla.nasdaq.ib', + src_shm=cast(ShmArray, shm), + ) + assert bounds == (10, 20) + assert len(history) == 10 + assert len(src_history) == 10 + with pytest.raises(StopAsyncIteration): + await anext(out_stream) + + trio.run(main) + + assert live_lengths == [11] + + @pytest.mark.parametrize('target', [wma, rsi, tina_vwap]) def test_builtin_fsp_stream_contract(target: Fsp) -> None: ''' @@ -131,3 +413,98 @@ def test_builtin_fsp_stream_contract(target: Fsp) -> None: await stream.aclose() trio.run(main) + + +class Viz: + def __init__(self, name: str, shm: Shm) -> None: + self.name: str = name + self.shm: Shm = shm + self.updates: int = 0 + self.force_redraws: list[bool] = [] + self._last_fsp_update_sig: ( + tuple[int, int, bytes]|None + ) = None + self._mxmns: dict[ + tuple[int, int], + tuple[int, int], + ] = {(10, 20): (1, 2)} + self.view = SimpleNamespace( + rescale_count=0, + ) + + def rescale( + *, + do_linked_charts: bool, + do_overlay_scaling: bool, + + ) -> None: + assert not do_linked_charts + assert do_overlay_scaling + self.view.rescale_count += 1 + + self.view.interact_graphics_cycle = rescale + self.plot: SimpleNamespace = SimpleNamespace( + getAxis=lambda name: SimpleNamespace(_stickies={}), + vb=self.view, + ) + + def update_graphics( + self, + force_redraw: bool = False, + + ) -> None: + self.updates += 1 + self.force_redraws.append(force_redraw) + + +def test_fsp_history_update_redraws_only_derived_vizs() -> None: + ''' + Keep a source-history prepend from refreshing the whole chart. + + FSP cascades recompute after each near-term backfill frame. The old + notification called the linked chart's full graphics cycle, which + redrew the primary market and every overlay and disturbed the live + view repeatedly. Arrange two visualizations sharing the rebuilt FSP + SHM plus unrelated source and overlay SHMs, then prove only the two + derived curves receive an update. + + ''' + fsp_shm = FspShm(10, 20, token='derived') + source_viz = Viz('source', Shm(10, 20, token='source')) + first_fsp_viz = Viz('flow', fsp_shm) + second_fsp_viz = Viz('dark_flow', fsp_shm) + overlay_viz = Viz('overlay', Shm(10, 20, token='overlay')) + + linked: LinkedSplits = cast( + LinkedSplits, + SimpleNamespace( + chart=SimpleNamespace( + _vizs={ + 'source': source_viz, + 'overlay': overlay_viz, + }, + ), + subplots={ + 'volume': SimpleNamespace( + _vizs={ + 'flow': first_fsp_viz, + 'dark_flow': second_fsp_viz, + }, + ), + }, + ), + ) + + updated = update_fsp_vizs(linked, fsp_shm._token) + + assert updated == 2 + assert first_fsp_viz.updates == 1 + assert second_fsp_viz.updates == 1 + assert first_fsp_viz.force_redraws == [True] + assert second_fsp_viz.force_redraws == [True] + assert first_fsp_viz._mxmns == {} + assert second_fsp_viz._mxmns == {} + assert first_fsp_viz.view.rescale_count == 1 + assert second_fsp_viz.view.rescale_count == 1 + assert source_viz.updates == 0 + assert overlay_viz.updates == 0