From d359b1cbfb6065b08a03bb46a0251803df6a5800 Mon Sep 17 00:00:00 2001 From: goodboy Date: Wed, 29 Jul 2026 13:30:51 -0400 Subject: [PATCH] Repair first-slot OHLC sentinels in `NativeDB` Filter known zero-initialized epoch slots before persisted history hydrates SHM or merges with a fresh provider frame. Deats, - reindex the repaired read view for same-session chart publication - treat sentinel-only files as empty history for fresh backfill - reject new first-slot sentinels but preserve modern zero-price bars - classify raw sentinel evidence separately in `store audit` Prompt-IO: ai/prompt-io/opencode/20260729T041723Z_ce33deb6_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- .../20260729T041723Z_ce33deb6_prompt_io.md | 40 +++++ ...20260729T041723Z_ce33deb6_prompt_io.raw.md | 31 ++++ piker/storage/_audit.py | 50 ++++++ piker/storage/nativedb.py | 81 ++++++++-- tests/test_storage_audit.py | 63 ++++++++ tests/test_storage_nativedb.py | 147 ++++++++++++++++-- 6 files changed, 381 insertions(+), 31 deletions(-) create mode 100644 ai/prompt-io/opencode/20260729T041723Z_ce33deb6_prompt_io.md create mode 100644 ai/prompt-io/opencode/20260729T041723Z_ce33deb6_prompt_io.raw.md diff --git a/ai/prompt-io/opencode/20260729T041723Z_ce33deb6_prompt_io.md b/ai/prompt-io/opencode/20260729T041723Z_ce33deb6_prompt_io.md new file mode 100644 index 00000000..2a299037 --- /dev/null +++ b/ai/prompt-io/opencode/20260729T041723Z_ce33deb6_prompt_io.md @@ -0,0 +1,40 @@ +--- +model: gpt-5.6-sol +provider: openai +service: opencode +session: 24fb9765-a550-4570-8350-f0fc9b7e17db +timestamp: 2026-07-29T04:17:23Z +git_ref: ce33deb6 +scope: code +substantive: true +raw_file: 20260729T041723Z_ce33deb6_prompt_io.raw.md +--- + +## Prompt + +Diagnose why the chart still showed a zero first OHLCV row after the +epoch-zero timestamp repair, causing auto-y-ranging to include zero. + +## Response summary + +Audited live 60s and 1s NativeDB files and found one exact 60s first-slot +sentinel at epoch 60 with a completely zero OHLCV payload. Added narrowly +scoped read and merge repair, matching incoming validation, sentinel-only +startup handling, canonical reindexing, and explicit audit classification. +Preserved modern all-zero bars as unclassified evidence rather than assuming +all zero-priced instruments are corrupt. + +## Files changed + +- `piker/storage/nativedb.py` - first-slot sentinel repair and validation +- `piker/storage/_audit.py` - sentinel classification and zero-row evidence +- `tests/test_storage_nativedb.py` - hydration, merge, and boundary regressions +- `tests/test_storage_audit.py` - sentinel and modern-zero audit regressions +- `ai/prompt-io/opencode/20260729T041723Z_ce33deb6_prompt_io.raw.md` + - unedited response record +- `ai/prompt-io/opencode/20260729T041723Z_ce33deb6_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/20260729T041723Z_ce33deb6_prompt_io.raw.md b/ai/prompt-io/opencode/20260729T041723Z_ce33deb6_prompt_io.raw.md new file mode 100644 index 00000000..8012c0da --- /dev/null +++ b/ai/prompt-io/opencode/20260729T041723Z_ce33deb6_prompt_io.raw.md @@ -0,0 +1,31 @@ +--- +model: gpt-5.6-sol +provider: openai +service: opencode +timestamp: 2026-07-29T04:17:23Z +git_ref: ce33deb6 +diff_cmd: git diff HEAD~1..HEAD +--- + +> `git diff HEAD~1..HEAD -- piker/storage/nativedb.py piker/storage/_audit.py tests/test_storage_nativedb.py tests/test_storage_audit.py` + +Fixed the remaining legacy first-slot sentinel observed in live MNQ 60s +history: `time=60` with zero open, high, low, close, and volume. NativeDB now +repairs the exact sentinel before read hydration and merge, reindexes the +filtered view, and persists the repaired frame on the next valid update. + +The repair remains narrowly scoped. Modern all-zero OHLC bars stay valid for +spreads or synthetic instruments, null-containing malformed rows still fail +validation, new writes can not recreate the first-slot sentinel, and a file +containing only sentinels loads as no history so fresh backfill can proceed. + +The raw audit reports total all-zero price rows, classifies first-slot +sentinels as structural violations, and leaves later all-zero rows visible as +unclassified warnings. + +Verification generated with the patch: + +- NativeDB, audit, history, xonsh, and IB regressions: 61 passed +- focused Ruff: passed +- `git diff --check`: passed +- final adversarial review: no findings diff --git a/piker/storage/_audit.py b/piker/storage/_audit.py index 4da02306..dc6a9ffb 100644 --- a/piker/storage/_audit.py +++ b/piker/storage/_audit.py @@ -30,6 +30,12 @@ _value_fields: tuple[str, ...] = ( 'close', 'volume', ) +_price_fields: tuple[str, ...] = ( + 'open', + 'high', + 'low', + 'close', +) def _utc_str(timestamp: float|int|None) -> str|None: @@ -494,6 +500,36 @@ def audit_ohlcv_frame( if counts['nonfinite'] != 0: all_values_finite = False + all_zero_price_rows: int|None = None + epoch_sentinel_rows: int|None = None + if all( + field in df.columns + and + df[field].dtype.is_numeric() + for field in _price_fields + ): + zero_prices: pl.Expr = pl.all_horizontal([ + pl.col(field) == 0 + for field in _price_fields + ]).fill_null(False) + all_zero_price_rows = int( + df.select(zero_prices).to_series().sum() + or 0 + ) + if ( + 'time' in df.columns + and + df['time'].dtype.is_numeric() + ): + epoch_sentinel_rows = int( + df.select( + (pl.col('time') <= period_s) + & + zero_prices + ).to_series().sum() + or 0 + ) + schema_ok: bool = bool( not missing and @@ -516,6 +552,8 @@ def audit_ohlcv_frame( violations.append('canonical_dtypes') if not all_values_finite: violations.append('nonfinite_ohlcv') + if epoch_sentinel_rows: + violations.append('epoch_ohlc_sentinel') if timestamps.get('nonfinite') != 0: violations.append('nonfinite_timestamps') if ( @@ -540,6 +578,15 @@ def audit_ohlcv_frame( warnings: list[str] = [] if gaps['count']: warnings.append('positive_time_gaps_unclassified') + unclassified_zero_rows: int|None = None + if all_zero_price_rows is not None: + unclassified_zero_rows = ( + all_zero_price_rows + - + (epoch_sentinel_rows or 0) + ) + if unclassified_zero_rows: + warnings.append('all_zero_ohlc_prices_unclassified') structural_ok: bool = not violations gap_free: bool|None = ( @@ -573,6 +620,9 @@ def audit_ohlcv_frame( 'nan_by_column': nans, 'infinity_by_column': infinities, 'all_finite': all_values_finite, + 'all_zero_price_rows': all_zero_price_rows, + 'epoch_sentinel_rows': epoch_sentinel_rows, + 'unclassified_zero_price_rows': unclassified_zero_rows, }, 'gaps': gaps, 'result': { diff --git a/piker/storage/nativedb.py b/piker/storage/nativedb.py index 40aba7c0..6ad51b00 100644 --- a/piker/storage/nativedb.py +++ b/piker/storage/nativedb.py @@ -74,6 +74,12 @@ from . import TimeseriesNotFound log = get_logger('storage.nativedb') +_price_fields: tuple[str, ...] = ( + 'open', + 'high', + 'low', + 'close', +) def detect_period(shm: ShmArray) -> float: @@ -233,6 +239,8 @@ class NativeStorageClient: ) from fnfe times = array['time'] + if array.size == 0: + return None return ( array, from_timestamp(times[0]), @@ -319,6 +327,41 @@ class NativeStorageClient: .cast(schema) ) + def _repair_stored_ohlcv( + self, + df: pl.DataFrame, + timeframe: int, + + ) -> pl.DataFrame: + ''' + Drop known legacy sentinels before hydration or merge. + + ''' + stored: pl.DataFrame = self._canonicalize_ohlcv(df) + zero_prices: pl.Expr = pl.all_horizontal([ + pl.col(field) == 0 + for field in _price_fields + ]).fill_null(False) + invalid: pl.Expr = ( + (pl.col('time') <= 0) + | + ( + (pl.col('time') <= timeframe) + & + zero_prices + ) + ) + stored_len: int = stored.height + stored = stored.filter(~invalid) + dropped: int = stored_len - stored.height + if dropped: + log.warning( + f'Dropping {dropped} invalid persisted OHLCV ' + f'row(s) during read repair' + ) + stored = self._canonicalize_ohlcv(stored) + return stored + async def read_ohlcv( self, fqme: str, @@ -331,7 +374,10 @@ class NativeStorageClient: fqme, period=int(timeframe), ) - df: pl.DataFrame = pl.read_parquet(path) + df: pl.DataFrame = self._repair_stored_ohlcv( + pl.read_parquet(path), + timeframe=int(timeframe), + ) self._cache_df( fqme=fqme, @@ -384,7 +430,7 @@ class NativeStorageClient: else: df = ohlcv df = self._canonicalize_ohlcv(df) - self._validate_ohlcv(df) + self._validate_ohlcv(df, timeframe) # TODO: in terms of managing the ultra long term data # -[ ] use a proper profiler to measure all this IO and @@ -405,7 +451,7 @@ class NativeStorageClient: try: df.write_parquet(tmp_path) committed: pl.DataFrame = pl.read_parquet(tmp_path) - self._validate_ohlcv(committed) + self._validate_ohlcv(committed, timeframe) if not committed.equals(df): raise IOError( 'Temporary parquet differs from input frame' @@ -442,6 +488,7 @@ class NativeStorageClient: def _validate_ohlcv( self, df: pl.DataFrame, + timeframe: int, ) -> None: ''' @@ -483,6 +530,17 @@ class NativeStorageClient: 'OHLCV timestamps must be finite and positive' ) + prices: np.ndarray = df.select(_price_fields).to_numpy() + epoch_sentinel: np.ndarray = ( + (times <= timeframe) + & + np.all(prices == 0, axis=1) + ) + if np.any(epoch_sentinel): + raise ValueError( + 'OHLCV frame contains a zero-initialized epoch sentinel' + ) + if np.any(np.diff(times) <= 0): raise ValueError( 'OHLCV timestamps must be strictly increasing' @@ -507,21 +565,14 @@ class NativeStorageClient: else: incoming = ohlcv incoming = self._canonicalize_ohlcv(incoming) - self._validate_ohlcv(incoming) + self._validate_ohlcv(incoming, timeframe) path: Path = self.mk_path(fqme, timeframe) if path.exists(): - stored: pl.DataFrame = self._canonicalize_ohlcv( - pl.read_parquet(path) + stored: pl.DataFrame = self._repair_stored_ohlcv( + pl.read_parquet(path), + timeframe=timeframe, ) - stored_len: int = stored.height - stored = stored.filter(pl.col('time') > 0) - dropped: int = stored_len - stored.height - if dropped: - log.warning( - f'Dropping {dropped} persisted OHLCV row(s) with ' - f'non-positive timestamps during merge repair' - ) merged: pl.DataFrame = pl.concat( [stored, incoming], how='diagonal_relaxed', @@ -537,7 +588,7 @@ class NativeStorageClient: else: merged = incoming - self._validate_ohlcv(merged) + self._validate_ohlcv(merged, timeframe) return self._write_ohlcv( fqme, merged, diff --git a/tests/test_storage_audit.py b/tests/test_storage_audit.py index 297cf843..b9590a83 100644 --- a/tests/test_storage_audit.py +++ b/tests/test_storage_audit.py @@ -129,6 +129,69 @@ def test_audit_preserves_raw_defect_evidence() -> None: assert report['values']['infinity_by_column']['volume'] == 1 +def test_audit_rejects_epoch_zero_price_sentinel() -> None: + ''' + The legacy first-slot sentinel is structural corruption. + + The repaired MNQ Parquet retained a row at epoch 60 whose OHLCV + payload was zero. Numeric and finite checks marked it valid even + though chart auto-ranging then included zero. Audit a canonical + frame with that sentinel and prove the report counts it and fails + structural qualification. + + ''' + frame = mk_frame((60, 120)).with_columns( + pl.Series('open', [0, 1], dtype=pl.Float64), + pl.Series('high', [0, 2], dtype=pl.Float64), + pl.Series('low', [0, 1], dtype=pl.Float64), + pl.Series('close', [0, 2], dtype=pl.Float64), + ) + + report: dict = audit_ohlcv_frame( + frame, + fqme='x.test', + period_s=60, + ) + + assert report['values']['all_zero_price_rows'] == 1 + assert report['values']['epoch_sentinel_rows'] == 1 + assert 'epoch_ohlc_sentinel' in report['result']['violations'] + assert report['result']['structural_ok'] is False + + +def test_audit_leaves_modern_zero_prices_unclassified() -> None: + ''' + Audit must not call every all-zero price bar corrupt. + + A spread or synthetic instrument may legitimately trade at zero. + The known corruption is tied to the first slot. Place an all-zero + bar later and prove it remains visible as a warning without + failing structural qualification. + + ''' + frame = mk_frame((120,)).with_columns( + pl.Series('open', [0], dtype=pl.Float64), + pl.Series('high', [0], dtype=pl.Float64), + pl.Series('low', [0], dtype=pl.Float64), + pl.Series('close', [0], dtype=pl.Float64), + ) + + report: dict = audit_ohlcv_frame( + frame, + fqme='x.test', + period_s=60, + ) + + assert report['values']['all_zero_price_rows'] == 1 + assert report['values']['epoch_sentinel_rows'] == 0 + assert report['values']['unclassified_zero_price_rows'] == 1 + assert report['result']['structural_ok'] is True + assert ( + 'all_zero_ohlc_prices_unclassified' + in report['result']['warnings'] + ) + + def test_gap_aggregates_are_not_truncated_with_details() -> None: ''' Limiting JSON detail must not undercount total missing coverage. diff --git a/tests/test_storage_nativedb.py b/tests/test_storage_nativedb.py index 2b1e22bf..0f991ef3 100644 --- a/tests/test_storage_nativedb.py +++ b/tests/test_storage_nativedb.py @@ -131,47 +131,162 @@ def test_update_preserves_history_and_resolves_conflicts( assert stored['index'].to_list() == [0, 1, 2, 3] -def test_update_repairs_nonpositive_persisted_timestamps( +def test_update_repairs_invalid_persisted_sentinel_rows( tmp_path: Path, ) -> None: ''' - Valid incoming history must repair a legacy epoch-zero row. + Valid incoming history must repair legacy sentinel rows. - The known MNQ baseline contains one zero timestamp plus extra derived - columns and noncanonical absolute indexes. ``update_ohlcv()`` used to - canonicalize those bytes but retain the zero row, then reject the - whole merged frame before fresh IB bars could publish. Write that - legacy shape directly, append a valid provider frame, and prove only - the invalid persisted row disappears while old and new valid bars - survive with canonical indexes. + The known MNQ baseline contained epoch-zero and all-zero rows, + plus extra columns and absolute indexes. ``update_ohlcv()`` first + retained both, then removed only epoch zero and left chart range + anchored at zero. Write both sentinels, append a valid provider + frame, and prove only valid bars survive with canonical + indexes. ''' client = NativeStorageClient(tmp_path) legacy = ( tsp.np2pl(mk_ohlcv( - (0, 60, 120), - (0, 1, 2), + (0, 60, 120, 180), + (0, 0, 2, 3), )) .with_columns( - pl.Series('index', [3137800, 3137801, 3137802]), - pl.Series('time_prev', [None, 0, 60]), + pl.Series( + 'index', + [3137800, 3137801, 3137802, 3137803], + ), + pl.Series('time_prev', [None, 0, 60, 120]), ) ) path: Path = client.mk_path('x.test', 60) legacy.write_parquet(path) + loaded = trio.run(client.read_ohlcv, 'x.test', 60) + assert loaded['time'].tolist() == [120, 180] + assert loaded['index'].tolist() == [0, 1] + assert pl.read_parquet(path).height == 4 + run(client.update_ohlcv( 'x.test', - mk_ohlcv((180,), (3,)), + mk_ohlcv((240,), (4,)), 60, )) stored: pl.DataFrame = pl.read_parquet(path) - assert stored['time'].to_list() == [60, 120, 180] - assert stored['close'].to_list() == [1, 2, 3] + assert stored['time'].to_list() == [120, 180, 240] + assert stored['close'].to_list() == [2, 3, 4] assert stored['index'].to_list() == [0, 1, 2] +def test_write_preserves_modern_all_zero_prices( + tmp_path: Path, +) -> None: + ''' + Zero prices outside the legacy sentinel range remain valid. + + Some spreads or synthetic instruments may legitimately trade at + zero. Repair targets the observed first-slot sentinel, not every + all-zero OHLC bar. Write a modern all-zero row and + prove it survives normal replacement unchanged. + + ''' + client = NativeStorageClient(tmp_path) + zero_bar = mk_ohlcv((1_800_000_000,), (0,)) + + run(client.write_ohlcv('x.test', zero_bar, 60)) + stored = pl.read_parquet(client.mk_path('x.test', 60)) + + assert stored['time'].to_list() == [1_800_000_000] + assert stored['close'].to_list() == [0] + + +def test_write_rejects_first_slot_zero_sentinel( + tmp_path: Path, +) -> None: + ''' + New writes must not recreate the repaired legacy sentinel. + + Read repair removes all-zero OHLC first-slot rows. Accepting one from + new input would recreate a file whose audit fails and whose read view + hides a row. Submit the exact 60-second + sentinel and prove no file publishes. + + ''' + client = NativeStorageClient(tmp_path) + sentinel = mk_ohlcv((60,), (0,)) + + with pytest.raises(ValueError, match='epoch sentinel'): + run(client.write_ohlcv('x.test', sentinel, 60)) + + assert not client.mk_path('x.test', 60).exists() + + +def test_sentinel_only_storage_loads_as_no_history( + tmp_path: Path, +) -> None: + ''' + A fully repairable legacy file must permit fresh startup. + + Filtering a file containing only sentinels yields an empty array. + ``load()`` previously indexed its endpoint timestamps and crashed + before valid data could replace it. Persist only sentinel rows, + prove load reports no history, then append a valid bar and + verify durable storage becomes canonical. + + ''' + client = NativeStorageClient(tmp_path) + path: Path = client.mk_path('x.test', 60) + tsp.np2pl(mk_ohlcv( + (0, 60), + (0, 0), + )).write_parquet(path) + + assert trio.run(client.load, 'x.test', 60) is None + run(client.update_ohlcv( + 'x.test', + mk_ohlcv((120,), (1,)), + 60, + )) + stored: pl.DataFrame = pl.read_parquet(path) + + assert stored['time'].to_list() == [120] + assert stored['index'].to_list() == [0] + + +def test_merge_does_not_hide_null_price_rows( + tmp_path: Path, +) -> None: + ''' + Sentinel repair must not silently discard other malformed rows. + + Polars boolean filters drop null predicates. Without an explicit + false fill, one null plus three zeros looked neither valid nor + all-zero but disappeared before validation. Persist that shape, + append valid data, and prove merge still rejects the null + while preserving the original evidence file. + + ''' + client = NativeStorageClient(tmp_path) + path: Path = client.mk_path('x.test', 60) + malformed = tsp.np2pl( + mk_ohlcv((1_800_000_000,), (0,)) + ).with_columns( + pl.Series('open', [None], dtype=pl.Float64) + ) + malformed.write_parquet(path) + before: bytes = path.read_bytes() + + with pytest.raises(ValueError, match='finite numeric'): + run(client.update_ohlcv( + 'x.test', + mk_ohlcv((1_800_000_060,), (1,)), + 60, + )) + + assert path.read_bytes() == before + + def test_write_ohlcv_remains_an_explicit_replacement( tmp_path: Path, ) -> None: