diff --git a/.claude/skills/run-tests/test-harness-reference.md b/.claude/skills/run-tests/test-harness-reference.md index 3b2f9cb6..1ff5a75d 100644 --- a/.claude/skills/run-tests/test-harness-reference.md +++ b/.claude/skills/run-tests/test-harness-reference.md @@ -105,6 +105,7 @@ Deterministic or local first-pass targets: - `tests/test_watchlists.py` - `tests/test_storage_audit.py` +- `tests/test_store_cli.py` - `tests/test_backfill_audit_snippet.py` - `tests/test_ib_history.py` - `tests/test_history_backfill.py` @@ -196,6 +197,7 @@ tests/ test_ldshm.py ldshm unpublished-slot guard test_questrade.py obsolete credentialed tests; skipped test_services.py pikerd/datad/feed/EMS actor lifecycle + test_store_cli.py storage command help and diagnostics UX test_storage_audit.py read-only NativeDB audit and JSON CLI test_backfill_audit_snippet.py disposable xonsh qualification helpers @@ -208,6 +210,7 @@ tests/ |---|---|---| | `piker/watchlists/` | `tests/test_watchlists.py` | CLI suite is skipped | | `piker/storage/_audit.py`, `piker/storage/cli.py` | `tests/test_storage_audit.py` | direct Typer app, no actor | +| `piker/storage/cli.py` command UX | `tests/test_store_cli.py` | fake SHM/runtime, no mutation | | `snippets/nativedb_backfill_audit.xsh` | `tests/test_backfill_audit_snippet.py` | disposable paths only | | `piker/brokers/ib/api.py`, `feed.py` history | `tests/test_ib_history.py` | fake client, no network | | `piker/tsp/_history.py` | `tests/test_history_backfill.py` | fake provider/storage/SHM | diff --git a/ai/prompt-io/opencode/20260731T165215Z_0846cbd4_prompt_io.md b/ai/prompt-io/opencode/20260731T165215Z_0846cbd4_prompt_io.md new file mode 100644 index 00000000..3761852f --- /dev/null +++ b/ai/prompt-io/opencode/20260731T165215Z_0846cbd4_prompt_io.md @@ -0,0 +1,41 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +session: ses_0799212ebffe42arY96czXn89F +timestamp: 2026-07-31T16:52:15Z +git_ref: 0846cbd4 +scope: code +substantive: true +raw_file: 20260731T165215Z_0846cbd4_prompt_io.raw.md +--- + +## Prompt + +Evaluate and update `piker store` so bare groups and endpoints immediately +show help; prepare for eventual optional fuzzy selection in `anal` and +`ldshm`; aggregate recent time-series diagnostics into CLI endpoints. The +user selected the safe-foundation scope. + +## Response summary + +Added immediate help UX, exact NativeDB and SHM identities, read-only +`series` and `shm` diagnostics, selector-ready exact SHM names, and robust +discovery/attachment. Fixed related `ldshm` option and no-match failures. +Deferred optional `fzf` and mutation-default changes. + +## Files changed + +- `piker/storage/cli.py` - help UX and read-only series/SHM diagnostics. +- `piker/storage/nativedb.py` - canonical compound series identities. +- `piker/tsp/_history.py` - exact SHM discovery and dtype resolution. +- `piker/data/_sharedmem.py` - attach-only existing-segment helper. +- `piker/tsp/__init__.py` - SHM identity helper exports. +- `tests/test_store_cli.py` - command UX and endpoint regressions. +- `tests/test_ldshm.py` - SHM diagnostics and operational regressions. +- `tests/test_storage_nativedb.py` - series parsing/index regressions. +- `docs/manual_backfill_qualification.rst` - triage command documentation. +- `.claude/skills/run-tests/test-harness-reference.md` - test mapping. + +## Human edits + +None - generated output remains uncommitted. diff --git a/ai/prompt-io/opencode/20260731T165215Z_0846cbd4_prompt_io.raw.md b/ai/prompt-io/opencode/20260731T165215Z_0846cbd4_prompt_io.raw.md new file mode 100644 index 00000000..eac6fd83 --- /dev/null +++ b/ai/prompt-io/opencode/20260731T165215Z_0846cbd4_prompt_io.raw.md @@ -0,0 +1,93 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +timestamp: 2026-07-31T16:52:15Z +git_ref: 0846cbd4 +diff_cmd: git diff HEAD~1..HEAD +--- + +The user requested an evaluation and possible update of `piker store`, +including immediate help output when a group or endpoint is invoked with +no arguments. The user also proposed eventual optional `fzf` selection for +`anal` and `ldshm`, and authorized aggregating the recent time-series and +SHM debugging work into CLI endpoints. After reviewing three scopes, the +user selected the recommended safe foundation rather than implementing +interactive fuzzy selection or changing mutation defaults immediately. + +> `git diff HEAD~1..HEAD -- piker/storage/cli.py` + +Generated no-argument help behavior for the `store` group and every +endpoint. Added read-only `series` and `shm` commands, JSON and Rich output, +exact `--shm-name` plumbing, SHM cadence/invalid/order/gap summaries, and +safe corrupt-epoch formatting. Fixed existing `ldshm` no-write/no-reload +control flow and no-match debugger behavior. + +> `git diff HEAD~1..HEAD -- piker/storage/nativedb.py` + +Generated canonical `NativeSeriesRef` parsing and deterministic listing. +Changed the internal index to preserve `(fqme, period)` identities while +keeping unique-FQME compatibility listing and cardinality. Hardened +malformed, sidecar, Unicode-period, symlink, and traversal handling. + +> `git diff HEAD~1..HEAD -- piker/tsp/_history.py` + +Generated exact `ShmBufferRef` parsing and stable discovery, replacing +unsafe substring globbing. Added dtype resolution that mirrors history +allocation, exact-ref attachment, disappearance handling, and optional +exact-name iteration. + +> `git diff HEAD~1..HEAD -- piker/data/_sharedmem.py` + +Generated attach-only SHM access which never creates a segment when a +discovered object disappears. + +> `git diff HEAD~1..HEAD -- piker/tsp/__init__.py` + +Exported selector-ready SHM identity, parsing, listing, and dtype helpers. + +> `git diff HEAD~1..HEAD -- tests/test_store_cli.py` + +Generated command-level help, exact durable-series JSON, read-only SHM JSON, +and unknown exact-name regressions. + +> `git diff HEAD~1..HEAD -- tests/test_ldshm.py` + +Generated SHM identity, summary, ordering, dtype, forged-name, disabled +mutation-flag, and gap-markup regressions. + +> `git diff HEAD~1..HEAD -- tests/test_storage_nativedb.py` + +Generated compound index, canonical parser, malformed filename, and dotted +FQME regressions. + +> `git diff HEAD~1..HEAD -- docs/manual_backfill_qualification.rst` + +Documented `store series` and `store shm` as separate read-only durable and +live-memory triage layers. + +> `git diff HEAD~1..HEAD -- .claude/skills/run-tests/test-harness-reference.md` + +Registered the deterministic storage CLI UX target and mapping. + +Verification output: + +```text +........................................................................ [ 71%] +............................. [100%] +101 passed in 1.72s +``` + +Real `piker store` and `piker store audit` invocations printed complete +help without `--help`, exiting with Click's normal no-args help status 2. +`git diff --check` passed. Ruff was unavailable in the existing worktree +environment. + +Adversarial review drove fixes for provider dtype reconstruction, observed +versus expected cadence, exact SHM grammar, discovery races, Typer direct +defaults, malformed NativeDB periods, compound cardinality compatibility, +undefined `ldshm` reload payloads, no-match debugger hangs, ordering +corruption reporting, and dtype precedence. Final review found no issues. + +Optional external `fzf` or built-in numbered selection remains deferred. +The exact series and SHM identities added here are the intended foundation +for that later patch. diff --git a/docs/manual_backfill_qualification.rst b/docs/manual_backfill_qualification.rst index 7f67ae54..7ccd183c 100644 --- a/docs/manual_backfill_qualification.rst +++ b/docs/manual_backfill_qualification.rst @@ -193,6 +193,18 @@ visible until a venue-aware classifier proves their session alignment. Gap Layer Triage ---------------- +List exact durable series before choosing a timeframe, and inspect live +SHM separately:: + + piker store series FQME_SUBSTRING + piker store shm FQME --max-gaps 20 + +Both commands are read-only. ``series`` reports each ``(FQME, period)`` +file independently. ``shm`` reports every exact actor generation and +``hist``/``rt`` buffer for the FQME, including invalid rows, ordering +defects, inferred versus expected cadence, and the largest timestamp gaps. +Use ``--shm-name EXACT_NAME`` to inspect only one reported buffer. + Use the persisted report and chart together: =============================== ========================================= diff --git a/piker/data/_sharedmem.py b/piker/data/_sharedmem.py index d894d392..269a2985 100644 --- a/piker/data/_sharedmem.py +++ b/piker/data/_sharedmem.py @@ -63,6 +63,28 @@ def _make_token( ) +def attach_existing_shm_array( + key: str, + size: int, + dtype: np.dtype|None = None, + readonly: bool = True, + +) -> ShmArray: + ''' + Attach to an existing segment without creating one on races. + + ''' + token: NDToken = _known_tokens.get(key) or _make_token( + key, + size=size, + dtype=dtype, + ) + return attach_shm_ndarray( + token=token, + readonly=readonly, + ) + + def maybe_open_shm_array( key: str, size: int, diff --git a/piker/storage/cli.py b/piker/storage/cli.py index 8129a4d9..663dfd78 100644 --- a/piker/storage/cli.py +++ b/piker/storage/cli.py @@ -19,12 +19,17 @@ Storage middle-ware CLIs. """ from __future__ import annotations +from datetime import ( + UTC, + datetime, +) import json from pathlib import Path import sys import time from types import ModuleType from typing import ( + Annotated, TYPE_CHECKING, ) @@ -51,13 +56,39 @@ from . import ( StorageClient, ) from ._audit import audit_ohlcv_parquet -from .nativedb import mk_ohlcv_shm_keyed_filepath +from .nativedb import ( + iter_native_series, + mk_ohlcv_shm_keyed_filepath, + NativeSeriesRef, +) if TYPE_CHECKING: from piker.ui._remote_ctl import AnnotCtl -store = typer.Typer() +store = typer.Typer(no_args_is_help=True) +_shm_periods: dict[str, int] = { + 'hist': 60, + 'rt': 1, +} + + +def _format_utc(timestamp: float) -> str|None: + ''' + Format a timestamp without aborting on corrupt epochs. + + ''' + try: + return datetime.fromtimestamp( + timestamp, + tz=UTC, + ).isoformat() + except ( + OSError, + OverflowError, + ValueError, + ): + return None def _shm_period_and_invalid_count( @@ -91,6 +122,95 @@ def _shm_period_and_invalid_count( return period, invalid_count +def _summarize_shm_frame( + ref: tsp.ShmBufferRef, + frame: np.ndarray, + max_gaps: int, + +) -> dict: + ''' + Summarize one immutable OHLCV SHM snapshot. + + ''' + times: np.ndarray = frame['time'] + observed_period_s, invalid_count = ( + _shm_period_and_invalid_count(times) + ) + period_s: int = _shm_periods[ref.kind] + valid: np.ndarray = ( + np.isfinite(times) + & + (times > 0) + ) + published: np.ndarray = times[valid] + steps: np.ndarray = np.diff(published) + duplicate_steps: int = int(np.count_nonzero(steps == 0)) + reversed_steps: int = int(np.count_nonzero(steps < 0)) + gaps: list[dict] = [] + if published.size > 1: + indexes: np.ndarray = np.flatnonzero(steps > period_s) + ranked: list[int] = sorted( + indexes.tolist(), + key=lambda index: steps[index], + reverse=True, + ) + for index in ranked[:max_gaps]: + left: float = float(published[index]) + right: float = float(published[index + 1]) + gaps.append({ + 'delta_s': float(steps[index]), + 'left_timestamp': left, + 'left_utc': _format_utc(left), + 'right_timestamp': right, + 'right_utc': _format_utc(right), + }) + + invalid: np.ndarray = np.flatnonzero(~valid) + return { + 'shm_name': ref.path.name, + 'service': ref.service, + 'generation': ref.generation, + 'fqme': ref.fqme, + 'kind': ref.kind, + 'rows': len(frame), + 'first_index': ( + int(frame['index'][0]) + if len(frame) + else None + ), + 'last_index': ( + int(frame['index'][-1]) + if len(frame) + else None + ), + 'first_timestamp': ( + float(published[0]) + if published.size + else None + ), + 'last_timestamp': ( + float(published[-1]) + if published.size + else None + ), + 'period_s': period_s, + 'observed_period_s': observed_period_s, + 'invalid_count': invalid_count, + 'duplicate_step_count': duplicate_steps, + 'reversed_step_count': reversed_steps, + 'invalid_index_bounds': ( + [ + int(frame['index'][invalid[0]]), + int(frame['index'][invalid[-1]]), + ] + if invalid.size + else None + ), + 'gap_count': len(np.flatnonzero(steps > period_s)), + 'largest_gaps': gaps, + } + + def _render_audit_report(report: dict) -> None: ''' Render a compact human summary and explicit gap endpoints. @@ -150,17 +270,34 @@ def _render_audit_report(report: dict) -> None: console.print(gap_table) -@store.command() +@store.command(no_args_is_help=True) def ls( - backends: list[str] = typer.Argument( - default=None, - help='Storage backends to query, default is all.' - ), + backends: Annotated[ + list[str]|None, + typer.Argument(help='Storage backends to query.'), + ] = None, + all_backends: Annotated[ + bool, + typer.Option( + '--all', + help='Query every configured storage backend.', + ), + ] = False, ): from rich.table import Table - if not backends: - backends: list[str] = __tsdbs__ + if ( + all_backends + and + backends + ): + raise typer.BadParameter( + 'Pass backend names or --all, not both' + ) + if all_backends: + backends = list(__tsdbs__) + elif not backends: + raise typer.BadParameter('Pass backend names or --all') console = Console() @@ -191,7 +328,88 @@ def ls( trio.run(query_all) -@store.command() +@store.command(no_args_is_help=True) +def series( + pattern: Annotated[ + str|None, + typer.Argument( + help='Optional case-insensitive FQME substring.', + ), + ] = None, + all_series: Annotated[ + bool, + typer.Option( + '--all', + help='List every canonical NativeDB series.', + ), + ] = False, + json_output: Annotated[ + bool, + typer.Option( + '--json', + help='Emit machine-readable JSON.', + ), + ] = False, +) -> None: + ''' + List exact NativeDB series identities without opening a runtime. + + ''' + if ( + all_series + and + pattern is not None + ): + raise typer.BadParameter('Pass a pattern or --all, not both') + if ( + not all_series + and + pattern is None + ): + raise typer.BadParameter('Pass a pattern or --all') + + datadir: Path = config.get_conf_dir() / 'nativedb' + refs: list[NativeSeriesRef] = list(iter_native_series(datadir)) + if pattern is not None: + needle: str = pattern.casefold() + refs = [ + ref + for ref in refs + if needle in ref.fqme.casefold() + ] + + payload: list[dict] = [] + for ref in refs: + try: + size_bytes: int = ref.path.stat().st_size + except FileNotFoundError: + continue + payload.append({ + 'fqme': ref.fqme, + 'period_s': ref.period_s, + 'path': str(ref.path), + 'size_bytes': size_bytes, + }) + if json_output: + typer.echo(json.dumps(payload, indent=2, sort_keys=True)) + return + + table = Table(title=f'NativeDB series @ {datadir}') + table.add_column('FQME') + table.add_column('Period') + table.add_column('Bytes', justify='right') + table.add_column('Path') + for item in payload: + table.add_row( + item['fqme'], + f'{item["period_s"]}s', + str(item['size_bytes']), + item['path'], + ) + Console().print(table) + + +@store.command(no_args_is_help=True) def audit( fqme: str, period: int = typer.Option( @@ -367,7 +585,7 @@ def audit( # ... -@store.command() +@store.command(no_args_is_help=True) def delete( symbols: list[str], @@ -405,7 +623,7 @@ def delete( trio.run(main, symbols) -@store.command() +@store.command(no_args_is_help=True) def anal( fqme: str, period: int = 60, @@ -494,12 +712,146 @@ def anal( trio.run(main) -@store.command() +@store.command('shm', no_args_is_help=True) +def inspect_shm( + fqme: str, + shm_name: Annotated[ + str|None, + typer.Option( + '--shm-name', + help='Inspect only this exact SHM object name.', + ), + ] = None, + max_gaps: Annotated[ + int, + typer.Option( + '--max-gaps', + min=0, + help='Maximum largest gaps to show per buffer.', + ), + ] = 10, + json_output: Annotated[ + bool, + typer.Option( + '--json', + help='Emit machine-readable JSON.', + ), + ] = False, +) -> None: + ''' + Inspect matching OHLCV SHM buffers without mutating them. + + ''' + refs: list[tsp.ShmBufferRef] = [ + ref + for ref in tsp.iter_shm_buffer_refs(fqme=fqme) + if ( + shm_name is None + or + ref.path.name == shm_name + ) + ] + if not refs: + typer.echo( + f'No exact OHLCV SHM buffers found for {fqme!r}', + err=True, + ) + raise typer.Exit(code=2) + + ref_by_name: dict[str, tsp.ShmBufferRef] = { + ref.path.name: ref + for ref in refs + } + reports: list[dict] = [] + + async def main() -> None: + async with open_piker_runtime('shm_inspector'): + for ( + shmfile, + shm, + _shm_df, + ) in tsp.iter_dfs_from_shms( + fqme, + shm_name=shm_name, + refs=refs, + ): + ref: tsp.ShmBufferRef = ref_by_name[shmfile.name] + reports.append(_summarize_shm_frame( + ref, + shm.array.copy(), + max_gaps, + )) + + trio.run(main) + if not reports: + typer.echo( + 'Selected SHM buffers disappeared before inspection', + err=True, + ) + raise typer.Exit(code=2) + if json_output: + typer.echo(json.dumps(reports, indent=2, sort_keys=True)) + return + + table = Table(title=f'OHLCV SHM buffers for {fqme}') + table.add_column('Kind') + table.add_column('Generation') + table.add_column('Rows', justify='right') + table.add_column('Period') + table.add_column('Invalid', justify='right') + table.add_column('Order', justify='right') + table.add_column('Gaps', justify='right') + table.add_column('SHM name') + for report in reports: + period_s: int = report['period_s'] + table.add_row( + report['kind'], + report['generation'], + str(report['rows']), + f'{period_s}s', + str(report['invalid_count']), + str( + report['duplicate_step_count'] + + + report['reversed_step_count'] + ), + str(report['gap_count']), + report['shm_name'], + ) + Console().print(table) + + for report in reports: + gaps: list[dict] = report['largest_gaps'] + if not gaps: + continue + gap_table = Table( + title=f'Largest gaps: {report["shm_name"]}' + ) + gap_table.add_column('Delta (s)') + gap_table.add_column('Left UTC') + gap_table.add_column('Right UTC') + for gap in gaps: + gap_table.add_row( + str(gap['delta_s']), + str(gap['left_utc']), + str(gap['right_utc']), + ) + Console().print(gap_table) + + +@store.command(no_args_is_help=True) def ldshm( fqme: str, write_parquet: bool = True, reload_parquet_to_shm: bool = True, pdb: bool = False, # --pdb passed? + shm_name: Annotated[ + str|None, + typer.Option( + '--shm-name', + help='Process only this exact SHM object name.', + ), + ] = None, ) -> None: ''' @@ -508,7 +860,21 @@ def ldshm( optionally write to offline storage via `.parquet` file. ''' - async def main(): + if ( + shm_name is not None + and + not any( + ref.path.name == shm_name + for ref in tsp.iter_shm_buffer_refs(fqme=fqme) + ) + ): + typer.echo( + f'No exact OHLCV SHM buffer named {shm_name!r}', + err=True, + ) + raise typer.Exit(code=2) + + async def main() -> bool: from piker.ui._remote_ctl import ( open_annot_ctl, ) @@ -530,24 +896,35 @@ def ldshm( shm_df: pl.DataFrame | None = None tf2aids: dict[float, dict] = {} + iter_kwargs: dict = {} + if shm_name is not None: + iter_kwargs['shm_name'] = shm_name for ( shmfile, shm, # parquet_path, shm_df, - ) in tsp.iter_dfs_from_shms(fqme): + ) in tsp.iter_dfs_from_shms(fqme, **iter_kwargs): times: np.ndarray = shm_df['time'].to_numpy() ( - period_s, + observed_period_s, invalid_count, ) = _shm_period_and_invalid_count(times) - if period_s is None: + if observed_period_s is None: log.warning( f'Could not infer a positive sample period ' f'for {shmfile.name}; skipping buffer\n' ) continue + kind: str = shmfile.suffix.removeprefix('.') + period_s: int = _shm_periods[kind] + if observed_period_s != period_s: + log.warning( + f'Modal step {observed_period_s}s in ' + f'{period_s}s {shmfile.name}; using kind ' + f'for storage identity\n' + ) log.info( f'Processing shm buffer:\n' f' file: {shmfile.name}\n' @@ -696,12 +1073,11 @@ def ldshm( do_markup_gaps: bool = True if do_markup_gaps: - new_df: pl.DataFrame = tsp.np2pl(new) aids: dict = await tsp._annotate.markup_gaps( fqme, period_s, actl, - new_df, + deduped, step_gaps, ) # last chance manual overwrites in REPL @@ -724,16 +1100,19 @@ def ldshm( 'but no significant time gaps!\n' ) - await tractor.pause() - log.info('Exiting TSP shm anal-izer!') - if shm_df is None: log.error( f'No matching shm buffers for {fqme} ?' ) + return False - trio.run(main) + await tractor.pause() + log.info('Exiting TSP shm anal-izer!') + return True + + if not trio.run(main): + raise typer.Exit(code=2) typer_click_object = typer.main.get_command(store) diff --git a/piker/storage/nativedb.py b/piker/storage/nativedb.py index 6ad51b00..077dab04 100644 --- a/piker/storage/nativedb.py +++ b/piker/storage/nativedb.py @@ -51,7 +51,9 @@ YET! # - https://github.com/spslater/borgapi # - https://nixos.wiki/wiki/ZFS from __future__ import annotations +from collections.abc import Iterator from contextlib import asynccontextmanager as acm +from dataclasses import dataclass from datetime import datetime import os from pathlib import Path @@ -82,6 +84,17 @@ _price_fields: tuple[str, ...] = ( ) +@dataclass(frozen=True) +class NativeSeriesRef: + ''' + Identify one canonical NativeDB OHLCV file. + + ''' + fqme: str + period_s: int + path: Path + + def detect_period(shm: ShmArray) -> float: ''' Attempt to detect the series time step sampling period @@ -107,8 +120,14 @@ def mk_ohlcv_shm_keyed_filepath( ) -> Path: - if period < 1.: - raise ValueError('Sample period should be >= 1.!?') + if ( + period < 1. + or + int(period) != period + ): + raise ValueError( + 'Sample period must be positive whole seconds' + ) path: Path = ( datadir @@ -118,12 +137,98 @@ def mk_ohlcv_shm_keyed_filepath( return path -def unpack_fqme_from_parquet_filepath(path: Path) -> str: +def parse_ohlcv_parquet_path( + path: Path, - filename: str = str(path.name) - fqme, fmt_descr, suffix = filename.split('.') - assert suffix == 'parquet' - return fqme +) -> NativeSeriesRef|None: + ''' + Parse one canonical NativeDB OHLCV filename. + + ''' + name: str = path.name + if not name.endswith('.parquet'): + return None + + stem: str = name.removesuffix('.parquet') + fqme, marker, period_text = stem.rpartition('.ohlcv') + if ( + marker != '.ohlcv' + or + not fqme + or + not period_text.endswith('s') + ): + return None + + digits: str = period_text.removesuffix('s') + if ( + not digits.isascii() + or + not digits.isdecimal() + or + digits.startswith('0') + or + '..' in path.parts + ): + return None + + period_s: int = int(digits) + if ( + fqme in {'.', '..'} + or + Path(fqme).name != fqme + or + '/' in fqme + or + '\\' in fqme + ): + return None + + expected: Path = mk_ohlcv_shm_keyed_filepath( + fqme, + period_s, + path.parent, + ) + if expected.name != name: + return None + + return NativeSeriesRef( + fqme=fqme, + period_s=period_s, + path=path, + ) + + +def iter_native_series( + datadir: Path, +) -> Iterator[NativeSeriesRef]: + ''' + Yield canonical NativeDB series in deterministic order. + + ''' + if not datadir.is_dir(): + return + + for path in sorted( + datadir.iterdir(), + key=lambda candidate: candidate.name, + ): + if ( + path.is_symlink() + or + not path.is_file() + ): + continue + ref: NativeSeriesRef|None = parse_ohlcv_parquet_path(path) + if ref is not None: + yield ref + + +def unpack_fqme_from_parquet_filepath(path: Path) -> str: + ref: NativeSeriesRef|None = parse_ohlcv_parquet_path(path) + if ref is None: + raise ValueError(f'Invalid NativeDB OHLCV path: {path}') + return ref.fqme # Only hydrate fields shared by canonical storage and provider SHM. @@ -152,7 +257,7 @@ class NativeStorageClient: ) -> None: self._datadir = datadir - self._index: dict[str, dict] = {} + self._index: dict[tuple[str, int], NativeSeriesRef] = {} # series' cache from tsdb reads self._dfs: dict[str, dict[str, pl.DataFrame]] = {} @@ -163,6 +268,10 @@ class NativeStorageClient: @property def cardinality(self) -> int: + return len({fqme for fqme, _ in self._index}) + + @property + def series_cardinality(self) -> int: return len(self._index) # @property @@ -170,29 +279,15 @@ class NativeStorageClient: # ... async def list_keys(self) -> list[str]: - return list(self._index) + return sorted({fqme for fqme, _ in self._index}) + + async def list_series(self) -> list[NativeSeriesRef]: + return list(self._index.values()) def index_files(self): - for path in self._datadir.iterdir(): - if ( - path.is_dir() - or - path.suffix != '.parquet' - # or - # path.name in {'borked', 'expired',} - ): - continue - - key: str = path.name.removesuffix('.parquet') - fqme, _, descr = key.rpartition('.') - prefix, _, suffix = descr.partition('ohlcv') - period: int = int(suffix.strip('s')) - - # cache description data - self._index[fqme] = { - 'path': path, - 'period': period, - } + self._index.clear() + for ref in iter_native_series(self._datadir): + self._index[(ref.fqme, ref.period_s)] = ref return self._index @@ -227,11 +322,11 @@ class NativeStorageClient: bs_fqme, _, *_ = fqme.rpartition('.') possible_matches: list[str] = [] - for tskey in self._index: - if bs_fqme in tskey: - possible_matches.append(tskey) + for tsfqme, _period in self._index: + if bs_fqme in tsfqme: + possible_matches.append(tsfqme) - match_str: str = '\n'.join(sorted(possible_matches)) + match_str: str = '\n'.join(sorted(set(possible_matches))) raise TimeseriesNotFound( f'No entry for `{fqme}`?\n' f'Maybe you need a more specific fqme-key like:\n\n' diff --git a/piker/tsp/__init__.py b/piker/tsp/__init__.py index 02c1a953..b56a996d 100644 --- a/piker/tsp/__init__.py +++ b/piker/tsp/__init__.py @@ -45,8 +45,12 @@ from ._dedupe_smart import ( dedupe_ohlcv_smart as dedupe_ohlcv_smart, ) from ._history import ( + iter_shm_buffer_refs as iter_shm_buffer_refs, iter_dfs_from_shms as iter_dfs_from_shms, manage_history as manage_history, + parse_shm_buffer_path as parse_shm_buffer_path, + resolve_shm_buffer_dtype as resolve_shm_buffer_dtype, + ShmBufferRef as ShmBufferRef, ) from ._annotate import ( markup_gaps as markup_gaps, diff --git a/piker/tsp/_history.py b/piker/tsp/_history.py index df31b849..f9edc727 100644 --- a/piker/tsp/_history.py +++ b/piker/tsp/_history.py @@ -29,15 +29,18 @@ Historical TSP (time-series processing) lowlevel mgmt machinery and biz logic fo ''' from __future__ import annotations +from dataclasses import dataclass from datetime import datetime from functools import partial from pathlib import Path import platform from pprint import pformat +from string import hexdigits from types import ModuleType from typing import ( Callable, Generator, + Literal, TYPE_CHECKING, ) @@ -56,6 +59,7 @@ import numpy as np import polars as pl from piker.brokers import NoData +from piker.brokers import get_brokermod from piker.accounting import ( MktPair, ) @@ -64,7 +68,10 @@ from piker.log import ( get_console_log, ) from tractor.ipc._shm import ShmArray -from ..data._sharedmem import maybe_open_shm_array +from ..data._sharedmem import ( + attach_existing_shm_array, + maybe_open_shm_array, +) from piker.data._source import ( def_iohlcv_fields, ) @@ -104,6 +111,19 @@ log = get_logger( ) +@dataclass(frozen=True) +class ShmBufferRef: + ''' + Identify one Piker OHLCV shared-memory buffer. + + ''' + service: str + generation: str + fqme: str + kind: Literal['hist', 'rt'] + path: Path + + # `ShmArray` buffer sizing configuration: _mins_in_day = int(60 * 24) # how much is probably dependent on lifestyle @@ -747,7 +767,7 @@ async def start_backfill( prepend_until_dt=backfill_until_dt, ) # The requested end bar is already the oldest published - # sample. Keep it for idempotent storage merge, but do not + # sample. Keep it for storage merge, but do not # consume another physical SHM row for the overlap. to_push: np.ndarray = to_store[ to_store['time'] < last_start_dt.timestamp() @@ -1681,8 +1701,182 @@ async def manage_history( await trio.sleep_forever() +def parse_shm_buffer_path( + path: Path, + +) -> ShmBufferRef|None: + ''' + Parse one Piker OHLCV shared-memory filename. + + ''' + kind: Literal['hist', 'rt'] + if path.name.endswith('.hist'): + kind = 'hist' + elif path.name.endswith('.rt'): + kind = 'rt' + else: + return None + + stem: str = path.name.removesuffix(f'.{kind}') + service_prefix, open_sep, generation_tail = stem.partition('[') + generation, close_sep, fqme = generation_tail.partition('].') + if ( + open_sep != '[' + or + close_sep != '].' + or + not service_prefix.startswith('piker.') + or + not generation + or + not fqme + ): + return None + + service: str = service_prefix.removeprefix('piker.') + generation_valid: bool = ( + len(generation) == 16 + and + generation[8] == '-' + and + generation[13] == '-' + and + all( + char in hexdigits + for index, char in enumerate(generation) + if index not in {8, 13} + ) + ) + service_valid: bool = ( + bool(service) + and + service[0].isalnum() + and + service[-1].isalnum() + and + all( + char.isalnum() + or + char in {'-', '_', '.'} + for char in service + ) + ) + fqme_valid: bool = ( + fqme not in {'.', '..'} + and + Path(fqme).name == fqme + and + '/' not in fqme + and + '\\' not in fqme + and + '[' not in fqme + and + ']' not in fqme + ) + if not ( + generation_valid + and + service_valid + and + fqme_valid + ): + return None + + return ShmBufferRef( + service=service, + generation=generation, + fqme=fqme, + kind=kind, + path=path, + ) + + +def resolve_shm_buffer_dtype( + ref: ShmBufferRef, + size: int, +) -> np.dtype: + ''' + Resolve a buffer dtype from its size and backend declarations. + + ''' + canonical: np.dtype = np.dtype(def_iohlcv_fields) + provider_candidates: list[np.dtype] = [] + brokername: str = ref.fqme.rpartition('.')[2] + try: + brokermod: ModuleType|None = get_brokermod(brokername) + except ModuleNotFoundError: + brokermod = None + + if brokermod is not None: + provider_dtype = getattr(brokermod, '_ohlc_dtype', None) + if provider_dtype is not None: + provider_candidates.append(np.dtype(provider_dtype)) + + candidates: list[np.dtype] = ( + provider_candidates + if provider_candidates + else [canonical] + ) + + actual_bytes: int = ref.path.stat().st_size + matches: dict[tuple, np.dtype] = { + tuple(candidate.descr): candidate + for candidate in candidates + if candidate.itemsize * size == actual_bytes + } + if len(matches) != 1: + descriptions: list[str] = [ + f'{candidate.itemsize}B:{candidate.names}' + for candidate in candidates + ] + raise ValueError( + f'Can not resolve dtype for {ref.path.name}: ' + f'{actual_bytes} bytes over {size} rows; ' + f'candidates={descriptions}' + ) + return next(iter(matches.values())) + + +def iter_shm_buffer_refs( + fqme: str|None = None, + shmdir: Path = Path('/dev/shm/'), +) -> Generator[ShmBufferRef, None, None]: + ''' + Yield exact Piker OHLCV SHM identities in stable order. + + ''' + if not shmdir.is_dir(): + return + + for path in sorted( + shmdir.iterdir(), + key=lambda candidate: candidate.name, + ): + if ( + path.is_symlink() + or + not path.is_file() + ): + continue + ref: ShmBufferRef|None = parse_shm_buffer_path(path) + if ( + ref is not None + and + ( + fqme is None + or + ref.fqme == fqme + ) + ): + yield ref + + def iter_dfs_from_shms( - fqme: str + fqme: str, + shm_name: str|None = None, + shmdir: Path = Path('/dev/shm/'), + refs: list[ShmBufferRef]|None = None, ) -> Generator[ tuple[Path, ShmArray, pl.DataFrame], None, @@ -1694,46 +1888,48 @@ def iter_dfs_from_shms( 'rt': _default_rt_size, } - # load all detected shm buffer files which have the - # passed FQME pattern in the file name. - shmfiles: list[Path] = [] - shmdir = Path('/dev/shm/') - - for shmfile in shmdir.glob(f'*{fqme}*'): - filename: str = shmfile.name - - # skip index files - if ( - '_first' in filename - or '_last' in filename - ): - continue - - assert shmfile.is_file() - log.debug(f'Found matching shm buffer file: {filename}') - shmfiles.append(shmfile) - - for shmfile in shmfiles: + selected_refs: list[ShmBufferRef] = ( + refs + if refs is not None + else [ + candidate + for candidate in iter_shm_buffer_refs( + fqme=fqme, + shmdir=shmdir, + ) + if ( + shm_name is None + or + candidate.path.name == shm_name + ) + ] + ) + for ref in selected_refs: + shmfile: Path = ref.path + log.debug(f'Found matching shm buffer file: {shmfile.name}') # lookup array buffer size based on file suffix # being either .rt or .hist - key: str = shmfile.name.rsplit('.')[-1] - - # skip FSP buffers for now.. - if key not in sizes: - continue - - size: int = sizes[key] + size: int = sizes[ref.kind] # attach to any shm buffer, load array into polars df, # write to local parquet file. - shm, opened = maybe_open_shm_array( - key=shmfile.name, - size=size, - dtype=def_iohlcv_fields, - readonly=True, - ) - assert not opened + try: + dtype: np.dtype = resolve_shm_buffer_dtype(ref, size) + shm: ShmArray = attach_existing_shm_array( + key=shmfile.name, + size=size, + dtype=dtype, + readonly=True, + ) + except ( + FileNotFoundError, + ValueError, + ) as err: + log.warning( + f'Could not attach SHM {shmfile.name}: {err}' + ) + continue ohlcv: np.ndarray = shm.array df: pl.DataFrame = np2pl(ohlcv) diff --git a/tests/test_ldshm.py b/tests/test_ldshm.py index 2548f578..3502c0d7 100644 --- a/tests/test_ldshm.py +++ b/tests/test_ldshm.py @@ -16,9 +16,11 @@ import polars as pl import pytest from piker import tsp +from piker.data import def_iohlcv_fields from piker.storage import cli as storage_cli from piker.storage.cli import ( _shm_period_and_invalid_count, + _summarize_shm_frame, ldshm, ) @@ -209,6 +211,153 @@ def test_ldshm_invalid_snapshot_never_reaches_storage( ldshm('qqq.nasdaq.ib') +@pytest.mark.parametrize('write_parquet', [False, True]) +def test_ldshm_no_reload_uses_deduped_markup_frame( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + write_parquet: bool, +) -> None: + ''' + Disabling SHM reload must not require a reload payload. + + ``ldshm`` created ``new`` only in reload but used it for markup. + No-write and no-reload modes crashed, the latter after a durable + write. Arrange one gap, exercise + write modes with reload disabled, and prove markup receives the + deduplicated frame without any SHM mutation. + + ''' + array = np.zeros( + 3, + dtype=np.dtype(def_iohlcv_fields), + ) + array['index'] = np.arange(3) + array['time'] = [60, 120, 240] + array['open'] = [10, 11, 12] + array['high'] = [12, 13, 14] + array['low'] = [8, 9, 10] + array['close'] = [11, 12, 13] + array['volume'] = [100, 101, 102] + frame = tsp.np2pl(array) + shm = SimpleNamespace(array=array) + shmfile = Path('piker.datad[aaaaaaaa-aaaa-aa].x.test.hist') + writes: list[pl.DataFrame] = [] + markups: list[pl.DataFrame] = [] + + @asynccontextmanager + async def open_runtime( + *args: Any, + **kwargs: Any, + ) -> AsyncIterator[None]: + ''' + Yield a runtime-free command context. + + ''' + yield + + class Client: + ''' + Record optional durable writes. + + ''' + async def write_ohlcv( + self, + fqme: str, + ohlcv: pl.DataFrame, + timeframe: int, + ) -> Path: + ''' + Persist the frame for the command's read-back step. + + ''' + writes.append(ohlcv) + path = tmp_path / 'frame.parquet' + ohlcv.write_parquet(path) + return path + + @asynccontextmanager + async def open_storage( + *args: Any, + **kwargs: Any, + ) -> AsyncIterator[tuple[SimpleNamespace, Client]]: + ''' + Yield the recording storage client. + + ''' + yield SimpleNamespace(), Client() + + @asynccontextmanager + async def open_annotations( + *args: Any, + **kwargs: Any, + ) -> AsyncIterator[SimpleNamespace]: + ''' + Yield an inert annotation controller. + + ''' + yield SimpleNamespace() + + def iter_shms( + fqme: str, + ) -> Iterator[tuple[Path, SimpleNamespace, pl.DataFrame]]: + ''' + Yield the one valid history snapshot. + + ''' + yield shmfile, shm, frame + + async def markup_gaps( + fqme: str, + period_s: int, + actl: SimpleNamespace, + new_df: pl.DataFrame, + step_gaps: pl.DataFrame, + ) -> dict: + ''' + Capture the frame used for non-mutating markup. + + ''' + markups.append(new_df) + return {} + + async def pause() -> None: + ''' + Replace the command's final interactive pause. + + ''' + + monkeypatch.setattr( + storage_cli, + 'open_piker_runtime', + open_runtime, + ) + monkeypatch.setattr( + storage_cli, + 'open_storage_client', + open_storage, + ) + monkeypatch.setattr(tsp, 'iter_dfs_from_shms', iter_shms) + monkeypatch.setattr(tsp._annotate, 'markup_gaps', markup_gaps) + monkeypatch.setattr(storage_cli.tractor, 'pause', pause) + + from piker.ui import _remote_ctl + monkeypatch.setattr( + _remote_ctl, + 'open_annot_ctl', + open_annotations, + ) + + ldshm( + 'x.test', + write_parquet=write_parquet, + reload_parquet_to_shm=False, + ) + + assert len(writes) == int(write_parquet) + assert len(markups) == 1 + assert markups[0]['time'].to_list() == [60, 120, 240] + + @pytest.mark.parametrize( 'times', [ @@ -233,3 +382,176 @@ def test_ldshm_skips_frame_without_positive_cadence( period_s, _ = _shm_period_and_invalid_count(times) assert period_s is None + + +def test_shm_buffer_discovery_uses_exact_identities( + tmp_path: Path, +) -> None: + ''' + SHM discovery must not mix substring matches or index companions. + + ``ldshm`` globbed ``*fqme*`` and could process stale and + active objects for another instrument whose name merely contained + requested text. Create two generations, both OHLCV kinds, index + companions, and a substring collision. Prove discovery returns + exact parsed identities in deterministic filename order. + + ''' + names = [ + 'piker.datad[bbbbbbbb-bbbb-bb].qqq.nasdaq.ib.rt', + 'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.hist', + 'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.hist_first', + 'piker.datad[aaaaaaaa-aaaa-aa].myqqq.nasdaq.ib.rt', + 'unrelated.qqq.nasdaq.ib.rt', + ] + for name in names: + (tmp_path / name).touch() + + refs = list(tsp.iter_shm_buffer_refs( + fqme='qqq.nasdaq.ib', + shmdir=tmp_path, + )) + + assert [ref.path.name for ref in refs] == [ + 'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.hist', + 'piker.datad[bbbbbbbb-bbbb-bb].qqq.nasdaq.ib.rt', + ] + assert [ + (ref.service, ref.generation, ref.kind) + for ref in refs + ] == [ + ('datad', 'aaaaaaaa-aaaa-aa', 'hist'), + ('datad', 'bbbbbbbb-bbbb-bb', 'rt'), + ] + + +def test_shm_summary_reports_invalid_rows_and_ranked_gaps() -> None: + ''' + Read-only SHM diagnostics must preserve the live triage evidence. + + The QQQ investigation needed cadence, invalid bounds, and largest + timestamp jumps without mutation. Build one zero row and two gaps; + prove summary excludes invalid timestamps and ranks durations. + + ''' + frame = np.zeros( + 5, + dtype=np.dtype(def_iohlcv_fields), + ) + frame['index'] = np.arange(100, 105) + frame['time'] = [1, 0, 3, 4, 10] + ref = tsp.ShmBufferRef( + service='datad', + generation='aaaaaaaa-aaaa-aa', + fqme='qqq.nasdaq.ib', + kind='rt', + path=Path( + 'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.rt' + ), + ) + + report = _summarize_shm_frame(ref, frame, max_gaps=2) + + assert report['period_s'] == 1 + assert report['observed_period_s'] == 1 + assert report['invalid_count'] == 1 + assert report['invalid_index_bounds'] == [101, 101] + assert report['gap_count'] == 2 + assert [ + gap['delta_s'] + for gap in report['largest_gaps'] + ] == [6, 2] + + +def test_shm_summary_reports_ordering_corruption() -> None: + ''' + Duplicate and reversed timestamps remain visible in diagnostics. + + Positive-gap detection considers both defects gap-free. Build + one exact RT snapshot with a duplicate and reversal and prove the + read-only report distinguishes both without classifying values as + unpublished slots. + + ''' + frame = np.zeros( + 5, + dtype=np.dtype(def_iohlcv_fields), + ) + frame['index'] = np.arange(5) + frame['time'] = [1, 2, 2, 1, 3] + ref = tsp.ShmBufferRef( + service='datad', + generation='aaaaaaaa-aaaa-aa', + fqme='qqq.nasdaq.ib', + kind='rt', + path=Path( + 'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.rt' + ), + ) + + report = _summarize_shm_frame(ref, frame, max_gaps=2) + + assert report['invalid_count'] == 0 + assert report['duplicate_step_count'] == 1 + assert report['reversed_step_count'] == 1 + + +def test_shm_dtype_resolution_matches_allocator_layout( + tmp_path: Path, +) -> None: + ''' + Inspector actors must mirror history allocation dtype lookup. + + Allocation reads ``_ohlc_dtype`` from the package; IB does not + export its API-internal extended dtype there. + Create four production-shaped canonical rows and prove inspection + resolves the same layout without searching unrelated submodules. + + ''' + path = tmp_path / ( + 'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.rt' + ) + path.write_bytes(b'\0' * (4 * 56)) + ref = tsp.ShmBufferRef( + service='datad', + generation='aaaaaaaa-aaaa-aa', + fqme='qqq.nasdaq.ib', + kind='rt', + path=path, + ) + + dtype = tsp.resolve_shm_buffer_dtype(ref, size=4) + + assert dtype.itemsize == 56 + assert dtype.names == ( + 'index', + 'time', + 'open', + 'high', + 'low', + 'close', + 'volume', + ) + + +@pytest.mark.parametrize( + 'name', + [ + 'piker..datad[aaaaaaaa-aaaa-aa].x.test.rt', + 'piker.datad[short].x.test.rt', + 'piker.datad[aaaaaaaa-aaaa-aa].x\\test.rt', + 'piker.datad[aaaaaaaa-aaaa-aa].x[y].test.rt', + ], +) +def test_shm_parser_rejects_forged_identities( + name: str, +) -> None: + ''' + SHM selectors must accept only generated Piker identities. + + Reject malformed service, generation, and FQME components so a + same-user object that merely resembles OHLCV can not become an + ``ldshm --shm-name`` target. + + ''' + assert tsp.parse_shm_buffer_path(Path(name)) is None diff --git a/tests/test_storage_nativedb.py b/tests/test_storage_nativedb.py index 0f991ef3..bd4f6d18 100644 --- a/tests/test_storage_nativedb.py +++ b/tests/test_storage_nativedb.py @@ -12,7 +12,11 @@ import trio from piker import tsp from piker.data._source import def_iohlcv_fields -from piker.storage.nativedb import NativeStorageClient +from piker.storage.nativedb import ( + iter_native_series, + NativeStorageClient, + parse_ohlcv_parquet_path, +) def mk_ohlcv( @@ -540,8 +544,96 @@ def test_index_files_ignores_sidecar_files( index = client.index_files() - assert list(index) == ['x.test'] - assert index['x.test']['period'] == 60 + assert list(index) == [('x.test', 60)] + ref = index[('x.test', 60)] + assert ref.fqme == 'x.test' + assert ref.period_s == 60 + + +def test_index_files_preserves_every_series_timeframe( + tmp_path: Path, +) -> None: + ''' + NativeDB discovery must identify FQME and timeframe together. + + The old index keyed only by FQME, so filesystem iteration silently + discarded either the 1s or 60s file. Write both periods plus another + instrument, rebuild the index, and prove exact series remain stable + while compatibility key listing still returns unique FQMEs. + + ''' + client = NativeStorageClient(tmp_path) + run(client.write_ohlcv('x.test', mk_ohlcv((1, 2)), 1)) + run(client.write_ohlcv('x.test', mk_ohlcv((60,)), 60)) + run(client.write_ohlcv('y.test', mk_ohlcv((60,)), 60)) + + index = client.index_files() + keys = trio.run(client.list_keys) + series = trio.run(client.list_series) + + assert list(index) == [ + ('x.test', 1), + ('x.test', 60), + ('y.test', 60), + ] + assert keys == ['x.test', 'y.test'] + assert [ + (ref.fqme, ref.period_s) + for ref in series + ] == list(index) + + +@pytest.mark.parametrize( + 'name', + [ + 'x.test.ohlcv0s.parquet', + 'x.test.ohlcv01s.parquet', + 'x.test.ohlcv²s.parquet', + 'x.test.ohlcv1.parquet', + 'x.test.ohlcv1s.parquet.tmp', + 'x.test.parquet', + '.ohlcv1s.parquet', + ], +) +def test_native_series_parser_rejects_noncanonical_names( + tmp_path: Path, + name: str, +) -> None: + ''' + Malformed files must not become selectable storage identities. + + NativeDB directories can contain crash files, legacy sidecars, and + unrelated Parquet. Exercise ambiguous period and suffix forms and + prove parser and directory discovery ignore them deterministically. + + ''' + path: Path = tmp_path / name + path.touch() + + assert parse_ohlcv_parquet_path(path) is None + assert list(iter_native_series(tmp_path)) == [] + + +def test_native_series_parser_round_trips_dotted_fqme( + tmp_path: Path, +) -> None: + ''' + Dotted market identities must survive canonical filename parsing. + + The legacy parser split on every dot and failed for ordinary FQMEs. + Parse a futures-style identity and prove every selector field is + preserved exactly. + + ''' + path = tmp_path / 'mnq.cme.20260918.ib.ohlcv1s.parquet' + path.touch() + + ref = parse_ohlcv_parquet_path(path) + + assert ref is not None + assert ref.fqme == 'mnq.cme.20260918.ib' + assert ref.period_s == 1 + assert ref.path == path def test_writes_create_no_lock_sidecars( diff --git a/tests/test_store_cli.py b/tests/test_store_cli.py new file mode 100644 index 00000000..313315fd --- /dev/null +++ b/tests/test_store_cli.py @@ -0,0 +1,251 @@ +''' +Storage command UX regressions. + +''' +from collections.abc import ( + AsyncIterator, + Iterator, +) +from contextlib import asynccontextmanager +import json +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import numpy as np +import pytest +from typer.testing import CliRunner + +from piker import tsp +from piker import config +from piker.data import def_iohlcv_fields +from piker.storage import cli as storage_cli +from piker.storage.cli import store + + +def test_store_group_shows_help_without_arguments() -> None: + ''' + Bare ``piker store`` must present its command map immediately. + + The Typer group emitted only a missing-command error and + required another ``--help`` invocation. Invoke the group + without arguments and prove complete help replaces that error. + + ''' + result = CliRunner().invoke(store, []) + + assert result.exit_code == 2 + assert 'Usage:' in result.output + assert 'Missing command' not in result.output + for command in ( + 'anal', + 'audit', + 'delete', + 'ldshm', + 'ls', + 'series', + 'shm', + ): + assert command in result.output + + +def test_store_commands_show_help_without_arguments() -> None: + ''' + Bare endpoints must not open runtimes or report missing args. + + Every endpoint is discoverable by typing its name once. Exercise + both required-input and explicitly-triggered listing commands and + prove help rendering exits before any callback can run. + + ''' + runner = CliRunner() + for command in ( + 'anal', + 'audit', + 'delete', + 'ldshm', + 'ls', + 'series', + 'shm', + ): + result = runner.invoke(store, [command]) + assert result.exit_code == 2 + assert 'Usage:' in result.output + assert 'Missing argument' not in result.output + + +def test_series_lists_exact_native_periods( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Durable listing must expose separate 1s and 60s identities. + + The old FQME-only index hid one timeframe. Create canonical files + without storage or actor services, invoke the JSON endpoint, + and prove both exact paths and periods are returned. + + ''' + nativedb: Path = tmp_path / 'nativedb' + nativedb.mkdir() + one = nativedb / 'qqq.nasdaq.ib.ohlcv1s.parquet' + sixty = nativedb / 'qqq.nasdaq.ib.ohlcv60s.parquet' + one.write_bytes(b'1') + sixty.write_bytes(b'60') + monkeypatch.setattr(config, 'get_conf_dir', lambda: tmp_path) + + result = CliRunner().invoke( + store, + ['series', '--all', '--json'], + ) + + assert result.exit_code == 0 + payload = json.loads(result.output) + assert [item['period_s'] for item in payload] == [1, 60] + assert [item['path'] for item in payload] == [ + str(one), + str(sixty), + ] + + +def test_shm_endpoint_reports_immutable_snapshot( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + SHM diagnostics report exact buffers without storage mutation. + + Build one RT identity and snapshot, then replace runtime + attachment with fakes. Invoke JSON output and prove it carries + generation, cadence, and invalid-slot + evidence without opening a storage client or writer. + + ''' + path = Path( + 'piker.datad[aaaaaaaa-aaaa-aa].qqq.nasdaq.ib.rt' + ) + ref = tsp.ShmBufferRef( + service='datad', + generation='aaaaaaaa-aaaa-aa', + fqme='qqq.nasdaq.ib', + kind='rt', + path=path, + ) + frame = np.zeros( + 4, + dtype=np.dtype(def_iohlcv_fields), + ) + frame['index'] = np.arange(4) + frame['time'] = [1, 0, 3, 4] + + @asynccontextmanager + async def open_runtime( + *args: Any, + **kwargs: Any, + ) -> AsyncIterator[None]: + ''' + Yield a runtime-free diagnostic context. + + ''' + yield + + def iter_refs( + fqme: str|None = None, + ) -> Iterator[tsp.ShmBufferRef]: + ''' + Yield the selected exact SHM identity. + + ''' + yield ref + + def iter_frames( + fqme: str, + shm_name: str|None = None, + refs: list[tsp.ShmBufferRef]|None = None, + ) -> Iterator[tuple[Path, SimpleNamespace, None]]: + ''' + Yield an immutable snapshot through the attachment interface. + + ''' + shm = SimpleNamespace(array=frame) + yield path, shm, None + + def fail_storage(*args: Any, **kwargs: Any) -> None: + ''' + Fail if read-only SHM inspection reaches storage. + + ''' + raise AssertionError('SHM diagnostics opened storage') + + monkeypatch.setattr( + storage_cli, + 'open_piker_runtime', + open_runtime, + ) + monkeypatch.setattr( + storage_cli, + 'open_storage_client', + fail_storage, + ) + monkeypatch.setattr(tsp, 'iter_shm_buffer_refs', iter_refs) + monkeypatch.setattr(tsp, 'iter_dfs_from_shms', iter_frames) + + result = CliRunner().invoke( + store, + ['shm', 'qqq.nasdaq.ib', '--json'], + ) + + assert result.exit_code == 0 + payload = json.loads(result.output) + assert len(payload) == 1 + assert payload[0]['generation'] == 'aaaaaaaa-aaaa-aa' + assert payload[0]['period_s'] == 1 + assert payload[0]['invalid_count'] == 1 + + +def test_ldshm_rejects_unknown_exact_name_before_runtime( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + A stale exact SHM selection must fail before opening services. + + The old no-match path paused after opening runtime and storage + setup. Return no identity, install a runtime fail-spy, + and prove exact selection exits immediately + with a deterministic error. + + ''' + def iter_refs( + fqme: str|None = None, + ) -> Iterator[tsp.ShmBufferRef]: + ''' + Yield no matching SHM identities. + + ''' + return iter(()) + + def fail_runtime(*args: Any, **kwargs: Any) -> None: + ''' + Fail if no-match handling opens a runtime. + + ''' + raise AssertionError('unknown SHM opened runtime') + + monkeypatch.setattr(tsp, 'iter_shm_buffer_refs', iter_refs) + monkeypatch.setattr( + storage_cli, + 'open_piker_runtime', + fail_runtime, + ) + + result = CliRunner().invoke( + store, + [ + 'ldshm', + 'qqq.nasdaq.ib', + '--shm-name', + 'missing', + ], + ) + + assert result.exit_code == 2 + assert 'No exact OHLCV SHM buffer' in result.output