From 19c6d98e6d2b4c943600fd073541397226ba9439 Mon Sep 17 00:00:00 2001 From: goodboy Date: Mon, 27 Jul 2026 18:08:33 -0400 Subject: [PATCH] Normalize provider OHLCV for `NativeDB` IB history frames omit the derived `index` and append `count`. Normalize provider data to durable names, order, and dtypes before merge, and reject fractional epoch truncation. Also, - map canonical storage fields into provider-specific SHM on reload - select Polars columns by name instead of position - exercise first-start and restart through actor, SHM, and Parquet Prompt-IO: ai/prompt-io/opencode/20260727T212454Z_689df816_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- .../20260727T212454Z_689df816_prompt_io.md | 42 ++++++ ...20260727T212454Z_689df816_prompt_io.raw.md | 34 +++++ piker/storage/nativedb.py | 69 +++++++++- piker/tsp/_anal.py | 7 +- tests/test_history_backfill.py | 125 ++++++++++++++++++ tests/test_storage_nativedb.py | 60 +++++++++ 6 files changed, 327 insertions(+), 10 deletions(-) create mode 100644 ai/prompt-io/opencode/20260727T212454Z_689df816_prompt_io.md create mode 100644 ai/prompt-io/opencode/20260727T212454Z_689df816_prompt_io.raw.md diff --git a/ai/prompt-io/opencode/20260727T212454Z_689df816_prompt_io.md b/ai/prompt-io/opencode/20260727T212454Z_689df816_prompt_io.md new file mode 100644 index 00000000..151fa195 --- /dev/null +++ b/ai/prompt-io/opencode/20260727T212454Z_689df816_prompt_io.md @@ -0,0 +1,42 @@ +--- +model: gpt-5.6-sol +provider: openai +service: opencode +session: ses_0799212ebffe42arY96czXn89F +timestamp: 2026-07-27T21:24:54Z +git_ref: 689df816 +scope: code +substantive: true +raw_file: 20260727T212454Z_689df816_prompt_io.raw.md +--- + +## Prompt + +The user reran the known IB FQME on `backfiller_deep_fixes` and reported +that `publish_latest_frame()` still failed because NativeDB required an +`index` absent from IB's provider frame. The user called out the missed +tabular schema boundary and requested an actual end-to-end test suite +rather than storage fakes. + +## Response summary + +Repaired the IB-provider-to-NativeDB boundary by canonicalizing durable +fields and mapping reloads into provider-specific SHM buffers. Added a +local actor/SHM/Parquet integration regression covering first startup and +restart, plus focused conversion and timestamp-coercion tests. Two review +rounds found and resolved restart hydration and fractional-time hazards. + +## Files changed + +- `piker/storage/nativedb.py` - canonical storage and SHM field mapping +- `piker/tsp/_anal.py` - name-based Polars-to-NumPy conversion +- `tests/test_history_backfill.py` - actor/SHM/Parquet integration test +- `tests/test_storage_nativedb.py` - schema conversion regressions +- `ai/prompt-io/opencode/20260727T212454Z_689df816_prompt_io.raw.md` + - unedited response record +- `ai/prompt-io/opencode/20260727T212454Z_689df816_prompt_io.md` + - provenance metadata and response summary + +## Human edits + +None - generated changes have not been edited by the human. diff --git a/ai/prompt-io/opencode/20260727T212454Z_689df816_prompt_io.raw.md b/ai/prompt-io/opencode/20260727T212454Z_689df816_prompt_io.raw.md new file mode 100644 index 00000000..53acc69c --- /dev/null +++ b/ai/prompt-io/opencode/20260727T212454Z_689df816_prompt_io.raw.md @@ -0,0 +1,34 @@ +--- +model: gpt-5.6-sol +provider: openai +service: opencode +timestamp: 2026-07-27T21:24:54Z +git_ref: 689df816 +diff_cmd: git diff HEAD~1..HEAD +--- + +> `git diff HEAD~1..HEAD -- piker/storage/nativedb.py piker/tsp/_anal.py` + +Normalized index-less provider frames to NativeDB's canonical names, +order, and dtypes before merge and publication. Provider-only columns +are excluded from durable files, derived indexes are regenerated, and +fractional timestamps are rejected before integer conversion. Added a +canonical field map for hydrating provider-specific SHM buffers while +leaving provider-only fields at their defaults. Changed Polars-to-NumPy +conversion to select columns by field name rather than position. + +> `git diff HEAD~1..HEAD -- tests/test_history_backfill.py tests/test_storage_nativedb.py` + +Added deterministic coverage using IB's actual provider dtype and a real +local `tractor` root actor, IB-typed shared-memory buffers, NativeDB, +Parquet persistence, first-start publication, and restart hydration. +Added focused regressions for reordered/extra Polars columns and +fractional timestamp rejection. + +Verification generated with the patch: + +- NativeDB and history integration set: 23 passed +- Ruff: passed +- `git diff --check`: passed +- second adversarial review: no findings +- no live gateway or network tests run diff --git a/piker/storage/nativedb.py b/piker/storage/nativedb.py index 54e89116..0806b4ee 100644 --- a/piker/storage/nativedb.py +++ b/piker/storage/nativedb.py @@ -58,6 +58,7 @@ from pathlib import Path from tempfile import NamedTemporaryFile import time +from bidict import bidict import numpy as np import polars as pl from pendulum import ( @@ -119,7 +120,13 @@ def unpack_fqme_from_parquet_filepath(path: Path) -> str: return fqme -ohlc_key_map = None +# Only hydrate fields shared by canonical storage and provider SHM. +# Provider-only fields remain at their buffer defaults on restart. +ohlc_key_map = bidict({ + name: name + for name, _ in def_iohlcv_fields + if name != 'index' +}) class NativeStorageClient: @@ -261,6 +268,57 @@ class NativeStorageClient: {}, )[fqme] = df + def _canonicalize_ohlcv( + self, + df: pl.DataFrame, + + ) -> pl.DataFrame: + ''' + Normalize a provider frame to the durable OHLCV schema. + + Provider frames may omit the derived ``index`` field and add + provider-only fields. Durable frames contain only canonical + fields in their declared order. + + ''' + required: tuple[str, ...] = tuple( + name + for name, _ in def_iohlcv_fields + ) + schema = { + name: pl.Int64 if field_type is int else pl.Float64 + for name, field_type in def_iohlcv_fields + } + missing: set[str] = set(required).difference(df.columns) + missing.discard('index') + if missing: + raise ValueError( + f'OHLCV frame is missing columns: {sorted(missing)}' + ) + + times: np.ndarray = df['time'].to_numpy() + if ( + not np.issubdtype(times.dtype, np.number) + or + not np.all(np.isfinite(times)) + ): + raise ValueError( + "OHLCV column 'time' must contain finite numeric data" + ) + if np.any(times != np.floor(times)): + raise ValueError( + 'OHLCV timestamps must use whole-second values' + ) + + return ( + df + .with_columns( + pl.Series('index', np.arange(df.height)) + ) + .select(required) + .cast(schema) + ) + async def read_ohlcv( self, fqme: str, @@ -325,9 +383,7 @@ class NativeStorageClient: df: pl.DataFrame = tsp.np2pl(ohlcv) else: df = ohlcv - df = df.with_columns( - pl.Series('index', np.arange(df.height)) - ) + df = self._canonicalize_ohlcv(df) self._validate_ohlcv(df) # TODO: in terms of managing the ultra long term data @@ -450,11 +506,14 @@ class NativeStorageClient: incoming: pl.DataFrame = tsp.np2pl(ohlcv) else: incoming = ohlcv + incoming = self._canonicalize_ohlcv(incoming) self._validate_ohlcv(incoming) path: Path = self.mk_path(fqme, timeframe) if path.exists(): - stored: pl.DataFrame = pl.read_parquet(path) + stored: pl.DataFrame = self._canonicalize_ohlcv( + pl.read_parquet(path) + ) merged: pl.DataFrame = pl.concat( [stored, incoming], how='diagonal_relaxed', diff --git a/piker/tsp/_anal.py b/piker/tsp/_anal.py index b2e000c3..580d4853 100644 --- a/piker/tsp/_anal.py +++ b/piker/tsp/_anal.py @@ -682,10 +682,7 @@ def pl2np( df.height, dtype, ) - for field, col in zip( - dtype.fields, - df.columns, - ): - array[field] = df.get_column(col).to_numpy() + for field in dtype.fields: + array[field] = df.get_column(field).to_numpy() return array diff --git a/tests/test_history_backfill.py b/tests/test_history_backfill.py index 0a26ea52..56ba7674 100644 --- a/tests/test_history_backfill.py +++ b/tests/test_history_backfill.py @@ -3,18 +3,31 @@ Deterministic history-backfill regressions. ''' from functools import partial +from pathlib import Path from types import SimpleNamespace +from uuid import uuid4 import numpy as np from pendulum import ( datetime, from_timestamp, ) +import polars as pl import pytest +import tractor import trio from piker.brokers import DataUnavailable +from piker.brokers.ib.api import ( + _bar_load_dtype, + _ohlc_dtype, +) +from piker.data._sharedmem import maybe_open_shm_array from piker.data._source import def_iohlcv_fields +from piker.storage.nativedb import ( + NativeStorageClient, + ohlc_key_map, +) from piker.tsp._history import ( notify_backfill, publish_latest_frame, @@ -194,6 +207,118 @@ def test_latest_frame_is_persisted_before_shm() -> None: assert events == ['storage', 'shm'] +def test_ib_latest_frame_round_trips_through_nativedb( + tmp_path: Path, +) -> None: + ''' + Persist and reload IB's provider schema through history startup. + + ``publish_latest_frame()`` previously passed IB's index-less bars + directly to NativeDB, whose durable-schema validator rejected the + missing derived ``index``. IB also appends ``count`` after OHLCV, + which exposed positional Polars-to-NumPy conversion to field + corruption. Build the actual IB load dtype with distinct values, + run the real history publication, NativeDB Parquet, and ``ShmArray`` + paths, then hydrate a fresh IB buffer from storage. Assertions prove + first-start publication preserves provider fields while restart maps + canonical fields and leaves provider-only ``count`` at its default. + + ''' + frame = np.zeros( + 2, + dtype=np.dtype(_bar_load_dtype), + ) + frame['time'] = [60, 120] + frame['open'] = [1.1, 2.1] + frame['high'] = [1.2, 2.2] + frame['low'] = [1.0, 2.0] + frame['close'] = [1.15, 2.15] + frame['volume'] = [10, 20] + frame['count'] = [3, 4] + + storage = NativeStorageClient(tmp_path) + shm_key = f'test_ib_history_{uuid4().hex}' + mkt = SimpleNamespace( + fqme='mnq.cme.20260918.ib', + dst=SimpleNamespace(atype='continuous_future'), + src=SimpleNamespace(atype='fiat'), + get_fqme=lambda **kwargs: 'mnq.cme.20260918.ib', + ) + + async def main() -> None: + with trio.fail_after(2): + async with tractor.open_root_actor( + name=shm_key, + tpt_bind_addrs=[('127.0.0.1', 0)], + ): + shm, opened = maybe_open_shm_array( + key=f'{shm_key}_first', + size=16, + dtype=np.dtype(_ohlc_dtype), + append_start_index=8, + ) + restart_shm, restart_opened = maybe_open_shm_array( + key=f'{shm_key}_restart', + size=16, + dtype=np.dtype(_ohlc_dtype), + append_start_index=8, + ) + assert opened + assert restart_opened + + await publish_latest_frame( + storage=storage, + mkt=mkt, + shm=shm, + array=frame, + timeframe=60, + ) + loaded = await storage.read_ohlcv( + mkt.fqme, + timeframe=60, + ) + restart_shm.push( + loaded, + prepend=True, + field_map=ohlc_key_map, + ) + stored = pl.read_parquet( + storage.mk_path(mkt.fqme, 60) + ) + canonical = [ + name + for name, _ in def_iohlcv_fields + ] + canonical_schema = { + name: ( + pl.Int64 + if field_type is int + else pl.Float64 + ) + for name, field_type in def_iohlcv_fields + } + + assert stored.columns == canonical + assert dict(stored.schema) == canonical_schema + assert list(loaded.dtype.fields) == canonical + assert loaded['index'].tolist() == [0, 1] + for field in canonical[1:]: + assert ( + loaded[field].tolist() + == + frame[field].tolist() + ) + assert ( + restart_shm.array[field].tolist() + == + frame[field].tolist() + ) + assert shm.array['count'].tolist() == [3, 4] + assert restart_shm.array['count'].tolist() == [0, 0] + + trio.run(main) + + def test_backfill_notification_timeout_is_bounded( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/tests/test_storage_nativedb.py b/tests/test_storage_nativedb.py index 46ec0c20..0e7f9d1f 100644 --- a/tests/test_storage_nativedb.py +++ b/tests/test_storage_nativedb.py @@ -69,6 +69,42 @@ def test_numpy_and_polars_round_trip( assert loaded['close'].tolist() == [1, 2, 3] +def test_pl2np_maps_fields_by_name() -> None: + ''' + Polars conversion must not depend on DataFrame column positions. + + Provider and legacy Parquet frames can include extra columns or + present canonical fields in a different order. The old ``zip()`` + conversion paired NumPy field names with DataFrame positions, + silently assigning unrelated values. Build distinct canonical + values, prepend provider-only ``count``, reverse canonical order, + and prove every structured-array field is selected by its name. + + ''' + expected = mk_ohlcv( + (60, 120), + (1.15, 2.15), + ) + expected['open'] = [1.1, 2.1] + expected['high'] = [1.2, 2.2] + expected['low'] = [1.0, 2.0] + expected['volume'] = [10, 20] + canonical = [name for name, _ in def_iohlcv_fields] + reordered = ( + tsp.np2pl(expected) + .with_columns(pl.Series('count', [3, 4])) + .select(['count', *reversed(canonical)]) + ) + + actual = tsp.pl2np( + reordered, + dtype=expected.dtype, + ) + + for field in canonical: + assert actual[field].tolist() == expected[field].tolist() + + def test_update_preserves_history_and_resolves_conflicts( tmp_path: Path, ) -> None: @@ -262,6 +298,30 @@ def test_replacement_rejects_invalid_timestamps( assert not client.mk_path('x.test', 60).exists() +def test_fractional_timestamps_are_not_truncated( + tmp_path: Path, +) -> None: + ''' + Durable timestamp coercion must not silently alter provider data. + + NativeDB declares integer epoch seconds, while IB delivers its + timestamps in a floating dtype. Casting before validation truncated + fractional values and could collapse distinct rows at one second. + Supply otherwise-valid half-second values and prove the write rejects + them before publishing any Parquet path. + + ''' + client = NativeStorageClient(tmp_path) + frame = tsp.np2pl(mk_ohlcv((60, 120))).with_columns( + pl.Series('time', [60.5, 120.5]) + ) + + with pytest.raises(ValueError, match='whole-second'): + run(client.write_ohlcv('x.test', frame, 60)) + + assert not client.mk_path('x.test', 60).exists() + + def test_write_rejects_invalid_schema_and_values( tmp_path: Path, ) -> None: