diff --git a/ai/prompt-io/opencode/20260805T195001Z_f0f3c806_prompt_io.md b/ai/prompt-io/opencode/20260805T195001Z_f0f3c806_prompt_io.md new file mode 100644 index 00000000..5c511214 --- /dev/null +++ b/ai/prompt-io/opencode/20260805T195001Z_f0f3c806_prompt_io.md @@ -0,0 +1,33 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +session: ses_0799212ebffe42arY96czXn89F +timestamp: 2026-08-05T19:50:01Z +git_ref: f0f3c806 +scope: code +substantive: true +raw_file: 20260805T195001Z_f0f3c806_prompt_io.raw.md +--- + +## Prompt + +Inspect the completed NVDA 60s SHM backfill before workspace restart, explain +the persistent zero region, and continue fixing the indexing logic so sparse +IB history does not leave physical null reservations. + +## Response summary + +Captured a 42,720-row live SHM null span whose valid margins shared the same +timestamp. Replaced wall-clock row reservation with actual provider-boundary +placement, fixed stale SHM first-index, endpoint insertion and field-map +semantics, published the final storage prepend, and added a full multi-frame +`tsdb_backfill()` regression using a real `ShmArray`. + +## Files changed + +- `piker/tsp/_history.py` - contiguous provider and NativeDB prepends. +- `tests/test_history_backfill.py` - sparse multi-frame SHM regression. + +## Human edits + +None - generated output incorporates the user's live observations. diff --git a/ai/prompt-io/opencode/20260805T195001Z_f0f3c806_prompt_io.raw.md b/ai/prompt-io/opencode/20260805T195001Z_f0f3c806_prompt_io.raw.md new file mode 100644 index 00000000..dc14023e --- /dev/null +++ b/ai/prompt-io/opencode/20260805T195001Z_f0f3c806_prompt_io.raw.md @@ -0,0 +1,71 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +timestamp: 2026-08-05T19:50:01Z +git_ref: f0f3c806 +diff_cmd: git diff HEAD~1..HEAD +--- + +The user reported that a completed NVDA 60s backfill left a large zero +region in live SHM and asked for immediate inspection before restarting the +workspace. The captured buffer contained 42,720 invalid rows at absolute +indexes `3065895..3108614`. The valid rows on both sides had the identical +timestamp `1781271360`, proving the region was an erroneous physical-index +reservation rather than a real temporal gap. + +The user then asked the agent to continue fixing the backfill logic. + +> `git diff HEAD~1..HEAD -- piker/tsp/_history.py` + +Removed wall-clock-based SHM reservation from `tsdb_backfill()`. Reverse +provider frames now extend the visible `ShmArray._first` boundary on every +prepend, beginning exactly at the current first index after inclusive query +endpoints are removed. Added `_prepend_tsdb_history()` to trim storage rows +against the actual earliest provider timestamp and prepend them directly +adjacent only after reverse provider retrieval completes. + +This preserves sparse venue history as ordinal samples rather than treating +every elapsed calendar period as a physical SHM row. It also prevents the +old `_first + 1` insertion point from overwriting the earliest published bar. + +> `git diff HEAD~1..HEAD -- tests/test_history_backfill.py` + +Added a full `tsdb_backfill()` regression modeled on the live NVDA failure. +It uses a real `ShmArray`, NativeDB's field map, two inclusive reverse-provider +frames, and overlapping stored history. Assertions prove the complete +timestamp sequence is contiguous and lossless, contains no non-positive +timestamps, sends complete provider frames to durable storage, and notifies +chart/FSP consumers after the final storage prepend. + +Live evidence captured before workspace restart: + +```text +rows 1575296 first_index 1568233 last_index 3143528 +invalid_count 42720 +left boundary +[(3065893, 1781271300, 204.03, 204.82) + (3065894, 1781271360, 204.81, 204.72) + (3065895, 0, 0. , 0. )] +right boundary +[(3108614, 0, 0. , 0. ) + (3108615, 1781271360, 204.81, 205.44) + (3108616, 1781271420, 205.46, 205.48)] +``` + +Verification output: + +```text +........................................................................ [ 68%] +................................. [100%] +105 passed in 4.80s +``` + +Adversarial review caught stale `_first` semantics and an off-by-one +insertion point in the initial patch. The first live restart then exposed an +undefined `time_key` local at the new `tsdb_backfill()` call site, which +crashed `datad.ib` and propagated to the chart as `RemoteActorError`. The +helper now derives the source timestamp field from its field map, and the +test enters the full enclosing lifecycle so the original `NameError` can not +recur unnoticed. A final review also drove explicit publication after stored +history is prepended and deterministic final-stage synchronization. Final +review found no remaining issues; another live restart remains pending. diff --git a/piker/tsp/_history.py b/piker/tsp/_history.py index f9edc727..06ef76eb 100644 --- a/piker/tsp/_history.py +++ b/piker/tsp/_history.py @@ -260,7 +260,7 @@ async def shm_push_in_between( prepend_index: int, backfill_until_dt: datetime, - update_start_on_prepend: bool = False, + update_start_on_prepend: bool = True, ) -> int: @@ -282,16 +282,12 @@ async def shm_push_in_between( to_push, prepend=True, - # XXX: only update the ._first index if no tsdb - # segment was previously prepended by the - # parent task. + # Keep the readable `ShmArray.array` boundary aligned with + # every completed provider prepend. update_first=update_start_on_prepend, - # XXX: only prepend from a manually calculated shm - # index if there was already a tsdb history - # segment prepended (since then the - # ._first.value is going to be wayyy in the - # past!) + # An explicit index is only needed by callers retaining a + # separately placed segment before the readable boundary. start=( prepend_index if not update_start_on_prepend @@ -558,7 +554,7 @@ async def start_backfill( task_status.started() # based on the sample step size, maybe load a certain amount history - update_start_on_prepend: bool = False + update_start_on_prepend: bool = True if ( _until_was_none := (backfill_until_dt is None) ): @@ -579,8 +575,6 @@ async def start_backfill( 60: {'years': 6}, } period_duration: int = periods[timeframe] - update_start_on_prepend: bool = True - # NOTE: manually set the "latest" datetime which we intend to # backfill history "until" so as to adhere to the history # settings above when the tsdb is detected as being empty. @@ -1164,6 +1158,74 @@ async def load_tsdb_hist( return None +def _prepend_tsdb_history( + shm: ShmArray, + tsdb_history: np.ndarray, + field_map: bidict|None, + +) -> np.ndarray: + ''' + Prepend stored rows beside the actual provider boundary. + + ''' + available: int = max(0, int(shm._first.value)) + if ( + not available + or + not len(tsdb_history) + ): + return tsdb_history[:0] + + frame: np.ndarray = shm.array + if not len(frame): + return tsdb_history[:0] + + # XXX EDGE CASEs: the most recent provider frame can overlap + # prior TSDB history when its start is earlier than the latest + # stored sample. + # + # Alternatively, this can occur when the venue was closed (say + # over a weekend), causing a wall-clock time-series gap while the + # provider's query frame is larger than the current market's + # operating session. For example, IB's 1s default returns around + # 2k datums (~33.33m), which can include samples from before a + # closure. The wall-clock interval between frames then differs + # from the number of actual venue samples and may even look like + # a negative overlap when compared only by endpoints. + # + # The previous placement converted that elapsed interval into a + # physical SHM offset. Sparse venue series contain samples, not + # one row per wall-clock period, so closures accumulated as null + # reservations between otherwise adjacent data. Instead, wait + # until reverse provider retrieval establishes the actual first + # sample, retain its values for any overlap, and prepend only TSDB + # samples older than that boundary. + # + # TODO: assert overlapping TSDB and provider segments contain the + # same values before preferring the provider's copy. + time_key: str = ( + field_map.inverse.get('time', 'time') + if field_map + else 'time' + ) + earliest_t: float = frame['time'][0] + older: np.ndarray = tsdb_history[ + tsdb_history[time_key] < earliest_t + ] + + # Stored history may extend further into the past than remaining + # SHM capacity. Keep only its most recent rows in that case. + to_push: np.ndarray = older[-available:] + if len(to_push): + shm.push( + to_push, + prepend=True, + field_map=field_map, + ) + + return to_push + + async def tsdb_backfill( mod: ModuleType, storemod: ModuleType, @@ -1249,7 +1311,7 @@ async def tsdb_backfill( # NOTE: iabs to start backfilling from, reverse chronological, # ONLY AFTER the first history frame has been pushed to # mem! - backfill_gap_from_shm_index: int = shm._first.value + 1 + backfill_gap_from_shm_index: int = shm._first.value # Prepend any tsdb history into the rt-shm-buffer which # should NOW be getting filled with the most recent history @@ -1338,76 +1400,29 @@ async def tsdb_backfill( write_tsdb=True, ) ) - if last_tsdb_dt is not None: - - # calc the index from which the tsdb data should be - # prepended, presuming there is a gap between the - # latest frame (loaded/read above) and the latest - # sample loaded from the tsdb. - backfill_diff: Duration = mr_start_dt - last_tsdb_dt - offset_s: float = backfill_diff.in_seconds() - - # XXX EDGE CASEs: the most recent frame overlaps with - # prior tsdb history!! - # - so the latest frame's start time is earlier then - # the tsdb's latest sample. - # - alternatively this may also more generally occur - # when the venue was closed (say over the weeknd) - # causing a timeseries gap, AND the query frames size - # (eg. for ib's 1s we rx 2k datums ~= 33.33m) IS - # GREATER THAN the current venue-market's operating - # session (time) we will receive datums from BEFORE THE - # CLOSURE GAP and thus the `offset_s` value will be - # NEGATIVE! In this case we need to ensure we don't try - # to push datums that have already been recorded in the - # tsdb. In this case we instead only retreive and push - # the series portion missing from the db's data set. - # if offset_s < 0: - # non_overlap_diff: Duration = mr_end_dt - last_tsdb_dt - # non_overlap_offset_s: float = backfill_diff.in_seconds() - - offset_samples: int = round(offset_s / timeframe) - - # TODO: see if there's faster multi-field reads: - # https://numpy.org/doc/stable/user/basics.rec.html#accessing-multiple-fields - # re-index with a `time` and index field - if offset_s > 0: - # NOTE XXX: ONLY when there is an actual gap - # between the earliest sample in the latest history - # frame do we want to NOT stick the latest tsdb - # history adjacent to that latest frame! - prepend_start = shm._first.value - offset_samples + 1 - to_push = tsdb_history[-prepend_start:] - else: - # when there is overlap we want to remove the - # overlapping samples from the tsdb portion (taking - # instead the latest frame's values since THEY - # SHOULD BE THE SAME) and prepend DIRECTLY adjacent - # to the latest frame! - # TODO: assert the overlap segment array contains - # the same values!?! - prepend_start = shm._first.value - to_push = tsdb_history[-(shm._first.value):offset_samples - 1] - - # tsdb history is so far in the past we can't fit it in - # shm buffer space so simply don't load it! - if prepend_start > 0: - shm.push( - to_push, - - # insert the history pre a "days worth" of samples - # to leave some real-time buffer space at the end. - prepend=True, - # update_first=False, - start=prepend_start, - field_map=storemod.ohlc_key_map, - ) - - log.info(f'Loaded {to_push.shape} datums from storage') - # 2nd nursery END if last_tsdb_dt is not None: + # Provider frames contain venue samples, not one row + # per wall-clock period. Wait until reverse retrieval + # establishes the actual earliest sample, then prepend + # storage directly beside it without reserving closure + # rows. + to_push: np.ndarray = _prepend_tsdb_history( + shm, + tsdb_history, + field_map=storemod.ohlc_key_map, + ) + log.info( + f'Loaded {to_push.shape} datums from storage' + ) + if len(to_push): + await notify_backfill( + sampler_stream, + mkt, + timeframe, + ) + # Repair only after reverse backfill releases providers # such as IB which permit one history request at a time. # TODO: ideally these null segments can never exist! diff --git a/tests/test_history_backfill.py b/tests/test_history_backfill.py index 099fdc6b..2255f585 100644 --- a/tests/test_history_backfill.py +++ b/tests/test_history_backfill.py @@ -2,6 +2,7 @@ Deterministic history-backfill regressions. ''' +from contextlib import asynccontextmanager from functools import partial from pathlib import Path from types import SimpleNamespace @@ -34,6 +35,7 @@ from piker.tsp._history import ( notify_backfill, publish_latest_frame, start_backfill, + tsdb_backfill, ) @@ -216,6 +218,197 @@ def test_latest_frame_is_persisted_before_shm() -> None: assert events == ['storage', 'shm'] +def test_tsdb_prepend_uses_actual_provider_boundary( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Sparse venue history must not reserve wall-clock rows in SHM. + + NVDA's 60s startup converted elapsed calendar minutes between + NativeDB and the latest IB frame into physical SHM offsets. IB + returned only trading bars, leaving 42,720 zero rows even though + both valid margins ended at the same timestamp. Model completed + reverse retrieval with a provider boundary overlapping NativeDB. + Exercise real ``ShmArray.push()`` indexing and prove the stored, + reverse-provider, and latest frames form one contiguous lossless + sequence without an explicit wall-clock offset. + + ''' + stored = np.zeros( + 3, + dtype=np.dtype(def_iohlcv_fields), + ) + stored['time'] = [60, 120, 180] + older_reverse = np.zeros( + 3, + dtype=np.dtype(def_iohlcv_fields), + ) + older_reverse['time'] = [180, 240, 300] + newer_reverse = np.zeros( + 3, + dtype=np.dtype(def_iohlcv_fields), + ) + newer_reverse['time'] = [300, 360, 420] + latest = np.zeros( + 2, + dtype=np.dtype(def_iohlcv_fields), + ) + latest['time'] = [420, 480] + storage_frames: list[np.ndarray] = [] + notifications: list[dict] = [] + startup_done = trio.Event() + + class Storage: + ''' + Capture the complete provider delta written to NativeDB. + + ''' + async def update_ohlcv( + self, + fqme: str, + ohlcv: np.ndarray, + timeframe: int, + ) -> None: + ''' + Record one reverse provider response. + + ''' + storage_frames.append(ohlcv.copy()) + + async def load( + self, + fqme: str, + timeframe: int, + ) -> tuple: + ''' + Return overlapping NativeDB history and its boundaries. + + ''' + return ( + stored, + from_timestamp(60), + from_timestamp(180), + ) + + class Sampler: + ''' + Accept deterministic backfill notifications. + + ''' + async def send(self, msg) -> None: + ''' + Capture notification without actor IPC. + + ''' + notifications.append(msg) + + mkt = SimpleNamespace( + fqme='nvda.nasdaq.ib', + dst=SimpleNamespace(atype='stock'), + src=SimpleNamespace(atype='fiat'), + get_fqme=lambda **kwargs: 'nvda.nasdaq.ib', + ) + + async def get_hist(*args, **kwargs): + ''' + Return inclusive reverse frames overlapping each SHM + boundary. + + ''' + end_dt = kwargs['end_dt'] + if end_dt is None: + return ( + latest, + from_timestamp(420), + from_timestamp(480), + ) + + end_t: float = end_dt.timestamp() + reverse: np.ndarray = ( + newer_reverse + if end_t == 420 + else older_reverse + ) + return ( + reverse, + from_timestamp(reverse['time'][0]), + from_timestamp(reverse['time'][-1]), + ) + + @asynccontextmanager + async def open_history_client(mkt): + ''' + Yield the deterministic provider and default frame config. + + ''' + yield get_hist, {} + + async def finish_null_repair(**kwargs) -> None: + ''' + Signal that provider and NativeDB prepends both completed. + + ''' + startup_done.set() + + monkeypatch.setattr( + 'piker.tsp._history.maybe_fill_null_segments', + finish_null_repair, + ) + + async def main() -> None: + shm_key = f'test_sparse_prepend_{uuid4().hex}' + async with tractor.open_root_actor( + name=shm_key, + tpt_bind_addrs=[('127.0.0.1', 0)], + ): + shm, opened = maybe_open_shm_array( + key=shm_key, + size=16, + dtype=np.dtype(def_iohlcv_fields), + append_start_index=12, + ) + assert opened + storage = Storage() + async with trio.open_nursery() as nursery: + nursery.start_soon(partial( + tsdb_backfill, + mod=SimpleNamespace( + name='fake', + open_history_client=open_history_client, + ), + storemod=SimpleNamespace( + ohlc_key_map=ohlc_key_map, + ), + storage=storage, + mkt=mkt, + shm=shm, + timeframe=60, + sampler_stream=Sampler(), + )) + with trio.fail_after(0.5): + await startup_done.wait() + nursery.cancel_scope.cancel() + + assert shm.array['time'].tolist() == [ + 60, + 120, + 180, + 240, + 300, + 360, + 420, + 480, + ] + assert not np.any(shm.array['time'] <= 0) + + trio.run(main) + assert len(storage_frames) == 3 + assert storage_frames[0]['time'].tolist() == [420, 480] + assert storage_frames[1]['time'].tolist() == [300, 360, 420] + assert storage_frames[2]['time'].tolist() == [180, 240, 300] + assert len(notifications) == 3 + + def test_ib_latest_frame_round_trips_through_nativedb( tmp_path: Path, ) -> None: