From d7d537d5317f83a133d8789b8bb9c6608affd2ae Mon Sep 17 00:00:00 2001 From: goodboy Date: Tue, 28 Jul 2026 19:38:50 -0400 Subject: [PATCH] Add raw OHLCV audit to `NativeDB` Inspect Parquet bytes directly without opening storage or normalizing malformed evidence. Deats, - report canonical schema, timestamp, index and finite-value state - retain exact cadence aggregates with bounded endpoint details - snapshot bytes before parsing and clean up partial copy failures - reject unsafe FQMEs, links, collisions and non-regular sources Prompt-IO: ai/prompt-io/opencode/20260728T005034Z_ed85721c_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- .../20260728T005034Z_ed85721c_prompt_io.md | 40 ++ ...20260728T005034Z_ed85721c_prompt_io.raw.md | 27 + piker/storage/_audit.py | 652 ++++++++++++++++++ piker/storage/cli.py | 230 ++++++ tests/test_storage_audit.py | 554 +++++++++++++++ 5 files changed, 1503 insertions(+) create mode 100644 ai/prompt-io/opencode/20260728T005034Z_ed85721c_prompt_io.md create mode 100644 ai/prompt-io/opencode/20260728T005034Z_ed85721c_prompt_io.raw.md create mode 100644 piker/storage/_audit.py create mode 100644 tests/test_storage_audit.py diff --git a/ai/prompt-io/opencode/20260728T005034Z_ed85721c_prompt_io.md b/ai/prompt-io/opencode/20260728T005034Z_ed85721c_prompt_io.md new file mode 100644 index 00000000..bbeeb1e9 --- /dev/null +++ b/ai/prompt-io/opencode/20260728T005034Z_ed85721c_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-28T00:50:34Z +git_ref: ed85721c +scope: code +substantive: true +raw_file: 20260728T005034Z_ed85721c_prompt_io.raw.md +--- + +## Prompt + +Add reusable read-only tooling to qualify raw NativeDB Parquet against a +known gappy IB chart, beginning with `mnq.cme.20260918`. Preserve malformed +evidence and distinguish structural validity from unclassified positive +gaps without opening or mutating the storage runtime. + +## Response summary + +Implemented raw Parquet audit reporting with human and JSON output, bounded +cadence details, exact integer arithmetic, strict status, collision-safe +snapshot capture, and source path protections. Regressions cover malformed +schema and values, timestamp precision, gap arithmetic, corrupt bytes, +copy failures, symlinks, traversal, and read-only CLI behavior. + +## Files changed + +- `piker/storage/_audit.py` - raw Parquet audit and snapshot implementation +- `piker/storage/cli.py` - `piker store audit` command +- `tests/test_storage_audit.py` - audit and CLI regressions +- `ai/prompt-io/opencode/20260728T005034Z_ed85721c_prompt_io.raw.md` + - unedited response record +- `ai/prompt-io/opencode/20260728T005034Z_ed85721c_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/20260728T005034Z_ed85721c_prompt_io.raw.md b/ai/prompt-io/opencode/20260728T005034Z_ed85721c_prompt_io.raw.md new file mode 100644 index 00000000..7cbcdb24 --- /dev/null +++ b/ai/prompt-io/opencode/20260728T005034Z_ed85721c_prompt_io.raw.md @@ -0,0 +1,27 @@ +--- +model: gpt-5.6-sol +provider: openai +service: opencode +timestamp: 2026-07-28T00:50:34Z +git_ref: ed85721c +diff_cmd: git diff HEAD~1..HEAD +--- + +> `git diff HEAD~1..HEAD -- piker/storage/_audit.py piker/storage/cli.py tests/test_storage_audit.py` + +Added a raw, read-only NativeDB Parquet audit with human and JSON output, +bounded cadence evidence, exact integer arithmetic, pre-parse snapshot +capture, and strict qualification status. + +The command rejects unsafe FQME paths, symlinked sources and outputs, +destination collisions, non-regular source nodes, malformed canonical +schema, non-finite values, invalid timestamps, noncanonical indexes, and +sub-period cadence. Copy-failure cleanup preserves source bytes while +completed corrupt snapshots remain available for diagnosis. + +Verification generated with the patch: + +- storage audit and CLI regressions: passed +- Ruff: passed +- `git diff --check`: passed +- final adversarial review: no findings diff --git a/piker/storage/_audit.py b/piker/storage/_audit.py new file mode 100644 index 00000000..4da02306 --- /dev/null +++ b/piker/storage/_audit.py @@ -0,0 +1,652 @@ +# piker: trading gear for hackers +# Copyright (C) Tyler Goodlet (in stewardship for pikers) + +''' +Read-only OHLCV persistence qualification. + +''' +from datetime import ( + UTC, + datetime, +) +from hashlib import sha256 +import os +from pathlib import Path +import shutil +from stat import S_ISREG +from typing import BinaryIO + +import numpy as np +import polars as pl + +from piker.data import def_iohlcv_fields + + +AUDIT_SCHEMA: str = 'piker.nativedb.audit/v1' +_value_fields: tuple[str, ...] = ( + 'open', + 'high', + 'low', + 'close', + 'volume', +) + + +def _utc_str(timestamp: float|int|None) -> str|None: + ''' + Render an epoch timestamp as deterministic UTC. + + ''' + if timestamp is None: + return None + try: + dt: datetime = datetime.fromtimestamp(timestamp, tz=UTC) + except ( + OSError, + OverflowError, + ValueError, + ): + return None + return dt.isoformat().replace('+00:00', 'Z') + + +def _number(value: float|int) -> float|int: + ''' + Convert a NumPy number to a JSON-native scalar. + + ''' + if isinstance(value, (int, np.integer)): + return int(value) + value = float(value) + return int(value) if value.is_integer() else value + + +def _sha256(file: BinaryIO) -> str: + ''' + Hash and rewind one already-open file snapshot. + + ''' + digest = sha256() + for chunk in iter( + lambda: file.read(1024 * 1024), + b'', + ): + digest.update(chunk) + file.seek(0) + return digest.hexdigest() + + +def _numeric_counts(series: pl.Series) -> dict[str, int|None]: + ''' + Count null and non-finite values without coercing text. + + ''' + null_count: int = series.null_count() + if not series.dtype.is_numeric(): + return { + 'null': null_count, + 'nan': None, + 'positive_infinity': None, + 'negative_infinity': None, + 'nonfinite': None, + } + + numeric: pl.Series = series.cast(pl.Float64, strict=False) + nan_count: int = int(numeric.is_nan().sum() or 0) + pos_inf_count: int = int( + (numeric == float('inf')).sum() or 0 + ) + neg_inf_count: int = int( + (numeric == float('-inf')).sum() or 0 + ) + cast_null_count: int = max( + 0, + numeric.null_count() - null_count, + ) + return { + 'null': null_count, + 'nan': nan_count, + 'positive_infinity': pos_inf_count, + 'negative_infinity': neg_inf_count, + 'nonfinite': ( + null_count + + nan_count + + pos_inf_count + + neg_inf_count + + cast_null_count + ), + } + + +def _numeric_values( + series: pl.Series, +) -> tuple[np.ndarray, np.ndarray]: + ''' + Preserve integer precision and return an explicit valid mask. + + ''' + if series.dtype.is_integer(): + valid: np.ndarray = series.is_not_null().to_numpy() + values: np.ndarray = series.fill_null(0).to_numpy() + return values, valid + + values = series.to_numpy() + try: + valid = np.isfinite(values) + except TypeError: + values = series.cast(pl.Float64).to_numpy() + valid = np.isfinite(values) + return values, valid + + +def _empty_gaps( + period_s: int, + verifiable: bool = False, +) -> dict: + ''' + Return the stable gap report shape for unauditable timestamps. + + ''' + return { + 'basis': 'sorted_unique_valid_timestamps', + 'expected_period_s': period_s, + 'verifiable': verifiable, + 'count': 0, + 'aligned_count': 0, + 'misaligned_count': 0, + 'subperiod_count': 0, + 'missing_samples_total': 0, + 'details_count': 0, + 'details_truncated': False, + 'intervals': [], + 'subperiod_intervals': [], + } + + +def _audit_timestamps( + df: pl.DataFrame, + period_s: int, + max_gaps: int, + +) -> tuple[dict, dict]: + ''' + Audit physical timestamp ordering and chronological gaps. + + ''' + if 'time' not in df.columns: + return ( + { + 'present': False, + 'numeric': False, + 'strictly_increasing': False, + }, + _empty_gaps(period_s), + ) + + series: pl.Series = df['time'] + counts: dict[str, int|None] = _numeric_counts(series) + if not series.dtype.is_numeric(): + return ( + { + 'present': True, + 'numeric': False, + **counts, + 'strictly_increasing': False, + }, + _empty_gaps(period_s), + ) + + values, finite = _numeric_values(series) + valid: np.ndarray = values[finite] + unique: np.ndarray = np.unique(valid) + adjacent_valid: np.ndarray = finite[:-1] & finite[1:] + deltas: np.ndarray = np.diff(values) + physical_deltas: np.ndarray = deltas[adjacent_valid] + + zero_delta_count: int = int(np.count_nonzero( + physical_deltas == 0 + )) + negative_delta_count: int = int(np.count_nonzero( + physical_deltas < 0 + )) + first: float|int|None = None + last: float|int|None = None + if values.size: + if finite[0]: + first = _number(values[0]) + if finite[-1]: + last = _number(values[-1]) + + minimum: float|int|None = None + maximum: float|int|None = None + if valid.size: + minimum = _number(np.min(valid)) + maximum = _number(np.max(valid)) + + chronological: np.ndarray = unique[unique > 0] + gap_deltas: np.ndarray = np.diff(chronological) + gap_indexes: np.ndarray = np.flatnonzero( + gap_deltas > period_s + ) + gap_values: np.ndarray = gap_deltas[gap_indexes] + integer_gaps: bool = np.issubdtype( + gap_values.dtype, + np.integer, + ) + if integer_gaps: + aligned: np.ndarray = gap_values % period_s == 0 + else: + aligned = np.isclose( + gap_values % period_s, + 0, + ) + aligned_count: int = int(np.count_nonzero(aligned)) + if integer_gaps: + missing_total: int = sum( + int(delta) // period_s - 1 + for delta in gap_values[aligned] + ) + else: + missing_total = sum( + int(round(float(delta) / period_s)) - 1 + for delta in gap_values[aligned] + ) + subperiod_indexes: np.ndarray = np.flatnonzero( + gap_deltas < period_s + ) + intervals: list[dict] = [] + for gap_index in gap_indexes[:max_gaps]: + left: float = chronological[gap_index] + right: float = chronological[gap_index + 1] + delta: float = right - left + period_multiple: bool = bool(np.isclose( + delta % period_s, + 0, + )) + missing_samples: int|None = None + if period_multiple: + missing_samples = ( + int(delta) // period_s - 1 + if integer_gaps + else int(round(float(delta) / period_s)) - 1 + ) + + left_value: float|int = _number(left) + right_value: float|int = _number(right) + intervals.append({ + 'left_timestamp': left_value, + 'right_timestamp': right_value, + 'left_utc': _utc_str(left_value), + 'right_utc': _utc_str(right_value), + 'delta_s': _number(delta), + 'period_multiple': period_multiple, + 'missing_samples': missing_samples, + 'classification': 'unclassified', + }) + + subperiod_intervals: list[dict] = [] + remaining_details: int = max(0, max_gaps - len(intervals)) + for step_index in subperiod_indexes[:remaining_details]: + left = _number(chronological[step_index]) + right = _number(chronological[step_index + 1]) + subperiod_intervals.append({ + 'left_timestamp': left, + 'right_timestamp': right, + 'left_utc': _utc_str(left), + 'right_utc': _utc_str(right), + 'delta_s': _number(right - left), + }) + + gap_count: int = gap_indexes.size + gaps: dict = { + 'basis': 'sorted_unique_valid_timestamps', + 'expected_period_s': period_s, + 'verifiable': chronological.size >= 2, + 'count': gap_count, + 'aligned_count': aligned_count, + 'misaligned_count': gap_count - aligned_count, + 'missing_samples_total': missing_total, + 'subperiod_count': subperiod_indexes.size, + 'details_count': ( + len(intervals) + + len(subperiod_intervals) + ), + 'details_truncated': bool( + gap_count > len(intervals) + or + subperiod_indexes.size > len(subperiod_intervals) + ), + 'intervals': intervals, + 'subperiod_intervals': subperiod_intervals, + } + timestamps: dict = { + 'present': True, + 'numeric': True, + 'first_in_file': first, + 'last_in_file': last, + 'minimum': minimum, + 'maximum': maximum, + 'minimum_utc': _utc_str(minimum), + 'maximum_utc': _utc_str(maximum), + **counts, + 'zero': int(np.count_nonzero(valid == 0)), + 'negative': int(np.count_nonzero(valid < 0)), + 'fractional': int(np.count_nonzero( + valid != np.floor(valid) + )), + 'unique': unique.size, + 'duplicate_excess': valid.size - unique.size, + 'adjacent_zero_delta': zero_delta_count, + 'negative_delta': negative_delta_count, + 'non_positive_delta': ( + zero_delta_count + + negative_delta_count + ), + 'strictly_increasing': bool( + values.size > 0 + and + values.size == valid.size + and + np.all(np.diff(values) > 0) + ), + } + return timestamps, gaps + + +def _audit_index(df: pl.DataFrame) -> dict: + ''' + Verify persisted indexes match physical row positions. + + ''' + if 'index' not in df.columns: + return { + 'present': False, + 'numeric': False, + 'canonical_from_zero': False, + } + + series: pl.Series = df['index'] + counts: dict[str, int|None] = _numeric_counts(series) + if not series.dtype.is_numeric(): + return { + 'present': True, + 'numeric': False, + 'dtype': str(series.dtype), + **counts, + 'canonical_from_zero': False, + } + + values, finite = _numeric_values(series) + expected: np.ndarray = np.arange(df.height) + mismatch_count: int = int(np.count_nonzero( + ~finite + | + (values != expected) + )) + valid: np.ndarray = values[finite] + unique_count: int = np.unique(valid).size + return { + 'present': True, + 'numeric': True, + 'dtype': str(series.dtype), + 'first': ( + _number(values[0]) + if values.size and finite[0] + else None + ), + 'last': ( + _number(values[-1]) + if values.size and finite[-1] + else None + ), + **counts, + 'duplicate_excess': valid.size - unique_count, + 'non_unit_step': int(np.count_nonzero( + np.diff(valid) != 1 + )), + 'row_position_mismatch': mismatch_count, + 'contiguous': bool( + values.size > 0 + and + valid.size == values.size + and + np.all(np.diff(values) == 1) + ), + 'canonical_from_zero': bool( + values.size > 0 + and + values.size == expected.size + and + mismatch_count == 0 + ), + } + + +def audit_ohlcv_frame( + df: pl.DataFrame, + fqme: str, + period_s: int, + max_gaps: int = 100, + +) -> dict: + ''' + Inspect a raw persisted frame without normalizing evidence. + + ''' + if period_s < 1: + raise ValueError('Audit period must be positive') + if max_gaps < 0: + raise ValueError('Maximum gap details must be non-negative') + + expected_columns: list[str] = [ + name + for name, _ in def_iohlcv_fields + ] + expected_dtypes: dict[str, str] = { + name: str(pl.Int64 if field_type is int else pl.Float64) + for name, field_type in def_iohlcv_fields + } + actual_dtypes: dict[str, str] = { + name: str(dtype) + for name, dtype in df.schema.items() + } + missing: list[str] = sorted( + set(expected_columns).difference(df.columns) + ) + extra: list[str] = sorted( + set(df.columns).difference(expected_columns) + ) + column_order_ok: bool = df.columns == expected_columns + dtypes_ok: bool = all( + actual_dtypes.get(name) == dtype + for name, dtype in expected_dtypes.items() + ) + + timestamps, gaps = _audit_timestamps( + df, + period_s, + max_gaps, + ) + index: dict = _audit_index(df) + + nulls: dict[str, int|None] = {} + nans: dict[str, int|None] = {} + infinities: dict[str, int|None] = {} + all_values_finite: bool = True + for field in _value_fields: + if field not in df.columns: + nulls[field] = None + nans[field] = None + infinities[field] = None + all_values_finite = False + continue + + counts = _numeric_counts(df[field]) + nulls[field] = counts['null'] + nans[field] = counts['nan'] + pos_inf: int|None = counts['positive_infinity'] + neg_inf: int|None = counts['negative_infinity'] + infinities[field] = ( + None + if pos_inf is None or neg_inf is None + else pos_inf + neg_inf + ) + if counts['nonfinite'] != 0: + all_values_finite = False + + schema_ok: bool = bool( + not missing + and + not extra + and + column_order_ok + and + dtypes_ok + ) + violations: list[str] = [] + if df.is_empty(): + violations.append('empty_frame') + if missing: + violations.append('missing_columns') + if extra: + violations.append('extra_columns') + if not column_order_ok: + violations.append('column_order') + if not dtypes_ok: + violations.append('canonical_dtypes') + if not all_values_finite: + violations.append('nonfinite_ohlcv') + if timestamps.get('nonfinite') != 0: + violations.append('nonfinite_timestamps') + if ( + timestamps.get('zero', 0) + or + timestamps.get('negative', 0) + ): + violations.append('nonpositive_timestamps') + if timestamps.get('fractional', 0): + violations.append('fractional_timestamps') + if timestamps.get('duplicate_excess', 0): + violations.append('duplicate_timestamps') + if not timestamps.get('strictly_increasing', False): + violations.append('timestamp_order') + if not index.get('canonical_from_zero', False): + violations.append('index_not_canonical') + if gaps.get('misaligned_count', 0): + violations.append('misaligned_time_gap') + if gaps.get('subperiod_count', 0): + violations.append('subperiod_time_step') + + warnings: list[str] = [] + if gaps['count']: + warnings.append('positive_time_gaps_unclassified') + + structural_ok: bool = not violations + gap_free: bool|None = ( + gaps['count'] == 0 + if gaps['verifiable'] + else None + ) + return { + 'audit_schema': AUDIT_SCHEMA, + 'fqme': fqme, + 'period_s': period_s, + 'schema': { + 'columns': df.columns, + 'dtypes': actual_dtypes, + 'expected_columns': expected_columns, + 'expected_dtypes': expected_dtypes, + 'missing_columns': missing, + 'extra_columns': extra, + 'column_order_ok': column_order_ok, + 'canonical_dtypes_ok': dtypes_ok, + 'canonical': schema_ok, + }, + 'rows': { + 'count': df.height, + 'empty': df.is_empty(), + }, + 'timestamps': timestamps, + 'index': index, + 'values': { + 'null_by_column': nulls, + 'nan_by_column': nans, + 'infinity_by_column': infinities, + 'all_finite': all_values_finite, + }, + 'gaps': gaps, + 'result': { + 'structural_ok': structural_ok, + 'gap_free': gap_free, + 'qualification_ok': bool(structural_ok and gap_free), + 'violations': violations, + 'warnings': warnings, + }, + } + + +def audit_ohlcv_parquet( + path: Path, + fqme: str, + period_s: int, + max_gaps: int = 100, + snapshot: Path|None = None, + +) -> dict: + ''' + Read and audit one Parquet file without touching storage state. + + ''' + path = path.expanduser().absolute() + descriptor: int = os.open( + path, + os.O_RDONLY + | os.O_CLOEXEC + | os.O_NOFOLLOW + | os.O_NONBLOCK, + ) + stat = os.fstat(descriptor) + if not S_ISREG(stat.st_mode): + os.close(descriptor) + raise ValueError('Audit source must be a regular file') + with os.fdopen(descriptor, 'rb') as file: + checksum: str = _sha256(file) + if snapshot is not None: + snapshot_is_symlink: bool = snapshot.is_symlink() + snapshot = snapshot.resolve() + if snapshot == path: + raise ValueError( + 'Snapshot path can not replace source' + ) + if snapshot_is_symlink: + raise ValueError( + 'Snapshot path can not be a symlink' + ) + file.seek(0) + created: bool = False + try: + with snapshot.open('xb') as snapshot_file: + created = True + os.fchmod(snapshot_file.fileno(), 0o444) + shutil.copyfileobj(file, snapshot_file) + except BaseException: + if created: + snapshot.unlink(missing_ok=True) + raise + file.seek(0) + frame: pl.DataFrame = pl.read_parquet(file) + report: dict = audit_ohlcv_frame( + frame, + fqme, + period_s, + max_gaps, + ) + generated_at: float = datetime.now(tz=UTC).timestamp() + report['generated_at_utc'] = _utc_str(generated_at) + report['source'] = { + 'path': str(path), + 'size_bytes': stat.st_size, + 'mtime_ns': stat.st_mtime_ns, + 'sha256': checksum, + } + return report diff --git a/piker/storage/cli.py b/piker/storage/cli.py index 78138de1..f6db254d 100644 --- a/piker/storage/cli.py +++ b/piker/storage/cli.py @@ -19,7 +19,9 @@ Storage middle-ware CLIs. """ from __future__ import annotations +import json from pathlib import Path +import sys import time from types import ModuleType from typing import ( @@ -31,20 +33,25 @@ import numpy as np import tractor # import pendulum from rich.console import Console +from rich.table import Table import trio # from rich.markdown import Markdown import typer +import piker as piker_pkg from piker.service import open_piker_runtime from piker.cli import cli from tractor.ipc._shm import ShmArray from piker import tsp +from piker import config from . import log from . import ( __tsdbs__, open_storage_client, StorageClient, ) +from ._audit import audit_ohlcv_parquet +from .nativedb import mk_ohlcv_shm_keyed_filepath if TYPE_CHECKING: from piker.ui._remote_ctl import AnnotCtl @@ -53,6 +60,65 @@ if TYPE_CHECKING: store = typer.Typer() +def _render_audit_report(report: dict) -> None: + ''' + Render a compact human summary and explicit gap endpoints. + + ''' + result: dict = report['result'] + timestamps: dict = report['timestamps'] + gaps: dict = report['gaps'] + source: dict = report['source'] + + fqme: str = report['fqme'] + period_s: int = report['period_s'] + table = Table(title=f'{fqme} @ {period_s}s') + table.add_column('Check') + table.add_column('Value') + table.add_row('Path', source['path']) + table.add_row('SHA-256', source['sha256']) + table.add_row('Rows', str(report['rows']['count'])) + table.add_row('Minimum UTC', str(timestamps.get('minimum_utc'))) + table.add_row('Maximum UTC', str(timestamps.get('maximum_utc'))) + table.add_row('Structural OK', str(result['structural_ok'])) + table.add_row('Gap free', str(result['gap_free'])) + table.add_row('Gap count', str(gaps['count'])) + table.add_row('Sub-period steps', str(gaps['subperiod_count'])) + table.add_row('Gap details truncated', str( + gaps['details_truncated'] + )) + violations: list[str] = result['violations'] + table.add_row( + 'Violations', + ', '.join(violations) if violations else 'none', + ) + + console = Console() + console.print(table) + intervals: list[dict] = [ + *gaps['intervals'], + *gaps['subperiod_intervals'], + ] + if not intervals: + return + + gap_table = Table(title='Timestamp cadence deviations') + gap_table.add_column('Left UTC') + gap_table.add_column('Right UTC') + gap_table.add_column('Delta (s)') + gap_table.add_column('Missing') + gap_table.add_column('Aligned') + for gap in intervals: + gap_table.add_row( + str(gap['left_utc']), + str(gap['right_utc']), + str(gap['delta_s']), + str(gap.get('missing_samples')), + str(gap.get('period_multiple')), + ) + console.print(gap_table) + + @store.command() def ls( backends: list[str] = typer.Argument( @@ -94,6 +160,170 @@ def ls( trio.run(query_all) +@store.command() +def audit( + fqme: str, + period: int = typer.Option( + 60, + '--period', + min=1, + help='Expected sampling period in seconds.', + ), + json_output: bool = typer.Option( + False, + '--json', + help='Emit machine-readable JSON.', + ), + output: Path|None = typer.Option( + None, + '--output', + '-o', + help='Write JSON to a new file without replacing it.', + ), + snapshot: Path|None = typer.Option( + None, + '--snapshot', + help='Copy the exact audited bytes to a new file.', + ), + max_gaps: int = typer.Option( + 100, + '--max-gaps', + min=0, + help='Maximum gap details to include.', + ), + strict: bool = typer.Option( + False, + '--strict', + help='Exit nonzero for structural defects or any gap.', + ), +) -> None: + ''' + Audit one NativeDB Parquet without normalizing or mutating it. + + Positive gaps are reported as unclassified evidence. They are not + automatically treated as corruption or expected venue closures. + + ''' + if ( + fqme in {'.', '..'} + or + Path(fqme).name != fqme + or + '/' in fqme + or + '\\' in fqme + ): + typer.echo(f'Unsafe FQME path component: {fqme!r}', err=True) + raise typer.Exit(code=2) + datadir: Path = config.get_conf_dir() / 'nativedb' + path: Path = mk_ohlcv_shm_keyed_filepath( + fqme, + period, + datadir, + ) + destinations: list[Path] = [ + destination + for destination in (output, snapshot) + if destination is not None + ] + resolved_source: Path = path.resolve() + if path.is_symlink(): + typer.echo( + f'Refusing symlinked audit source: {path}', + err=True, + ) + raise typer.Exit(code=2) + resolved_destinations: list[Path] = [] + for destination in destinations: + if destination.is_symlink(): + typer.echo( + f'Refusing symlinked output: {destination}', + err=True, + ) + raise typer.Exit(code=2) + resolved: Path = destination.resolve() + if resolved == resolved_source: + typer.echo( + f'Refusing to replace audited source: {resolved}', + err=True, + ) + raise typer.Exit(code=2) + if destination.exists(): + typer.echo( + f'Refusing to replace existing output: ' + f'{destination}', + err=True, + ) + raise typer.Exit(code=2) + if not destination.parent.is_dir(): + typer.echo( + f'Output directory does not exist: ' + f'{destination.parent}', + err=True, + ) + raise typer.Exit(code=2) + resolved_destinations.append(resolved) + if len(set(resolved_destinations)) != len(resolved_destinations): + typer.echo( + 'JSON output and snapshot paths must differ', + err=True, + ) + raise typer.Exit(code=2) + + try: + report: dict = audit_ohlcv_parquet( + path, + fqme, + period, + max_gaps, + snapshot, + ) + except ( + FileNotFoundError, + OSError, + pl.exceptions.PolarsError, + ValueError, + ) as err: + typer.echo( + f'Unable to audit {path}: {err}', + err=True, + ) + raise typer.Exit(code=2) from err + + report['runtime'] = { + 'piker_file': str(Path(piker_pkg.__file__).resolve()), + 'executable': str(Path(sys.argv[0]).resolve()), + } + payload: str = json.dumps( + report, + allow_nan=False, + indent=2, + sort_keys=True, + ) + if output is not None: + try: + with output.open('x') as output_file: + output_file.write(f'{payload}\n') + except OSError as err: + typer.echo( + f'Unable to write audit output: {err}', + err=True, + ) + raise typer.Exit(code=2) from err + + if json_output: + typer.echo(payload) + else: + _render_audit_report(report) + + if ( + strict + and + not report['result']['qualification_ok'] + ): + raise typer.Exit(code=1) + + # TODO: like ls but takes in a pattern and matches # @store.command() # def search( diff --git a/tests/test_storage_audit.py b/tests/test_storage_audit.py new file mode 100644 index 00000000..297cf843 --- /dev/null +++ b/tests/test_storage_audit.py @@ -0,0 +1,554 @@ +''' +Read-only NativeDB audit regressions. + +''' +from hashlib import sha256 +import json +from pathlib import Path + +import numpy as np +import polars as pl +import pytest +from typer.testing import CliRunner + +from piker import config +from piker.data import def_iohlcv_fields +from piker.storage import _audit as audit_mod +from piker.storage._audit import ( + audit_ohlcv_frame, + audit_ohlcv_parquet, +) +from piker.storage.cli import store +from piker.storage.nativedb import ( + mk_ohlcv_shm_keyed_filepath, +) + + +def mk_frame(times: tuple[int, ...]) -> pl.DataFrame: + ''' + Build one canonical persisted frame with finite values. + + ''' + size: int = len(times) + return pl.DataFrame({ + 'index': pl.Series(np.arange(size), dtype=pl.Int64), + 'time': pl.Series(times, dtype=pl.Int64), + 'open': pl.Series(np.arange(size) + 1, dtype=pl.Float64), + 'high': pl.Series(np.arange(size) + 2, dtype=pl.Float64), + 'low': pl.Series(np.arange(size), dtype=pl.Float64), + 'close': pl.Series(np.arange(size) + 1, dtype=pl.Float64), + 'volume': pl.Series(np.arange(size) + 10, dtype=pl.Float64), + }) + + +def test_audit_separates_structure_from_positive_gaps() -> None: + ''' + Expected closures must not masquerade as structural corruption. + + Persisted CME and session-market history can be canonical while + containing positive timestamp gaps. Build a structurally valid + frame with one aligned interval and prove validity remains green + while exact endpoints and unresolved coverage remain visible. + + ''' + report = audit_ohlcv_frame( + mk_frame((60, 120, 300)), + fqme='mnq.cme.20260918.ib', + period_s=60, + ) + + assert report['result'] == { + 'structural_ok': True, + 'gap_free': False, + 'qualification_ok': False, + 'violations': [], + 'warnings': ['positive_time_gaps_unclassified'], + } + assert report['gaps']['count'] == 1 + assert report['gaps']['missing_samples_total'] == 2 + assert report['gaps']['intervals'][0] == { + 'left_timestamp': 120, + 'right_timestamp': 300, + 'left_utc': '1970-01-01T00:02:00Z', + 'right_utc': '1970-01-01T00:05:00Z', + 'delta_s': 180, + 'period_multiple': True, + 'missing_samples': 2, + 'classification': 'unclassified', + } + + +def test_audit_preserves_raw_defect_evidence() -> None: + ''' + Audit must count malformed rows without repairing or sorting. + + A baseline can contain extra provider columns, duplicated and + reversed timestamps, a zero epoch, broken indexes, and non-finite + values simultaneously. Construct all defects in physical file + order and prove stable violations report each layer instead of + hiding them through NativeDB canonicalization or dedupe. + + ''' + frame = mk_frame((60, 60, 30, 0)).with_columns( + pl.Series('index', [0, 2, 2, 4]), + pl.Series('open', [1, 2, 3, 4], dtype=pl.Int64), + pl.Series( + 'close', + [1, float('nan'), 3, 4], + dtype=pl.Float64, + ), + pl.Series( + 'volume', + [1, 2, float('inf'), 4], + dtype=pl.Float64, + ), + pl.Series('count', [1, 1, 1, 1]), + ) + report = audit_ohlcv_frame( + frame, + fqme='x.test', + period_s=60, + ) + + assert set(report['result']['violations']) == { + 'extra_columns', + 'column_order', + 'canonical_dtypes', + 'nonfinite_ohlcv', + 'nonpositive_timestamps', + 'duplicate_timestamps', + 'timestamp_order', + 'index_not_canonical', + 'subperiod_time_step', + } + assert report['timestamps']['duplicate_excess'] == 1 + assert report['timestamps']['adjacent_zero_delta'] == 1 + assert report['timestamps']['negative_delta'] == 2 + assert report['index']['row_position_mismatch'] == 2 + assert report['values']['nan_by_column']['close'] == 1 + assert report['values']['infinity_by_column']['volume'] == 1 + + +def test_gap_aggregates_are_not_truncated_with_details() -> None: + ''' + Limiting JSON detail must not undercount total missing coverage. + + Long-lived session markets can contain thousands of expected + closure intervals. Arrange two gaps but request one detail, + record, then prove aggregate counts still describe the complete + frame while the bounded detail list is marked truncated. + + ''' + report = audit_ohlcv_frame( + mk_frame((60, 180, 300)), + fqme='x.test', + period_s=60, + max_gaps=1, + ) + gaps: dict = report['gaps'] + + assert gaps['count'] == 2 + assert gaps['aligned_count'] == 2 + assert gaps['missing_samples_total'] == 2 + assert gaps['details_count'] == 1 + assert gaps['details_truncated'] is True + + +def test_subperiod_steps_fail_cadence_qualification() -> None: + ''' + Short positive deltas must not bypass expected-period validation. + + Gap detection alone considers deltas larger than the expected + period. That let a 30-second step in 60-second data qualify. + Arrange that defect and prove it remains separate from positive + gaps while making structural qualification fail. + + ''' + report = audit_ohlcv_frame( + mk_frame((60, 90)), + fqme='x.test', + period_s=60, + ) + + assert report['gaps']['count'] == 0 + assert report['gaps']['subperiod_count'] == 1 + assert report['gaps']['subperiod_intervals'][0]['delta_s'] == 30 + assert 'subperiod_time_step' in report['result']['violations'] + assert report['result']['qualification_ok'] is False + + +def test_integer_evidence_preserves_values_above_float_precision( +) -> None: + ''' + Int64 timestamp and index evidence must not round through float. + + Adjacent integers above ``2**53`` collapse when coerced to + Float64, inventing duplicate times and index mismatches. Audit a + canonical two-row frame at that boundary and prove exact + endpoints, uniqueness, and indexes remain JSON-native integers. + + ''' + start: int = 2**53 + frame = mk_frame((start, start + 1)).with_columns( + pl.Series('index', [start, start + 1], dtype=pl.Int64) + ) + report = audit_ohlcv_frame( + frame, + fqme='x.test', + period_s=1, + ) + + assert report['timestamps']['minimum'] == start + assert report['timestamps']['maximum'] == start + 1 + assert report['timestamps']['duplicate_excess'] == 0 + assert report['timestamps']['strictly_increasing'] is True + assert report['index']['first'] == start + assert report['index']['last'] == start + 1 + assert report['index']['duplicate_excess'] == 0 + assert report['index']['contiguous'] is True + assert report['index']['canonical_from_zero'] is False + + +def test_large_integer_gap_counts_remain_exact() -> None: + ''' + Missing-sample totals must not divide Int64 deltas through float. + + A near-Int64 interval exceeds Float64's exact range. Audit it at + one-second cadence and prove all counts agree with exact integer + division rather than rounded evidence. + + ''' + right: int = 2**63 - 1 + report = audit_ohlcv_frame( + mk_frame((1, right)), + fqme='x.test', + period_s=1, + ) + expected: int = right - 2 + + assert report['gaps']['missing_samples_total'] == expected + assert ( + report['gaps']['intervals'][0]['missing_samples'] + == + expected + ) + + +def test_decimal_columns_remain_auditable_defect_evidence() -> None: + ''' + Numeric but noncanonical Polars dtypes must produce a report. + + Decimal Parquet columns are readable numeric evidence but do not + support Polars ``is_nan()``. Cast timestamps and an OHLC field to + Decimal and prove audit reports the dtype violation instead of + crashing before malformed baseline evidence can be saved. + + ''' + frame = mk_frame((60, 120)).with_columns( + pl.col('time').cast(pl.Decimal(scale=0)), + pl.col('close').cast(pl.Decimal(scale=2)), + ) + report = audit_ohlcv_frame( + frame, + fqme='x.test', + period_s=60, + ) + + assert report['timestamps']['numeric'] is True + assert report['timestamps']['strictly_increasing'] is True + assert report['values']['all_finite'] is True + assert 'canonical_dtypes' in report['result']['violations'] + + +def test_empty_frame_has_unverifiable_gap_coverage() -> None: + ''' + Empty storage must not claim valid cadence or index results. + + An empty canonical schema is structurally invalid and has no pair + of timestamps from which gap coverage can be inferred. Prove the + report retains its stable gap shape while gap-free status remains + unknown and canonical index qualification is rejected. + + ''' + report = audit_ohlcv_frame( + mk_frame(()), + fqme='x.test', + period_s=60, + ) + + assert report['gaps']['verifiable'] is False + assert report['gaps']['aligned_count'] == 0 + assert report['result']['gap_free'] is None + assert report['index']['canonical_from_zero'] is False + assert report['result']['qualification_ok'] is False + + +def test_detail_limit_bounds_all_cadence_deviations() -> None: + ''' + One detail budget must cover gaps and short steps together. + + Human output combines both anomaly types. Slicing each list can + exceed ``--max-gaps`` and misreport the detail count. Arrange one + short step and one long gap with a budget of one and prove the + combined detail output remains bounded and marked truncated. + + ''' + report = audit_ohlcv_frame( + mk_frame((60, 90, 210)), + fqme='x.test', + period_s=60, + max_gaps=1, + ) + gaps: dict = report['gaps'] + + assert gaps['count'] == 1 + assert gaps['subperiod_count'] == 1 + assert gaps['details_count'] == 1 + assert ( + len(gaps['intervals']) + + len(gaps['subperiod_intervals']) + ) == 1 + assert gaps['details_truncated'] is True + + +def test_store_audit_json_is_read_only( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + The CLI must inspect exact bytes without storage side effects. + + Operational storage openers can create config directories and + mutate caches. Point config at an existing disposable NativeDB, + invoke the real Typer command, and prove its JSON references the + exact file while bytes and modification time stay unchanged. + + ''' + monkeypatch.setattr(config, '_config_dir', tmp_path) + datadir: Path = tmp_path / 'nativedb' + datadir.mkdir() + fqme: str = 'x.test' + path: Path = mk_ohlcv_shm_keyed_filepath( + fqme, + 60, + datadir, + ) + mk_frame((60, 120)).write_parquet(path) + before_bytes: bytes = path.read_bytes() + before_mtime: int = path.stat().st_mtime_ns + output: Path = tmp_path / 'audit.json' + snapshot: Path = tmp_path / 'audit.parquet' + + result = CliRunner().invoke( + store, + [ + 'audit', + fqme, + '--period', + '60', + '--output', + str(output), + '--snapshot', + str(snapshot), + '--json', + ], + ) + report: dict = json.loads(result.stdout) + + assert result.exit_code == 0 + assert report['source']['path'] == str(path) + assert report['source']['sha256'] == sha256( + before_bytes + ).hexdigest() + assert report['result']['qualification_ok'] is True + assert path.read_bytes() == before_bytes + assert path.stat().st_mtime_ns == before_mtime + assert snapshot.read_bytes() == before_bytes + assert json.loads(output.read_text()) == report + assert report['schema']['columns'] == [ + name + for name, _ in def_iohlcv_fields + ] + + +def test_store_audit_refuses_output_collisions( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Evidence output must not replace source or existing files. + + The first CLI accepted any ``--output`` path and could overwrite + audited Parquet with JSON. Point output at the source, a report, + and a directory; prove each exits with original bytes unchanged. + + ''' + monkeypatch.setattr(config, '_config_dir', tmp_path) + datadir: Path = tmp_path / 'nativedb' + datadir.mkdir() + fqme: str = 'x.test' + path: Path = mk_ohlcv_shm_keyed_filepath( + fqme, + 60, + datadir, + ) + mk_frame((60, 120)).write_parquet(path) + before: bytes = path.read_bytes() + runner = CliRunner() + + for output in (path, tmp_path / 'existing.json', tmp_path): + if output.name == 'existing.json': + output.write_text('keep\n') + result = runner.invoke( + store, + ['audit', fqme, '--output', str(output)], + ) + assert result.exit_code == 2 + + assert path.read_bytes() == before + assert (tmp_path / 'existing.json').read_text() == 'keep\n' + + +def test_direct_snapshot_collision_preserves_existing_file( + tmp_path: Path, +) -> None: + ''' + Exclusive snapshot races must never unlink another writer's file. + + The helper opens snapshots with ``xb``. Its first cleanup path + unlinked the destination when exclusive open failed. Pre-create a + destination and prove the collision preserves its existing bytes. + + ''' + source: Path = tmp_path / 'source.parquet' + snapshot: Path = tmp_path / 'snapshot.parquet' + mk_frame((60, 120)).write_parquet(source) + snapshot.write_bytes(b'existing evidence') + + with pytest.raises(FileExistsError): + audit_ohlcv_parquet( + source, + fqme='x.test', + period_s=60, + snapshot=snapshot, + ) + + assert snapshot.read_bytes() == b'existing evidence' + + +def test_invalid_parquet_is_snapshotted_before_parse( + tmp_path: Path, +) -> None: + ''' + Unreadable phase bytes must survive a failed structural audit. + + A truncated or non-Parquet file previously failed during parsing + before ``--snapshot`` captured anything, discarding exact failure + evidence from repair and restart. Supply invalid bytes to the + low-level audit and prove it raises only after creating an exact, + read-only snapshot for later diagnosis. + + ''' + source: Path = tmp_path / 'broken.parquet' + snapshot: Path = tmp_path / 'snapshot.parquet' + evidence: bytes = b'not a parquet file\n' + source.write_bytes(evidence) + + with pytest.raises(pl.exceptions.PolarsError): + audit_ohlcv_parquet( + source, + fqme='x.test', + period_s=60, + snapshot=snapshot, + ) + + assert snapshot.read_bytes() == evidence + assert snapshot.stat().st_mode & 0o222 == 0 + + +def test_snapshot_copy_failure_removes_partial_output( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Copy failure must not leave partial authoritative evidence. + + Snapshot bytes are captured before parsing so malformed evidence + survives. Failed copies leave partial output that blocks retry. + Raise after a prefix. Prove partial output is removed while the + source stays whole. + + ''' + source: Path = tmp_path / 'source.parquet' + snapshot: Path = tmp_path / 'snapshot.parquet' + mk_frame((60, 120)).write_parquet(source) + before: bytes = source.read_bytes() + + def fail_copy(source_file, snapshot_file) -> None: + snapshot_file.write(source_file.read(10)) + raise OSError('simulated full filesystem') + + monkeypatch.setattr(audit_mod.shutil, 'copyfileobj', fail_copy) + with pytest.raises(OSError, match='full filesystem'): + audit_ohlcv_parquet( + source, + fqme='x.test', + period_s=60, + snapshot=snapshot, + ) + + assert not snapshot.exists() + assert source.read_bytes() == before + + +def test_store_audit_rejects_unsafe_or_symlinked_sources( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + User-controlled FQME text must stay inside NativeDB storage. + + Absolute and parent-relative market names previously flowed into + path construction, while a normal market could be a symlink to + external bytes. Point config at a disposable NativeDB and prove + traversal and source links fail before external bytes change. + + ''' + monkeypatch.setattr(config, '_config_dir', tmp_path) + datadir: Path = tmp_path / 'nativedb' + datadir.mkdir() + outside: Path = tmp_path / 'outside.parquet' + mk_frame((60, 120)).write_parquet(outside) + before: bytes = outside.read_bytes() + source: Path = mk_ohlcv_shm_keyed_filepath( + 'x.test', + 60, + datadir, + ) + source.symlink_to(outside) + runner = CliRunner() + + for fqme in ('../../outside', str(outside), 'x.test'): + result = runner.invoke(store, ['audit', fqme]) + assert result.exit_code == 2 + + assert outside.read_bytes() == before + + +def test_direct_audit_rejects_nonregular_source( + tmp_path: Path, +) -> None: + ''' + Audit must not read devices, directories, or blocking pipes. + + ``O_NOFOLLOW`` rejects links but opens other node types. Pass a + a directory to the reader and prove descriptor metadata + rejects it before hashing or handing it to the Parquet parser. + + ''' + with pytest.raises(ValueError, match='regular file'): + audit_ohlcv_parquet( + tmp_path, + fqme='x.test', + period_s=60, + )