Compare commits

..

No commits in common. "main" and "to_asyncio_api_update" have entirely different histories.

40 changed files with 797 additions and 925 deletions

View File

@ -1,291 +0,0 @@
# Test Harness Reference: piker
This repository-local file supplements the canonical `/run-tests` skill.
Keep shared execution, worktree, failure-inspection, and cleanup policy in the
canonical `SKILL.md`; keep Piker commands, paths, fixtures, and known outcomes
here.
## Project And Environment
- Project/import: `piker`
- Test root: `tests/`
- Supported Python: `>=3.12,<3.14`
- Preferred complete environment: worktree-local `py313` inside the current
`nix develop` shell
- The flake shell pins CPython 3.13 and sets
`UV_PROJECT_ENVIRONMENT=py313`.
- Verify the interpreter, package resolution, and dependency import before
running tests. A bare `py313` may lack the Qt binding supplied by Nix.
Use an already-provisioned `py313` only when `import piker` succeeds:
```text
py313/bin/python -m pytest -p no:xonsh
```
If direct environment paths are unavailable, an existing uv environment can
be used without changing it, subject to the same import check:
```text
UV_PROJECT_ENVIRONMENT=py313 uv run --frozen --no-sync python -m pytest -p no:xonsh
```
Ask before running provisioning commands such as:
```text
UV_PROJECT_ENVIRONMENT=py313 uv sync --dev --all-extras --no-group lint
nix develop
nix-shell default.nix
```
`nix develop` is the current Wayland/Qt 6 shell. `default.nix` is the current
X11 shell. Do not use `develop.nix` for current testing; it retains the old
Python 3.11, Poetry, and Qt 5 stack.
The current root-checkout `py313` resolves `piker` locally but fails
`import piker` outside Nix because PyQtGraph cannot import PyQt or PySide. Do
not treat that environment as test-ready and do not enter `nix develop`
without approval: its shell hook may recreate and sync `py313`.
Plain `uv sync` does not include the testing group. The `dbs` dependency group
is also absent from normal dev-shell provisioning.
## Commands
Base command in the preferred environment:
```text
py313/bin/python -m pytest -p no:xonsh
```
The explicit `-p no:xonsh` is required. The tracked comments-only
`pytest.ini` takes precedence over `pyproject.toml`, so the intended
`addopts = "-p no:xonsh"` and `testpaths = ["tests"]` are inactive.
Always pass a test path or node ID explicitly.
Package-resolution check that does not import Piker's dependencies:
```text
py313/bin/python -c 'import importlib.util, pathlib, sys; root = pathlib.Path.cwd().resolve(); spec = importlib.util.find_spec("piker"); mod = pathlib.Path(spec.origin).resolve(); print(sys.executable); print(mod); assert mod.is_relative_to(root)'
```
Dependency import check, required before collection or execution:
```text
py313/bin/python -c 'import pathlib, piker, sys; root = pathlib.Path.cwd().resolve(); mod = pathlib.Path(piker.__file__).resolve(); print(sys.executable); print(mod); assert mod.is_relative_to(root)'
```
Safe core collection check:
```text
py313/bin/python -m pytest -p no:xonsh -q --collect-only tests/test_watchlists.py tests/test_accounting.py tests/test_services.py tests/test_ems.py tests/test_feeds.py tests/test_cli.py
```
Default first-pass flags are `-q -x --tb=short --no-header` unless the user
requests otherwise. For actor-heavy tests, use one file or node per process
with an outer timeout:
```text
timeout -k 5 300 py313/bin/python -m pytest -p no:xonsh -q <one-file-or-node>
```
If an actor-heavy command exits `124` or `143`, retry that exact command once
and report both attempts. Never convert a retry pass into an unconditional
pass, and do not retry assertion, import, collection, or configuration
failures.
## Scope And Path Resolution
Resolve bare test filenames beneath `tests/`. Preserve complete node IDs and
apply `-k` only within explicitly selected paths. There is no marker-based
offline/full-suite split, so never use `pytest tests` as a deterministic
default.
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`
- `tests/test_services.py::test_runtime_boot`
- `tests/test_services.py::test_datad_spawn`
- `tests/test_ems.py::test_ems_err_on_bad_broker`
Require explicit authorization before running:
- `tests/test_feeds.py` - live Binance/Kraken feeds;
- `tests/test_services.py::test_ensure_datafeed_actors` - live Kraken feed;
- `tests/test_services.py::test_ensure_ems_in_paper_actors` - paper EMS with
live Kraken symbology/feed access;
- `tests/test_ems.py::test_multi_fill_positions` - live backend setup and
persisted state;
- `tests/test_accounting.py::test_paper_ledger_position_calcs` - tracked
fixtures plus possible live symcache generation;
- `tests/test_accounting.py::test_ib_account_with_duplicated_mktids` - active
broker/account configuration and state writes;
- `tests/test_dpi_font.py` - Qt/UI import and user-config side effects;
- `tests/test_docker_services.py` - optional dependencies and containers;
- `tests/test_questrade.py` - obsolete credentialed imports.
`tests/test_cli.py` is currently hard-skipped. It is not an active CLI
regression gate.
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
| Flag | Purpose |
|---|---|
| `--ll LEVEL` | Piker log level |
| `--confdir PATH` | Override `piker.config._config_dir` |
| `--spawn-backend trio|mp_spawn|mp_forkserver` | Tractor process backend |
| `--tpt-proto PROTO` | Tractor transport; one protocol per session |
| `--tpdb` / `--debug-mode` | Tractor plugin debug mode |
| `--pdb` | Standard pytest debugger |
| `-s` | No capture; required with `--pdb` and `open_test_pikerd` |
Do not invent `network`, `offline`, `docker`, `gui`, or `broker` markers; none
currently exists. `CI=1` is not an offline selector: feed tests still leave a
live Kraken case enabled. Current Piker tests exercise TCP; do not assume the
whole suite supports UDS merely because the Tractor plugin exposes it.
## Fixture Invariants
- The session `confdir` fixture does nothing unless `--confdir` is passed.
Its claimed `tests/data` fallback is not implemented and that directory is
absent.
- Function-scoped `tmpconfdir` changes process-global config state and does
not restore the previous path. Use separate pytest processes when
diagnosing state leakage.
- `open_test_pikerd` passes the temporary config path to child actors through
`tractor_runtime_overrides`.
- `tests/_inputs/trades_binance_paper.toml` and
`tests/_inputs/account.binance.paper.toml` are used in place. Accounting
contexts can write them on exit. Inspect `git diff -- tests/_inputs` after
any selected accounting case.
- Importing `tests/test_dpi_font.py` constructs module-global font objects
before fixtures can isolate config. If explicitly requested, isolate
`XDG_CONFIG_HOME` before Python starts and use the proper Qt/Nix shell.
- Piker and the installed Tractor pytest plugin do not provide a
repository-local process or socket reaper. Never apply historical broad
`pkill -f tractor._child` guidance automatically.
- The function-scoped autouse `shm_leak_tracker` fixture wraps Tractor's
current-process `SharedMemory` factory. It tracks only successful
`create=True` calls, restores the pre-test `_known_tokens` cache, and
unlinks exact surviving names before failing the leaking test. It never
scans `/dev/shm` or unlinks attachments created by another process.
## Test Layout
```text
tests/
conftest.py Piker options, config fixtures, Tractor plugin
_inputs/ tracked ledger/account fixtures
test_accounting.py config, ledgers, accounts, and position math
test_cli.py legacy CLI suite; hard-skipped
test_docker_services.py container integrations; optional deps, skipped
test_dpi_font.py Qt DPI/font behavior
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 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
test_watchlists.py deterministic watchlist JSON operations
```
## Change-To-Test Mapping
| Changed area | Run first | Caveat |
|---|---|---|
| `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` 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 |
| `piker/service/_actor_runtime.py`, `_registry.py`, `_mngr.py` | `test_runtime_boot` | then `test_datad_spawn` |
| `piker/service/`, `piker/data/_daemon.py` | `test_datad_spawn` | feed lifecycle cases are live |
| `piker/data/feed.py`, `flows.py`, `_sharedmem.py`, `_sampling.py` | collect first | feed execution needs live permission |
| `piker/clearing/` | `test_ems_err_on_bad_broker` | multi-fill case is live/persisted |
| `piker/brokers/binance/`, `kraken/` | selected feed/accounting node | live network |
| `piker/brokers/ib/` | duplicated-market-ID node | controlled account config required |
| Docker/service adapters | `tests/test_docker_services.py` | optional deps and containers |
| project, lock, or Nix files | import check and safe collection | full collection is not safe by default |
Prefer deterministic filesystem/config tests, then local actor-runtime nodes,
then explicitly approved live broker, GUI, or container coverage.
## Quick Checks
```text
py313/bin/python -c 'import importlib.util, pathlib, sys; root = pathlib.Path.cwd().resolve(); spec = importlib.util.find_spec("piker"); mod = pathlib.Path(spec.origin).resolve(); print(sys.executable); print(mod); assert mod.is_relative_to(root)'
py313/bin/python -c 'import pathlib, piker, sys; root = pathlib.Path.cwd().resolve(); mod = pathlib.Path(piker.__file__).resolve(); print(sys.executable); print(mod); assert mod.is_relative_to(root)'
py313/bin/python -m pytest -p no:xonsh -q tests/test_watchlists.py
py313/bin/python -m pytest -p no:xonsh -q tests/test_accounting.py::test_account_file_default_empty
timeout -k 5 300 py313/bin/python -m pytest -p no:xonsh -q tests/test_services.py::test_runtime_boot
timeout -k 5 300 py313/bin/python -m pytest -p no:xonsh -q tests/test_services.py::test_datad_spawn
timeout -k 5 300 py313/bin/python -m pytest -p no:xonsh -q tests/test_ems.py::test_ems_err_on_bad_broker
```
## Known Outcomes
- The current root-checkout `py313` fails `import piker` outside the Nix shell
with `ImportError: PyQtGraph requires one of PyQt5, PyQt6, PySide2 or
PySide6`. This is an incomplete environment, not an application regression.
- `tests/test_accounting.py::test_root_conf_networking_section` currently
expects `network.tsdb`, which is absent from the tracked config template.
Match the current `KeyError: 'tsdb'` before classifying it as the known
repository mismatch.
- `tests/test_docker_services.py` is marked skipped but imports
`elasticsearch` first. Without the `dbs` group it fails collection rather
than skipping.
- `tests/test_questrade.py` is marked skipped but imports undeclared `asks`
through the legacy broker module before marks apply.
- `test_open_orders_reloaded` and `test_dark_order_clearing` in
`tests/test_ems.py` contain only ellipsis bodies. A pass does not verify the
named behavior.
- Treat `.pytest_cache` feed IDs using old FQME forms as stale historical
cache, not current known failures.
Do not classify every `TooSlowError`, timeout, or child survivor as the known
Tractor teardown wedge. Match the selected node and the documented
second-runtime/lingering-child signature.
## Tractor Runtime Notes
The suite loads `tractor._testing.pytest`. Defaults are the `trio` spawn
backend and TCP transport. Registry addresses are normally session-unique,
except `tests/test_services.py::test_runtime_boot`, which binds
`127.0.0.1:6666`. Check that fixed port only for that node; ordinary tests do
not require Piker's production `127.0.0.1:6116` registry address.
Current repository concurrency notes document an intermittent second
in-process `pikerd` boot wedge with a lingering broker child and unread parent
IPC bytes. Use the outer timeout for actor-heavy nodes. Retry an exact command
once only after status `124` or `143`, and report both attempts.
Ordinary actor tests support normal capture. Diagnose a capture-dependent hang
by retrying only the exact node with `-s`. Standard `--pdb` with
`open_test_pikerd` requires `-s`; Tractor's `--tpdb` is a separate option.
After abnormal exit, inspect only descendants, sockets, and shared-memory
objects attributable to the exact pytest session. Ask before signaling or
unlinking anything.

View File

@ -1 +0,0 @@
{"version": 1, "backend": "neutral"}

View File

@ -0,0 +1,11 @@
{
"permissions": {
"allow": [
"Bash(chmod:*)",
"Bash(/tmp/piker_commits.txt)",
"Bash(python:*)"
],
"deny": [],
"ask": []
}
}

View File

@ -0,0 +1,84 @@
---
name: commit-msg
description: >
Generate piker-style git commit messages from
staged changes or prompt input, following the
style guide learned from 500 repo commits.
argument-hint: "[optional-scope-or-description]"
disable-model-invocation: true
allowed-tools: Bash(git *), Read, Grep, Glob, Write
---
## Current staged changes
!`git diff --staged --stat`
## Recent commit style reference
!`git log --oneline -10`
# Piker Git Commit Message Generator
Generate a commit message from the staged diff above
following the piker project's conventions (learned from
analyzing 500 repo commits).
If `$ARGUMENTS` is provided, use it as scope or
description context for the commit message.
For the full style guide with verb frequencies,
section markers, abbreviations, piker-specific terms,
and examples, see
[style-guide-reference.md](./style-guide-reference.md).
## Quick Reference
- **Subject**: ~50 chars, present tense verb, use
backticks for code refs
- **Body**: only for complex/multi-file changes,
67 char line max
- **Section markers**: Also, / Deats, / Other,
- **Bullets**: use `-` style
- **Tone**: technical but casual (piker style)
## Claude-code Footer
When the written **patch** was assisted by
claude-code, include:
```
(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
```
When only the **commit msg** was written by
claude-code (human wrote the patch), use:
```
(this commit msg was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
```
## Output Instructions
When generating a commit message:
1. Analyze the staged diff (injected above via
dynamic context) to understand all changes.
2. If `$ARGUMENTS` provides a scope (e.g.,
`.ib.feed`) or description, incorporate it into
the subject line.
3. Write the subject line following verb + backtick
conventions from the
[style guide](./style-guide-reference.md).
4. Add body only for multi-file or complex changes.
5. Write the message to a file in the repo's
`.claude/` subdir with filename format:
`<timestamp>_<first-7-chars-of-last-commit-hash>_commit_msg.md`
where `<timestamp>` is from `date --iso-8601=seconds`.
Also write a copy to
`.claude/git_commit_msg_LATEST.md`
(overwrite if exists).
---
**Analysis date:** 2026-01-27
**Commits analyzed:** 500 from piker repository
**Maintained by:** Tyler Goodlet

View File

@ -150,18 +150,21 @@ Common in piker commits (33.0% use colons):
- File:line references not used (0 occurrences)
- No WIP commits in analyzed set
### Coding-harness Footer
When a coding harness assisted with the written patch,
identify the active harness, model, and provider:
### Claude-code Footer
When the written **patch** was assisted by claude-code,
include:
```
(this patch was generated in some part by `<harness>` using `<model>` (`<provider>`))
(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
```
When it generated only the commit message, use:
When only the **commit msg** was written by claude-code
(human wrote the patch), use:
```
(this commit msg was generated in some part by `<harness>` using `<model>` (`<provider>`))
(this commit msg was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
```
## Piker-Specific Terms

View File

@ -1 +0,0 @@
../../.agents/skills/piker-profiling

View File

@ -1 +0,0 @@
../../.agents/skills/piker-slang

View File

@ -1 +0,0 @@
../../.agents/skills/pyqtgraph-optimization

View File

@ -1 +0,0 @@
../../.agents/skills/timeseries-optimization

362
.gitignore vendored
View File

@ -107,9 +107,6 @@ ENV/
.git/
# any commit-msg gen tmp files
.claude/skills/commit-msg/msgs/
.claude/skills/commit-msg/conf.toml
.claude/git_commit_msg_LATEST.md
.claude/*_commit_*.md
.claude/*_commit*.toml
@ -133,362 +130,3 @@ gitea/
# LLM conversations that should remain private
docs/conversations/
# BEGIN ai.skillz: direct:symlink:claude:gish
/.claude/skills/gish
# END ai.skillz: direct:symlink:claude:gish
# BEGIN ai.skillz: direct:symlink:opencode:gish
/.opencode/skills/gish
# END ai.skillz: direct:symlink:opencode:gish
# BEGIN ai.skillz: runtime:open-wkt
/wkts/
# END ai.skillz: runtime:open-wkt
# BEGIN ai.skillz: direct:symlink:claude:open-wkt
/.claude/skills/open-wkt
# END ai.skillz: direct:symlink:claude:open-wkt
# BEGIN ai.skillz: direct:symlink:opencode:open-wkt
/.opencode/skills/open-wkt
# END ai.skillz: direct:symlink:opencode:open-wkt
# BEGIN ai.skillz: direct:symlink:claude:run-tests
/.claude/skills/run-tests/SKILL.md
# END ai.skillz: direct:symlink:claude:run-tests
# BEGIN ai.skillz: direct:symlink:opencode:run-tests
/.opencode/skills/run-tests/SKILL.md
# END ai.skillz: direct:symlink:opencode:run-tests
# BEGIN ai.skillz: direct:symlink:claude:close-wkt
/.claude/skills/close-wkt
# END ai.skillz: direct:symlink:claude:close-wkt
# BEGIN ai.skillz: direct:symlink:opencode:close-wkt
/.opencode/skills/close-wkt
# END ai.skillz: direct:symlink:opencode:close-wkt
# BEGIN ai.skillz: runtime:code-review
.ai/code-review/reports/
# END ai.skillz: runtime:code-review
# BEGIN ai.skillz: direct:symlink:claude:code-review
/.claude/skills/code-review
# END ai.skillz: direct:symlink:claude:code-review
# BEGIN ai.skillz: direct:symlink:opencode:code-review
/.opencode/skills/code-review
# END ai.skillz: direct:symlink:opencode:code-review
# BEGIN ai.skillz: runtime:code-review-changes
.ai/state/review/
.claude/review_context.md
.claude/review_regression.md
.claude/review_replies/
# END ai.skillz: runtime:code-review-changes
# BEGIN ai.skillz: direct:symlink:claude:code-review-changes
/.claude/skills/code-review-changes
# END ai.skillz: direct:symlink:claude:code-review-changes
# BEGIN ai.skillz: direct:symlink:opencode:code-review-changes
/.opencode/skills/code-review-changes
# END ai.skillz: direct:symlink:opencode:code-review-changes
# BEGIN ai.skillz: runtime:commit-msg
.ai/state/commit-msg/
.ai/state/migrations/
.ai/commit-msg/conf.toml
.claude/skills/commit-msg/msgs/
.claude/git_commit_msg_LATEST.md
# END ai.skillz: runtime:commit-msg
# BEGIN ai.skillz: direct:symlink:claude:commit-msg
/.claude/skills/commit-msg/SKILL.md
# END ai.skillz: direct:symlink:claude:commit-msg
# BEGIN ai.skillz: direct:symlink:opencode:commit-msg
/.opencode/skills/commit-msg/SKILL.md
# END ai.skillz: direct:symlink:opencode:commit-msg
# BEGIN ai.skillz: direct:symlink:claude:commit-plan
/.claude/skills/commit-plan
# END ai.skillz: direct:symlink:claude:commit-plan
# BEGIN ai.skillz: direct:symlink:opencode:commit-plan
/.opencode/skills/commit-plan
# END ai.skillz: direct:symlink:opencode:commit-plan
# BEGIN ai.skillz: direct:symlink:claude:dep-supersede-scan
/.claude/skills/dep-supersede-scan
# END ai.skillz: direct:symlink:claude:dep-supersede-scan
# BEGIN ai.skillz: direct:symlink:opencode:dep-supersede-scan
/.opencode/skills/dep-supersede-scan
# END ai.skillz: direct:symlink:opencode:dep-supersede-scan
# BEGIN ai.skillz: direct:symlink:claude:resolve-conflicts
/.claude/skills/resolve-conflicts
# END ai.skillz: direct:symlink:claude:resolve-conflicts
# BEGIN ai.skillz: direct:symlink:opencode:resolve-conflicts
/.opencode/skills/resolve-conflicts
# END ai.skillz: direct:symlink:opencode:resolve-conflicts
# BEGIN ai.skillz: direct:symlink:claude:git-mgmt
/.claude/skills/git-mgmt
# END ai.skillz: direct:symlink:claude:git-mgmt
# BEGIN ai.skillz: direct:symlink:opencode:git-mgmt
/.opencode/skills/git-mgmt
# END ai.skillz: direct:symlink:opencode:git-mgmt
# BEGIN ai.skillz: direct:symlink:claude:harness-perf
/.claude/skills/harness-perf
# END ai.skillz: direct:symlink:claude:harness-perf
# BEGIN ai.skillz: direct:symlink:opencode:harness-perf
/.opencode/skills/harness-perf
# END ai.skillz: direct:symlink:opencode:harness-perf
# BEGIN ai.skillz: direct:symlink:claude:inter-skill-review
/.claude/skills/inter-skill-review
# END ai.skillz: direct:symlink:claude:inter-skill-review
# BEGIN ai.skillz: direct:symlink:opencode:inter-skill-review
/.opencode/skills/inter-skill-review
# END ai.skillz: direct:symlink:opencode:inter-skill-review
# BEGIN ai.skillz: direct:symlink:claude:opencode-cleaning
/.claude/skills/opencode-cleaning
# END ai.skillz: direct:symlink:claude:opencode-cleaning
# BEGIN ai.skillz: direct:symlink:opencode:opencode-cleaning
/.opencode/skills/opencode-cleaning
# END ai.skillz: direct:symlink:opencode:opencode-cleaning
# BEGIN ai.skillz: direct:symlink:claude:plan-io
/.claude/skills/plan-io
# END ai.skillz: direct:symlink:claude:plan-io
# BEGIN ai.skillz: direct:symlink:opencode:plan-io
/.opencode/skills/plan-io
# END ai.skillz: direct:symlink:opencode:plan-io
# BEGIN ai.skillz: runtime:pr-msg
.ai/state/pr-msg/
.ai/pr-msg/conf.toml
.claude/skills/pr-msg/msgs/
.claude/skills/pr-msg/pr_msg_LATEST.md
# END ai.skillz: runtime:pr-msg
# BEGIN ai.skillz: direct:symlink:claude:pr-msg
/.claude/skills/pr-msg/SKILL.md
/.claude/skills/pr-msg/references
/.claude/skills/pr-msg/scripts
# END ai.skillz: direct:symlink:claude:pr-msg
# BEGIN ai.skillz: direct:symlink:opencode:pr-msg
/.opencode/skills/pr-msg/SKILL.md
/.opencode/skills/pr-msg/references
/.opencode/skills/pr-msg/scripts
# END ai.skillz: direct:symlink:opencode:pr-msg
# BEGIN ai.skillz: direct:symlink:claude:prompt-io
/.claude/skills/prompt-io
# END ai.skillz: direct:symlink:claude:prompt-io
# BEGIN ai.skillz: direct:symlink:opencode:prompt-io
/.opencode/skills/prompt-io
# END ai.skillz: direct:symlink:opencode:prompt-io
# BEGIN ai.skillz: direct:symlink:claude:py-codestyle
/.claude/skills/py-codestyle
# END ai.skillz: direct:symlink:claude:py-codestyle
# BEGIN ai.skillz: direct:symlink:opencode:py-codestyle
/.opencode/skills/py-codestyle
# END ai.skillz: direct:symlink:opencode:py-codestyle
# BEGIN ai.skillz: runtime:taken-export
.ai/taken/exports/
# END ai.skillz: runtime:taken-export
# BEGIN ai.skillz: direct:symlink:claude:taken-export
/.claude/skills/taken-export
# END ai.skillz: direct:symlink:claude:taken-export
# BEGIN ai.skillz: direct:symlink:opencode:taken-export
/.opencode/skills/taken-export
# END ai.skillz: direct:symlink:opencode:taken-export
# BEGIN ai.skillz: direct:symlink:claude:yt-url-lookup
/.claude/skills/yt-url-lookup
# END ai.skillz: direct:symlink:claude:yt-url-lookup
# BEGIN ai.skillz: direct:symlink:opencode:yt-url-lookup
/.opencode/skills/yt-url-lookup
# END ai.skillz: direct:symlink:opencode:yt-url-lookup
# BEGIN ai.skillz: direct:symlink:opencode:command:run-tests
/.opencode/commands/run-tests.md
# END ai.skillz: direct:symlink:opencode:command:run-tests
# BEGIN ai.skillz: direct:symlink:opencode:command:code-review
/.opencode/commands/code-review.md
# END ai.skillz: direct:symlink:opencode:command:code-review
# BEGIN ai.skillz: direct:symlink:opencode:command:code-review-changes
/.opencode/commands/code-review-changes.md
# END ai.skillz: direct:symlink:opencode:command:code-review-changes
# BEGIN ai.skillz: direct:symlink:opencode:command:commit-msg
/.opencode/commands/commit-msg.md
# END ai.skillz: direct:symlink:opencode:command:commit-msg
# BEGIN ai.skillz: direct:symlink:opencode:command:commit-plan
/.opencode/commands/commit-plan.md
# END ai.skillz: direct:symlink:opencode:command:commit-plan
# BEGIN ai.skillz: direct:symlink:opencode:command:opencode-cleaning
/.opencode/commands/opencode-cleaning.md
# END ai.skillz: direct:symlink:opencode:command:opencode-cleaning
# BEGIN ai.skillz: direct:symlink:opencode:command:pr-msg
/.opencode/commands/pr-msg.md
# END ai.skillz: direct:symlink:opencode:command:pr-msg
# BEGIN ai.skillz: direct:symlink:opencode:command:taken-export
/.opencode/commands/taken-export.md
# END ai.skillz: direct:symlink:opencode:command:taken-export
# BEGIN ai.skillz: direct:symlink:claude:code-nav-refs
/.claude/skills/code-nav-refs
# END ai.skillz: direct:symlink:claude:code-nav-refs
# BEGIN ai.skillz: direct:symlink:opencode:code-nav-refs
/.opencode/skills/code-nav-refs
# END ai.skillz: direct:symlink:opencode:code-nav-refs
# BEGIN ai.skillz: direct:symlink:opencode:command:gish
/.opencode/commands/gish.md
# END ai.skillz: direct:symlink:opencode:command:gish
# BEGIN ai.skillz: direct:symlink:opencode:command:resolve-conflicts
/.opencode/commands/resolve-conflicts.md
# END ai.skillz: direct:symlink:opencode:command:resolve-conflicts
# BEGIN ai.skillz: direct:symlink:opencode:command:git-mgmt
/.opencode/commands/git-mgmt.md
# END ai.skillz: direct:symlink:opencode:command:git-mgmt
# BEGIN ai.skillz: direct:symlink:opencode:command:open-wkt
/.opencode/commands/open-wkt.md
# END ai.skillz: direct:symlink:opencode:command:open-wkt
# BEGIN ai.skillz: direct:symlink:opencode:command:close-wkt
/.opencode/commands/close-wkt.md
# END ai.skillz: direct:symlink:opencode:command:close-wkt
# BEGIN ai.skillz: direct:symlink:opencode:command:dep-supersede-scan
/.opencode/commands/dep-supersede-scan.md
# END ai.skillz: direct:symlink:opencode:command:dep-supersede-scan
# BEGIN ai.skillz: direct:symlink:opencode:command:harness-perf
/.opencode/commands/harness-perf.md
# END ai.skillz: direct:symlink:opencode:command:harness-perf
# BEGIN ai.skillz: direct:symlink:opencode:command:yt-url-lookup
/.opencode/commands/yt-url-lookup.md
# END ai.skillz: direct:symlink:opencode:command:yt-url-lookup
# BEGIN ai.skillz: direct:symlink:agents:gish
/.agents/skills/gish
# END ai.skillz: direct:symlink:agents:gish
# BEGIN ai.skillz: direct:symlink:agents:resolve-conflicts
/.agents/skills/resolve-conflicts
# END ai.skillz: direct:symlink:agents:resolve-conflicts
# BEGIN ai.skillz: direct:symlink:agents:git-mgmt
/.agents/skills/git-mgmt
# END ai.skillz: direct:symlink:agents:git-mgmt
# BEGIN ai.skillz: direct:symlink:agents:open-wkt
/.agents/skills/open-wkt
# END ai.skillz: direct:symlink:agents:open-wkt
# BEGIN ai.skillz: direct:symlink:agents:run-tests
/.agents/skills/run-tests
# END ai.skillz: direct:symlink:agents:run-tests
# BEGIN ai.skillz: direct:symlink:agents:close-wkt
/.agents/skills/close-wkt
# END ai.skillz: direct:symlink:agents:close-wkt
# BEGIN ai.skillz: direct:symlink:agents:code-review
/.agents/skills/code-review
# END ai.skillz: direct:symlink:agents:code-review
# BEGIN ai.skillz: direct:symlink:agents:code-nav-refs
/.agents/skills/code-nav-refs
# END ai.skillz: direct:symlink:agents:code-nav-refs
# BEGIN ai.skillz: direct:symlink:agents:code-review-changes
/.agents/skills/code-review-changes
# END ai.skillz: direct:symlink:agents:code-review-changes
# BEGIN ai.skillz: direct:symlink:agents:commit-msg
/.agents/skills/commit-msg
# END ai.skillz: direct:symlink:agents:commit-msg
# BEGIN ai.skillz: direct:symlink:agents:commit-plan
/.agents/skills/commit-plan
# END ai.skillz: direct:symlink:agents:commit-plan
# BEGIN ai.skillz: direct:symlink:agents:dep-supersede-scan
/.agents/skills/dep-supersede-scan
# END ai.skillz: direct:symlink:agents:dep-supersede-scan
# BEGIN ai.skillz: direct:symlink:agents:harness-perf
/.agents/skills/harness-perf
# END ai.skillz: direct:symlink:agents:harness-perf
# BEGIN ai.skillz: direct:symlink:agents:inter-skill-review
/.agents/skills/inter-skill-review
# END ai.skillz: direct:symlink:agents:inter-skill-review
# BEGIN ai.skillz: direct:symlink:agents:opencode-cleaning
/.agents/skills/opencode-cleaning
# END ai.skillz: direct:symlink:agents:opencode-cleaning
# BEGIN ai.skillz: direct:symlink:agents:plan-io
/.agents/skills/plan-io
# END ai.skillz: direct:symlink:agents:plan-io
# BEGIN ai.skillz: direct:symlink:agents:pr-msg
/.agents/skills/pr-msg
# END ai.skillz: direct:symlink:agents:pr-msg
# BEGIN ai.skillz: direct:symlink:agents:prompt-io
/.agents/skills/prompt-io
# END ai.skillz: direct:symlink:agents:prompt-io
# BEGIN ai.skillz: direct:symlink:agents:py-codestyle
/.agents/skills/py-codestyle
# END ai.skillz: direct:symlink:agents:py-codestyle
# BEGIN ai.skillz: direct:symlink:agents:taken-export
/.agents/skills/taken-export
# END ai.skillz: direct:symlink:agents:taken-export
# BEGIN ai.skillz: direct:symlink:agents:yt-url-lookup
/.agents/skills/yt-url-lookup
# END ai.skillz: direct:symlink:agents:yt-url-lookup
# BEGIN ai.skillz: workflow-state
/.ai/state/
/.ai/commit-msg/conf.toml
/.ai/pr-msg/conf.toml
# END ai.skillz: workflow-state

View File

@ -1,34 +0,0 @@
---
model: openai/gpt-5.6-sol
service: opencode
session: unavailable
timestamp: 2026-09-07T21:40:00Z
git_ref: 9115d24
scope: config
substantive: true
raw_file: 20260907T214000Z_9115d24_prompt_io.raw.md
---
## Prompt
Deploy the provider-neutral `.agents` skill tree for Codex first and
then migrate Piker's workflow state without losing archived helpers.
## Response summary
Deployed 21 ignored `.agents/skills` links, copied 132 legacy runtime
files into `.ai/state`, selected the neutral backend, and preserved the
original archive and staged `ruff.toml` change.
## Files changed
- `.gitignore` - ignore shared links and generated neutral state.
- `.ai/workflow-state.json` - select the neutral backend.
- `.ai/commit-msg/style-guide-reference.md` - port commit guidance.
- `.ai/run-tests/test-harness-reference.md` - port test guidance.
## Human edits
The human prioritized immediate Codex deployment and directed helper
archives to migrate as opaque runtime data instead of blocking the
operation. No direct source-line edits were made by the human.

View File

@ -1,15 +0,0 @@
---
model: openai/gpt-5.6-sol
service: opencode
timestamp: 2026-09-07T21:40:00Z
git_ref: 9115d24
diff_cmd: git diff HEAD~1..HEAD
---
Deploy all shared `ai.skillz` skills through `.agents/skills` for
Codex, then migrate Piker's generated workflow state from the legacy
Claude layout to provider-neutral `.ai/state` storage.
Preserve every legacy source and the existing staged `ruff.toml`
change. Commit only the neutral backend selection, repository guidance,
ignore rules, and this provenance record.

View File

@ -1,37 +0,0 @@
---
model: openai/gpt-5.6-sol
service: opencode
session: unavailable
timestamp: 2026-09-08T04:31:34Z
git_ref: f148fbcc
scope: config
substantive: true
raw_file: 20260908T043134Z_f148fbcc_prompt_io.raw.md
---
## Prompt
Audit tracked legacy skill files under `.claude/`, then migrate Piker's
repository-specific skills into the provider-neutral discovery tree.
Keep Claude compatibility and remove stale Claude-local configuration.
## Response summary
Moves four Piker-specific skill packages to `.agents/skills/`, retains
Claude discovery through relative adapter links, and drops two stale
tracked Claude-local files.
## Files changed
- `.agents/skills/piker-*/` - provider-neutral Piker skill sources.
- `.claude/skills/piker-*` - relative Claude compatibility links.
- `.claude/settings.local.json` - remove tracked local permissions.
- `.claude/skills/commit-msg/style-guide-reference.md` - remove the
superseded legacy copy now represented under `.ai/commit-msg/`.
## Human edits
The human identified that repository-specific skills were not migrated
by the shared `ai.skillz` deployment, selected `.agents/skills/` as the
neutral source, and requested that the migration land in PR 93 before
following up on the generic migration tooling.

View File

@ -1,34 +0,0 @@
---
model: openai/gpt-5.6-sol
service: opencode
timestamp: 2026-09-08T04:31:34Z
git_ref: f148fbcc
diff_cmd: git diff --cached
---
## Prompt
Audit tracked legacy skill files under `.claude/`, then migrate Piker's
repository-specific skills into the provider-neutral discovery tree.
Keep Claude compatibility and remove stale Claude-local configuration.
## Generated changes
> `git diff --cached -- .agents/skills/`
Moves the four Piker-owned skill packages into the neutral
`.agents/skills/` source tree without changing their contents.
> `git diff --cached -- .claude/skills/ .claude/settings.local.json`
Adds relative Claude adapter links to the neutral skill sources and
removes the tracked local settings file and superseded commit-message
style guide.
## Validation
- All four Claude adapter links resolve to a `SKILL.md`.
- `git diff --cached --check` passes.
- No tracked files match the repository's ignore rules.
- Shared deployment validation retains one unrelated stale local
`commit-plan` hybrid-link error for an `ai.skillz` follow-up.

View File

@ -32,7 +32,7 @@ import tractor
from piker.brokers import open_cached_client
from piker.log import get_logger, get_console_log
from tractor.ipc._shm import ShmArray
from piker.data import ShmArray
from piker.brokers._util import (
BrokerError,
DataUnavailable,

View File

@ -23,13 +23,13 @@ sharing live streams over a network.
"""
from .ticktools import iterticks
from tractor.ipc._shm import (
ShmArray,
from ._sharedmem import (
maybe_open_shm_array,
attach_shm_array,
open_shm_array,
get_shm_token,
open_shm_ndarray as open_shm_array,
attach_shm_ndarray as attach_shm_array,
ShmArray,
)
from ._sharedmem import maybe_open_shm_array
from ._source import (
def_iohlcv_fields,
def_ohlcv_fields,

View File

@ -28,7 +28,9 @@ from msgspec import field
import numpy as np
from numpy.lib import recfunctions as rfn
from tractor.ipc._shm import ShmArray
from ._sharedmem import (
ShmArray,
)
from ._pathops import (
path_arrays_from_ohlc,
)

View File

@ -55,7 +55,9 @@ from ._util import (
from ..service import maybe_spawn_daemon
if TYPE_CHECKING:
from tractor.ipc._shm import ShmArray
from ._sharedmem import (
ShmArray,
)
from .feed import (
_FeedsBus,
Sub,
@ -376,16 +378,16 @@ async def register_with_sampler(
# feed_is_live.is_set()
# ^TODO? pass it in instead?
):
from tractor.ipc._shm import (
attach_shm_ndarray,
NDToken,
from ._sharedmem import (
attach_shm_array,
_Token,
)
for period in shms_by_period:
# load and register shm handles
shm_token_msg = shms_by_period[period]
shm = attach_shm_ndarray(
NDToken.from_msg(shm_token_msg),
shm = attach_shm_array(
_Token.from_msg(shm_token_msg),
readonly=False,
)
shms_by_period[period] = shm

View File

@ -1,106 +1,661 @@
# piker: trading gear for hackers
# Copyright (C) Tyler Goodlet (in stewardship for pikers)
# This program is free software: you can redistribute it
# and/or modify it under the terms of the GNU Affero General
# Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your
# option) any later version.
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be
# useful, but WITHOUT ANY WARRANTY; without even the implied
# warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Affero General Public License for
# more details.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
# You should have received a copy of the GNU Affero General
# Public License along with this program. If not, see
# <https://www.gnu.org/licenses/>.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
'''
Piker-specific shared memory helpers.
"""
NumPy compatible shared memory buffers for real-time IPC streaming.
Thin shim providing piker-only wrappers around
``tractor.ipc._shm``; all core types and functions
are now imported directly from tractor throughout
the codebase.
"""
from __future__ import annotations
from sys import byteorder
import time
from typing import Optional
from multiprocessing.shared_memory import SharedMemory, _USE_POSIX
'''
if _USE_POSIX:
from _posixshmem import shm_unlink
# import msgspec
import numpy as np
from tractor.ipc._shm import (
NDToken,
ShmArray,
_known_tokens,
_make_token as _tractor_make_token,
open_shm_ndarray,
attach_shm_ndarray,
)
from numpy.lib import recfunctions as rfn
import tractor
from ._util import log
from ._source import def_iohlcv_fields
from piker.types import Struct
def cuckoff_mantracker():
'''
Disable all ``multiprocessing``` "resource tracking" machinery since
it's an absolute multi-threaded mess of non-SC madness.
'''
from multiprocessing import resource_tracker as mantracker
# Tell the "resource tracker" thing to fuck off.
class ManTracker(mantracker.ResourceTracker):
def register(self, name, rtype):
pass
def unregister(self, name, rtype):
pass
def ensure_running(self):
pass
# "know your land and know your prey"
# https://www.dailymotion.com/video/x6ozzco
mantracker._resource_tracker = ManTracker()
mantracker.register = mantracker._resource_tracker.register
mantracker.ensure_running = mantracker._resource_tracker.ensure_running
mantracker.unregister = mantracker._resource_tracker.unregister
mantracker.getfd = mantracker._resource_tracker.getfd
cuckoff_mantracker()
class SharedInt:
"""Wrapper around a single entry shared memory array which
holds an ``int`` value used as an index counter.
"""
def __init__(
self,
shm: SharedMemory,
) -> None:
self._shm = shm
@property
def value(self) -> int:
return int.from_bytes(self._shm.buf, byteorder)
@value.setter
def value(self, value) -> None:
self._shm.buf[:] = value.to_bytes(self._shm.size, byteorder)
def destroy(self) -> None:
if _USE_POSIX:
# We manually unlink to bypass all the "resource tracker"
# nonsense meant for non-SC systems.
name = self._shm.name
try:
shm_unlink(name)
except FileNotFoundError:
# might be a teardown race here?
log.warning(f'Shm for {name} already unlinked?')
class _Token(Struct, frozen=True):
'''
Internal represenation of a shared memory "token"
which can be used to key a system wide post shm entry.
'''
shm_name: str # this servers as a "key" value
shm_first_index_name: str
shm_last_index_name: str
dtype_descr: tuple
size: int # in struct-array index / row terms
@property
def dtype(self) -> np.dtype:
return np.dtype(list(map(tuple, self.dtype_descr))).descr
def as_msg(self):
return self.to_dict()
@classmethod
def from_msg(cls, msg: dict) -> _Token:
if isinstance(msg, _Token):
return msg
# TODO: native struct decoding
# return _token_dec.decode(msg)
msg['dtype_descr'] = tuple(map(tuple, msg['dtype_descr']))
return _Token(**msg)
# _token_dec = msgspec.msgpack.Decoder(_Token)
# TODO: this api?
# _known_tokens = tractor.ActorVar('_shm_tokens', {})
# _known_tokens = tractor.ContextStack('_known_tokens', )
# _known_tokens = trio.RunVar('shms', {})
# process-local store of keys to tokens
_known_tokens = {}
def get_shm_token(key: str) -> _Token:
"""Convenience func to check if a token
for the provided key is known by this process.
"""
return _known_tokens.get(key)
def _make_token(
key: str,
size: int,
dtype: np.dtype|None = None,
) -> NDToken:
dtype: Optional[np.dtype] = None,
) -> _Token:
'''
Wrap tractor's ``_make_token()`` with piker's
default dtype fallback to ``def_iohlcv_fields``.
Create a serializable token that can be used
to access a shared array.
'''
from ._source import def_iohlcv_fields
dtype = (
def_iohlcv_fields
if dtype is None
else dtype
dtype = def_iohlcv_fields if dtype is None else dtype
return _Token(
shm_name=key,
shm_first_index_name=key + "_first",
shm_last_index_name=key + "_last",
dtype_descr=tuple(np.dtype(dtype).descr),
size=size,
)
return _tractor_make_token(
class ShmArray:
'''
A shared memory ``numpy`` (compatible) array API.
An underlying shared memory buffer is allocated based on
a user specified ``numpy.ndarray``. This fixed size array
can be read and written to by pushing data both onto the "front"
or "back" of a set index range. The indexes for the "first" and
"last" index are themselves stored in shared memory (accessed via
``SharedInt`` interfaces) values such that multiple processes can
interact with the same array using a synchronized-index.
'''
def __init__(
self,
shmarr: np.ndarray,
first: SharedInt,
last: SharedInt,
shm: SharedMemory,
# readonly: bool = True,
) -> None:
self._array = shmarr
# indexes for first and last indices corresponding
# to fille data
self._first = first
self._last = last
self._len = len(shmarr)
self._shm = shm
self._post_init: bool = False
# pushing data does not write the index (aka primary key)
dtype = shmarr.dtype
if dtype.fields:
self._write_fields = list(shmarr.dtype.fields.keys())[1:]
else:
self._write_fields = None
# TODO: ringbuf api?
@property
def _token(self) -> _Token:
return _Token(
shm_name=self._shm.name,
shm_first_index_name=self._first._shm.name,
shm_last_index_name=self._last._shm.name,
dtype_descr=tuple(self._array.dtype.descr),
size=self._len,
)
@property
def token(self) -> dict:
"""Shared memory token that can be serialized and used by
another process to attach to this array.
"""
return self._token.as_msg()
@property
def index(self) -> int:
return self._last.value % self._len
@property
def array(self) -> np.ndarray:
'''
Return an up-to-date ``np.ndarray`` view of the
so-far-written data to the underlying shm buffer.
'''
a = self._array[self._first.value:self._last.value]
# first, last = self._first.value, self._last.value
# a = self._array[first:last]
# TODO: eventually comment this once we've not seen it in the
# wild in a long time..
# XXX: race where first/last indexes cause a reader
# to load an empty array..
if len(a) == 0 and self._post_init:
raise RuntimeError('Empty array race condition hit!?')
return a
def ustruct(
self,
fields: Optional[list[str]] = None,
# type that all field values will be cast to
# in the returned view.
common_dtype: np.dtype = float,
) -> np.ndarray:
array = self._array
if fields:
selection = array[fields]
# fcount = len(fields)
else:
selection = array
# fcount = len(array.dtype.fields)
# XXX: manual ``.view()`` attempt that also doesn't work.
# uview = selection.view(
# dtype='<f16',
# ).reshape(-1, 4, order='A')
# assert len(selection) == len(uview)
u = rfn.structured_to_unstructured(
selection,
# dtype=float,
copy=True,
)
# unstruct = np.ndarray(u.shape, dtype=a.dtype, buffer=shm.buf)
# array[:] = a[:]
return u
# return ShmArray(
# shmarr=u,
# first=self._first,
# last=self._last,
# shm=self._shm
# )
def last(
self,
length: int = 1,
) -> np.ndarray:
'''
Return the last ``length``'s worth of ("row") entries from the
array.
'''
return self.array[-length:]
def push(
self,
data: np.ndarray,
field_map: Optional[dict[str, str]] = None,
prepend: bool = False,
update_first: bool = True,
start: int | None = None,
) -> int:
'''
Ring buffer like "push" to append data
into the buffer and return updated "last" index.
NB: no actual ring logic yet to give a "loop around" on overflow
condition, lel.
'''
length = len(data)
if prepend:
index = (start or self._first.value) - length
if index < 0:
raise ValueError(
f'Array size of {self._len} was overrun during prepend.\n'
f'You have passed {abs(index)} too many datums.'
)
else:
index = start if start is not None else self._last.value
end = index + length
if field_map:
src_names, dst_names = zip(*field_map.items())
else:
dst_names = src_names = self._write_fields
try:
self._array[
list(dst_names)
][index:end] = data[list(src_names)][:]
# NOTE: there was a race here between updating
# the first and last indices and when the next reader
# tries to access ``.array`` (which due to the index
# overlap will be empty). Pretty sure we've fixed it now
# but leaving this here as a reminder.
if (
prepend
and update_first
and length
):
assert index < self._first.value
if (
index < self._first.value
and update_first
):
assert prepend, 'prepend=True not passed but index decreased?'
self._first.value = index
elif not prepend:
self._last.value = end
self._post_init = True
return end
except ValueError as err:
if field_map:
raise
# should raise if diff detected
self.diff_err_fields(data)
raise err
def diff_err_fields(
self,
data: np.ndarray,
) -> None:
# reraise with any field discrepancy
our_fields, their_fields = (
set(self._array.dtype.fields),
set(data.dtype.fields),
)
only_in_ours = our_fields - their_fields
only_in_theirs = their_fields - our_fields
if only_in_ours:
raise TypeError(
f"Input array is missing field(s): {only_in_ours}"
)
elif only_in_theirs:
raise TypeError(
f"Input array has unknown field(s): {only_in_theirs}"
)
# TODO: support "silent" prepends that don't update ._first.value?
def prepend(
self,
data: np.ndarray,
) -> int:
end = self.push(data, prepend=True)
assert end
def close(self) -> None:
self._first._shm.close()
self._last._shm.close()
self._shm.close()
def destroy(self) -> None:
if _USE_POSIX:
# We manually unlink to bypass all the "resource tracker"
# nonsense meant for non-SC systems.
shm_unlink(self._shm.name)
self._first.destroy()
self._last.destroy()
def flush(self) -> None:
# TODO: flush to storage backend like markestore?
...
def open_shm_array(
size: int,
key: str | None = None,
dtype: np.dtype | None = None,
append_start_index: int | None = None,
readonly: bool = False,
) -> ShmArray:
'''Open a memory shared ``numpy`` using the standard library.
This call unlinks (aka permanently destroys) the buffer on teardown
and thus should be used from the parent-most accessor (process).
'''
# create new shared mem segment for which we
# have write permission
a = np.zeros(size, dtype=dtype)
a['index'] = np.arange(len(a))
shm = SharedMemory(
name=key,
create=True,
size=a.nbytes
)
array = np.ndarray(
a.shape,
dtype=a.dtype,
buffer=shm.buf
)
array[:] = a[:]
array.setflags(write=int(not readonly))
token = _make_token(
key=key,
size=size,
dtype=dtype,
)
# create single entry arrays for storing an first and last indices
first = SharedInt(
shm=SharedMemory(
name=token.shm_first_index_name,
create=True,
size=4, # std int
)
)
last = SharedInt(
shm=SharedMemory(
name=token.shm_last_index_name,
create=True,
size=4, # std int
)
)
# start the "real-time" updated section after 3-days worth of 1s
# sampled OHLC. this allows appending up to a days worth from
# tick/quote feeds before having to flush to a (tsdb) storage
# backend, and looks something like,
# -------------------------
# | | i
# _________________________
# <-------------> <------->
# history real-time
#
# Once fully "prepended", the history section will leave the
# ``ShmArray._start.value: int = 0`` and the yet-to-be written
# real-time section will start at ``ShmArray.index: int``.
# this sets the index to nearly 2/3rds into the the length of
# the buffer leaving at least a "days worth of second samples"
# for the real-time section.
if append_start_index is None:
append_start_index = round(size * 0.616)
last.value = first.value = append_start_index
shmarr = ShmArray(
array,
first,
last,
shm,
)
assert shmarr._token == token
_known_tokens[key] = shmarr.token
# "unlink" created shm on process teardown by
# pushing teardown calls onto actor context stack
stack = tractor.current_actor(
err_on_no_runtime=False,
).lifetime_stack
if stack:
stack.callback(shmarr.close)
stack.callback(shmarr.destroy)
return shmarr
def attach_shm_array(
token: tuple[str, str, tuple[str, str]],
readonly: bool = True,
) -> ShmArray:
'''
Attach to an existing shared memory array previously
created by another process using ``open_shared_array``.
No new shared mem is allocated but wrapper types for read/write
access are constructed.
'''
token = _Token.from_msg(token)
key = token.shm_name
if key in _known_tokens:
assert _Token.from_msg(_known_tokens[key]) == token, "WTF"
# XXX: ugh, looks like due to the ``shm_open()`` C api we can't
# actually place files in a subdir, see discussion here:
# https://stackoverflow.com/a/11103289
# attach to array buffer and view as per dtype
_err: Optional[Exception] = None
for _ in range(3):
try:
shm = SharedMemory(
name=key,
create=False,
)
break
except OSError as oserr:
_err = oserr
time.sleep(0.1)
else:
if _err:
raise _err
shmarr = np.ndarray(
(token.size,),
dtype=token.dtype,
buffer=shm.buf
)
shmarr.setflags(write=int(not readonly))
first = SharedInt(
shm=SharedMemory(
name=token.shm_first_index_name,
create=False,
size=4, # std int
),
)
last = SharedInt(
shm=SharedMemory(
name=token.shm_last_index_name,
create=False,
size=4, # std int
),
)
# make sure we can read
first.value
sha = ShmArray(
shmarr,
first,
last,
shm,
)
# read test
sha.array
# Stash key -> token knowledge for future queries
# via `maybe_opepn_shm_array()` but only after we know
# we can attach.
if key not in _known_tokens:
_known_tokens[key] = token
# "close" attached shm on actor teardown
if (actor := tractor.current_actor(
err_on_no_runtime=False,
)):
actor.lifetime_stack.callback(sha.close)
return sha
def maybe_open_shm_array(
key: str,
size: int,
dtype: np.dtype|None = None,
append_start_index: int|None = None,
dtype: np.dtype | None = None,
append_start_index: int | None = None,
readonly: bool = False,
**kwargs,
) -> tuple[ShmArray, bool]:
'''
Attempt to attach to a shared memory block
using a "key" lookup to registered blocks in
the user's overall "system" registry (presumes
you don't have the block's explicit token).
Attempt to attach to a shared memory block using a "key" lookup
to registered blocks in the users overall "system" registry
(presumes you don't have the block's explicit token).
This is a thin wrapper around tractor's
``maybe_open_shm_ndarray()`` preserving piker's
historical defaults (``readonly=False``,
``append_start_index=None``).
This function is meant to solve the problem of discovering whether
a shared array token has been allocated or discovered by the actor
running in **this** process. Systems where multiple actors may seek
to access a common block can use this function to attempt to acquire
a token as discovered by the actors who have previously stored
a "key" -> ``_Token`` map in an actor local (aka python global)
variable.
If you know the explicit ``NDToken`` for your
memory segment instead use
``tractor.ipc._shm.attach_shm_ndarray()``.
If you know the explicit ``_Token`` for your memory segment instead
use ``attach_shm_array``.
'''
try:
# see if we already know this key
token = _known_tokens[key]
return (
attach_shm_ndarray(
attach_shm_array(
token=token,
readonly=readonly,
),
False,
)
except KeyError:
log.debug(
f'Could not find {key} in shms cache'
)
log.debug(f"Could not find {key} in shms cache")
if dtype:
token = _make_token(
key,
@ -108,18 +663,9 @@ def maybe_open_shm_array(
dtype=dtype,
)
try:
return (
attach_shm_ndarray(
token=token,
**kwargs,
),
False,
)
return attach_shm_array(token=token, **kwargs), False
except FileNotFoundError:
log.debug(
f'Could not attach to shm'
f' with token {token}'
)
log.debug(f"Could not attach to shm with token {token}")
# This actor does not know about memory
# associated with the provided "key".
@ -127,7 +673,7 @@ def maybe_open_shm_array(
# to fail if a block has been allocated
# on the OS by someone else.
return (
open_shm_ndarray(
open_shm_array(
key=key,
size=size,
dtype=dtype,
@ -137,20 +683,18 @@ def maybe_open_shm_array(
True,
)
def try_read(
array: np.ndarray,
) -> np.ndarray|None:
'''
Try to read the last row from a shared mem
array or ``None`` if the array read returns
a zero-length array result.
array: np.ndarray
Can be used to check for backfilling race
conditions where an array is currently being
(re-)written by a writer actor but the reader
is unaware and reads during the window where
the first and last indexes are being updated.
) -> Optional[np.ndarray]:
'''
Try to read the last row from a shared mem array or ``None``
if the array read returns a zero-length array result.
Can be used to check for backfilling race conditions where an array
is currently being (re-)written by a writer actor but the reader is
unaware and reads during the window where the first and last indexes
are being updated.
'''
try:
@ -158,13 +702,14 @@ def try_read(
except IndexError:
# XXX: race condition with backfilling shm.
#
# the underlying issue is that a backfill
# (aka prepend) and subsequent shm array
# first/last index update could result in an
# empty array read here since the indices may
# be updated in such a way that a read delivers
# an empty array (though it seems like we
# *should* be able to prevent that?).
# the underlying issue is that a backfill (aka prepend) and subsequent
# shm array first/last index update could result in an empty array
# read here since the indices may be updated in such a way that
# a read delivers an empty array (though it seems like we
# *should* be able to prevent that?). also, as and alt and
# something we need anyway, maybe there should be some kind of
# signal that a prepend is taking place and this consumer can
# respond (eg. redrawing graphics) accordingly.
# the array read was empty
# the array read was emtpy
return None

View File

@ -31,10 +31,10 @@ import pendulum
import numpy as np
from piker.types import Struct
from tractor.ipc._shm import (
from ._sharedmem import (
attach_shm_array,
ShmArray,
NDToken,
attach_shm_ndarray,
_Token,
)
from piker.accounting import MktPair
@ -64,11 +64,11 @@ class Flume(Struct):
'''
mkt: MktPair
first_quote: dict
_rt_shm_token: NDToken
_rt_shm_token: _Token
# optional since some data flows won't have a "downsampled" history
# buffer/stream (eg. FSPs).
_hist_shm_token: NDToken|None = None
_hist_shm_token: _Token | None = None
# private shm refs loaded dynamically from tokens
_hist_shm: ShmArray | None = None
@ -88,7 +88,7 @@ class Flume(Struct):
def rt_shm(self) -> ShmArray:
if self._rt_shm is None:
self._rt_shm = attach_shm_ndarray(
self._rt_shm = attach_shm_array(
token=self._rt_shm_token,
readonly=self._readonly,
)
@ -104,7 +104,7 @@ class Flume(Struct):
)
if self._hist_shm is None:
self._hist_shm = attach_shm_ndarray(
self._hist_shm = attach_shm_array(
token=self._hist_shm_token,
readonly=self._readonly,
)

View File

@ -37,12 +37,12 @@ import numpy as np
import tractor
from tractor.msg import NamespacePath
from tractor.ipc._shm import (
from ..data._sharedmem import (
ShmArray,
NDToken,
attach_shm_ndarray,
maybe_open_shm_array,
attach_shm_array,
_Token,
)
from ..data._sharedmem import maybe_open_shm_array
from ..log import get_logger
log = get_logger(__name__)
@ -78,8 +78,8 @@ class Fsp:
# + the consuming fsp *to* the consumers output
# shm flow.
_flow_registry: dict[
tuple[NDToken, str],
tuple[NDToken, Optional[ShmArray]],
tuple[_Token, str],
tuple[_Token, Optional[ShmArray]],
] = {}
def __init__(
@ -148,7 +148,7 @@ class Fsp:
# times as possible as per:
# - https://github.com/pikers/piker/issues/359
# - https://github.com/pikers/piker/issues/332
maybe_array := attach_shm_ndarray(dst_token)
maybe_array := attach_shm_array(dst_token)
)
return maybe_array

View File

@ -40,7 +40,7 @@ from ..log import (
)
from .. import data
from ..data.flows import Flume
from tractor.ipc._shm import ShmArray
from ..data._sharedmem import ShmArray
from ..data._sampling import (
_default_delay_s,
open_sample_stream,
@ -49,7 +49,7 @@ from ..accounting import MktPair
from ._api import (
Fsp,
_load_builtins,
NDToken,
_Token,
)
from ..toolz import Profiler
@ -414,7 +414,7 @@ async def cascade(
dst_flume_addr: dict,
ns_path: NamespacePath,
shm_registry: dict[str, NDToken],
shm_registry: dict[str, _Token],
zero_on_step: bool = False,
loglevel: str|None = None,
@ -465,9 +465,9 @@ async def cascade(
# not sure how else to do it.
for (token, fsp_name, dst_token) in shm_registry:
Fsp._flow_registry[(
NDToken.from_msg(token),
_Token.from_msg(token),
fsp_name,
)] = NDToken.from_msg(dst_token), None
)] = _Token.from_msg(dst_token), None
fsp: Fsp = reg.get(
NamespacePath(ns_path)

View File

@ -25,7 +25,7 @@ from numba import jit, float64, optional, int64
from ._api import fsp
from ..data import iterticks
from tractor.ipc._shm import ShmArray
from ..data._sharedmem import ShmArray
@jit(

View File

@ -21,7 +21,7 @@ from tractor.trionics._broadcast import AsyncReceiver
from ._api import fsp
from ..data import iterticks
from tractor.ipc._shm import ShmArray
from ..data._sharedmem import ShmArray
from ._momo import _wma
from ..log import get_logger

View File

@ -37,7 +37,9 @@ import typer
from piker.service import open_piker_runtime
from piker.cli import cli
from tractor.ipc._shm import ShmArray
from piker.data import (
ShmArray,
)
from piker import tsp
from . import log
from . import (

View File

@ -64,8 +64,10 @@ from pendulum import (
from piker import config
from piker import tsp
from tractor.ipc._shm import ShmArray
from piker.data import def_iohlcv_fields
from piker.data import (
def_iohlcv_fields,
ShmArray,
)
from piker.log import get_logger
from . import TimeseriesNotFound

View File

@ -59,12 +59,11 @@ from piker.brokers import NoData
from piker.accounting import (
MktPair,
)
from piker.log import (
get_logger,
get_console_log,
from piker.log import get_logger
from ..data._sharedmem import (
maybe_open_shm_array,
ShmArray,
)
from tractor.ipc._shm import ShmArray
from ..data._sharedmem import maybe_open_shm_array
from piker.data._source import (
def_iohlcv_fields,
)
@ -1387,10 +1386,6 @@ async def manage_history(
engages.
'''
get_console_log(
name=__name__,
level=loglevel,
)
# TODO: is there a way to make each shm file key
# actor-tree-discovery-addr unique so we avoid collisions
# when doing tests which also allocate shms for certain instruments

View File

@ -49,7 +49,7 @@ from ._cursor import (
Cursor,
ContentsLabel,
)
from tractor.ipc._shm import ShmArray
from ..data._sharedmem import ShmArray
from ._ohlc import BarItems
from ._curve import (
Curve,

View File

@ -42,7 +42,9 @@ from numpy import (
import pyqtgraph as pg
from piker.ui.qt import QLineF
from tractor.ipc._shm import ShmArray
from ..data._sharedmem import (
ShmArray,
)
from ..data.flows import Flume
from ..data._formatters import (
IncrementalFormatter,

View File

@ -44,12 +44,14 @@ from piker.fsp import (
dolla_vlm,
flow_rates,
)
from tractor.ipc._shm import (
from piker.data import (
Flume,
ShmArray,
NDToken,
)
from piker.data import Flume
from piker.data._sharedmem import try_read
from piker.data._sharedmem import (
_Token,
try_read,
)
from piker.log import get_logger
from piker.toolz import Profiler
from piker.types import Struct
@ -380,7 +382,7 @@ class FspAdmin:
tuple,
tuple[tractor.MsgStream, ShmArray]
] = {}
self._flow_registry: dict[NDToken, str] = {}
self._flow_registry: dict[_Token, str] = {}
# TODO: make this a `.src_flume` and add
# a `dst_flume`?