Compare commits
4 Commits
0846cbd4a4
...
2aa66c2cf6
| Author | SHA1 | Date |
|---|---|---|
 Gud Boi
|
2aa66c2cf6 | |
|
Gud Boi
|
06d5ea5dc0 | |
|
Gud Boi
|
f0f3c80657 | |
|
Gud Boi
|
74ea2152f9 |
|
|
@ -47,7 +47,7 @@ that's why it's like 1000x faster ya know?"
|
|||
- Wrap all code symbols: `function()`,
|
||||
`ClassName`, `field_name`
|
||||
- File paths: `piker/ui/_remote_ctl.py`
|
||||
- Commands: `git status`, `piker store ldshm`
|
||||
- Commands: `git status`, `piker store shm`
|
||||
|
||||
**Explain like you're pair programming:**
|
||||
```
|
||||
|
|
|
|||
|
|
@ -105,8 +105,10 @@ 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_ib_method_proxy.py`
|
||||
- `tests/test_history_backfill.py`
|
||||
- `tests/test_ldshm.py`
|
||||
- `tests/test_accounting.py::test_account_file_default_empty`
|
||||
|
|
@ -133,8 +135,8 @@ Require explicit authorization before running:
|
|||
`tests/test_cli.py` is currently hard-skipped. It is not an active CLI
|
||||
regression gate.
|
||||
|
||||
Never execute `piker store anal` or `piker store ldshm` as tests. They are
|
||||
mutating or interactive operational commands.
|
||||
Never execute `piker store anal` or `piker store shm --write-parquet`
|
||||
as tests. They are mutating or interactive operational commands.
|
||||
|
||||
## Project-Specific Flags And Backend Matrix
|
||||
|
||||
|
|
@ -192,10 +194,12 @@ tests/
|
|||
test_ems.py actor, EMS, and paper-position behavior
|
||||
test_feeds.py live Binance/Kraken feeds and shared memory
|
||||
test_ib_history.py deterministic IB history request formatting
|
||||
test_ib_method_proxy.py deterministic IB asyncio proxy routing
|
||||
test_history_backfill.py deterministic history/SHM orchestration
|
||||
test_ldshm.py ldshm unpublished-slot guard
|
||||
test_ldshm.py SHM 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,10 +212,12 @@ 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/brokers/ib/api.py` method proxy | `tests/test_ib_method_proxy.py` | fake channel, no network |
|
||||
| `piker/tsp/_history.py` | `tests/test_history_backfill.py` | fake provider/storage/SHM |
|
||||
| `piker/storage/cli.py` ldshm null-slot guard | `tests/test_ldshm.py` | synthetic timestamps, no SHM mutation |
|
||||
| `piker/storage/cli.py` SHM null-slot guard | `tests/test_ldshm.py` | synthetic timestamps, no SHM mutation |
|
||||
| `piker/config.py` | `test_account_file_default_empty` | root-network test has a known mismatch |
|
||||
| `piker/accounting/` | targeted accounting node | some cases use live/configured state |
|
||||
| `piker/ui/_style.py`, `piker/ui/qt.py` | `tests/test_dpi_font.py` | GUI/config-isolated opt-in |
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
@ -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.
|
||||
|
|
@ -0,0 +1,39 @@
|
|||
---
|
||||
model: openai/gpt-5.6-sol
|
||||
service: opencode
|
||||
session: ses_0799212ebffe42arY96czXn89F
|
||||
timestamp: 2026-08-01T01:53:21Z
|
||||
git_ref: 74ea2152
|
||||
scope: code
|
||||
substantive: true
|
||||
raw_file: 20260801T015321Z_74ea2152_prompt_io.raw.md
|
||||
---
|
||||
|
||||
## Prompt
|
||||
|
||||
Make `piker store ls` default to all NativeDB series as a compact counterpart
|
||||
to detailed `series`, with sensible no-flag output and a visible help
|
||||
description. Consolidate SHM inspection and repair under canonical `shm`,
|
||||
because attachment to volatile segments is implied, and make inspection the
|
||||
safe default while retaining explicit persistence and reload controls.
|
||||
|
||||
## Response summary
|
||||
|
||||
Replaced backend-oriented `ls` with compact NativeDB discovery and made both
|
||||
listing commands default to all series. Consolidated SHM behavior under a
|
||||
read-only-by-default `shm` command, guarded explicit repair options, updated
|
||||
executable references and docs, and added command regressions.
|
||||
|
||||
## Files changed
|
||||
|
||||
- `piker/storage/cli.py` - listing defaults and unified SHM command modes.
|
||||
- `tests/test_store_cli.py` - listing, help, and SHM mode regressions.
|
||||
- `tests/test_ldshm.py` - explicit persistence-path regressions.
|
||||
- `docs/manual_backfill_qualification.rst` - revised safe CLI workflow.
|
||||
- `snippets/claude_debug_helper.py` - explicit SHM persistence commands.
|
||||
- `.claude/skills/run-tests/test-harness-reference.md` - safety and test map.
|
||||
- `.claude/skills/piker-slang/examples.md` - canonical command example.
|
||||
|
||||
## Human edits
|
||||
|
||||
None - generated output incorporates the user's design refinements.
|
||||
|
|
@ -0,0 +1,78 @@
|
|||
---
|
||||
model: openai/gpt-5.6-sol
|
||||
service: opencode
|
||||
timestamp: 2026-08-01T01:53:21Z
|
||||
git_ref: 74ea2152
|
||||
diff_cmd: git diff HEAD~1..HEAD
|
||||
---
|
||||
|
||||
The user requested command-surface refinements after reviewing the new
|
||||
storage CLI. Bare `ls` should default to all series and act as a compact
|
||||
form of detailed `series`; commands should produce sensible default output
|
||||
without requiring trigger flags; and `ls` needed a parent-help description.
|
||||
The user first proposed folding read-only SHM inspection into `ldshm -r`,
|
||||
then clarified that `shm` is the better canonical command because attaching
|
||||
to volatile segments is implied by the resource name.
|
||||
|
||||
The resulting design uses `piker store ls` as a compact NativeDB FQME and
|
||||
period view, while `piker store series` reports each exact sub-series with
|
||||
size and path details. Both default to all records and accept an optional
|
||||
FQME substring.
|
||||
|
||||
`piker store shm FQME` is read-only by default. Persistence and reload enter
|
||||
the prior repair workflow only through explicit `--write-parquet` and
|
||||
`--reload-parquet-to-shm` flags. Reload requires persistence, and read-only
|
||||
formatting flags are rejected in repair mode instead of being ignored.
|
||||
|
||||
> `git diff HEAD~1..HEAD -- piker/storage/cli.py`
|
||||
|
||||
Generated shared NativeDB payload collection, compact and detailed output,
|
||||
numeric period ordering, default-all semantics, the canonical `shm` command,
|
||||
safe mode dispatch, explicit repair validation, and debugger propagation in
|
||||
read-only mode.
|
||||
|
||||
> `git diff HEAD~1..HEAD -- tests/test_store_cli.py`
|
||||
|
||||
Generated regressions for no-flag listing, parent-help descriptions, numeric
|
||||
period display, default read-only SHM JSON, reload dependencies, and repair
|
||||
versus read-only option conflicts.
|
||||
|
||||
> `git diff HEAD~1..HEAD -- tests/test_ldshm.py`
|
||||
|
||||
Adjusted persistence regressions for the renamed command and explicit repair
|
||||
entry while preserving the historical failure rationale.
|
||||
|
||||
> `git diff HEAD~1..HEAD -- docs/manual_backfill_qualification.rst`
|
||||
|
||||
Documented compact versus detailed NativeDB discovery, default read-only SHM
|
||||
inspection, and the explicitly mutating repair invocation.
|
||||
|
||||
> `git diff HEAD~1..HEAD -- snippets/claude_debug_helper.py`
|
||||
|
||||
Updated executable debug helper commands to select explicit SHM persistence.
|
||||
|
||||
> `git diff HEAD~1..HEAD -- .claude/skills/run-tests/test-harness-reference.md`
|
||||
|
||||
Updated operational safety guidance and SHM regression mappings.
|
||||
|
||||
> `git diff HEAD~1..HEAD -- .claude/skills/piker-slang/examples.md`
|
||||
|
||||
Updated the command example to the canonical `shm` spelling.
|
||||
|
||||
Verification output:
|
||||
|
||||
```text
|
||||
........................................................................ [ 69%]
|
||||
................................ [100%]
|
||||
104 passed in 5.77s
|
||||
```
|
||||
|
||||
Real command-help rendering was checked for the parent group, `ls`, `series`,
|
||||
and `shm`. Real read-only `ls` and `series --json` output was also exercised
|
||||
against the configured NativeDB. `git diff --check` passed.
|
||||
|
||||
Adversarial review identified and drove fixes for lexicographic period
|
||||
ordering, stale runbook safety wording, ignored mode-specific options, and an
|
||||
explicit-default `--max-gaps 10` validation bypass. Final review found no
|
||||
remaining issues. Real OS-backed SHM attachment and successful repair/reload
|
||||
remain integration-level validation concerns.
|
||||
|
|
@ -0,0 +1,33 @@
|
|||
---
|
||||
model: openai/gpt-5.6-sol
|
||||
service: opencode
|
||||
session: ses_0799212ebffe42arY96czXn89F
|
||||
timestamp: 2026-08-05T19:50:01Z
|
||||
git_ref: f0f3c806
|
||||
scope: code
|
||||
substantive: true
|
||||
raw_file: 20260805T195001Z_f0f3c806_prompt_io.raw.md
|
||||
---
|
||||
|
||||
## Prompt
|
||||
|
||||
Inspect the completed NVDA 60s SHM backfill before workspace restart, explain
|
||||
the persistent zero region, and continue fixing the indexing logic so sparse
|
||||
IB history does not leave physical null reservations.
|
||||
|
||||
## Response summary
|
||||
|
||||
Captured a 42,720-row live SHM null span whose valid margins shared the same
|
||||
timestamp. Replaced wall-clock row reservation with actual provider-boundary
|
||||
placement, fixed stale SHM first-index, endpoint insertion and field-map
|
||||
semantics, published the final storage prepend, and added a full multi-frame
|
||||
`tsdb_backfill()` regression using a real `ShmArray`.
|
||||
|
||||
## Files changed
|
||||
|
||||
- `piker/tsp/_history.py` - contiguous provider and NativeDB prepends.
|
||||
- `tests/test_history_backfill.py` - sparse multi-frame SHM regression.
|
||||
|
||||
## Human edits
|
||||
|
||||
None - generated output incorporates the user's live observations.
|
||||
|
|
@ -0,0 +1,71 @@
|
|||
---
|
||||
model: openai/gpt-5.6-sol
|
||||
service: opencode
|
||||
timestamp: 2026-08-05T19:50:01Z
|
||||
git_ref: f0f3c806
|
||||
diff_cmd: git diff HEAD~1..HEAD
|
||||
---
|
||||
|
||||
The user reported that a completed NVDA 60s backfill left a large zero
|
||||
region in live SHM and asked for immediate inspection before restarting the
|
||||
workspace. The captured buffer contained 42,720 invalid rows at absolute
|
||||
indexes `3065895..3108614`. The valid rows on both sides had the identical
|
||||
timestamp `1781271360`, proving the region was an erroneous physical-index
|
||||
reservation rather than a real temporal gap.
|
||||
|
||||
The user then asked the agent to continue fixing the backfill logic.
|
||||
|
||||
> `git diff HEAD~1..HEAD -- piker/tsp/_history.py`
|
||||
|
||||
Removed wall-clock-based SHM reservation from `tsdb_backfill()`. Reverse
|
||||
provider frames now extend the visible `ShmArray._first` boundary on every
|
||||
prepend, beginning exactly at the current first index after inclusive query
|
||||
endpoints are removed. Added `_prepend_tsdb_history()` to trim storage rows
|
||||
against the actual earliest provider timestamp and prepend them directly
|
||||
adjacent only after reverse provider retrieval completes.
|
||||
|
||||
This preserves sparse venue history as ordinal samples rather than treating
|
||||
every elapsed calendar period as a physical SHM row. It also prevents the
|
||||
old `_first + 1` insertion point from overwriting the earliest published bar.
|
||||
|
||||
> `git diff HEAD~1..HEAD -- tests/test_history_backfill.py`
|
||||
|
||||
Added a full `tsdb_backfill()` regression modeled on the live NVDA failure.
|
||||
It uses a real `ShmArray`, NativeDB's field map, two inclusive reverse-provider
|
||||
frames, and overlapping stored history. Assertions prove the complete
|
||||
timestamp sequence is contiguous and lossless, contains no non-positive
|
||||
timestamps, sends complete provider frames to durable storage, and notifies
|
||||
chart/FSP consumers after the final storage prepend.
|
||||
|
||||
Live evidence captured before workspace restart:
|
||||
|
||||
```text
|
||||
rows 1575296 first_index 1568233 last_index 3143528
|
||||
invalid_count 42720
|
||||
left boundary
|
||||
[(3065893, 1781271300, 204.03, 204.82)
|
||||
(3065894, 1781271360, 204.81, 204.72)
|
||||
(3065895, 0, 0. , 0. )]
|
||||
right boundary
|
||||
[(3108614, 0, 0. , 0. )
|
||||
(3108615, 1781271360, 204.81, 205.44)
|
||||
(3108616, 1781271420, 205.46, 205.48)]
|
||||
```
|
||||
|
||||
Verification output:
|
||||
|
||||
```text
|
||||
........................................................................ [ 68%]
|
||||
................................. [100%]
|
||||
105 passed in 4.80s
|
||||
```
|
||||
|
||||
Adversarial review caught stale `_first` semantics and an off-by-one
|
||||
insertion point in the initial patch. The first live restart then exposed an
|
||||
undefined `time_key` local at the new `tsdb_backfill()` call site, which
|
||||
crashed `datad.ib` and propagated to the chart as `RemoteActorError`. The
|
||||
helper now derives the source timestamp field from its field map, and the
|
||||
test enters the full enclosing lifecycle so the original `NameError` can not
|
||||
recur unnoticed. A final review also drove explicit publication after stored
|
||||
history is prepended and deterministic final-stage synchronization. Final
|
||||
review found no remaining issues; another live restart remains pending.
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
---
|
||||
model: openai/gpt-5.6-sol
|
||||
service: opencode
|
||||
session: ses_0799212ebffe42arY96czXn89F
|
||||
timestamp: 2026-08-11T20:58:26Z
|
||||
git_ref: 06d5ea5d
|
||||
scope: code
|
||||
substantive: true
|
||||
raw_file: 20260811T205826Z_06d5ea5d_prompt_io.raw.md
|
||||
---
|
||||
|
||||
## Prompt
|
||||
|
||||
Diagnose and fix an IB live-order failure where lazy NVDA contract
|
||||
qualification raised `tractor.trionics.Lagged` after the order reached
|
||||
`brokerd.ib`. Determine whether recent history changes or cancelled symbol
|
||||
searches caused the stale state.
|
||||
|
||||
## Response summary
|
||||
|
||||
Traced the failure to competing consumers of a one-slot Trio broadcast, not
|
||||
the recent SHM change or stale asyncio client state. Replaced broadcast
|
||||
fan-out with one response/status demultiplexer, retained request-ID
|
||||
correlation, handled cancellation and teardown, documented the task/channel
|
||||
ownership model, warned on diagnostic metadata eviction, and added
|
||||
deterministic proxy regressions.
|
||||
|
||||
## Files changed
|
||||
|
||||
- `piker/brokers/ib/api.py` - sole-reader IB proxy response routing.
|
||||
- `tests/test_ib_method_proxy.py` - status, concurrency, cancellation and EOF regressions.
|
||||
- `.claude/skills/run-tests/test-harness-reference.md` - proxy test mapping.
|
||||
|
||||
## Human edits
|
||||
|
||||
None - generated output follows the user's live failure diagnosis.
|
||||
|
|
@ -0,0 +1,66 @@
|
|||
---
|
||||
model: openai/gpt-5.6-sol
|
||||
service: opencode
|
||||
timestamp: 2026-08-11T20:58:26Z
|
||||
git_ref: 06d5ea5d
|
||||
diff_cmd: git diff HEAD~1..HEAD
|
||||
---
|
||||
|
||||
The user reported that certain IB order entries might have broken and
|
||||
provided a live failure for a first NVDA sell order. EMS delivered the
|
||||
request to `brokerd.ib`, but lazy `get_mkt_info('nvda.nasdaq')` contract
|
||||
qualification failed in `MethodProxy._run_method()` with:
|
||||
|
||||
```text
|
||||
tractor.trionics._broadcast.Lagged:
|
||||
Task `piker.brokers.ib.broker.handle_order_requests` overrun and
|
||||
dropped `0` values
|
||||
```
|
||||
|
||||
The user suspected a cancelled symbol-search task might have left stale
|
||||
asyncio state. Cancellation can leave an orphaned method response, but the
|
||||
observed failure was stale Trio broadcast state. `open_client_proxy()`
|
||||
broadcast a one-slot `LinkedTaskChannel` to an always-running IB status relay
|
||||
and an idle method receiver. Status traffic advanced only the relay, so the
|
||||
first method call for an uncached symbol entered an already-lagged receiver.
|
||||
Orders for eagerly cached contracts bypassed this path.
|
||||
|
||||
The immediately preceding sparse-history commit did not modify IB order,
|
||||
symbol, or proxy code; it exposed no direct dependency on this failure.
|
||||
|
||||
> `git diff HEAD~1..HEAD -- piker/brokers/ib/api.py`
|
||||
|
||||
Generated a sole-reader message demultiplexer for each IB client proxy.
|
||||
`relay_client_proxy_messages()` owns every asyncio-to-Trio receive, handles
|
||||
status/error traffic, and dispatches `mid`-tagged responses into per-call
|
||||
`MethodProxy` wait slots. Cancelled calls remove their pending slot, causing
|
||||
late responses to be logged and dropped without skewing later calls. Clean
|
||||
channel EOF is accepted during normal proxy teardown.
|
||||
|
||||
Follow-up review restored the masked reconnect TODO and per-message case
|
||||
rationale, documented the one-reader/many-writer task hierarchy and separate
|
||||
per-proxy channels, and added a warning with request method metadata when the
|
||||
bounded unresolved-request diagnostic table evicts its oldest entry.
|
||||
|
||||
> `git diff HEAD~1..HEAD -- tests/test_ib_method_proxy.py`
|
||||
|
||||
Generated deterministic regressions for idle status traffic before first
|
||||
qualification, two concurrent calls with reverse-order responses, cancelled
|
||||
calls with late responses followed by a successful call, and graceful
|
||||
channel EOF. Every synchronization wait is timeout-bounded.
|
||||
|
||||
> `git diff HEAD~1..HEAD -- .claude/skills/run-tests/test-harness-reference.md`
|
||||
|
||||
Registered the deterministic IB method-proxy target and source mapping.
|
||||
|
||||
Verification output:
|
||||
|
||||
```text
|
||||
........................ [100%]
|
||||
25 passed, 4 warnings in 3.44s
|
||||
```
|
||||
|
||||
The warnings are existing Tractor/Trio deprecations from the EMS test.
|
||||
Adversarial review drove a clean-EOF guard and bounded polling. Final review
|
||||
found no actionable issues. Live IB order validation remains pending and
|
||||
requires restarting `brokerd.ib`.
|
||||
|
|
@ -24,8 +24,8 @@ automatically corruption or an expected venue closure.
|
|||
|
||||
Do not run qualification against normal user storage. The helper below
|
||||
requires a marked disposable root before it will seed, archive, or clear
|
||||
a Parquet. ``piker store anal`` and ``piker store ldshm`` are interactive
|
||||
and potentially mutating; they are not audit commands.
|
||||
a Parquet. ``piker store anal`` and ``piker store shm --write-parquet``
|
||||
are interactive and potentially mutating; they are not audit commands.
|
||||
|
||||
Keep the disposable root private and do not rename, replace, or symlink
|
||||
files under it while a helper is running. Stop the chart cleanly before
|
||||
|
|
@ -193,6 +193,20 @@ 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 ls
|
||||
piker store series FQME_SUBSTRING
|
||||
piker store shm FQME --max-gaps 20
|
||||
|
||||
All three forms are read-only. ``ls`` summarizes periods by FQME, while
|
||||
``series`` reports each ``(FQME, period)`` file independently. Read-only
|
||||
``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:
|
||||
|
||||
=============================== =========================================
|
||||
|
|
|
|||
|
|
@ -1643,6 +1643,61 @@ class MethodProxy:
|
|||
# to the next caller causing an off-by-one skew of
|
||||
# every result thereafter!
|
||||
self._mids = itertools.count()
|
||||
self._pending: dict[
|
||||
int,
|
||||
tuple[trio.Event, list[dict]],
|
||||
] = {}
|
||||
self._request_methods: dict[int, str|None] = {}
|
||||
|
||||
def _deliver_method_response(
|
||||
self,
|
||||
msg: dict,
|
||||
|
||||
) -> bool:
|
||||
'''
|
||||
Deliver one correlated response to its waiting caller.
|
||||
|
||||
'''
|
||||
mid: int = msg['mid']
|
||||
meth: str|None = self._request_methods.pop(mid, None)
|
||||
pending = self._pending.get(mid)
|
||||
if pending is None:
|
||||
log.warning(
|
||||
f'Dropping stale method-resp,\n'
|
||||
f'mid: {mid}\n'
|
||||
f'meth: {meth!r}\n'
|
||||
f'(caller prolly got cancelled '
|
||||
f'before its resp?)\n'
|
||||
)
|
||||
return False
|
||||
|
||||
event, slot = pending
|
||||
slot.append(msg)
|
||||
event.set()
|
||||
return True
|
||||
|
||||
def _track_request_method(
|
||||
self,
|
||||
mid: int,
|
||||
meth: str|None,
|
||||
|
||||
) -> None:
|
||||
'''
|
||||
Retain bounded request metadata for stale-response logs.
|
||||
|
||||
'''
|
||||
self._request_methods[mid] = meth
|
||||
if len(self._request_methods) > 256:
|
||||
oldest_mid: int = next(iter(self._request_methods))
|
||||
oldest_meth: str|None = self._request_methods.pop(
|
||||
oldest_mid
|
||||
)
|
||||
log.warning(
|
||||
f'Evicting unresolved method-request metadata,\n'
|
||||
f'mid: {oldest_mid}\n'
|
||||
f'meth: {oldest_meth!r}\n'
|
||||
f'pending metadata limit: 256\n'
|
||||
)
|
||||
|
||||
async def _run_method(
|
||||
self,
|
||||
|
|
@ -1656,65 +1711,30 @@ class MethodProxy:
|
|||
``tractor.to_asyncio`` layer.
|
||||
|
||||
'''
|
||||
chan = self.chan
|
||||
mid: int = next(self._mids)
|
||||
await chan.send((meth, kwargs, mid))
|
||||
event = trio.Event()
|
||||
slot: list[dict] = []
|
||||
self._pending[mid] = event, slot
|
||||
self._track_request_method(mid, meth)
|
||||
try:
|
||||
await self.chan.send((meth, kwargs, mid))
|
||||
await event.wait()
|
||||
finally:
|
||||
self._pending.pop(mid, None)
|
||||
|
||||
while not chan.closed():
|
||||
msg = await chan.receive()
|
||||
msg: dict = slot[0]
|
||||
match msg:
|
||||
# OUR method-call response B)
|
||||
case {'result': res}:
|
||||
return res
|
||||
|
||||
# TODO: implement reconnect functionality like
|
||||
# in our `.data._web_bs.NoBsWs`
|
||||
# try:
|
||||
# msg = await chan.receive()
|
||||
# except ConnectionError:
|
||||
# self.reset()
|
||||
case {'exception': err}:
|
||||
raise err
|
||||
|
||||
match msg:
|
||||
# OUR method-call response B)
|
||||
case {'mid': resp_mid, 'result': res} if (
|
||||
resp_mid == mid
|
||||
):
|
||||
return res
|
||||
|
||||
case {'mid': resp_mid, 'exception': err} if (
|
||||
resp_mid == mid
|
||||
):
|
||||
raise err
|
||||
|
||||
# an "orphaned" response to some prior
|
||||
# (cancelled) caller's request; drop it and
|
||||
# keep waiting for ours.
|
||||
case {'mid': resp_mid}:
|
||||
log.warning(
|
||||
f'Dropping stale method-resp,\n'
|
||||
f'mid: {resp_mid} (ours: {mid})\n'
|
||||
f'(a prior caller prolly got '
|
||||
f'cancelled before its resp?)\n'
|
||||
)
|
||||
continue
|
||||
|
||||
# out-of-band (inline) client error: raise
|
||||
# to the current caller as before.
|
||||
case {'exception': err}:
|
||||
raise err
|
||||
|
||||
case ('error', emsg):
|
||||
log.warning(f'IB error relay: {emsg}')
|
||||
continue
|
||||
|
||||
# routine (api-farm conn) status events
|
||||
# relayed inline by `Client.inline_errors()`;
|
||||
# not a response to our method call so just
|
||||
# log at info and keep waiting.
|
||||
case ('event', emsg):
|
||||
log.info(
|
||||
f'IB status event relay: {emsg}'
|
||||
)
|
||||
continue
|
||||
|
||||
case _:
|
||||
log.warning(f'UNKNOWN IB MSG: {msg}')
|
||||
case _:
|
||||
raise RuntimeError(
|
||||
f'Invalid method response for mid={mid}: {msg!r}'
|
||||
)
|
||||
|
||||
def status_event(
|
||||
self,
|
||||
|
|
@ -1800,6 +1820,73 @@ async def open_aio_client_method_relay(
|
|||
raise ValueError(f'Unhandled msg {msg}')
|
||||
|
||||
|
||||
async def relay_client_proxy_messages(
|
||||
chan: tractor.to_asyncio.LinkedTaskChannel,
|
||||
proxy: MethodProxy,
|
||||
|
||||
) -> None:
|
||||
'''
|
||||
Route one proxy's asyncio-client msgs from a single reader.
|
||||
|
||||
Each `open_client_proxy()` allocates its own linked channel and
|
||||
starts this task in its relay nursery. Any number of Trio caller
|
||||
tasks may concurrently write tagged requests through
|
||||
`MethodProxy._run_method()` and wait on their per-`mid` slots.
|
||||
This task alone reads the channel and dispatches each response to
|
||||
its waiting caller.
|
||||
|
||||
Separate proxy contexts own separate channels and reader tasks;
|
||||
no cross-proxy response broadcast is intended. IB status events
|
||||
still reach each proxy through its own `Client.inline_errors()`
|
||||
handler.
|
||||
|
||||
'''
|
||||
try:
|
||||
while not chan.closed():
|
||||
# TODO: implement reconnect functionality like
|
||||
# in our `.data._web_bs.NoBsWs`
|
||||
# try:
|
||||
msg = await chan.receive()
|
||||
# except ConnectionError:
|
||||
# proxy.reset()
|
||||
|
||||
match msg:
|
||||
# Correlated response from a method call.
|
||||
case {'mid': _}:
|
||||
proxy._deliver_method_response(msg)
|
||||
|
||||
# routine (api-farm conn) status events
|
||||
# relayed inline by `Client.inline_errors()`;
|
||||
# not a response to a method call so just
|
||||
# log at info and keep waiting.
|
||||
case ('event', status_msg):
|
||||
reason = status_msg['reason']
|
||||
event = proxy.event_table.pop(reason, None)
|
||||
if (
|
||||
event
|
||||
and
|
||||
event.statistics().tasks_waiting
|
||||
):
|
||||
log.info(f'Relaying ib status message: {msg}')
|
||||
event.set()
|
||||
|
||||
# Inline API error with no request correlation.
|
||||
case ('error', emsg):
|
||||
log.warning(f'IB error relay: {emsg}')
|
||||
|
||||
# Out-of-band client error with no method `mid`.
|
||||
case {'exception': err}:
|
||||
log.error(
|
||||
f'Uncorrelated IB client error: {err!r}'
|
||||
)
|
||||
|
||||
# Preserve visibility into unexpected relay traffic.
|
||||
case _:
|
||||
log.warning(f'UNKNOWN IB MSG: {msg}')
|
||||
except trio.EndOfChannel:
|
||||
log.info('IB client proxy relay closed')
|
||||
|
||||
|
||||
@acm
|
||||
async def open_client_proxy(
|
||||
client: Client,
|
||||
|
|
@ -1836,28 +1923,11 @@ async def open_client_proxy(
|
|||
continue
|
||||
setattr(proxy, name, partial(proxy._run_method, meth=name))
|
||||
|
||||
async def relay_events():
|
||||
|
||||
async with chan.subscribe() as msg_stream:
|
||||
|
||||
async for msg in msg_stream:
|
||||
if 'event' not in msg:
|
||||
continue
|
||||
|
||||
# if 'event' in msg:
|
||||
# wake up any system event waiters.
|
||||
etype, status_msg = msg
|
||||
reason = status_msg['reason']
|
||||
|
||||
ev = proxy.event_table.pop(reason, None)
|
||||
|
||||
if ev and ev.statistics().tasks_waiting:
|
||||
log.info(f'Relaying ib status message: {msg}')
|
||||
ev.set()
|
||||
|
||||
continue
|
||||
|
||||
relay_tn.start_soon(relay_events)
|
||||
relay_tn.start_soon(
|
||||
relay_client_proxy_messages,
|
||||
chan,
|
||||
proxy,
|
||||
)
|
||||
|
||||
yield proxy
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
)
|
||||
|
||||
|
|
@ -46,18 +51,43 @@ 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
|
||||
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 +121,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,48 +269,146 @@ def _render_audit_report(report: dict) -> None:
|
|||
console.print(gap_table)
|
||||
|
||||
|
||||
def _native_series_payload(
|
||||
pattern: str|None,
|
||||
|
||||
) -> tuple[Path, list[dict]]:
|
||||
'''
|
||||
Collect exact NativeDB series metadata for CLI presentation.
|
||||
|
||||
'''
|
||||
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,
|
||||
})
|
||||
payload.sort(key=lambda item: (
|
||||
item['fqme'],
|
||||
item['period_s'],
|
||||
))
|
||||
|
||||
return datadir, payload
|
||||
|
||||
|
||||
@store.command()
|
||||
def ls(
|
||||
backends: list[str] = typer.Argument(
|
||||
default=None,
|
||||
help='Storage backends to query, default is all.'
|
||||
),
|
||||
):
|
||||
from rich.table import Table
|
||||
pattern: Annotated[
|
||||
str|None,
|
||||
typer.Argument(
|
||||
help='Optional case-insensitive FQME substring.',
|
||||
),
|
||||
] = None,
|
||||
all_series: Annotated[
|
||||
bool,
|
||||
typer.Option(
|
||||
'--all',
|
||||
help='List all FQMEs (the default).',
|
||||
),
|
||||
] = False,
|
||||
) -> None:
|
||||
'''
|
||||
List NativeDB FQMEs and their available sample periods.
|
||||
|
||||
if not backends:
|
||||
backends: list[str] = __tsdbs__
|
||||
'''
|
||||
if (
|
||||
all_series
|
||||
and
|
||||
pattern is not None
|
||||
):
|
||||
raise typer.BadParameter('Pass a pattern or --all, not both')
|
||||
|
||||
console = Console()
|
||||
datadir, payload = _native_series_payload(pattern)
|
||||
periods_by_fqme: dict[str, list[int]] = {}
|
||||
for item in payload:
|
||||
periods_by_fqme.setdefault(item['fqme'], []).append(
|
||||
item['period_s']
|
||||
)
|
||||
|
||||
async def query_all():
|
||||
nonlocal backends
|
||||
|
||||
async with (
|
||||
open_piker_runtime(
|
||||
'tsdb_storage',
|
||||
table = Table(title=f'NativeDB series @ {datadir}')
|
||||
table.add_column('FQME')
|
||||
table.add_column('Periods')
|
||||
for fqme, periods in periods_by_fqme.items():
|
||||
table.add_row(
|
||||
fqme,
|
||||
', '.join(
|
||||
f'{period}s'
|
||||
for period in sorted(periods)
|
||||
),
|
||||
):
|
||||
for i, backend in enumerate(backends):
|
||||
table = Table()
|
||||
try:
|
||||
async with open_storage_client(backend=backend) as (
|
||||
mod,
|
||||
client,
|
||||
):
|
||||
table.add_column(f'{mod.name}@{client.address}')
|
||||
keys: list[str] = await client.list_keys()
|
||||
for key in keys:
|
||||
table.add_row(key)
|
||||
|
||||
console.print(table)
|
||||
except Exception:
|
||||
log.error(f'Unable to connect to storage engine: `{backend}`')
|
||||
|
||||
trio.run(query_all)
|
||||
)
|
||||
Console().print(table)
|
||||
|
||||
|
||||
@store.command()
|
||||
def series(
|
||||
pattern: Annotated[
|
||||
str|None,
|
||||
typer.Argument(
|
||||
help='Optional case-insensitive FQME substring.',
|
||||
),
|
||||
] = None,
|
||||
all_series: Annotated[
|
||||
bool,
|
||||
typer.Option(
|
||||
'--all',
|
||||
help='List all series (the default).',
|
||||
),
|
||||
] = False,
|
||||
json_output: Annotated[
|
||||
bool,
|
||||
typer.Option(
|
||||
'--json',
|
||||
help='Emit machine-readable JSON.',
|
||||
),
|
||||
] = False,
|
||||
) -> None:
|
||||
'''
|
||||
List detailed NativeDB sub-series without opening a runtime.
|
||||
|
||||
'''
|
||||
if (
|
||||
all_series
|
||||
and
|
||||
pattern is not None
|
||||
):
|
||||
raise typer.BadParameter('Pass a pattern or --all, not both')
|
||||
datadir, payload = _native_series_payload(pattern)
|
||||
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 +584,7 @@ def audit(
|
|||
# ...
|
||||
|
||||
|
||||
@store.command()
|
||||
@store.command(no_args_is_help=True)
|
||||
def delete(
|
||||
symbols: list[str],
|
||||
|
||||
|
|
@ -405,7 +622,7 @@ def delete(
|
|||
trio.run(main, symbols)
|
||||
|
||||
|
||||
@store.command()
|
||||
@store.command(no_args_is_help=True)
|
||||
def anal(
|
||||
fqme: str,
|
||||
period: int = 60,
|
||||
|
|
@ -494,21 +711,234 @@ def anal(
|
|||
trio.run(main)
|
||||
|
||||
|
||||
@store.command()
|
||||
def ldshm(
|
||||
def _inspect_shm(
|
||||
fqme: str,
|
||||
write_parquet: bool = True,
|
||||
reload_parquet_to_shm: bool = True,
|
||||
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,
|
||||
pdb: bool = 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',
|
||||
debug_mode=pdb,
|
||||
):
|
||||
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 shm(
|
||||
fqme: str,
|
||||
write_parquet: Annotated[
|
||||
bool,
|
||||
typer.Option(
|
||||
'--write-parquet',
|
||||
help='Persist repaired frames to NativeDB.',
|
||||
),
|
||||
] = False,
|
||||
reload_parquet_to_shm: Annotated[
|
||||
bool,
|
||||
typer.Option(
|
||||
'--reload-parquet-to-shm',
|
||||
help='Reload persisted repairs into SHM.',
|
||||
),
|
||||
] = False,
|
||||
pdb: bool = False, # --pdb passed?
|
||||
max_gaps: Annotated[
|
||||
int|None,
|
||||
typer.Option(
|
||||
'--max-gaps',
|
||||
min=0,
|
||||
help=(
|
||||
'Maximum largest gaps in read-only output '
|
||||
'(default: 10).'
|
||||
),
|
||||
show_default=False,
|
||||
),
|
||||
] = None,
|
||||
json_output: Annotated[
|
||||
bool,
|
||||
typer.Option(
|
||||
'--json',
|
||||
help='Emit read-only output as JSON.',
|
||||
),
|
||||
] = False,
|
||||
shm_name: Annotated[
|
||||
str|None,
|
||||
typer.Option(
|
||||
'--shm-name',
|
||||
help='Select only this exact SHM object name.',
|
||||
),
|
||||
] = None,
|
||||
|
||||
) -> None:
|
||||
'''
|
||||
Linux ONLY: load any fqme file name matching shm buffer from
|
||||
/dev/shm/ into an OHLCV numpy array and polars DataFrame,
|
||||
optionally write to offline storage via `.parquet` file.
|
||||
Inspect volatile OHLCV SHM, with explicit repair options.
|
||||
|
||||
'''
|
||||
async def main():
|
||||
if (
|
||||
reload_parquet_to_shm
|
||||
and
|
||||
not write_parquet
|
||||
):
|
||||
raise typer.BadParameter(
|
||||
'--reload-parquet-to-shm requires --write-parquet'
|
||||
)
|
||||
if (
|
||||
write_parquet
|
||||
and (
|
||||
json_output
|
||||
or
|
||||
max_gaps is not None
|
||||
)
|
||||
):
|
||||
raise typer.BadParameter(
|
||||
'--json and --max-gaps apply only to read-only output'
|
||||
)
|
||||
if not write_parquet:
|
||||
_inspect_shm(
|
||||
fqme,
|
||||
shm_name=shm_name,
|
||||
max_gaps=(
|
||||
10
|
||||
if max_gaps is None
|
||||
else max_gaps
|
||||
),
|
||||
json_output=json_output,
|
||||
pdb=pdb,
|
||||
)
|
||||
return
|
||||
|
||||
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 +960,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 +1137,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 +1164,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)
|
||||
|
|
|
|||
|
|
@ -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'
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -240,7 +260,7 @@ async def shm_push_in_between(
|
|||
prepend_index: int,
|
||||
backfill_until_dt: datetime,
|
||||
|
||||
update_start_on_prepend: bool = False,
|
||||
update_start_on_prepend: bool = True,
|
||||
|
||||
) -> int:
|
||||
|
||||
|
|
@ -262,16 +282,12 @@ async def shm_push_in_between(
|
|||
to_push,
|
||||
prepend=True,
|
||||
|
||||
# XXX: only update the ._first index if no tsdb
|
||||
# segment was previously prepended by the
|
||||
# parent task.
|
||||
# Keep the readable `ShmArray.array` boundary aligned with
|
||||
# every completed provider prepend.
|
||||
update_first=update_start_on_prepend,
|
||||
|
||||
# XXX: only prepend from a manually calculated shm
|
||||
# index if there was already a tsdb history
|
||||
# segment prepended (since then the
|
||||
# ._first.value is going to be wayyy in the
|
||||
# past!)
|
||||
# An explicit index is only needed by callers retaining a
|
||||
# separately placed segment before the readable boundary.
|
||||
start=(
|
||||
prepend_index
|
||||
if not update_start_on_prepend
|
||||
|
|
@ -538,7 +554,7 @@ async def start_backfill(
|
|||
task_status.started()
|
||||
|
||||
# based on the sample step size, maybe load a certain amount history
|
||||
update_start_on_prepend: bool = False
|
||||
update_start_on_prepend: bool = True
|
||||
if (
|
||||
_until_was_none := (backfill_until_dt is None)
|
||||
):
|
||||
|
|
@ -559,8 +575,6 @@ async def start_backfill(
|
|||
60: {'years': 6},
|
||||
}
|
||||
period_duration: int = periods[timeframe]
|
||||
update_start_on_prepend: bool = True
|
||||
|
||||
# NOTE: manually set the "latest" datetime which we intend to
|
||||
# backfill history "until" so as to adhere to the history
|
||||
# settings above when the tsdb is detected as being empty.
|
||||
|
|
@ -747,7 +761,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()
|
||||
|
|
@ -1144,6 +1158,74 @@ async def load_tsdb_hist(
|
|||
return None
|
||||
|
||||
|
||||
def _prepend_tsdb_history(
|
||||
shm: ShmArray,
|
||||
tsdb_history: np.ndarray,
|
||||
field_map: bidict|None,
|
||||
|
||||
) -> np.ndarray:
|
||||
'''
|
||||
Prepend stored rows beside the actual provider boundary.
|
||||
|
||||
'''
|
||||
available: int = max(0, int(shm._first.value))
|
||||
if (
|
||||
not available
|
||||
or
|
||||
not len(tsdb_history)
|
||||
):
|
||||
return tsdb_history[:0]
|
||||
|
||||
frame: np.ndarray = shm.array
|
||||
if not len(frame):
|
||||
return tsdb_history[:0]
|
||||
|
||||
# XXX EDGE CASEs: the most recent provider frame can overlap
|
||||
# prior TSDB history when its start is earlier than the latest
|
||||
# stored sample.
|
||||
#
|
||||
# Alternatively, this can occur when the venue was closed (say
|
||||
# over a weekend), causing a wall-clock time-series gap while the
|
||||
# provider's query frame is larger than the current market's
|
||||
# operating session. For example, IB's 1s default returns around
|
||||
# 2k datums (~33.33m), which can include samples from before a
|
||||
# closure. The wall-clock interval between frames then differs
|
||||
# from the number of actual venue samples and may even look like
|
||||
# a negative overlap when compared only by endpoints.
|
||||
#
|
||||
# The previous placement converted that elapsed interval into a
|
||||
# physical SHM offset. Sparse venue series contain samples, not
|
||||
# one row per wall-clock period, so closures accumulated as null
|
||||
# reservations between otherwise adjacent data. Instead, wait
|
||||
# until reverse provider retrieval establishes the actual first
|
||||
# sample, retain its values for any overlap, and prepend only TSDB
|
||||
# samples older than that boundary.
|
||||
#
|
||||
# TODO: assert overlapping TSDB and provider segments contain the
|
||||
# same values before preferring the provider's copy.
|
||||
time_key: str = (
|
||||
field_map.inverse.get('time', 'time')
|
||||
if field_map
|
||||
else 'time'
|
||||
)
|
||||
earliest_t: float = frame['time'][0]
|
||||
older: np.ndarray = tsdb_history[
|
||||
tsdb_history[time_key] < earliest_t
|
||||
]
|
||||
|
||||
# Stored history may extend further into the past than remaining
|
||||
# SHM capacity. Keep only its most recent rows in that case.
|
||||
to_push: np.ndarray = older[-available:]
|
||||
if len(to_push):
|
||||
shm.push(
|
||||
to_push,
|
||||
prepend=True,
|
||||
field_map=field_map,
|
||||
)
|
||||
|
||||
return to_push
|
||||
|
||||
|
||||
async def tsdb_backfill(
|
||||
mod: ModuleType,
|
||||
storemod: ModuleType,
|
||||
|
|
@ -1229,7 +1311,7 @@ async def tsdb_backfill(
|
|||
# NOTE: iabs to start backfilling from, reverse chronological,
|
||||
# ONLY AFTER the first history frame has been pushed to
|
||||
# mem!
|
||||
backfill_gap_from_shm_index: int = shm._first.value + 1
|
||||
backfill_gap_from_shm_index: int = shm._first.value
|
||||
|
||||
# Prepend any tsdb history into the rt-shm-buffer which
|
||||
# should NOW be getting filled with the most recent history
|
||||
|
|
@ -1318,76 +1400,29 @@ async def tsdb_backfill(
|
|||
write_tsdb=True,
|
||||
)
|
||||
)
|
||||
if last_tsdb_dt is not None:
|
||||
|
||||
# calc the index from which the tsdb data should be
|
||||
# prepended, presuming there is a gap between the
|
||||
# latest frame (loaded/read above) and the latest
|
||||
# sample loaded from the tsdb.
|
||||
backfill_diff: Duration = mr_start_dt - last_tsdb_dt
|
||||
offset_s: float = backfill_diff.in_seconds()
|
||||
|
||||
# XXX EDGE CASEs: the most recent frame overlaps with
|
||||
# prior tsdb history!!
|
||||
# - so the latest frame's start time is earlier then
|
||||
# the tsdb's latest sample.
|
||||
# - alternatively this may also more generally occur
|
||||
# when the venue was closed (say over the weeknd)
|
||||
# causing a timeseries gap, AND the query frames size
|
||||
# (eg. for ib's 1s we rx 2k datums ~= 33.33m) IS
|
||||
# GREATER THAN the current venue-market's operating
|
||||
# session (time) we will receive datums from BEFORE THE
|
||||
# CLOSURE GAP and thus the `offset_s` value will be
|
||||
# NEGATIVE! In this case we need to ensure we don't try
|
||||
# to push datums that have already been recorded in the
|
||||
# tsdb. In this case we instead only retreive and push
|
||||
# the series portion missing from the db's data set.
|
||||
# if offset_s < 0:
|
||||
# non_overlap_diff: Duration = mr_end_dt - last_tsdb_dt
|
||||
# non_overlap_offset_s: float = backfill_diff.in_seconds()
|
||||
|
||||
offset_samples: int = round(offset_s / timeframe)
|
||||
|
||||
# TODO: see if there's faster multi-field reads:
|
||||
# https://numpy.org/doc/stable/user/basics.rec.html#accessing-multiple-fields
|
||||
# re-index with a `time` and index field
|
||||
if offset_s > 0:
|
||||
# NOTE XXX: ONLY when there is an actual gap
|
||||
# between the earliest sample in the latest history
|
||||
# frame do we want to NOT stick the latest tsdb
|
||||
# history adjacent to that latest frame!
|
||||
prepend_start = shm._first.value - offset_samples + 1
|
||||
to_push = tsdb_history[-prepend_start:]
|
||||
else:
|
||||
# when there is overlap we want to remove the
|
||||
# overlapping samples from the tsdb portion (taking
|
||||
# instead the latest frame's values since THEY
|
||||
# SHOULD BE THE SAME) and prepend DIRECTLY adjacent
|
||||
# to the latest frame!
|
||||
# TODO: assert the overlap segment array contains
|
||||
# the same values!?!
|
||||
prepend_start = shm._first.value
|
||||
to_push = tsdb_history[-(shm._first.value):offset_samples - 1]
|
||||
|
||||
# tsdb history is so far in the past we can't fit it in
|
||||
# shm buffer space so simply don't load it!
|
||||
if prepend_start > 0:
|
||||
shm.push(
|
||||
to_push,
|
||||
|
||||
# insert the history pre a "days worth" of samples
|
||||
# to leave some real-time buffer space at the end.
|
||||
prepend=True,
|
||||
# update_first=False,
|
||||
start=prepend_start,
|
||||
field_map=storemod.ohlc_key_map,
|
||||
)
|
||||
|
||||
log.info(f'Loaded {to_push.shape} datums from storage')
|
||||
|
||||
# 2nd nursery END
|
||||
|
||||
if last_tsdb_dt is not None:
|
||||
# Provider frames contain venue samples, not one row
|
||||
# per wall-clock period. Wait until reverse retrieval
|
||||
# establishes the actual earliest sample, then prepend
|
||||
# storage directly beside it without reserving closure
|
||||
# rows.
|
||||
to_push: np.ndarray = _prepend_tsdb_history(
|
||||
shm,
|
||||
tsdb_history,
|
||||
field_map=storemod.ohlc_key_map,
|
||||
)
|
||||
log.info(
|
||||
f'Loaded {to_push.shape} datums from storage'
|
||||
)
|
||||
if len(to_push):
|
||||
await notify_backfill(
|
||||
sampler_stream,
|
||||
mkt,
|
||||
timeframe,
|
||||
)
|
||||
|
||||
# Repair only after reverse backfill releases providers
|
||||
# such as IB which permit one history request at a time.
|
||||
# TODO: ideally these null segments can never exist!
|
||||
|
|
@ -1681,8 +1716,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 +1903,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)
|
||||
|
||||
|
|
|
|||
|
|
@ -59,7 +59,10 @@ def expect(
|
|||
|
||||
def run_pdb_commands(
|
||||
commands: list[str],
|
||||
initial_cmd: str = 'piker store ldshm xmrusdt.usdtm.perp.binance',
|
||||
initial_cmd: str = (
|
||||
'piker store shm xmrusdt.usdtm.perp.binance '
|
||||
'--write-parquet'
|
||||
),
|
||||
timeout: int = 30,
|
||||
print_output: bool = True,
|
||||
) -> dict[str, str]:
|
||||
|
|
@ -145,7 +148,10 @@ class InteractivePdbSession:
|
|||
'''
|
||||
def __init__(
|
||||
self,
|
||||
cmd: str = 'piker store ldshm xmrusdt.usdtm.perp.binance',
|
||||
cmd: str = (
|
||||
'piker store shm xmrusdt.usdtm.perp.binance '
|
||||
'--write-parquet'
|
||||
),
|
||||
timeout: int = 30,
|
||||
):
|
||||
self.cmd: str = cmd
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
Deterministic history-backfill regressions.
|
||||
|
||||
'''
|
||||
from contextlib import asynccontextmanager
|
||||
from functools import partial
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
|
@ -34,6 +35,7 @@ from piker.tsp._history import (
|
|||
notify_backfill,
|
||||
publish_latest_frame,
|
||||
start_backfill,
|
||||
tsdb_backfill,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -216,6 +218,197 @@ def test_latest_frame_is_persisted_before_shm() -> None:
|
|||
assert events == ['storage', 'shm']
|
||||
|
||||
|
||||
def test_tsdb_prepend_uses_actual_provider_boundary(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
'''
|
||||
Sparse venue history must not reserve wall-clock rows in SHM.
|
||||
|
||||
NVDA's 60s startup converted elapsed calendar minutes between
|
||||
NativeDB and the latest IB frame into physical SHM offsets. IB
|
||||
returned only trading bars, leaving 42,720 zero rows even though
|
||||
both valid margins ended at the same timestamp. Model completed
|
||||
reverse retrieval with a provider boundary overlapping NativeDB.
|
||||
Exercise real ``ShmArray.push()`` indexing and prove the stored,
|
||||
reverse-provider, and latest frames form one contiguous lossless
|
||||
sequence without an explicit wall-clock offset.
|
||||
|
||||
'''
|
||||
stored = np.zeros(
|
||||
3,
|
||||
dtype=np.dtype(def_iohlcv_fields),
|
||||
)
|
||||
stored['time'] = [60, 120, 180]
|
||||
older_reverse = np.zeros(
|
||||
3,
|
||||
dtype=np.dtype(def_iohlcv_fields),
|
||||
)
|
||||
older_reverse['time'] = [180, 240, 300]
|
||||
newer_reverse = np.zeros(
|
||||
3,
|
||||
dtype=np.dtype(def_iohlcv_fields),
|
||||
)
|
||||
newer_reverse['time'] = [300, 360, 420]
|
||||
latest = np.zeros(
|
||||
2,
|
||||
dtype=np.dtype(def_iohlcv_fields),
|
||||
)
|
||||
latest['time'] = [420, 480]
|
||||
storage_frames: list[np.ndarray] = []
|
||||
notifications: list[dict] = []
|
||||
startup_done = trio.Event()
|
||||
|
||||
class Storage:
|
||||
'''
|
||||
Capture the complete provider delta written to NativeDB.
|
||||
|
||||
'''
|
||||
async def update_ohlcv(
|
||||
self,
|
||||
fqme: str,
|
||||
ohlcv: np.ndarray,
|
||||
timeframe: int,
|
||||
) -> None:
|
||||
'''
|
||||
Record one reverse provider response.
|
||||
|
||||
'''
|
||||
storage_frames.append(ohlcv.copy())
|
||||
|
||||
async def load(
|
||||
self,
|
||||
fqme: str,
|
||||
timeframe: int,
|
||||
) -> tuple:
|
||||
'''
|
||||
Return overlapping NativeDB history and its boundaries.
|
||||
|
||||
'''
|
||||
return (
|
||||
stored,
|
||||
from_timestamp(60),
|
||||
from_timestamp(180),
|
||||
)
|
||||
|
||||
class Sampler:
|
||||
'''
|
||||
Accept deterministic backfill notifications.
|
||||
|
||||
'''
|
||||
async def send(self, msg) -> None:
|
||||
'''
|
||||
Capture notification without actor IPC.
|
||||
|
||||
'''
|
||||
notifications.append(msg)
|
||||
|
||||
mkt = SimpleNamespace(
|
||||
fqme='nvda.nasdaq.ib',
|
||||
dst=SimpleNamespace(atype='stock'),
|
||||
src=SimpleNamespace(atype='fiat'),
|
||||
get_fqme=lambda **kwargs: 'nvda.nasdaq.ib',
|
||||
)
|
||||
|
||||
async def get_hist(*args, **kwargs):
|
||||
'''
|
||||
Return inclusive reverse frames overlapping each SHM
|
||||
boundary.
|
||||
|
||||
'''
|
||||
end_dt = kwargs['end_dt']
|
||||
if end_dt is None:
|
||||
return (
|
||||
latest,
|
||||
from_timestamp(420),
|
||||
from_timestamp(480),
|
||||
)
|
||||
|
||||
end_t: float = end_dt.timestamp()
|
||||
reverse: np.ndarray = (
|
||||
newer_reverse
|
||||
if end_t == 420
|
||||
else older_reverse
|
||||
)
|
||||
return (
|
||||
reverse,
|
||||
from_timestamp(reverse['time'][0]),
|
||||
from_timestamp(reverse['time'][-1]),
|
||||
)
|
||||
|
||||
@asynccontextmanager
|
||||
async def open_history_client(mkt):
|
||||
'''
|
||||
Yield the deterministic provider and default frame config.
|
||||
|
||||
'''
|
||||
yield get_hist, {}
|
||||
|
||||
async def finish_null_repair(**kwargs) -> None:
|
||||
'''
|
||||
Signal that provider and NativeDB prepends both completed.
|
||||
|
||||
'''
|
||||
startup_done.set()
|
||||
|
||||
monkeypatch.setattr(
|
||||
'piker.tsp._history.maybe_fill_null_segments',
|
||||
finish_null_repair,
|
||||
)
|
||||
|
||||
async def main() -> None:
|
||||
shm_key = f'test_sparse_prepend_{uuid4().hex}'
|
||||
async with tractor.open_root_actor(
|
||||
name=shm_key,
|
||||
tpt_bind_addrs=[('127.0.0.1', 0)],
|
||||
):
|
||||
shm, opened = maybe_open_shm_array(
|
||||
key=shm_key,
|
||||
size=16,
|
||||
dtype=np.dtype(def_iohlcv_fields),
|
||||
append_start_index=12,
|
||||
)
|
||||
assert opened
|
||||
storage = Storage()
|
||||
async with trio.open_nursery() as nursery:
|
||||
nursery.start_soon(partial(
|
||||
tsdb_backfill,
|
||||
mod=SimpleNamespace(
|
||||
name='fake',
|
||||
open_history_client=open_history_client,
|
||||
),
|
||||
storemod=SimpleNamespace(
|
||||
ohlc_key_map=ohlc_key_map,
|
||||
),
|
||||
storage=storage,
|
||||
mkt=mkt,
|
||||
shm=shm,
|
||||
timeframe=60,
|
||||
sampler_stream=Sampler(),
|
||||
))
|
||||
with trio.fail_after(0.5):
|
||||
await startup_done.wait()
|
||||
nursery.cancel_scope.cancel()
|
||||
|
||||
assert shm.array['time'].tolist() == [
|
||||
60,
|
||||
120,
|
||||
180,
|
||||
240,
|
||||
300,
|
||||
360,
|
||||
420,
|
||||
480,
|
||||
]
|
||||
assert not np.any(shm.array['time'] <= 0)
|
||||
|
||||
trio.run(main)
|
||||
assert len(storage_frames) == 3
|
||||
assert storage_frames[0]['time'].tolist() == [420, 480]
|
||||
assert storage_frames[1]['time'].tolist() == [300, 360, 420]
|
||||
assert storage_frames[2]['time'].tolist() == [180, 240, 300]
|
||||
assert len(notifications) == 3
|
||||
|
||||
|
||||
def test_ib_latest_frame_round_trips_through_nativedb(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,271 @@
|
|||
'''
|
||||
IB asyncio method-proxy regressions.
|
||||
|
||||
'''
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
import trio
|
||||
|
||||
from piker.brokers.ib import api as ib_api
|
||||
from piker.brokers.ib.api import (
|
||||
MethodProxy,
|
||||
relay_client_proxy_messages,
|
||||
)
|
||||
|
||||
|
||||
class FakeChannel:
|
||||
'''
|
||||
Model the Trio side of an asyncio linked-task channel.
|
||||
|
||||
'''
|
||||
def __init__(self) -> None:
|
||||
self.requests: list[tuple] = []
|
||||
(
|
||||
self._tx,
|
||||
self._rx,
|
||||
) = trio.open_memory_channel[Any](10)
|
||||
|
||||
async def send(self, msg: tuple) -> None:
|
||||
'''
|
||||
Capture one Trio-to-asyncio method request.
|
||||
|
||||
'''
|
||||
self.requests.append(msg)
|
||||
|
||||
async def receive(self) -> Any:
|
||||
'''
|
||||
Receive one simulated asyncio-to-Trio response.
|
||||
|
||||
'''
|
||||
return await self._rx.receive()
|
||||
|
||||
def closed(self) -> bool:
|
||||
'''
|
||||
Report an open channel for the lifetime of each test.
|
||||
|
||||
'''
|
||||
return False
|
||||
|
||||
async def respond(self, msg: Any) -> None:
|
||||
'''
|
||||
Send one response or status event to the proxy relay.
|
||||
|
||||
'''
|
||||
await self._tx.send(msg)
|
||||
|
||||
async def close_responses(self) -> None:
|
||||
'''
|
||||
Close the simulated asyncio-to-Trio response channel.
|
||||
|
||||
'''
|
||||
await self._tx.aclose()
|
||||
|
||||
|
||||
def test_method_proxy_routes_after_idle_status_events() -> None:
|
||||
'''
|
||||
Idle IB status traffic must not break first-symbol qualification.
|
||||
|
||||
``open_client_proxy()`` previously broadcast a one-slot channel
|
||||
to an always-running status relay and an idle method receiver.
|
||||
Status events advanced only the relay, so a first order for an
|
||||
uncached symbol raised ``Lagged`` before receiving
|
||||
``get_sym_details()``. Feed several idle events through the
|
||||
sole-reader relay, then start two method calls and return their
|
||||
responses in reverse order. Both exact results prove status
|
||||
traffic cannot overrun a dormant caller and concurrent responses
|
||||
remain correlated by request ID.
|
||||
|
||||
'''
|
||||
async def main() -> None:
|
||||
chan = FakeChannel()
|
||||
proxy = MethodProxy(
|
||||
chan,
|
||||
event_table={},
|
||||
asyncio_ns=SimpleNamespace(),
|
||||
)
|
||||
results: dict[str, str] = {}
|
||||
|
||||
async def call(name: str) -> None:
|
||||
results[name] = await proxy._run_method(meth=name)
|
||||
|
||||
async with trio.open_nursery() as nursery:
|
||||
nursery.start_soon(
|
||||
relay_client_proxy_messages,
|
||||
chan,
|
||||
proxy,
|
||||
)
|
||||
for i in range(3):
|
||||
await chan.respond((
|
||||
'event',
|
||||
{'reason': f'farm-status-{i}'},
|
||||
))
|
||||
|
||||
nursery.start_soon(call, 'first')
|
||||
nursery.start_soon(call, 'second')
|
||||
with trio.fail_after(0.1):
|
||||
while len(chan.requests) < 2:
|
||||
await trio.lowlevel.checkpoint()
|
||||
|
||||
mids: dict[str, int] = {
|
||||
meth: mid
|
||||
for meth, _kwargs, mid in chan.requests
|
||||
}
|
||||
await chan.respond({
|
||||
'mid': mids['second'],
|
||||
'result': 'second-result',
|
||||
})
|
||||
await chan.respond({
|
||||
'mid': mids['first'],
|
||||
'result': 'first-result',
|
||||
})
|
||||
with trio.fail_after(0.1):
|
||||
while len(results) < 2:
|
||||
await trio.lowlevel.checkpoint()
|
||||
nursery.cancel_scope.cancel()
|
||||
|
||||
assert results == {
|
||||
'first': 'first-result',
|
||||
'second': 'second-result',
|
||||
}
|
||||
|
||||
trio.run(main)
|
||||
|
||||
|
||||
def test_method_proxy_relay_accepts_clean_eof() -> None:
|
||||
'''
|
||||
Normal asyncio relay completion must not fail proxy teardown.
|
||||
|
||||
The sole-reader relay blocks in ``LinkedTaskChannel.receive()``
|
||||
until the asyncio task exits and closes its Trio send channel.
|
||||
Close the fake response sender before entering the relay and
|
||||
prove the resulting ``EndOfChannel`` is graceful completion.
|
||||
|
||||
'''
|
||||
async def main() -> None:
|
||||
chan = FakeChannel()
|
||||
proxy = MethodProxy(
|
||||
chan,
|
||||
event_table={},
|
||||
asyncio_ns=SimpleNamespace(),
|
||||
)
|
||||
await chan.close_responses()
|
||||
with trio.fail_after(0.1):
|
||||
await relay_client_proxy_messages(chan, proxy)
|
||||
|
||||
trio.run(main)
|
||||
|
||||
|
||||
def test_method_proxy_drops_cancelled_call_response() -> None:
|
||||
'''
|
||||
A cancelled symbol search must not skew the next method response.
|
||||
|
||||
Search timeouts can cancel a Trio caller after its request
|
||||
reaches the asyncio client. Wait until one request is sent,
|
||||
cancel its caller, and then deliver the orphaned response. The
|
||||
sole-reader relay must drop that stale ``mid`` without failing; a
|
||||
subsequent call must still receive its own exact response.
|
||||
|
||||
'''
|
||||
async def main() -> None:
|
||||
chan = FakeChannel()
|
||||
proxy = MethodProxy(
|
||||
chan,
|
||||
event_table={},
|
||||
asyncio_ns=SimpleNamespace(),
|
||||
)
|
||||
call_scope = trio.CancelScope()
|
||||
cancelled = trio.Event()
|
||||
result: list[str] = []
|
||||
|
||||
async def cancelled_call() -> None:
|
||||
with call_scope:
|
||||
try:
|
||||
await proxy._run_method(meth='cancelled')
|
||||
finally:
|
||||
cancelled.set()
|
||||
|
||||
async def next_call() -> None:
|
||||
value = await proxy._run_method(meth='next')
|
||||
result.append(value)
|
||||
|
||||
async with trio.open_nursery() as nursery:
|
||||
nursery.start_soon(
|
||||
relay_client_proxy_messages,
|
||||
chan,
|
||||
proxy,
|
||||
)
|
||||
nursery.start_soon(cancelled_call)
|
||||
with trio.fail_after(0.1):
|
||||
while len(chan.requests) < 1:
|
||||
await trio.lowlevel.checkpoint()
|
||||
cancelled_mid: int = chan.requests[0][2]
|
||||
call_scope.cancel()
|
||||
with trio.fail_after(0.1):
|
||||
await cancelled.wait()
|
||||
assert proxy._pending == {}
|
||||
assert (
|
||||
proxy._request_methods[cancelled_mid]
|
||||
==
|
||||
'cancelled'
|
||||
)
|
||||
|
||||
await chan.respond({
|
||||
'mid': cancelled_mid,
|
||||
'result': 'stale',
|
||||
})
|
||||
with trio.fail_after(0.1):
|
||||
while cancelled_mid in proxy._request_methods:
|
||||
await trio.lowlevel.checkpoint()
|
||||
nursery.start_soon(next_call)
|
||||
with trio.fail_after(0.1):
|
||||
while len(chan.requests) < 2:
|
||||
await trio.lowlevel.checkpoint()
|
||||
next_mid: int = chan.requests[1][2]
|
||||
await chan.respond({
|
||||
'mid': next_mid,
|
||||
'result': 'fresh',
|
||||
})
|
||||
with trio.fail_after(0.1):
|
||||
while not result:
|
||||
await trio.lowlevel.checkpoint()
|
||||
nursery.cancel_scope.cancel()
|
||||
|
||||
assert result == ['fresh']
|
||||
|
||||
trio.run(main)
|
||||
|
||||
|
||||
def test_method_proxy_warns_when_metadata_is_evicted(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
'''
|
||||
Unresolved metadata eviction must remain operator-visible.
|
||||
|
||||
Cancelled calls whose asyncio responses never arrive retain a
|
||||
bounded method name solely for later stale-response diagnostics.
|
||||
Fill that table beyond its limit and prove the oldest entry is
|
||||
evicted with its request ID, method name, and configured bound in
|
||||
the warning.
|
||||
|
||||
'''
|
||||
proxy = MethodProxy(
|
||||
FakeChannel(),
|
||||
event_table={},
|
||||
asyncio_ns=SimpleNamespace(),
|
||||
)
|
||||
warnings: list[str] = []
|
||||
monkeypatch.setattr(ib_api.log, 'warning', warnings.append)
|
||||
|
||||
for mid in range(257):
|
||||
proxy._track_request_method(mid, f'method-{mid}')
|
||||
|
||||
assert len(proxy._request_methods) == 256
|
||||
assert 0 not in proxy._request_methods
|
||||
assert warnings == [
|
||||
'Evicting unresolved method-request metadata,\n'
|
||||
'mid: 0\n'
|
||||
"meth: 'method-0'\n"
|
||||
'pending metadata limit: 256\n'
|
||||
]
|
||||
|
|
@ -16,19 +16,21 @@ 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,
|
||||
ldshm,
|
||||
_summarize_shm_frame,
|
||||
shm as shm_cmd,
|
||||
)
|
||||
|
||||
|
||||
def test_ldshm_detects_nulls_without_skewing_period() -> None:
|
||||
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.
|
||||
``ldshm`` included them in period inference and then sent them to
|
||||
``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
|
||||
|
|
@ -52,7 +54,7 @@ def test_ldshm_detects_nulls_without_skewing_period() -> None:
|
|||
assert invalid_count == 2
|
||||
|
||||
|
||||
def test_ldshm_period_is_an_observed_step() -> None:
|
||||
def test_shm_period_is_an_observed_step() -> None:
|
||||
'''
|
||||
Sparse timestamps must not invent a storage timeframe.
|
||||
|
||||
|
|
@ -71,13 +73,13 @@ def test_ldshm_period_is_an_observed_step() -> None:
|
|||
assert invalid_count == 0
|
||||
|
||||
|
||||
def test_ldshm_counts_every_invalid_timestamp() -> None:
|
||||
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 ``ldshm``.
|
||||
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.
|
||||
|
||||
|
|
@ -98,7 +100,7 @@ def test_ldshm_counts_every_invalid_timestamp() -> None:
|
|||
assert invalid_count == 4
|
||||
|
||||
|
||||
def test_ldshm_invalid_snapshot_never_reaches_storage(
|
||||
def test_shm_invalid_snapshot_never_reaches_storage(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
'''
|
||||
|
|
@ -206,7 +208,156 @@ def test_ldshm_invalid_snapshot_never_reaches_storage(
|
|||
open_annotations,
|
||||
)
|
||||
|
||||
ldshm('qqq.nasdaq.ib')
|
||||
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_ctl',
|
||||
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(
|
||||
|
|
@ -217,7 +368,7 @@ def test_ldshm_invalid_snapshot_never_reaches_storage(
|
|||
np.array([60, 60]),
|
||||
],
|
||||
)
|
||||
def test_ldshm_skips_frame_without_positive_cadence(
|
||||
def test_shm_skips_frame_without_positive_cadence(
|
||||
times: np.ndarray,
|
||||
) -> None:
|
||||
'''
|
||||
|
|
@ -227,9 +378,182 @@ def test_ldshm_skips_frame_without_positive_cadence(
|
|||
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 ``ldshm`` to skip the buffer.
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -0,0 +1,326 @@
|
|||
'''
|
||||
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',
|
||||
'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
|
||||
required-input commands and prove help rendering exits before any
|
||||
callback can run.
|
||||
|
||||
'''
|
||||
runner = CliRunner()
|
||||
for command in (
|
||||
'anal',
|
||||
'audit',
|
||||
'delete',
|
||||
'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'
|
||||
three_hundred = nativedb / (
|
||||
'qqq.nasdaq.ib.ohlcv300s.parquet'
|
||||
)
|
||||
one.write_bytes(b'1')
|
||||
sixty.write_bytes(b'60')
|
||||
three_hundred.write_bytes(b'300')
|
||||
monkeypatch.setattr(config, 'get_conf_dir', lambda: tmp_path)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
store,
|
||||
['series', '--json'],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
payload = json.loads(result.output)
|
||||
assert [item['period_s'] for item in payload] == [1, 60, 300]
|
||||
assert [item['path'] for item in payload] == [
|
||||
str(one),
|
||||
str(sixty),
|
||||
str(three_hundred),
|
||||
]
|
||||
|
||||
compact = CliRunner().invoke(store, ['ls'])
|
||||
assert compact.exit_code == 0
|
||||
assert 'qqq.nasdaq.ib' in compact.output
|
||||
assert '1s, 60s, 300s' in compact.output
|
||||
|
||||
|
||||
def test_shm_reload_requires_explicit_persistence() -> None:
|
||||
'''
|
||||
SHM reload must not silently opt into durable persistence.
|
||||
|
||||
Reload reads back the Parquet produced by the repair path, so the
|
||||
reload flag alone has no valid input and must not open runtime or
|
||||
storage services. Invoke that invalid combination and prove CLI
|
||||
validation rejects it before inspecting an FQME.
|
||||
|
||||
'''
|
||||
result = CliRunner().invoke(
|
||||
store,
|
||||
['shm', 'qqq.nasdaq.ib', '--reload-parquet-to-shm'],
|
||||
)
|
||||
|
||||
assert result.exit_code == 2
|
||||
assert 'requires --write-parquet' in result.output
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'option',
|
||||
[
|
||||
['--json'],
|
||||
['--max-gaps', '10'],
|
||||
],
|
||||
)
|
||||
def test_shm_repair_rejects_read_only_formatting(
|
||||
option: list[str],
|
||||
) -> None:
|
||||
'''
|
||||
Repair mode must not silently ignore read-only output options.
|
||||
|
||||
Combining the old repair workflow with the new inspector made
|
||||
``--json`` and an explicitly default-valued ``--max-gaps``
|
||||
appeared accepted even though repair produced logs and entered an
|
||||
interactive pause. Invoke each conflicting option and prove
|
||||
validation stops before runtime or storage can be opened.
|
||||
|
||||
'''
|
||||
result = CliRunner().invoke(
|
||||
store,
|
||||
[
|
||||
'shm',
|
||||
'qqq.nasdaq.ib',
|
||||
'--write-parquet',
|
||||
*option,
|
||||
],
|
||||
)
|
||||
|
||||
assert result.exit_code == 2
|
||||
assert 'apply only to read-only output' in result.output
|
||||
|
||||
|
||||
def test_ls_has_description_in_store_help() -> None:
|
||||
'''
|
||||
Compact NativeDB discovery must explain itself in group help.
|
||||
|
||||
The backend-oriented ``ls`` callback had no docstring, leaving
|
||||
its command-map description blank. Render the parent help and
|
||||
prove the compact series purpose is visible without opening the
|
||||
command.
|
||||
|
||||
'''
|
||||
result = CliRunner().invoke(store, [])
|
||||
|
||||
assert result.exit_code == 2
|
||||
assert 'List NativeDB FQMEs' in result.output
|
||||
|
||||
|
||||
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_shm_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,
|
||||
[
|
||||
'shm',
|
||||
'qqq.nasdaq.ib',
|
||||
'--shm-name',
|
||||
'missing',
|
||||
],
|
||||
)
|
||||
|
||||
assert result.exit_code == 2
|
||||
assert 'No exact OHLCV SHM buffer' in result.output
|
||||
Loading…
Reference in New Issue