pikers
/
piker
4
1
Fork 0
Code Issues 20 Pull Requests 12 Packages Projects Releases Wiki Activity

Fix sparse IB history prepends in `ShmArray` 

IB returns trading samples, not one row per wall-clock period.
Converting elapsed time to a physical prepend offset left 42,720
zero rows in NVDA's 60s buffer even though both valid margins
shared the same timestamp.

Use the actual earliest provider sample as the NativeDB boundary.
After reverse retrieval, prepend older stored rows directly beside
it.

Deats,
- start reverse insertion at the current first index;
- update `ShmArray._first` after every provider prepend;
- trim provider/NativeDB overlap by its timestamp;
- publish the final storage prepend to chart and FSP consumers;
- derive the source `time` field from `ohlc_key_map`;
- cover the full lifecycle with a real multi-frame `ShmArray`.

Prompt-IO: ai/prompt-io/opencode/20260805T195001Z_f0f3c806_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
ib_methproxy_refinery
Gud Boi 2026-08-05 19:40:07 -04:00
parent f0f3c80657
commit 06d5ea5dc0
4 changed files with 392 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
ai/prompt-io/opencode/20260805T195001Z_f0f3c806_prompt_io.md 100644
View File

@ -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.

71
ai/prompt-io/opencode/20260805T195001Z_f0f3c806_prompt_io.raw.md 100644
View File

@ -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.

175
piker/tsp/_history.py
View File

@ -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!

193
tests/test_history_backfill.py
View File

@ -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: