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

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`))
backfiller_deep_fixes
Gud Boi 2026-07-29 13:30:51 -04:00
parent ce33deb6b9
commit d359b1cbfb
6 changed files with 381 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
ai/prompt-io/opencode/20260729T041723Z_ce33deb6_prompt_io.md 100644
View File

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

31
ai/prompt-io/opencode/20260729T041723Z_ce33deb6_prompt_io.raw.md 100644
View File

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

50
piker/storage/_audit.py
View File

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

81
piker/storage/nativedb.py
View File

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

63
tests/test_storage_audit.py
View File

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

147
tests/test_storage_nativedb.py
View File

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