560 lines
15 KiB
Python
560 lines
15 KiB
Python
'''
|
|
Shared-memory persistence regressions.
|
|
|
|
'''
|
|
from collections.abc import (
|
|
AsyncIterator,
|
|
Iterator,
|
|
)
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
import numpy as np
|
|
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,
|
|
shm as shm_cmd,
|
|
)
|
|
|
|
|
|
def test_shm_detects_nulls_without_skewing_period() -> None:
|
|
'''
|
|
Null SHM slots must stop persistence without skewing cadence.
|
|
|
|
A live QQQ buffer contained an interior run of zero-time rows.
|
|
``shm`` included them in period inference and then sent them to
|
|
NativeDB, which correctly rejected the replacement. Build a
|
|
60-second series with the same interior hole and enough
|
|
bars around the gap. Prove inspection counts every null while
|
|
inferring the published series' 60-second period; the
|
|
command uses that count to skip both persistence and SHM reload.
|
|
|
|
'''
|
|
times = np.array([
|
|
60,
|
|
120,
|
|
0,
|
|
0,
|
|
300,
|
|
360,
|
|
420,
|
|
])
|
|
|
|
period_s, invalid_count = _shm_period_and_invalid_count(times)
|
|
|
|
assert period_s == 60
|
|
assert invalid_count == 2
|
|
|
|
|
|
def test_shm_period_is_an_observed_step() -> None:
|
|
'''
|
|
Sparse timestamps must not invent a storage timeframe.
|
|
|
|
The arithmetic median of two steps can produce a cadence that
|
|
occurs nowhere in the source series. Such a value can select the
|
|
wrong NativeDB timeframe and hide the gap. Arrange tied 60 and
|
|
120-second steps and prove the deterministic tie-break chooses
|
|
the smaller observed period.
|
|
|
|
'''
|
|
times = np.array([60, 120, 240])
|
|
|
|
period_s, invalid_count = _shm_period_and_invalid_count(times)
|
|
|
|
assert period_s == 60
|
|
assert invalid_count == 0
|
|
|
|
|
|
def test_shm_counts_every_invalid_timestamp() -> None:
|
|
'''
|
|
Corrupt timestamps stop persistence just like null SHM slots.
|
|
|
|
Exact zero denotes an unpublished slot, while negative, NaN, and
|
|
infinite values indicate corruption. NativeDB rejects them all,
|
|
but allowing a replacement attempt would still abort ``shm``.
|
|
Mix each invalid class into a regular series and prove the same
|
|
snapshot inspection counts all four while preserving cadence.
|
|
|
|
'''
|
|
times = np.array([
|
|
60,
|
|
120,
|
|
0,
|
|
-1,
|
|
np.nan,
|
|
np.inf,
|
|
180,
|
|
])
|
|
|
|
period_s, invalid_count = _shm_period_and_invalid_count(times)
|
|
|
|
assert period_s == 60
|
|
assert invalid_count == 4
|
|
|
|
|
|
def test_shm_invalid_snapshot_never_reaches_storage(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
'''
|
|
Invalid snapshots stop before analyzers and storage mutation.
|
|
|
|
The original QQQ failure passed zero-time rows through dedupe and
|
|
into NativeDB. An intermediate fix called null-segment analysis
|
|
before its guard, allowing corrupt indexes to abort the command.
|
|
Supply one immutable Polars snapshot containing that combination,
|
|
then replace dedupe and writes with failing spies. Completing the
|
|
command proves its guard runs first, preventing persistence or
|
|
SHM reload without relying on task timing.
|
|
|
|
'''
|
|
shm_df = pl.DataFrame({
|
|
'index': [100, 500, 104],
|
|
'time': [60, 0, 180],
|
|
})
|
|
|
|
@asynccontextmanager
|
|
async def open_runtime(
|
|
*args: Any,
|
|
**kwargs: Any,
|
|
) -> AsyncIterator[None]:
|
|
'''
|
|
Yield a runtime-free command context.
|
|
|
|
'''
|
|
yield
|
|
|
|
async def fail_write(
|
|
*args: Any,
|
|
**kwargs: Any,
|
|
) -> None:
|
|
'''
|
|
Fail if invalid SHM reaches NativeDB.
|
|
|
|
'''
|
|
raise AssertionError('invalid SHM reached NativeDB')
|
|
|
|
client = SimpleNamespace(write_ohlcv=fail_write)
|
|
|
|
@asynccontextmanager
|
|
async def open_storage(
|
|
*args: Any,
|
|
**kwargs: Any,
|
|
) -> AsyncIterator[tuple[SimpleNamespace, SimpleNamespace]]:
|
|
'''
|
|
Yield storage objects with a failing writer.
|
|
|
|
'''
|
|
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 single captured invalid snapshot.
|
|
|
|
'''
|
|
yield Path('qqq.rt'), SimpleNamespace(), shm_df
|
|
|
|
def dedupe(*args: Any, **kwargs: Any) -> None:
|
|
'''
|
|
Fail if invalid SHM reaches deduplication.
|
|
|
|
'''
|
|
raise AssertionError('invalid SHM reached dedupe')
|
|
|
|
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, 'dedupe_ohlcv_smart', dedupe)
|
|
monkeypatch.setattr(storage_cli.tractor, 'pause', pause)
|
|
|
|
from piker.ui import _remote_ctl
|
|
monkeypatch.setattr(
|
|
_remote_ctl,
|
|
'open_annot_client',
|
|
open_annotations,
|
|
)
|
|
|
|
shm_cmd(
|
|
'qqq.nasdaq.ib',
|
|
write_parquet=True,
|
|
)
|
|
|
|
|
|
def test_shm_write_without_reload_uses_deduped_markup_frame(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
'''
|
|
A Parquet write without SHM reload must not need a reload
|
|
payload.
|
|
|
|
``ldshm`` created ``new`` only in reload but used it for markup.
|
|
Its no-reload mode crashed after a durable write. Arrange one
|
|
gap, explicitly enable persistence 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_client',
|
|
open_annotations,
|
|
)
|
|
|
|
shm_cmd(
|
|
'x.test',
|
|
write_parquet=True,
|
|
reload_parquet_to_shm=False,
|
|
)
|
|
|
|
assert len(writes) == 1
|
|
assert len(markups) == 1
|
|
assert markups[0]['time'].to_list() == [60, 120, 240]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'times',
|
|
[
|
|
np.array([0, 0]),
|
|
np.array([60]),
|
|
np.array([60, 60]),
|
|
],
|
|
)
|
|
def test_shm_skips_frame_without_positive_cadence(
|
|
times: np.ndarray,
|
|
) -> None:
|
|
'''
|
|
Degenerate buffers are skipped before indexing or persistence.
|
|
|
|
The old inference indexed final timestamps and left its median
|
|
undefined for short buffers. Fully unpublished, one-row, and
|
|
duplicate-only frames could therefore fail before the command's
|
|
guard. Exercise each shape and prove inspection
|
|
returns no cadence, which directs ``shm`` to skip the buffer.
|
|
|
|
'''
|
|
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.
|
|
|
|
``shm`` 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
|
|
``shm --shm-name`` target.
|
|
|
|
'''
|
|
assert tsp.parse_shm_buffer_path(Path(name)) is None
|