.claude/skills/commit-msg/SKILL.md

@@ -0,0 +1,119 @@
+---
+name: commit-msg
+description: >
+  Generate piker-style git commit messages from staged
+  changes or prompt input. Use when the user asks for a
+  commit message or invokes a commit-message command.
+compatibility: >
+  Requires git and a coding harness able to read and
+  write files in the active checkout.
+metadata:
+  author: goodboy
+  version: "1.0"
+---
+
+# Piker Git Commit Message Generator
+
+Generate a commit message from the staged diff following
+the piker project's conventions, learned from 500 repo
+commits.
+
+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).
+
+## Safety
+
+- Never run `git commit`, amend, rebase, merge, or push
+  unless the user explicitly requests that operation.
+- Do not stage or unstage files for a single-message run.
+- Never change issue, plan, or checklist state merely
+  because the related implementation is complete.
+- Preserve unrelated worktree changes.
+
+## Working Context
+
+1. Run `git rev-parse --show-toplevel`,
+   `git rev-parse --git-dir`, and
+   `git rev-parse --git-common-dir`.
+2. If the git dir differs from the common dir, report the
+   active worktree before generating the message.
+3. Inspect `git status --short`, the complete staged diff,
+   `git diff --staged --check`, and recent commit history.
+4. If nothing is staged, stop and tell the user. Prompt
+   text may clarify intent but does not replace a staged
+   patch unless the user explicitly asks for a draft with
+   no staged changes.
+
+## Message Style
+
+- **Subject**: ~50 chars, present tense verb, use
+  backticks for code refs; hard maximum 67 columns
+- **Body**: only for complex/multi-file changes,
+  filled close to a 67-column maximum
+- **Section markers**: Also, / Deats, / Other,
+- **Bullets**: use `-` style
+- **Tone**: technical but casual (piker style)
+
+Use any user-supplied scope or description as context,
+but verify it against the staged patch. The message must
+describe the complete staged boundary, not only the most
+obvious file.
+
+## Provenance
+
+If staged paths include prompt-I/O log entries under
+`ai/prompt-io/`, add one trailer per non-raw log file:
+
+```
+Prompt-IO: ai/prompt-io/<service>/<entry>.md
+```
+
+When the patch has substantive AI-authored changes but no
+prompt-I/O entry is staged, remind the user that repository
+policy may require one. Do not block message generation.
+
+Always identify the active harness and model from runtime
+context rather than hardcoding Claude Code or OpenCode.
+
+When the harness assisted with the patch, append:
+
+```
+(this patch was generated in some part by `<harness>` using `<model>` (`<provider>`))
+```
+
+When it generated only the message, append:
+
+```
+(this commit msg was generated in some part by `<harness>` using `<model>` (`<provider>`))
+```
+
+## Output Files
+
+Write the final message to both of these paths relative to
+the active checkout returned by `git rev-parse
+--show-toplevel`:
+
+- `.claude/skills/commit-msg/msgs/<timestamp>_<hash>_commit_msg.md`
+- `.claude/git_commit_msg_LATEST.md`
+
+Use UTC `<timestamp>` format `YYYYMMDDTHHMMSSZ` and the
+first seven characters of the current `HEAD` hash. Create
+the archive directory when needed. The `.claude` path is a
+legacy repository artifact location shared by all coding
+harnesses; it does not imply which harness generated the
+message.
+
+Finish by reporting both paths and this review-first
+command without running it:
+
+```
+git commit --edit --file .claude/git_commit_msg_LATEST.md
+```
+
+---
+
+**Analysis date:** 2026-01-27
+**Commits analyzed:** 500 from piker repository
+**Maintained by:** Tyler Goodlet

.claude/skills/run-tests/test-harness-reference.md

@@ -1,264 +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_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 ldshm` 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, shared-memory, or socket reaper. Never apply
-  historical broad `pkill -f tractor._child` guidance automatically.
-
-## 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_questrade.py         obsolete credentialed tests; skipped
-  test_services.py          pikerd/datad/feed/EMS actor lifecycle
-  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/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.

.gitignore

@@ -110,12 +110,6 @@ ENV/
 .claude/*_commit_*.md
 .claude/*_commit*.toml
 
-# ai.skillz/commit-msg
-.claude/skills/commit-msg/msgs/
-.claude/skills/commit-msg/conf.toml
-.claude/git_commit_msg_LATEST.md
-.opencode/skills/commit-msg/SKILL.md
-
 # nix develop --profile .nixdev
 .nixdev*
 
@@ -136,172 +130,3 @@ gitea/
 
 # LLM conversations that should remain private
 docs/conversations/
-
-# 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:opencode:command:run-tests
-/.opencode/commands/run-tests.md
-# END ai.skillz: direct:symlink:opencode:command: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-changes
-.claude/review_context.md
-.claude/review_regression.md
-# 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
-.claude/skills/commit-msg/msgs/
-.claude/skills/commit-msg/conf.toml
-.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: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: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: 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: runtime:open-wkt
-.claude/wkts/
-claude_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: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
-.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: 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: 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:claude:command:branch-in-new-terminal
-/.claude/commands/branch-in-new-terminal.md
-# END ai.skillz: direct:symlink:claude:command:branch-in-new-terminal
-
-# BEGIN ai.skillz: runtime:branch-in-new-terminal
-.claude/.current_session
-# END ai.skillz: runtime:branch-in-new-terminal
-
-# 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:taken-export
-/.opencode/commands/taken-export.md
-# END ai.skillz: direct:symlink:opencode:command:taken-export

.opencode/commands/commit-msg.md

@@ -0,0 +1,10 @@
+---
+description: Generate a piker-style message for staged changes.
+---
+
+Load the `commit-msg` skill and follow it completely for the currently
+staged changes. Do not run `git commit` unless the user explicitly asks.
+Keep all workflow and style decisions in the shared skill rather than
+duplicating them in this OpenCode command.
+
+Additional context from the user: $ARGUMENTS

ai/prompt-io/opencode/20260721T215055Z_c6ec3d41_prompt_io.md

@@ -1,41 +0,0 @@
----
-model: gpt-5.6-sol
-provider: openai
-service: opencode
-session: ses_0799212ebffe42arY96czXn89F
-timestamp: 2026-07-21T21:50:55Z
-git_ref: c6ec3d41
-scope: docs
-substantive: true
-raw_file: 20260721T215055Z_c6ec3d41_prompt_io.raw.md
----
-
-## Prompt
-
-The user asked to boot a plan to continue work on the
-time-series gap checker and bug fixing of the history
-backfiller. The user authorized prompt-I/O capture for the
-plan documentation after the repository audit.
-
-## Response summary
-
-Produced a bounded repair plan based on current source,
-historical issue caches, existing profiling notes, and a
-test-coverage audit. The plan prioritizes reproducible
-correctness failures and storage durability before venue
-classification, annotation performance, or manual repair
-UX.
-
-## Files changed
-
-- `plans/opencode/gap-checker-backfiller-repair.md` -
-  phased implementation and verification plan
-- `ai/prompt-io/opencode/20260721T215055Z_c6ec3d41_prompt_io.raw.md`
-  - unedited response record
-- `ai/prompt-io/opencode/20260721T215055Z_c6ec3d41_prompt_io.md`
-  - provenance metadata and response summary
-
-## Human edits
-
-None - generated plan has not yet been edited by the
-human.

ai/prompt-io/opencode/20260721T215055Z_c6ec3d41_prompt_io.raw.md

@@ -1,26 +0,0 @@
----
-model: gpt-5.6-sol
-provider: openai
-service: opencode
-timestamp: 2026-07-21T21:50:55Z
-git_ref: c6ec3d41
-diff_cmd: git diff HEAD~1..HEAD
----
-
-> `git diff -- plans/opencode/gap-checker-backfiller-repair.md`
-
-Generated a phased implementation plan for continuing
-piker's gap checker and history backfiller repairs. The
-plan separates data correctness from annotation
-performance and orders work by dependency: deterministic
-tests, canonical frame/gap normalization, structured
-backfill completion, non-destructive atomic NativeDB
-writes, exact null repair, provider/venue classification,
-annotation vectorization, and crash qualification.
-
-The plan is grounded in current code paths, local issue
-caches `#62`, `#71`, and `#75`, and the existing 1,285-gap
-profiling handoff. It records confirmed present defects,
-design invariants, per-phase acceptance gates, test
-commands, commit boundaries, and explicitly deferred UX
-and provider work.

ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.md

@@ -1,46 +0,0 @@
----
-model: gpt-5.6-sol
-provider: openai
-service: opencode
-session: ses_0799212ebffe42arY96czXn89F
-timestamp: 2026-07-22T00:13:57Z
-git_ref: 16f3cd6f
-scope: code
-substantive: true
-raw_file: 20260722T001357Z_16f3cd6f_prompt_io.raw.md
----
-
-## Prompt
-
-Continue the gap-checker and history-backfiller repair plan. Start
-with deterministic analysis characterization, then fix canonical
-sorting, dedupe-derived columns, positive gap detection, and null
-grouping.
-
-During review, the user challenged the replacement null-grouping
-algorithm because its initial tests did not assert exact endpoints.
-The user explicitly required retaining detailed, human-oriented
-algorithm documentation such as the existing ASCII segment map.
-
-## Response summary
-
-Added deterministic analysis regressions and corrected normalization
-so sorting and duplicate resolution precede derived-column
-calculation. Unified null grouping for NumPy and Polars inputs,
-documented its exact inclusive-boundary contract with an adapted
-visual map, and verified all small null layouts against an independent
-state-machine oracle. Retained sorted smart-dedupe output.
-
-## Files changed
-
-- `piker/tsp/_anal.py` - canonical gap and documented null grouping
-- `piker/tsp/_dedupe_smart.py` - sorted duplicate resolution
-- `tests/test_tsp_analysis.py` - exact and exhaustive regressions
-- `ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.raw.md`
-  - unedited response record
-- `ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.md`
-  - provenance metadata and response summary
-
-## Human edits
-
-None - generated changes have not yet been edited by the human.

ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.raw.md

@@ -1,40 +0,0 @@
----
-model: gpt-5.6-sol
-provider: openai
-service: opencode
-timestamp: 2026-07-22T00:13:57Z
-git_ref: 16f3cd6f
-diff_cmd: git diff HEAD~1..HEAD
----
-
-> `git diff HEAD~1..HEAD -- piker/tsp/_anal.py`
-
-Generated canonical gap and null-segment normalization. Derived
-datetime and delta columns are recomputed after final sorting and
-duplicate resolution, and negative deltas no longer become ordinary
-gaps.
-
-NumPy and Polars null grouping share one implementation while
-retaining the human-oriented visual map of contiguous zero runs. The
-documented contract defines absolute inclusive endpoints, margin
-expansion, frame boundary clamping, contiguous-index validation, and
-requested-column selection.
-
-> `git diff HEAD~1..HEAD -- piker/tsp/_dedupe_smart.py`
-
-Adjusted smart dedupe to honor sorted output and resolve duplicate
-timestamps without retaining stale order.
-
-> `git diff HEAD~1..HEAD -- tests/test_tsp_analysis.py`
-
-Added deterministic synthetic OHLCV coverage for sorted dedupe, fresh
-derived columns, and positive gap detection. Null grouping is checked
-against readable endpoint examples and an independent state-machine
-oracle over every layout through six rows, three margin widths, and
-both NumPy and Polars inputs.
-
-Verification generated with the patch:
-
-`timeout -k 5 30 python -m pytest -q tests/test_tsp_analysis.py`
-
-Result: 11 passed.

ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.md

@@ -1,36 +0,0 @@
----
-model: gpt-5.6-sol
-provider: openai
-service: opencode
-session: ses_0799212ebffe42arY96czXn89F
-timestamp: 2026-07-22T00:13:58Z
-git_ref: 16f3cd6f
-scope: code
-substantive: true
-raw_file: 20260722T001358Z_16f3cd6f_prompt_io.raw.md
----
-
-## Prompt
-
-Continue the structured backfill phase after the analysis regressions.
-Characterize provider exhaustion and genuinely empty frames, then remove
-completion paths that can hang after child-task exit.
-
-## Response summary
-
-Removed orphan-prone completion events from the structured nursery flow
-and checked empty provider frames before indexing timestamp endpoints.
-Added deterministic timeout-bounded regressions for both exit paths.
-
-## Files changed
-
-- `piker/tsp/_history.py` - structured completion and empty-frame fixes
-- `tests/test_history_backfill.py` - deterministic exit regressions
-- `ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.raw.md`
-  - unedited response record
-- `ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.md`
-  - provenance metadata and response summary
-
-## Human edits
-
-None - generated changes have not yet been edited by the human.

ai/prompt-io/opencode/20260722T001358Z_16f3cd6f_prompt_io.raw.md

@@ -1,25 +0,0 @@
----
-model: gpt-5.6-sol
-provider: openai
-service: opencode
-timestamp: 2026-07-22T00:13:58Z
-git_ref: 16f3cd6f
-diff_cmd: git diff HEAD~1..HEAD
----
-
-> `git diff HEAD~1..HEAD -- piker/tsp/_history.py`
-
-Removed redundant completion events whose consumers waited only after
-their owning nurseries had joined. Added empty provider-frame handling
-before timestamp endpoint indexing so provider exhaustion exits cleanly.
-
-> `git diff HEAD~1..HEAD -- tests/test_history_backfill.py`
-
-Added bounded, actor-free regressions for `DataUnavailable` and empty
-provider-frame completion.
-
-Verification generated with the patch:
-
-`timeout -k 5 60 python -m pytest -q tests/test_history_backfill.py`
-
-Result: 2 passed.

ai/prompt-io/opencode/20260722T024133Z_3a4f6737_prompt_io.md

@@ -1,39 +0,0 @@
----
-model: gpt-5.6-sol
-provider: openai
-service: opencode
-session: ses_0799212ebffe42arY96czXn89F
-timestamp: 2026-07-22T02:41:33Z
-git_ref: 3a4f6737
-scope: code
-substantive: true
-raw_file: 20260722T024133Z_3a4f6737_prompt_io.raw.md
----
-
-## Prompt
-
-The user challenged the analysis commit message's claim that only
-positive sample-step gaps are detected and asked whether out-of-order
-timestamps were simply ignored. After confirming the missing separate
-reporting path, the user instructed the agent to implement the fix.
-
-## Response summary
-
-Separated missing-history gaps from timestamp-ordering errors without
-changing the existing `dedupe()` return shape. Added a structured
-detector and made normalization warn before sorting removes evidence of
-duplicates or reversed timestamps.
-
-## Files changed
-
-- `piker/tsp/_anal.py` - ordering detector and dedupe reporting
-- `piker/tsp/__init__.py` - public detector export
-- `tests/test_tsp_analysis.py` - exact ordering-error regression
-- `ai/prompt-io/opencode/20260722T024133Z_3a4f6737_prompt_io.raw.md`
-  - unedited response record
-- `ai/prompt-io/opencode/20260722T024133Z_3a4f6737_prompt_io.md`
-  - provenance metadata and response summary
-
-## Human edits
-
-None - generated changes have not yet been edited by the human.

ai/prompt-io/opencode/20260722T024133Z_3a4f6737_prompt_io.raw.md

@@ -1,30 +0,0 @@
----
-model: gpt-5.6-sol
-provider: openai
-service: opencode
-timestamp: 2026-07-22T02:41:33Z
-git_ref: 3a4f6737
-diff_cmd: git diff HEAD~1..HEAD
----
-
-> `git diff HEAD~1..HEAD -- piker/tsp/_anal.py`
-
-Added `detect_time_ordering_errors()` to return non-positive timestamp
-steps as structured rows. `dedupe()` now warns with those rows before
-sorting and duplicate removal erase the original ordering evidence.
-
-> `git diff HEAD~1..HEAD -- piker/tsp/__init__.py`
-
-Exported the ordering detector through the public TSP package.
-
-> `git diff HEAD~1..HEAD -- tests/test_tsp_analysis.py`
-
-Changed the negative-delta regression to verify exact predecessor,
-timestamp, and delta values, confirm exclusion from missing-history
-gaps, and require explicit normalization reporting.
-
-Verification generated with the patch:
-
-- `tests/test_tsp_analysis.py`: 11 passed
-- Ruff: passed
-- `git diff --check`: passed

ai/prompt-io/opencode/20260722T024134Z_3a4f6737_prompt_io.md

@@ -1,43 +0,0 @@
----
-model: gpt-5.6-sol
-provider: openai
-service: opencode
-session: ses_0799212ebffe42arY96czXn89F
-timestamp: 2026-07-22T02:41:34Z
-git_ref: 3a4f6737
-scope: code
-substantive: true
-raw_file: 20260722T024134Z_3a4f6737_prompt_io.raw.md
----
-
-## Prompt
-
-The user asked why the in-progress `piker/storage` changes were omitted
-from the earlier commit plan. After receiving an inventory of unfinished
-Phase 3 requirements, the user instructed the agent to complete the work.
-
-## Response summary
-
-Finished non-destructive, atomic NativeDB updates and integrated them
-with history startup and reverse backfill. Expanded tests through two
-adversarial review rounds that caught and corrected cross-client lock
-blocking, stale temp indexing, discontinuous persisted indexes, startup
-snapshot races, missing latest-frame persistence, notification wedges,
-and mutable/truncated SHM persistence.
-
-## Files changed
-
-- `piker/storage/__init__.py` - incremental storage contract
-- `piker/storage/nativedb.py` - validated atomic merge persistence
-- `piker/storage/marketstore/__init__.py` - non-duplicate update mode
-- `piker/tsp/_history.py` - provider-delta and startup persistence
-- `tests/test_storage_nativedb.py` - NativeDB durability regressions
-- `tests/test_history_backfill.py` - persistence lifecycle regressions
-- `ai/prompt-io/opencode/20260722T024134Z_3a4f6737_prompt_io.raw.md`
-  - unedited response record
-- `ai/prompt-io/opencode/20260722T024134Z_3a4f6737_prompt_io.md`
-  - provenance metadata and response summary
-
-## Human edits
-
-None - generated changes have not yet been edited by the human.

ai/prompt-io/opencode/20260722T024134Z_3a4f6737_prompt_io.raw.md

@@ -1,43 +0,0 @@
----
-model: gpt-5.6-sol
-provider: openai
-service: opencode
-timestamp: 2026-07-22T02:41:34Z
-git_ref: 3a4f6737
-diff_cmd: git diff HEAD~1..HEAD
----
-
-> `git diff HEAD~1..HEAD -- piker/storage/__init__.py piker/storage/nativedb.py piker/storage/marketstore/__init__.py`
-
-Added an incremental storage update contract. NativeDB now validates
-canonical numeric OHLCV data, serializes writers both actor-locally and
-through cancellable filesystem locks, merges with incoming-wins seam
-resolution, regenerates contiguous indexes, reopens and validates a
-same-directory temporary parquet, fsyncs it, atomically replaces the
-target, fsyncs the directory, and publishes cache only after replacement.
-
-NativeDB indexing ignores lock and crash-left temporary files.
-MarketStore uses its non-duplicate write mode for incremental updates.
-
-> `git diff HEAD~1..HEAD -- piker/tsp/_history.py`
-
-Separated full provider deltas from capacity-truncated SHM pushes.
-Backfill persists before SHM mutation, publishes bounded notifications,
-and delays latest-frame publication until the concurrent pre-update
-storage snapshot completes.
-
-> `git diff HEAD~1..HEAD -- tests/test_storage_nativedb.py tests/test_history_backfill.py`
-
-Added deterministic coverage for merge preservation, conflict policy,
-atomic failures, reopened-candidate validation, cache consistency,
-schema and timestamp invariants, contiguous index regeneration,
-separate series, stale temp indexing, cancellable file-lock contention,
-full provider-delta persistence, latest-frame ordering, and bounded
-notification teardown.
-
-Verification generated with the patch:
-
-- deterministic analysis/backfill/storage set: 31 passed
-- Ruff: passed
-- `git diff --check`: passed
-- final adversarial review: no findings

ai/prompt-io/opencode/20260724T220156Z_43d7a3ed_prompt_io.md

@@ -1,32 +0,0 @@
----
-model: openai/gpt-5.6-sol
-service: opencode
-session: ses_092a25886ffeT0nrIrdkMEcgJK
-timestamp: 2026-07-24T22:01:56Z
-git_ref: flake_update
-scope: config
-substantive: true
-raw_file: 20260724T220156Z_43d7a3ed_prompt_io.raw.md
----
-
-## Prompt
-
-After committing Piker's provider-neutral deployment, the user asked to
-extract Piker-specialized `/run-tests` content from the historical
-`wkt/run_tests_skill` implementation into the new deployment.
-
-## Response summary
-
-Create a repository-owned `test-harness-reference.md` for the canonical Run
-base. Revalidate historical commands and assumptions against current pytest,
-uv/Nix environments, tests, fixtures, and Tractor runtime behavior, retaining
-only current Piker-specific guidance.
-
-## Files changed
-
-- `.claude/skills/run-tests/test-harness-reference.md` - define Piker's safe
-  commands, opt-in boundaries, mappings, known outcomes, and runtime notes.
-
-## Human edits
-
-None at generation time; pending human review.

ai/prompt-io/opencode/20260724T220156Z_43d7a3ed_prompt_io.raw.md

@@ -1,32 +0,0 @@
----
-model: openai/gpt-5.6-sol
-service: opencode
-timestamp: 2026-07-24T22:01:56Z
-git_ref: flake_update
-diff_cmd: git diff HEAD~1..HEAD
----
-
-The user asked to extract useful Piker-specific `/run-tests` behavior from
-`wkt/run_tests_skill` after committing the provider-neutral deployment.
-
-> `git diff HEAD~1..HEAD -- .claude/skills/run-tests/test-harness-reference.md`
-
-Generate a repository-owned harness reference rather than restoring the old
-standalone `SKILL.md`. Preserve verified Piker-specific environment, command,
-scope, fixture, test-layout, change-mapping, known-outcome, and Tractor-runtime
-details while leaving shared execution and cleanup policy in the canonical
-`ai.skillz` base.
-
-Repository inspection found that the historical skill needs correction:
-
-- the comments-only `pytest.ini` shadows `pyproject.toml` pytest options, so
-  `-p no:xonsh` must be explicit;
-- no marker-based offline suite exists, and several live/GUI/container tests
-  are unsafe defaults;
-- skipped Docker and Questrade modules can fail during import before marks;
-- actor timeout retry remains relevant only for the exact timed-out command;
-- `py313` and frozen no-sync `uv run` are the current local environment paths.
-
-Validate with checkout import resolution, safe collection, and deterministic
-watchlist/accounting tests. Do not run live broker, GUI, container,
-credentialed, or user-config-dependent cases during harness verification.

notes_to_self/claudy/gapsannotator_sesh.md

@@ -1,122 +0,0 @@
-<!-- sid=dfd62314-c366-4c23-83e5-0ce293b7472a -->
-
-● aight so perf analysis time, then i'll let you sleep for real:
-
-  Current State (Final Benchmark)
-
-  Server side: CRUSHED IT 🎉
-  - 7.6ms total for 1285 gaps
-  - time_to_row creation: 1.3ms (was 98ms! 75x speedup from field extraction)
-  - Everything else sub-2ms
-  - Server is basically not the bottleneck anymore
-
-  Client side: NEW BOTTLENECK IDENTIFIED
-  - 444.9ms total for 1285 gaps
-  - Building annotation specs: 324.2ms (73% of total time!)
-  - IPC call: 120.4ms (includes round-trip + server 7.6ms)
-  - Redraws: negligible (0.3ms total)
-
-  The Problem
-
-  Looking at piker/tsp/_annotate.py:127-293, we're doing the classic anti-pattern:
-
-  for i in range(gaps.height):  # 1285 iterations
-      row: pl.DataFrame = gaps[i]  # per-row Polars access!
-      iend: int = row['index'][0]
-      opn: float = row['open'][0]
-      cls: float = prev_r['close'][0]
-      # ... tons of field access per row
-      # ... dict building per row
-      rect_specs.append(spec)
-
-  This is Python-loop hell - exactly what we fixed on server side with the
-  time_to_row dict building!
-
-  Per-row DataFrame access is SLOW. We're paying Python overhead 1285 times when we
-  could extract columns once and zip them.
-
-  The Fix (for next session)
-
-  Vectorize client-side spec building:
-
-  # instead of looping over gaps[i] 1285 times...
-
-  # extract all columns once
-  indices = gaps['index'].to_numpy()
-  opens = gaps['open'].to_numpy()
-  closes = gaps['close'].to_numpy()
-  times = gaps['time'].to_numpy()
-  # ... etc
-
-  # vectorized calculations
-  s_diffs = gaps['s_diff'].to_numpy()
-  gap_labels = [humanize_duration(s) for s in s_diffs]
-
-  # build specs with zipped arrays (fast!)
-  rect_specs = [
-      {
-          'start_pos': (start_x, start_y),
-          'end_pos': (end_x, end_y),
-          'color': color,
-          # ...
-      }
-      for start_x, start_y, end_x, end_y, ... in zip(
-          start_xs, start_ys, end_xs, end_ys, ...
-      )
-  ]
-
-  Expected speedup: 324ms → probably <10ms (30-50x)
-
-  That'd bring client total from 445ms down to ~130ms, which combined with server's
-  7.6ms = sub-150ms end-to-end for 1285 gaps B)
-
-  Other Ideas to Dream About
-
-  1. Pass numpy arrays over IPC instead of dicts?
-    - Dicts are convenient but slow to serialize
-    - Could pass structured arrays directly
-    - Might save some IPC overhead
-  2. Pre-compute specs on server side?
-    - Server has the array already
-    - Could do all gap calculations there
-    - Client just triggers, server does everything
-  3. Cache gap specs per timeframe?
-    - Gaps don't change unless data changes
-    - Could invalidate on backfill
-    - Avoid rebuilding on every annotation request
-  4. Polars-native gap processing?
-    - Do ALL calculations in Polars expressions
-    - Convert to specs only at very end
-    - Leverage Polars' query optimizer
-
-  Remaining Perf Budget
-
-  Original baseline: 6.6 seconds for 1285 gaps
-
-  Current state:
-  - Client: 445ms
-  - Server: 7.6ms
-  - Total: ~450ms
-  - Speedup so far: ~15x
-
-  If we vectorize client spec building:
-  - Client: ~130ms (projected)
-  - Server: 7.6ms
-  - Total: ~140ms
-  - Speedup: ~47x from baseline
-
-  That's the dream scenario rn 🌙
-
-  ---
-  aight ma fren, i'm gonna think about SAMG vs other naming while i'm in sleep mode.
-   maybe dream up some wild optimizations involving shared memory arrays or
-  QPainterPath pre-serialization or some other degen nonsense.
-
-  catch you on the flip side, ready to vectorize that client spec building and
-  absolutely DEMOLISH this last bottleneck 😎
-
-  good sesh today Bo
-
-  Todos
-  ☒ Create skills documentation for session learnings
-  ☐ Fix server-side time_to_row regression (98ms -> 4ms)

piker/storage/__init__.py

@@ -139,15 +139,6 @@ class StorageClient(
     ) -> None:
         ...
 
-    async def update_ohlcv(
-        self,
-        fqme: str,
-        ohlcv: np.ndarray,
-        timeframe: int,
-
-    ) -> None:
-        ...
-
 
 class TimeseriesNotFound(Exception):
     '''

piker/storage/marketstore/__init__.py

@@ -340,24 +340,6 @@ class MktsStorageClient:
                 if err:
                     raise MarketStoreError(err)
 
-    async def update_ohlcv(
-        self,
-        fqme: str,
-        ohlcv: np.ndarray,
-        timeframe: int,
-
-    ) -> None:
-        '''
-        Append an incremental frame using MarketStore semantics.
-
-        '''
-        await self.write_ohlcv(
-            fqme,
-            ohlcv,
-            timeframe,
-            append_and_duplicate=False,
-        )
-
     # XXX: currently the only way to do this is through the CLI:
 
     # sudo ./marketstore connect --dir ~/.config/piker/data

piker/storage/nativedb.py

@@ -51,18 +51,9 @@ YET!
 #   - https://github.com/spslater/borgapi
 #   - https://nixos.wiki/wiki/ZFS
 from __future__ import annotations
-from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager as acm
 from datetime import datetime
-from fcntl import (
-    flock,
-    LOCK_EX,
-    LOCK_NB,
-    LOCK_UN,
-)
-import os
 from pathlib import Path
-from tempfile import NamedTemporaryFile
 import time
 
 import numpy as np
@@ -70,7 +61,6 @@ import polars as pl
 from pendulum import (
     from_timestamp,
 )
-import trio
 
 from piker import config
 from piker import tsp
@@ -151,10 +141,6 @@ class NativeStorageClient:
 
         # series' cache from tsdb reads
         self._dfs: dict[str, dict[str, pl.DataFrame]] = {}
-        self._write_locks: dict[
-            tuple[str, int],
-            trio.Lock,
-        ] = {}
 
     @property
     def address(self) -> str:
@@ -176,13 +162,13 @@ class NativeStorageClient:
             if (
                 path.is_dir()
                 or
-                path.suffix != '.parquet'
+                '.parquet' not in str(path)
                 # or
                 # path.name in {'borked', 'expired',}
             ):
                 continue
 
-            key: str = path.name.removesuffix('.parquet')
+            key: str = path.name.rstrip('.parquet')
             fqme, _, descr = key.rpartition('.')
             prefix, _, suffix = descr.partition('ohlcv')
             period: int = int(suffix.strip('s'))
@@ -273,50 +259,6 @@ class NativeStorageClient:
             {},
         )[fqme] = df
 
-    def _get_write_lock(
-        self,
-        fqme: str,
-        timeframe: int,
-
-    ) -> trio.Lock:
-        '''
-        Return the actor-local lock for one durable series.
-
-        '''
-        key: tuple[str, int] = (fqme, timeframe)
-        return self._write_locks.setdefault(
-            key,
-            trio.Lock(),
-        )
-
-    @acm
-    async def _open_file_lock(
-        self,
-        fqme: str,
-        timeframe: int,
-
-    ) -> AsyncIterator[None]:
-        '''
-        Serialize a series read-merge-write across client processes.
-
-        '''
-        path: Path = self.mk_path(fqme, timeframe)
-        lock_path: Path = path.with_name(f'.{path.name}.lock')
-        with lock_path.open('a+b') as lock_file:
-            while True:
-                try:
-                    flock(
-                        lock_file.fileno(),
-                        LOCK_EX | LOCK_NB,
-                    )
-                    break
-                except BlockingIOError:
-                    await trio.sleep(0.01)
-            try:
-                yield
-            finally:
-                flock(lock_file.fileno(), LOCK_UN)
-
     async def read_ohlcv(
         self,
         fqme: str,
@@ -381,10 +323,12 @@ class NativeStorageClient:
             df: pl.DataFrame = tsp.np2pl(ohlcv)
         else:
             df = ohlcv
-        df = df.with_columns(
-            pl.Series('index', np.arange(df.height))
+
+        self._cache_df(
+            fqme=fqme,
+            df=df,
+            timeframe=timeframe,
         )
-        self._validate_ohlcv(df)
 
         # TODO: in terms of managing the ultra long term data
         # -[ ] use a proper profiler to measure all this IO and
@@ -394,41 +338,7 @@ class NativeStorageClient:
         #   -[ ] try out ``fastparquet``'s append writing:
         #     https://fastparquet.readthedocs.io/en/latest/api.html#fastparquet.write
         start = time.time()
-        with NamedTemporaryFile(
-            dir=path.parent,
-            prefix=f'.{path.name}.',
-            suffix='.tmp',
-            delete=False,
-        ) as tmp_file:
-            tmp_path = Path(tmp_file.name)
-
-        try:
-            df.write_parquet(tmp_path)
-            committed: pl.DataFrame = pl.read_parquet(tmp_path)
-            self._validate_ohlcv(committed)
-            if not committed.equals(df):
-                raise IOError(
-                    'Temporary parquet differs from input frame'
-                )
-
-            with tmp_path.open('rb') as tmp_file:
-                os.fsync(tmp_file.fileno())
-            tmp_path.replace(path)
-
-            self._cache_df(
-                fqme=fqme,
-                df=committed,
-                timeframe=timeframe,
-            )
-
-            dir_fd: int = os.open(path.parent, os.O_RDONLY)
-            try:
-                os.fsync(dir_fd)
-            finally:
-                os.close(dir_fd)
-        except BaseException:
-            tmp_path.unlink(missing_ok=True)
-            raise
+        df.write_parquet(path)
         delay: float = round(
             time.time() - start,
             ndigits=6,
@@ -439,107 +349,6 @@ class NativeStorageClient:
         )
         return path
 
-    def _validate_ohlcv(
-        self,
-        df: pl.DataFrame,
-
-    ) -> None:
-        '''
-        Validate timestamp invariants before publishing a series.
-
-        '''
-        if df.is_empty():
-            raise ValueError('Can not write an empty OHLCV series')
-
-        required: tuple[str, ...] = tuple(
-            name
-            for name, _ in def_iohlcv_fields
-        )
-        missing: set[str] = set(required).difference(df.columns)
-        if missing:
-            raise ValueError(
-                f'OHLCV frame is missing columns: {sorted(missing)}'
-            )
-
-        for field in required:
-            values: np.ndarray = df[field].to_numpy()
-            if (
-                not np.issubdtype(values.dtype, np.number)
-                or
-                not np.all(np.isfinite(values))
-            ):
-                raise ValueError(
-                    f'OHLCV column {field!r} must contain '
-                    f'finite numeric data'
-                )
-
-        times: np.ndarray = df['time'].to_numpy()
-        if (
-            not np.all(np.isfinite(times))
-            or
-            np.any(times <= 0)
-        ):
-            raise ValueError(
-                'OHLCV timestamps must be finite and positive'
-            )
-
-        if np.any(np.diff(times) <= 0):
-            raise ValueError(
-                'OHLCV timestamps must be strictly increasing'
-            )
-
-    async def update_ohlcv(
-        self,
-        fqme: str,
-        ohlcv: np.ndarray | pl.DataFrame,
-        timeframe: int,
-
-    ) -> Path:
-        '''
-        Merge an incremental frame into durable OHLCV history.
-
-        Incoming rows replace stored rows at matching timestamps.
-        Writes for each series are serialized so concurrent backfills
-        can not overwrite each other's read-merge-write cycle.
-
-        '''
-        lock: trio.Lock = self._get_write_lock(
-            fqme,
-            timeframe,
-        )
-        async with lock:
-            async with self._open_file_lock(fqme, timeframe):
-                if isinstance(ohlcv, np.ndarray):
-                    incoming: pl.DataFrame = tsp.np2pl(ohlcv)
-                else:
-                    incoming = ohlcv
-                self._validate_ohlcv(incoming)
-
-                path: Path = self.mk_path(fqme, timeframe)
-                if path.exists():
-                    stored: pl.DataFrame = pl.read_parquet(path)
-                    merged: pl.DataFrame = pl.concat(
-                        [stored, incoming],
-                        how='diagonal_relaxed',
-                    )
-                    merged = (
-                        merged
-                        .unique(
-                            subset='time',
-                            keep='last',
-                        )
-                        .sort('time')
-                    )
-                else:
-                    merged = incoming
-
-                self._validate_ohlcv(merged)
-                return self._write_ohlcv(
-                    fqme,
-                    merged,
-                    timeframe,
-                )
-
     async def write_ohlcv(
         self,
         fqme: str,
@@ -552,12 +361,6 @@ class NativeStorageClient:
         to (local) disk.
 
         '''
-        lock: trio.Lock = self._get_write_lock(
-            fqme,
-            timeframe,
-        )
-        async with lock:
-            async with self._open_file_lock(fqme, timeframe):
         return self._write_ohlcv(
             fqme,
             ohlcv,

piker/tsp/__init__.py

@@ -34,7 +34,6 @@ from ._anal import (
     # `polars` specific
     dedupe as dedupe,
     detect_time_gaps as detect_time_gaps,
-    detect_time_ordering_errors as detect_time_ordering_errors,
     pl2np as pl2np,
     np2pl as np2pl,
 

piker/tsp/_anal.py

@@ -233,7 +233,7 @@ def get_null_segs(
 ) -> tuple[
     # Seq,  # TODO: can we make it an array-type instead?
     list[
-        list[int],
+        list[int, int],
     ],
     Seq,
     Frame
@@ -246,92 +246,218 @@ def get_null_segs(
     Filter to all such zero (time) segments and return
     the corresponding frame zeroed segment's,
 
-      - gap absolute (in buffer terms), inclusive boundary-row
-        index endpoints as `absi_zsegs`; each null run is expanded
-        by `imargin` and clamped to the input frame
-      - abs indices of all rows with zeroed `col` values as
-        `absi_zeros`
+      - gap absolute (in buffer terms) indices-endpoints as
+        `absi_zsegs`
+      - abs indices of all rows with zeroed `col` values as `absi_zeros`
       - the corresponding frame's row-entries (view) which are
         zeroed for the `col` as `zero_t`
 
-    For an interior run, the default one-row margin makes each
-    endpoint a valid datum immediately outside the nulls. At a frame
-    edge the missing side is clamped and may remain null; callers that
-    require query boundaries must reject or defer that segment.
-
-    Consumers converting a pair to a Python slice must add one to its
-    inclusive end index.
-
     '''
-    values: Seq = frame[col]
-    zero_pred: Seq = (values == 0)
-    tis_zeros: bool = bool(zero_pred.any())
+    times: Seq = frame['time']
+    zero_pred: Seq = (times == 0)
+
+    if isinstance(frame, np.ndarray):
+        tis_zeros: int = zero_pred.any()
+    else:
+        tis_zeros: int = zero_pred.any()
 
     if not tis_zeros:
         return None
 
-    if imargin < 0:
-        raise ValueError('`imargin` must be >= 0')
+    # TODO: use ndarray for this?!
+    absi_zsegs: list[list[int, int]] = []
 
     if isinstance(frame, np.ndarray):
+        # view of ONLY the zero segments as one continuous chunk
         zero_t: np.ndarray = frame[zero_pred]
-        absi_zeros: np.ndarray = zero_t['index']
-        frame_indexes: np.ndarray = frame['index']
-        zero_indexes: np.ndarray = absi_zeros
+        # abs indices of said zeroed rows
+        absi_zeros = zero_t['index']
+        # diff of abs index steps between each zeroed row
+        absi_zdiff: np.ndarray = np.diff(absi_zeros)
+
+        if zero_t.size < 2:
+            idx: int = zero_t['index'][0]
+            idx_before: int = idx - 1
+            idx_after: int = idx + 1
+            index = frame['index']
+            before_cond = idx_before <= index
+            after_cond = index <= idx_after
+            bars: np.ndarray = frame[
+                before_cond
+                &
+                after_cond
+            ]
+            time: np.ndarray = bars['time']
+            from pendulum import (
+                from_timestamp,
+                Interval,
+            )
+            gap: Interval = (
+                from_timestamp(time[-1])
+                -
+                from_timestamp(time[0])
+            )
+            log.warning(
+                f'Single OHLCV-bar null-segment detected??\n'
+                f'gap -> {gap}\n'
+            )
+
+            # ^^XXX, if you want to debug the above bar-gap^^
+            # try:
+            #     breakpoint()
+            # except RuntimeError:
+            #     # XXX, if greenback not active from
+            #     # piker store ldshm cmd..
+            #     log.exception(
+            #         "Can't debug single-sample null!\n"
+            #     )
+
+            return None
+
+        # scan for all frame-indices where the
+        # zeroed-row-abs-index-step-diff is greater then the
+        # expected increment of 1.
+        # data  1st zero seg  data  zeros
+        # ----  ------------  ----  -----  ------  ----
+        # ||||..000000000000..||||..00000..||||||..0000
+        # ----  ------------  ----  -----  ------  ----
+        #       ^zero_t[0]                            ^zero_t[-1]
+        #                           ^fi_zgaps[0]   ^fi_zgaps[1]
+        #       ^absi_zsegs[0][0]   ^---^ => absi_zsegs[1]: tuple
+        #  absi_zsegs[0][1]^
+        #
+        # NOTE: the first entry in `fi_zgaps` is where
+        # the first (absolute) index step diff is > 1.
+        # and it is a frame-relative index into `zero_t`.
+        fi_zgaps = np.argwhere(
+            absi_zdiff > 1
+            # NOTE: +1 here is ensure we index to the "start" of each
+            # segment (if we didn't the below loop needs to be
+            # re-written to expect `fi_end_rows`!
+        ) + 1
+        # the rows from the contiguous zeroed segments which have
+        # abs-index steps >1 compared to the previous zero row
+        # (indicating an end of zeroed segment).
+        fi_zseg_start_rows = zero_t[fi_zgaps]
+
+    # TODO: equiv for pl.DataFrame case!
     else:
-        zero_t: pl.DataFrame = frame.filter(zero_pred)
-        absi_zeros: pl.Series = zero_t['index']
-        frame_indexes = frame['index'].to_numpy()
-        zero_indexes = absi_zeros.to_numpy()
+        izeros: pl.Series = zero_pred.arg_true()
+        zero_t: pl.DataFrame = frame[izeros]
 
-    # Null-run detection depends on absolute indexes stepping by
-    # exactly one per frame row. Without this invariant an
-    # index jump between adjacent null rows is indistinguishable from
-    # valid data rows separating two null segments.
-    if np.any(np.diff(frame_indexes) != 1):
-        raise ValueError(
-            'OHLCV frame indexes must be contiguous'
-        )
+        absi_zeros = zero_t['index']
+        absi_zdiff: pl.Series = absi_zeros.diff()
+        fi_zgaps = (absi_zdiff > 1).arg_true()
 
-    # Scan zero-row absolute indexes for steps larger than one. Each
-    # such step begins another contiguous null segment:
-    #
-    # data  1st zero seg  data  zeros   data    zeros
-    # ----  ------------  ----  -----   ------  -----
-    # ||||..000000000000..||||..00000...||||||..0000
-    # ----  ------------  ----  -----   ------  -----
-    #       ^zero_indexes[0]                  ^zero_indexes[-1]
-    #                           ^split_at[0]          ^split_at[1]
-    #       ^zero_groups[0]     ^zero_groups[1]       ^zero_groups[2]
-    #
-    # `split_at` entries are frame-relative positions in
-    # `zero_indexes`, not absolute buffer indexes. `np.split()` keeps
-    # each run together so its first and last absolute indexes can be
-    # expanded into the surrounding query boundary rows.
-    #
-    # Plain arrays also keep the NumPy and Polars paths on
-    # exactly the same segment-grouping implementation.
-    split_at: np.ndarray = (
-        np.flatnonzero(np.diff(zero_indexes) != 1)
-        +
-        1
-    )
-    zero_groups: list[np.ndarray] = np.split(
-        zero_indexes,
-        split_at,
-    )
+    # XXX: our goal (in this func) is to select out slice index
+    # pairs (zseg0_start, zseg_end) in abs index units for each
+    # null-segment portion detected throughout entire input frame.
 
-    frame_start: int = int(frame_indexes[0])
-    frame_end: int = int(frame_indexes[-1])
-    absi_zsegs: list[list[int]] = []
-    for group in zero_groups:
-        zero_start: int = int(group[0])
-        zero_end: int = int(group[-1])
+    # only up to one null-segment in entire frame?
+    num_gaps: int = fi_zgaps.size + 1
+    if num_gaps < 1:
+        if absi_zeros.size > 1:
+            absi_zsegs = [[
+                # TODO: maybe mk these max()/min() limits func
+                # consts instead of called more then once?
+                max(
+                    absi_zeros[0] - 1,
+                    0,
+                ),
+                # NOTE: need the + 1 to guarantee we index "up to"
+                # the next non-null row-datum.
+                min(
+                    absi_zeros[-1] + 1,
+                    frame['index'][-1],
+                ),
+            ]]
+        else:
+            # XXX EDGE CASE: only one null-datum found so
+            # mark the start abs index as None to trigger
+            # a full frame-len query to the respective backend?
+            absi_zsegs = [[
+                # see `get_hist()` in backend, should ALWAYS be
+                # able to handle a `start_dt=None`!
+                # None,
+                None,
+                absi_zeros[0] + 1,
+            ]]
+
+    # XXX NOTE XXX: if >= 2 zeroed segments are found, there should
+    # ALWAYS be more then one zero-segment-abs-index-step-diff row
+    # in `absi_zdiff`, so loop through all such
+    # abs-index-step-diffs >1 (i.e. the entries of `absi_zdiff`)
+    # and add them as the "end index" entries for each segment.
+    # Then, iif NOT iterating the first such segment end, look back
+    # for the prior segments zero-segment start indext by relative
+    # indexing the `zero_t` frame by -1 and grabbing the abs index
+    # of what should be the prior zero-segment abs start index.
+    else:
+        # NOTE: since `absi_zdiff` will never have a row
+        # corresponding to the first zero-segment's row, we add it
+        # manually here.
         absi_zsegs.append([
-            max(frame_start, zero_start - imargin),
-            min(frame_end, zero_end + imargin),
+            max(
+                absi_zeros[0] - 1,
+                0,
+            ),
+            None,
         ])
 
+        # TODO: can we do it with vec ops?
+        for i, (
+            fi,  # frame index of zero-seg start
+            zseg_start_row,  # full row for ^
+        ) in enumerate(zip(
+            fi_zgaps,
+            fi_zseg_start_rows,
+        )):
+            assert (zseg_start_row == zero_t[fi]).all()
+            iabs: int = zseg_start_row['index'][0]
+            absi_zsegs.append([
+                iabs - 1,
+                None,  # backfilled on next iter
+            ])
+
+            # final iter case, backfill FINAL end iabs!
+            if (i + 1) == fi_zgaps.size:
+                absi_zsegs[-1][1] = absi_zeros[-1] + 1
+
+            # NOTE: only after the first segment (due to `.diff()`
+            # usage above) can we do a lookback to the prior
+            # segment's end row and determine it's abs index to
+            # retroactively insert to the prior
+            # `absi_zsegs[i-1][1]` entry Bo
+            last_end: int = absi_zsegs[i][1]
+            if last_end is None:
+                prev_zseg_row = zero_t[fi - 1]
+                absi_post_zseg = prev_zseg_row['index'][0] + 1
+                # XXX: MUST BACKFILL previous end iabs!
+                absi_zsegs[i][1] = absi_post_zseg
+
+        else:
+            if 0 < num_gaps < 2:
+                absi_zsegs[-1][1] = min(
+                    absi_zeros[-1] + 1,
+                    frame['index'][-1],
+                )
+
+            iabs_first: int = frame['index'][0]
+            for start, end in absi_zsegs:
+
+                ts_start: float = times[start - iabs_first]
+                ts_end: float = times[end - iabs_first]
+                if (
+                    (ts_start == 0 and not start == 0)
+                    or
+                    ts_end == 0
+                ):
+                    import pdbp
+                    pdbp.set_trace()
+
+                assert end
+                assert start < end
+
     log.warning(
         f'Frame has {len(absi_zsegs)} NULL GAPS!?\n'
         f'period: {period}\n'
@@ -366,12 +492,12 @@ def iter_null_segs(
     ],
     None,
 ]:
-    if null_segs is None:
-        null_segs = get_null_segs(
+    if not (
+        null_segs := get_null_segs(
             frame,
             period=timeframe,
         )
-    if not null_segs:
+    ):
         return
 
     absi_pairs_zsegs: list[list[float, float]]
@@ -483,7 +609,7 @@ def detect_time_gaps(
     # first select by any sample-period (in seconds unit) step size
     # greater then expected.
     step_gaps: pl.DataFrame = w_dts.filter(
-        pl.col('s_diff') > expect_period
+        pl.col('s_diff').abs() > expect_period
     )
 
     if gap_dt_unit is None:
@@ -512,22 +638,6 @@ def detect_time_gaps(
     )
 
 
-def detect_time_ordering_errors(
-    w_dts: pl.DataFrame,
-
-) -> pl.DataFrame:
-    '''
-    Return rows whose timestamp does not follow its predecessor.
-
-    Zero deltas identify duplicate timestamps and negative deltas
-    identify out-of-order rows. Neither is a missing-history gap.
-
-    '''
-    return w_dts.filter(
-        pl.col('s_diff') <= 0
-    )
-
-
 def detect_price_gaps(
     df: pl.DataFrame,
     gt_multiplier: float = 2.,
@@ -590,16 +700,11 @@ def dedupe(
 
     '''
     wdts: pl.DataFrame = with_dts(src_df)
-    ordering_errors = detect_time_ordering_errors(wdts)
-    if not ordering_errors.is_empty():
-        log.warning(
-            f'Found {ordering_errors.height} non-positive '
-            f'timestamp step(s) before normalization:\n'
-            f'{ordering_errors}'
-        )
+
+    deduped = wdts
 
     # remove duplicated datetime samples/sections
-    deduped: pl.DataFrame = src_df.unique(
+    deduped: pl.DataFrame = wdts.unique(
         # subset=['dt'],
         subset=['time'],
         maintain_order=True,
@@ -608,8 +713,9 @@ def dedupe(
     # maybe sort on any time field
     if sort:
         deduped = deduped.sort(by='time')
-
-    deduped = with_dts(deduped)
+        # TODO: detect out-of-order segments which were corrected!
+        # -[ ] report in log msg
+        # -[ ] possibly return segment sections which were moved?
 
     diff: int = (
         wdts.height

piker/tsp/_dedupe_smart.py

@@ -50,18 +50,8 @@ def dedupe_ohlcv_smart(
     )
 
     if dupes.is_empty():
-        deduped: pl.DataFrame = (
-            src_df.sort(time_col)
-            if sort
-            else src_df
-        )
-        return (
-            wdts,
-            with_dts(deduped),
-            0,
-            None,
-            None,
-        )
+        # No duplicates, return as-is
+        return (wdts, wdts, 0, None, None)
 
     # Analyze duplicate groups for validation
     dupe_analysis: pl.DataFrame = (
@@ -205,8 +195,6 @@ def dedupe_ohlcv_smart(
     if sort:
         deduped = deduped.sort(by=time_col)
 
-    deduped = with_dts(deduped.select(src_df.columns))
-
     diff: int = wdts.height - deduped.height
 
     return (

piker/tsp/_history.py

@@ -78,6 +78,7 @@ from piker.brokers._util import (
 )
 from piker.storage import TimeseriesNotFound
 from ._anal import (
+    dedupe,
     get_null_segs,
     iter_null_segs,
     Frame,
@@ -122,7 +123,6 @@ _default_rt_size: int = _days_worth * _secs_in_day
 # NOTE: start the append index in rt buffer such that 1 day's worth
 # can be appenened before overrun.
 _rt_buffer_start = int((_days_worth - 1) * _secs_in_day)
-_notify_timeout_s: float = 1
 
 
 def diff_history(
@@ -148,64 +148,6 @@ def diff_history(
         return array[times >= prepend_until_dt.timestamp()]
 
 
-def mk_storage_key(mkt: MktPair) -> str:
-    '''
-    Build the historical storage key for a market.
-
-    Always drop the source-asset token for non-currency-pair market
-    types. For historical reasons the table-key schema stores
-    `tsla.nasdaq.ib`, not `tsla/usd.nasdaq.ib`, while currency pairs
-    and crypto-settled futures retain both assets.
-
-    '''
-    if (
-        mkt.dst.atype not in {
-            'crypto',
-            'crypto_currency',
-            'fiat',  # a "forex pair"
-            'perpetual_future',  # stupid "perps" from cex land
-        }
-        and not (
-            mkt.src.atype == 'crypto_currency'
-            and
-            mkt.dst.atype == 'future'
-        )
-    ):
-        return mkt.get_fqme(
-            delim_char='',
-            without_src=True,
-        )
-
-    return mkt.get_fqme(delim_char='')
-
-
-async def notify_backfill(
-    sampler_stream: tractor.MsgStream,
-    mkt: MktPair,
-    timeframe: float,
-
-) -> None:
-    '''
-    Broadcast an SHM update without making teardown unbounded.
-
-    '''
-    with trio.move_on_after(
-        _notify_timeout_s,
-        shield=True,
-    ) as cs:
-        await sampler_stream.send({
-            'broadcast_all': {
-                'backfilling': (mkt.fqme, timeframe),
-            },
-        })
-
-    if cs.cancelled_caught:
-        log.warning(
-            f'Timed out broadcasting backfill for '
-            f'{timeframe}@{mkt.fqme}'
-        )
-
-
 async def shm_push_in_between(
     shm: ShmArray,
     to_push: np.ndarray,
@@ -260,11 +202,12 @@ async def maybe_fill_null_segments(
     mkt: MktPair,
     backfill_until_dt: datetime,
 
-    task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED,
+    task_status: TaskStatus[trio.Event] = trio.TASK_STATUS_IGNORED,
 
 ) -> list[Frame]:
 
-    task_status.started()
+    null_segs_detected = trio.Event()
+    task_status.started(null_segs_detected)
 
     frame: Frame = shm.array
 
@@ -365,6 +308,7 @@ async def maybe_fill_null_segments(
             await tractor.pause()
             raise
 
+    null_segs_detected.set()
     # RECHECK for more null-gaps
     frame: Frame = shm.array
     null_segs: tuple | None = get_null_segs(
@@ -463,13 +407,15 @@ async def start_backfill(
     storage: StorageClient|None = None,
     write_tsdb: bool = True,
 
-    task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED,
+    task_status: TaskStatus[tuple] = trio.TASK_STATUS_IGNORED,
 
 ) -> int:
 
-        # Let the caller place storage history while this task
-        # continues reverse retrieval in the owning nursery.
-        task_status.started()
+        # let caller unblock and deliver latest history frame
+        # and use to signal that backfilling the shm gap until
+        # the tsdb end is complete!
+        bf_done = trio.Event()
+        task_status.started(bf_done)
 
         # based on the sample step size, maybe load a certain amount history
         update_start_on_prepend: bool = False
@@ -589,15 +535,6 @@ async def start_backfill(
                 #     charge of solving such faults? yes, right?
                 return
 
-            if array.size == 0:
-                log.warning(
-                    f'Provider {mod.name!r} returned an empty frame\n'
-                    f'fqme: {mkt.fqme}\n'
-                    f'timeframe: {timeframe}\n'
-                    f'end_dt: {end_dt_param}\n'
-                )
-                break
-
             time: np.ndarray = array['time']
             assert (
                 time[0]
@@ -676,11 +613,10 @@ async def start_backfill(
                 )
                 # await tractor.pause()
 
-            to_store: np.ndarray = diff_history(
+            to_push = diff_history(
                 array,
                 prepend_until_dt=backfill_until_dt,
             )
-            to_push: np.ndarray = to_store
             ln: int = len(to_push)
             if ln:
                 log.info(
@@ -720,23 +656,6 @@ async def start_backfill(
                     )
                     break
 
-            if (
-                storage is not None
-                and
-                write_tsdb
-            ):
-                log.info(
-                    f'Writing {len(to_store)} frame to storage:\n'
-                    f'{next_start_dt} -> {last_start_dt}'
-                )
-                await storage.update_ohlcv(
-                    mk_storage_key(mkt),
-                    to_store,
-                    timeframe,
-                )
-
-            shm_full: bool = False
-
             # bail gracefully on shm allocation overrun/full
             # condition
             try:
@@ -747,6 +666,11 @@ async def start_backfill(
                     backfill_until_dt=backfill_until_dt,
                     update_start_on_prepend=update_start_on_prepend,
                 )
+                await sampler_stream.send({
+                    'broadcast_all': {
+                        'backfilling': (mkt.fqme, timeframe),
+                    },
+                })
 
                 # decrement next prepend point
                 next_prepend_index = next_prepend_index - ln
@@ -758,7 +682,7 @@ async def start_backfill(
                         f'Reached buffer start (index={next_prepend_index}), '
                         f'stopping backfill'
                     )
-                    shm_full = True
+                    break
 
             except ValueError as ve:
                 _ve = ve
@@ -773,7 +697,6 @@ async def start_backfill(
                     )
 
                 to_push = to_push[-next_prepend_index + 1:]
-                ln = len(to_push)
                 await shm_push_in_between(
                     shm,
                     to_push,
@@ -781,23 +704,89 @@ async def start_backfill(
                     backfill_until_dt=backfill_until_dt,
                     update_start_on_prepend=update_start_on_prepend,
                 )
-                next_prepend_index = next_prepend_index - ln
-                last_start_dt = next_start_dt
-                shm_full = True
+                await sampler_stream.send({
+                    'broadcast_all': {
+                        'backfilling': (mkt.fqme, timeframe),
+                    },
+                })
+
+                # XXX, can't push the entire frame? so
+                # push only the amount that can fit..
+                break
 
             log.info(
                 f'Shm pushed {ln} frame:\n'
                 f'{next_start_dt} -> {last_start_dt}'
             )
 
-            await notify_backfill(
-                sampler_stream,
-                mkt,
-                timeframe,
+            # FINALLY, maybe write immediately to the tsdb backend for
+            # long-term storage.
+            if (
+                storage is not None
+                and
+                write_tsdb
+            ):
+                log.info(
+                    f'Writing {ln} frame to storage:\n'
+                    f'{next_start_dt} -> {last_start_dt}'
                 )
 
-            if shm_full:
-                break
+                # NOTE, always drop the src asset token for
+                # non-currency-pair like market types (for now)
+                #
+                # THAT IS, for now our table key schema is NOT
+                # including the dst[/src] source asset token. SO,
+                # 'tsla.nasdaq.ib' over 'tsla/usd.nasdaq.ib' for
+                # historical reasons ONLY.
+                if (
+                    mkt.dst.atype not in {
+                        'crypto',
+                        'crypto_currency',
+                        'fiat',  # a "forex pair"
+                        'perpetual_future',  # stupid "perps" from cex land
+                    }
+                    and not (
+                        mkt.src.atype == 'crypto_currency'
+                        and
+                        mkt.dst.atype in {
+                            'future',
+                        }
+                    )
+                ):
+                    col_sym_key: str = mkt.get_fqme(
+                        delim_char='',
+                        without_src=True,
+                    )
+                else:
+                    col_sym_key: str = mkt.get_fqme(
+                        delim_char='',
+                    )
+
+                await storage.write_ohlcv(
+                    col_sym_key,
+                    shm.array,
+                    timeframe,
+                )
+                df: pl.DataFrame = await storage.as_df(
+                    fqme=mkt.fqme,
+                    period=timeframe,
+                    load_from_offline=False,
+                )
+                (
+                    wdts,
+                    deduped,
+                    diff,
+                ) = dedupe(df)
+                if diff:
+                    log.warning(
+                        f'Found {diff!r} duplicates in tsdb! '
+                        f'=> Overwriting with `deduped` data !! <=\n'
+                    )
+                    await storage.write_ohlcv(
+                        col_sym_key,
+                        deduped,
+                        timeframe,
+                    )
 
         else:
             # finally filled gap
@@ -811,6 +800,10 @@ async def start_backfill(
         # memory...
         # await sampler_stream.send('broadcast_all')
 
+        # short-circuit (for now)
+        bf_done.set()
+
+
 # NOTE: originally this was used to cope with a tsdb (marketstore)
 # which could not delivery very large frames of history over gRPC
 # (thanks goolag) due to corruption issues.
@@ -956,11 +949,11 @@ async def back_load_from_tsdb(
         # await sampler_stream.send('broadcast_all')
 
 
-async def query_latest_frame(
+async def push_latest_frame(
     # box-type only that should get packed with the datetime
     # objects received for the latest history frame
     dt_eps: list[DateTime, DateTime],
-    frames: list[np.ndarray],
+    shm: ShmArray,
     get_hist: Callable[
         [int, datetime, datetime],
         tuple[np.ndarray, str]
@@ -989,7 +982,7 @@ async def query_latest_frame(
             mr_start_dt,
             mr_end_dt,
         ])
-        frames.append(array)
+        task_status.started(dt_eps)
 
     # XXX: timeframe not supported for backend (since
     # above exception type), terminate immediately since
@@ -1003,38 +996,18 @@ async def query_latest_frame(
         # prolly tf not supported
         return None
 
-    task_status.started(dt_eps)
-
-    return dt_eps
-
-
-async def publish_latest_frame(
-    storage: StorageClient,
-    mkt: MktPair,
-    shm: ShmArray,
-    array: np.ndarray,
-    timeframe: float,
-
-) -> None:
-    '''
-    Persist then publish the most-recent provider frame.
-
-    '''
-    await storage.update_ohlcv(
-        mk_storage_key(mkt),
-        array,
-        timeframe,
-    )
-
-    # NOTE: on the first history, most recent history frame we
-    # PREPEND from the current shm ._last index. A gap may exist
-    # between its earliest datum and the latest loaded from the tsdb.
+    # NOTE: on the first history, most recent history
+    # frame we PREPEND from the current shm ._last index
+    # and thus a gap between the earliest datum loaded here
+    # and the latest loaded from the tsdb may exist!
     log.info(f'Pushing {array.size} to shm!')
     shm.push(
         array,
         prepend=True,  # append on first frame
     )
 
+    return dt_eps
+
 
 async def load_tsdb_hist(
     storage: StorageClient,
@@ -1120,15 +1093,14 @@ async def tsdb_backfill(
         # concurrently load the provider's most-recent-frame AND any
         # pre-existing tsdb history already saved in `piker` storage.
         dt_eps: list[DateTime, DateTime] = []
-        latest_frames: list[np.ndarray] = []
         async with (
             tractor.trionics.collapse_eg(),
             trio.open_nursery() as tn
         ):
             tn.start_soon(
-                query_latest_frame,
+                push_latest_frame,
                 dt_eps,
-                latest_frames,
+                shm,
                 get_hist,
                 timeframe,
                 config,
@@ -1139,15 +1111,6 @@ async def tsdb_backfill(
                 timeframe,
             )
 
-        if latest_frames:
-            await publish_latest_frame(
-                storage,
-                mkt,
-                shm,
-                latest_frames[0],
-                timeframe,
-            )
-
         # tell parent task to continue
         # TODO: really we'd want this the other way with the
         # tsdb load happening asap and the since the latest
@@ -1227,7 +1190,7 @@ async def tsdb_backfill(
                 trio.open_nursery() as tn,
             ):
 
-                await tn.start(
+                bf_done: trio.Event = await tn.start(
                     partial(
                         start_backfill,
                         get_hist=get_hist,
@@ -1247,6 +1210,8 @@ async def tsdb_backfill(
                         write_tsdb=True,
                     )
                 )
+                nulls_detected: trio.Event|None = None
+
                 if last_tsdb_dt is not None:
 
                     # calc the index from which the tsdb data should be
@@ -1323,7 +1288,7 @@ async def tsdb_backfill(
                     #    work PREVENTAVELY instead?
                     # -[ ] fill in non-zero epoch time values ALWAYS!
                     # await maybe_fill_null_segments(
-                    await tn.start(partial(
+                    nulls_detected: trio.Event = await tn.start(partial(
                         maybe_fill_null_segments,
 
                         shm=shm,
@@ -1336,6 +1301,11 @@ async def tsdb_backfill(
 
             # 2nd nursery END
 
+            # TODO: who would want to?
+            if nulls_detected:
+                await nulls_detected.wait()
+
+            await bf_done.wait()
         # TODO: maybe start history anal and load missing "history
         # gaps" via backend..
 

plans/opencode/gap-checker-backfiller-repair.md

@@ -1,436 +0,0 @@
-# Gap Checker and Backfiller Repair
-
-## Purpose
-
-Continue the time-series gap checker and backfiller work
-from PRs/issues `#62`, `#71`, and `#75`, prioritizing data
-correctness and crash safety before annotation UX or
-rendering performance.
-
-Source baseline:
-
-- branch: `flake_update`
-- commit: `c6ec3d41`
-- gap handoff:
-  `notes_to_self/claudy/gapsannotator_sesh.md`
-- primary code:
-  `piker/tsp/_anal.py`, `piker/tsp/_history.py`,
-  `piker/storage/nativedb.py`, and
-  `piker/tsp/_annotate.py`
-
-## Goals
-
-1. Make gap detection operate on sorted, unique,
-   validated OHLCV frames with fresh derived columns.
-2. Make backfill completion, frame boundaries, and null
-   repair deterministic under exhaustion and cancellation.
-3. Prevent a bounded or transient SHM view from
-   destructively replacing older persisted history.
-4. Persist normalized history through crash-safe atomic
-   file replacement.
-5. Distinguish expected venue closures from suspicious
-   missing data without production breakpoints.
-6. Generate correct annotation specs before optimizing
-   the remaining client-side Polars bottleneck.
-
-## Confirmed Baseline Defects
-
-These are current-code findings to characterize in tests,
-not assumptions inferred from historical task state.
-
-### Gap normalization
-
-- `dedupe()` computes `time_prev`, `s_diff`, and
-  `dt_diff` before sorting/deduplication, then returns
-  stale derived columns in the normalized frame
-  (`piker/tsp/_anal.py:683-729`).
-- Negative/out-of-order deltas are treated as ordinary
-  gaps because `detect_time_gaps()` filters on
-  `abs(s_diff)` (`piker/tsp/_anal.py:582-638`).
-- `get_null_segs()` drops single zero rows and its Polars
-  path does not define all state used by the shared path
-  (`piker/tsp/_anal.py:227-471`).
-- `iter_null_segs()` ignores a supplied `null_segs`
-  result and rescans the frame (`piker/tsp/_anal.py:474-546`).
-
-### Backfill control and boundaries
-
-- `DataUnavailable` returns before `bf_done.set()`, while
-  `tsdb_backfill()` later waits forever on that event
-  (`piker/tsp/_history.py:516-536,1193-1212,1302-1308`).
-- Empty arrays are indexed before the nominal null-frame
-  branch can handle them (`piker/tsp/_history.py:538-601`).
-- Provider frame seams have no explicit inclusive/exclusive
-  contract, allowing duplicate adjacent timestamps in SHM.
-- Cancellation may occur after SHM mutation but before
-  cursor and persistence bookkeeping
-  (`piker/tsp/_history.py:661-789`).
-
-### Null repair
-
-- Null query length is derived from an absolute endpoint,
-  not the null-span width (`piker/tsp/_history.py:278-287`).
-- Fallback repair includes and mutates the valid leading
-  boundary row (`piker/tsp/_history.py:345-367`).
-- Null repair and reverse backfill share one provider
-  client concurrently without enforcing the backend's
-  declared request limits.
-
-### Persistence
-
-- Every iteration passes bounded `shm.array` to
-  `write_ohlcv()`, while NativeDB replaces the complete
-  parquet file. Older rows that do not fit in SHM can be
-  lost (`piker/tsp/_history.py:722-789`,
-  `piker/storage/nativedb.py:305-368`).
-- NativeDB writes directly to the final path and updates
-  its cache before the write succeeds; interruption can
-  leave disk and cache partial or inconsistent.
-- Intentional zero-filled SHM space may be persisted while
-  null repair and reverse backfill run concurrently.
-
-### Annotation contract
-
-- `markup_gaps()` performs a full Polars filter per gap and
-  remains the measured client bottleneck
-  (`piker/tsp/_annotate.py:139-217`).
-- Its missing-prior-row fallback selects the current row,
-  and `copysign(1, 0)` makes the flat-price branch
-  unreachable (`piker/tsp/_annotate.py:166-239`).
-- Remote/reposition timestamp lookups index
-  `searchsorted()` results before excluding values equal
-  to `len(array)`.
-
-## Design Invariants
-
-All implementation phases preserve these rules:
-
-1. Persisted provider rows have nonzero, finite timestamps.
-2. Normalized timestamps are strictly increasing and
-   unique.
-3. Time-derived columns are computed only after final
-   sorting and duplicate resolution.
-4. A detected gap explicitly carries its left and right
-   endpoints; consumers do not infer the left row from
-   `index - 1`.
-5. Provider frame metadata equals the normalized array's
-   actual first and last timestamps.
-6. Internal middleware uses one documented boundary
-   convention; provider adapters translate into it.
-7. Expected sparse venue closures are represented by
-   timestamp distance, not persisted zero or synthetic
-   bars.
-8. Storage merges preserve all unaffected older rows.
-9. A storage commit becomes visible atomically, and cache
-   state changes only after disk commit succeeds.
-10. Rendering or annotation failure cannot change the
-    checker or persistence result.
-
-## Phase 0: Deterministic Characterization
-
-Add synthetic OHLCV fixtures based on
-`piker.data._source.def_iohlcv_fields`; avoid brokers,
-networking, real actor trees, and Qt in the core suite.
-
-### Analysis tests
-
-Create `tests/test_tsp_analysis.py` covering:
-
-- uniform 1-second and 60-second series;
-- one and multiple positive gaps;
-- duplicate plus out-of-order timestamps;
-- stale derived columns after sort/dedupe;
-- zero, one, and multiple null segments;
-- nulls at frame boundaries and nonzero absolute indexes;
-- NumPy and Polars inputs;
-- smart-dedupe conflict cases, including differing closes.
-
-### Backfill tests
-
-Create `tests/test_history_backfill.py` with scripted
-`get_hist`, fake SHM, fake sampler stream, and recording
-storage implementations. Characterize:
-
-- `DataUnavailable` completion without a hang;
-- a genuinely empty frame;
-- exact adjacency, overlap, and a positive storage gap;
-- inclusive provider seam duplication;
-- one-row and two-row startup history;
-- cancellation after query, after SHM mutation, and
-  before persistence;
-- one and multiple null rows with boundary preservation.
-
-### Storage tests
-
-Create `tests/test_storage_nativedb.py` using `tmp_path`:
-
-- NumPy and Polars round trips;
-- preservation of history older than the SHM window;
-- duplicate conflict policy;
-- write failure leaves the prior parquet readable;
-- cache changes only after a successful commit;
-- separate FQME/timeframe series do not interfere.
-
-### Annotation tests
-
-Create `tests/test_gap_annotations.py` around a fake
-`AnnotCtl`, without Qt initially:
-
-- left-close/right-open rectangle endpoints;
-- up, down, and flat price direction;
-- first-row and missing-prior-row behavior;
-- empty gaps and out-of-range timestamps;
-- deterministic batched rect/arrow/text payloads.
-
-Gate: every confirmed defect above has a focused failing
-test or a written reason it requires a later integration
-test.
-
-## Phase 1: Canonical Frame and Gap Model
-
-Refactor `piker/tsp/_anal.py` around one normalization
-order:
-
-1. validate required columns and finite timestamps;
-2. sort by `time`;
-3. resolve duplicate timestamps under an explicit policy;
-4. recompute all datetime/delta columns with `with_dts()`;
-5. reject/report non-positive ordering deltas;
-6. detect positive sample-step gaps.
-
-Keep the first implementation small: a normalized Polars
-frame may remain the contract, but each gap row must carry
-explicit `time_prev`, `index_prev`, `close_prev`, current
-`time/index/open`, duration, and missing-sample count.
-
-Rewrite null grouping from extracted index/time arrays so
-the NumPy and Polars paths share one implementation. Treat
-a single null row as a normal one-row segment and honor a
-precomputed `null_segs` argument without rescanning.
-
-Gate: analysis tests pass and no deduped/sorted frame
-retains stale delta columns.
-
-## Phase 2: Structured Backfill Completion
-
-Refactor `start_backfill()` and `tsdb_backfill()` before
-changing persistence semantics:
-
-- remove completion events that are awaited only after the
-  owning nursery has already joined its children, or make
-  completion a guaranteed `finally` outcome;
-- represent `completed`, `provider_exhausted`,
-  `cancelled`, and `failed` as explicit results;
-- check empty arrays before indexing endpoints;
-- normalize each provider frame and recompute endpoint
-  metadata from the normalized data;
-- define and enforce a disjoint seam between each new
-  reverse frame and the already-published frame;
-- update cursor state immediately with the SHM mutation,
-  before any optional notification checkpoint;
-- serialize reverse backfill and null repair unless a
-  backend capability explicitly permits concurrent
-  history requests;
-- enforce declared backend request/rate limits.
-
-Prefer structured task return values over actor-local
-events whose only consumer is the parent task.
-
-Gate: exhaustion and cancellation tests terminate within
-their timeout, and SHM timestamps remain strictly
-increasing across all frame seams.
-
-## Phase 3: Non-Destructive Atomic NativeDB Writes
-
-Stop treating the bounded SHM view as the full database.
-
-Add a NativeDB merge/update operation that:
-
-- accepts only the normalized provider frame being
-  committed;
-- loads or uses the cached persisted frame;
-- merges by timestamp while preserving unaffected rows;
-- uses the historical provider bar as authoritative at a
-  closed-bar seam;
-- validates ordering, uniqueness, schema, and nonzero
-  timestamps before writing;
-- writes to a same-directory temporary parquet;
-- reopens and validates the temporary file;
-- atomically replaces the final path;
-- updates `_dfs` only after replacement succeeds;
-- serializes writers per `(fqme, timeframe)`.
-
-Backfill commits the validated frame/delta, never a live
-`shm.array` snapshot containing mutable current bars or
-intentional zero space.
-
-Gate: a persisted series can only grow backward or replace
-explicitly overlapping timestamps; its earliest timestamp
-never moves forward during ordinary backfill.
-
-## Phase 4: Exact Null Repair
-
-Use the canonical null spans from Phase 1:
-
-- query using explicit boundary timestamps;
-- trim provider responses to the actual missing interval;
-- write only null rows, never either valid boundary row;
-- validate repaired timestamps and OHLC fields before
-  publication or persistence;
-- leave expected venue closures sparse;
-- do not persist synthetic flat bars without a durable
-  quality marker and an explicit schema decision.
-
-If provider data is unavailable, retain and report the
-missing interval instead of silently converting it into
-apparently valid history.
-
-Gate: single and multi-row repairs preserve both boundary
-bars and either eliminate exactly the target nulls or
-return an unresolved-gap result.
-
-## Phase 5: Venue and Provider Classification
-
-Apply gap classification after canonical detection rather
-than embedding it in individual warning paths.
-
-For IB:
-
-- correct expiry comparison and weekday constants;
-- classify internal positive gaps as well as trailing
-  request-coverage gaps;
-- require closure interval alignment, not merely the
-  presence of a weekend/holiday somewhere in the gap;
-- cover overnight sessions, weekends, holidays, DST,
-  active/expired futures, and unknown calendars;
-- replace production `breakpoint()` calls with structured
-  data-quality results/errors.
-
-Provider adapters must return normalized endpoint metadata
-and translate their native inclusive/exclusive APIs into
-the middleware boundary contract.
-
-Gate: every gap is classified as expected closure,
-suspicious missing data, provider exhaustion, or invalid
-ordering; no production path pauses interactively.
-
-## Phase 6: Annotation Correctness and Performance
-
-Extract a synchronous, Qt-free spec builder from
-`markup_gaps()`:
-
-```python
-build_gap_annotation_specs(frame, gaps)
-```
-
-It consumes the explicit gap endpoints from Phase 1 and:
-
-- extracts needed columns once;
-- replaces per-gap `wdts.filter()` with a vectorized join
-  or indexed lookup;
-- handles equal prices explicitly;
-- preserves previous-close/current-open semantics;
-- styles expected closures separately from suspicious
-  gaps;
-- emits deterministic rect/arrow/text batches.
-
-Then fix remote lookup/reposition bounds by masking
-`searchsorted()` results before indexing and enforce both
-FQME and timeframe ownership during redraw.
-
-Benchmark the same 1,285-gap workload from the Claude
-handoff. Treat these as local benchmark goals, not portable
-CI limits:
-
-- client spec build under 20 ms;
-- server preparation under 10 ms;
-- one composite gap graphics item;
-- no full-frame Polars filter per gap;
-- total local creation near or below 150 ms.
-
-Gate: annotation contract tests pass before performance
-numbers are accepted.
-
-## Phase 7: Crash and Provider Qualification
-
-Run deterministic tests first, then the historical manual
-matrix from issue `#62` on disposable storage:
-
-- fresh full backfill;
-- append to existing history;
-- graceful Ctrl-C;
-- network interruption;
-- SIGTERM and SIGKILL during query and parquet commit;
-- restart and revalidate after every interruption.
-
-Start with Binance/Kraken/Kucoin, then run IB only with an
-available account and gateway. Record before/after earliest
-timestamp, latest timestamp, row count, unique count, null
-count, non-positive deltas, and parquet readability.
-
-## Verification Commands
-
-Use an outer process timeout because actor teardown has a
-known second-runtime wedge:
-
-```bash
-timeout -k 5 30 python -m pytest -q \
-  tests/test_tsp_analysis.py \
-  tests/test_storage_nativedb.py
-
-timeout -k 5 60 python -m pytest -q \
-  tests/test_history_backfill.py
-
-timeout -k 5 60 python -m pytest -q \
-  tests/test_gap_annotations.py
-
-CI=1 timeout -k 20 120 python -m pytest -q -rs \
-  tests/test_feeds.py
-```
-
-Run live/network tests separately from deterministic tests.
-Do not automate `piker store anal` or `piker store ldshm`:
-both may pause interactively or mutate attached data.
-
-## Commit Boundaries
-
-Keep implementation reviewable in this order:
-
-1. synthetic fixtures and analysis regressions;
-2. canonical normalization and null grouping;
-3. backfill completion and frame-boundary fixes;
-4. NativeDB merge and atomic persistence;
-5. exact null repair;
-6. IB/provider gap classification;
-7. annotation contract fixes and vectorization;
-8. qualification notes and documentation.
-
-Each boundary must leave deterministic tests green. Do not
-mix renderer optimization into storage/backfill commits.
-
-## Deferred
-
-The following remain follow-ups until the correctness
-gates above pass:
-
-- chart context-menu/manual piecewise refill UX;
-- automatic checker integration into every live chart;
-- compact columnar annotation IPC;
-- persistent synthetic-bar quality metadata/schema;
-- full Deribit history repair and broad provider cleanup;
-- Marketstore behavior changes;
-- long-term incremental parquet partitioning.
-
-## Source References
-
-- `gitea/62.md`: workaround-era backfill and manual test
-  history
-- `gitea/71.md`: IB venue closure behavior and pending
-  account verification
-- `gitea/75.md`: gap annotation, storage races, and
-  backfiller follow-ups
-- `notes_to_self/claudy/gapsannotator_sesh.md`: current
-  annotation benchmark and client bottleneck
-- https://github.com/pikers/piker/pull/486
-- https://github.com/pikers/piker/pull/446
-- https://github.com/pikers/piker/issues/536

plans/opencode/known-fqme-backfill-qualification.md

@@ -1,221 +0,0 @@
-# Known-FQME Backfill Qualification
-
-## Purpose
-
-Move `backfiller_deep_fixes` into the main checkout, reproduce the
-historical backfill failures against one known-problem market, and turn
-every confirmed failure into deterministic regression coverage before
-expanding live qualification.
-
-This is a correctness-debugging pass, not a broad provider smoke test.
-One FQME and one timeframe remain fixed until their original failure is
-understood, covered, repaired, and requalified.
-
-## Current Repository State
-
-The branch is currently checked out at:
-
-`/home/goodboy/repos/piker/.claude/wkts/backfiller_deep_fixes`
-
-Its head is `10e0ced7`. The linked worktree has only generated,
-untracked commit-message metadata.
-
-The main checkout is on `flake_update` at `c6ec3d41` and has staged and
-untracked user work. Do not switch branches, stash, remove files, or
-close the linked worktree until that state has been explicitly
-preserved by the user.
-
-## Phase 1: Transfer Branch Ownership Safely
-
-1. Review the main checkout's staged and untracked state.
-2. Preserve it using the user's chosen mechanism: finish and commit it,
-   move it to another branch/worktree, or explicitly create a stash that
-   includes the required untracked files.
-3. Review the linked worktree's generated `.claude/skills/commit-msg/`
-   artifacts. Archive or discard them explicitly; they are not branch
-   source.
-4. Remove the linked `backfiller_deep_fixes` worktree while keeping its
-   exact branch.
-5. Switch the main checkout to `backfiller_deep_fixes`.
-6. Verify all entry points import from the main checkout and resolve to
-   head `10e0ced7` or its descendant.
-7. Run the 31-test deterministic analysis/backfill/storage set before
-   booting any daemon.
-
-Gate: the main checkout owns `backfiller_deep_fixes`, tracked status is
-clean, user work from `flake_update` is preserved, and deterministic
-tests pass.
-
-## Phase 2: Record The Known Failure
-
-Capture these details for the selected case:
-
-- provider and exact FQME;
-- timeframe, initially `60` or `1` seconds;
-- known-bad date/time range and timezone;
-- original symptom: destructive truncation, duplicate seam, null rows,
-  out-of-order timestamps, hang, or another observed failure;
-- command or chart workflow that exposed it;
-- whether a known-bad parquet already exists;
-- expected venue closure or continuous-session behavior;
-- credentials, gateway, and network requirements;
-- smallest reproducible provider response or persisted frame.
-
-Do not start with multiple symbols, providers, or timeframes. Preserve a
-checksum of any known-bad parquet before copying it into disposable
-storage.
-
-Gate: one case has an explicit expected result and enough source data to
-distinguish a real provider/venue gap from a piker persistence defect.
-
-## Phase 3: Build A Disposable Runtime
-
-Use the main checkout's Python environment and a dedicated XDG config
-root. The `pikerd` process and every client process must inherit the same
-config root and import the main checkout.
-
-The disposable directory contains only:
-
-- a minimal `conf.toml`;
-- credentials required for the selected provider;
-- a dedicated `nativedb/` directory;
-- an optional copy of the single known-bad parquet;
-- captured logs and before/after audit output.
-
-Ensure no older `pikerd` is listening on the selected registry address.
-Verify `piker.__file__`, the branch, and the commit before starting the
-daemon.
-
-Gate: deleting the disposable directory cannot affect normal user
-configuration, accounting data, or production history.
-
-## Phase 4: Establish A Read-Only Baseline
-
-Before backfill, record:
-
-- parquet checksum and readability;
-- earliest and latest timestamp;
-- total and unique timestamp counts;
-- duplicate count;
-- zero and non-finite timestamp counts;
-- non-positive timestamp deltas;
-- null OHLCV counts;
-- contiguous persisted index status;
-- positive sample gaps with explicit endpoints.
-
-Use a read-only audit helper or focused Python test. Do not automate
-`piker store anal` or `piker store ldshm`; both can pause or mutate
-attached data.
-
-Gate: the original symptom is visible in saved evidence, or the plan
-records why a fresh provider fetch is required to reproduce it.
-
-## Phase 5: Reproduce One Lifecycle At A Time
-
-Run in this order:
-
-1. Load the known-bad persisted history without a provider write.
-2. Start `pikerd` and request the chosen FQME/timeframe only.
-3. Observe startup latest-frame query, pre-update TSDB snapshot, reverse
-   backfill, provider-delta commit, SHM publication, and notification.
-4. Stop cleanly and rerun the read-only audit.
-5. Restart and append to the same disposable history.
-6. Compare earliest/latest timestamps, row counts, conflicts, and gaps
-   against the baseline.
-
-Capture logs around every `update_ohlcv()` call, provider frame boundary,
-SHM truncation, lock acquisition, temporary parquet, replacement, and
-cache publication.
-
-Gate: the run either preserves every unaffected row and normalizes the
-known defect, or yields a bounded failure with enough data for a test.
-
-## Phase 6: Convert Failures Into Tests
-
-For each reproduced defect:
-
-1. Save the smallest provider frames and persisted rows needed to
-   reproduce it. Prefer synthetic arrays; use a trimmed fixture only
-   when provider-specific boundaries matter.
-2. Write a failing deterministic test before changing production code.
-3. Assert exact timestamps, endpoint policy, persisted rows, and state
-   transition ordering rather than only asserting completion.
-4. Keep actor-free tests in the existing focused modules where possible.
-5. Add subprocess integration tests only for process death, advisory
-   locks, or atomic filesystem behavior that cannot be represented in
-   one Trio runtime.
-6. Apply the smallest production fix and rerun both the focused test and
-   the complete deterministic set.
-7. Rerun the exact manual FQME lifecycle that found the failure.
-
-Potential deterministic modules:
-
-- `tests/test_tsp_analysis.py` for normalization and classification;
-- `tests/test_history_backfill.py` for provider/SHM/storage ordering;
-- `tests/test_storage_nativedb.py` for merge and atomic durability;
-- a new subprocess fault module for kill/restart and cross-process locks;
-- provider-specific fixtures only when generic frames cannot reproduce
-  the behavior.
-
-Gate: no production fix lands without a failing-before, passing-after
-test or a written explanation for an unavoidable live-only condition.
-
-## Phase 7: Fault Qualification
-
-After the normal lifecycle is stable, exercise:
-
-1. graceful Ctrl-C during provider query and storage update;
-2. SIGTERM at deterministic query/write barriers;
-3. SIGKILL after temp write, before replace, and after replace;
-4. provider disconnect and recovery;
-5. concurrent NativeDB writers for the same series;
-6. restart after every interruption;
-7. actual sampler backpressure and actor teardown;
-8. MarketStore separately from NativeDB.
-
-Use explicit barriers or failpoints rather than timing sleeps whenever
-the failure position matters. Repeat the read-only audit after every
-restart.
-
-Gate: the final parquet is always old-valid or new-valid, unaffected
-history remains present, indexes are contiguous, cache agrees with the
-visible file, and shutdown remains bounded.
-
-## Expansion Order
-
-Only after the chosen FQME passes:
-
-1. qualify its second timeframe;
-2. qualify another continuous-session provider;
-3. run Binance, Kraken, and Kucoin cases;
-4. test IB with an available account and gateway;
-5. add expected venue-closure classification;
-6. widen to the remaining manual matrix.
-
-## Commit Boundaries
-
-Keep follow-up commits narrow:
-
-1. captured fixture and failing regression;
-2. minimal analysis/backfill/storage correction;
-3. subprocess fault-injection harness;
-4. provider-specific adapter correction;
-5. qualification notes and reproducible audit output.
-
-Every boundary must leave deterministic tests green. Do not combine
-annotation performance, unrelated providers, or chart UX with the known
-FQME debugging loop.
-
-## Exit Criteria
-
-The selected FQME/timeframe is complete when:
-
-- the original bug reproduces from preserved evidence;
-- a deterministic regression covers it where technically possible;
-- fresh and append backfills preserve unaffected history;
-- provider seam conflicts follow the documented policy;
-- no zero, duplicate, non-finite, or non-positive timestamps persist;
-- indexes remain contiguous;
-- clean shutdown and restart preserve readable parquet;
-- failure injection leaves an old-valid or new-valid file;
-- remaining live-only risks are explicitly recorded.

tests/test_history_backfill.py

@@ -1,221 +0,0 @@
-'''
-Deterministic history-backfill regressions.
-
-'''
-from functools import partial
-from types import SimpleNamespace
-
-import numpy as np
-from pendulum import (
-    datetime,
-    from_timestamp,
-)
-import pytest
-import trio
-
-from piker.brokers import DataUnavailable
-from piker.data._source import def_iohlcv_fields
-from piker.tsp._history import (
-    notify_backfill,
-    publish_latest_frame,
-    start_backfill,
-)
-
-
-def run_exhausted_backfill(get_hist) -> None:
-    '''
-    Run a provider-exhaustion case without actor services.
-
-    '''
-    async def main() -> None:
-        with trio.fail_after(0.5):
-            async with trio.open_nursery() as nursery:
-                await nursery.start(partial(
-                    start_backfill,
-                    get_hist=get_hist,
-                    def_frame_duration=None,
-                    mod=SimpleNamespace(name='fake'),
-                    mkt=SimpleNamespace(fqme='x.fake'),
-                    shm=object(),
-                    timeframe=60,
-                    backfill_from_shm_index=100,
-                    backfill_from_dt=datetime(2026, 1, 2),
-                    sampler_stream=object(),
-                    backfill_until_dt=datetime(2026, 1, 1),
-                    storage=None,
-                    write_tsdb=False,
-                ))
-
-    trio.run(main)
-
-
-def test_data_unavailable_completes() -> None:
-    '''
-    Provider exhaustion exits without orphaned completion waits.
-
-    '''
-    async def get_hist(*args, **kwargs):
-        raise DataUnavailable('history exhausted')
-
-    run_exhausted_backfill(get_hist)
-
-
-def test_empty_frame_completes() -> None:
-    '''
-    An empty provider frame is checked before endpoint indexing.
-
-    '''
-    frame = np.empty(
-        0,
-        dtype=np.dtype(def_iohlcv_fields),
-    )
-
-    async def get_hist(*args, **kwargs):
-        end_dt = kwargs['end_dt']
-        return frame, end_dt, end_dt
-
-    run_exhausted_backfill(get_hist)
-
-
-def test_storage_receives_full_provider_delta() -> None:
-    '''
-    SHM capacity truncation does not truncate durable history.
-
-    '''
-    frame = np.zeros(
-        2,
-        dtype=np.dtype(def_iohlcv_fields),
-    )
-    frame['index'] = [0, 1]
-    frame['time'] = [60, 120]
-
-    events: list[str] = []
-
-    class Shm:
-        def __init__(self) -> None:
-            self.pushed: list[np.ndarray] = []
-
-        def push(self, array, **kwargs) -> None:
-            self.pushed.append(array.copy())
-            events.append('shm')
-
-    class Sampler:
-        async def send(self, msg) -> None:
-            events.append('sampler')
-
-    class Storage:
-        def __init__(self) -> None:
-            self.frames: list[np.ndarray] = []
-
-        async def update_ohlcv(
-            self,
-            fqme: str,
-            ohlcv: np.ndarray,
-            timeframe: int,
-        ) -> None:
-            self.frames.append(ohlcv.copy())
-            events.append('storage')
-
-    shm = Shm()
-    storage = Storage()
-    mkt = SimpleNamespace(
-        fqme='x.test',
-        dst=SimpleNamespace(atype='crypto'),
-        src=SimpleNamespace(atype='crypto_currency'),
-        get_fqme=lambda **kwargs: 'x.test',
-    )
-
-    async def get_hist(*args, **kwargs):
-        return (
-            frame,
-            from_timestamp(60),
-            from_timestamp(120),
-        )
-
-    async def main() -> None:
-        with trio.fail_after(0.5):
-            await start_backfill(
-                get_hist=get_hist,
-                def_frame_duration=None,
-                mod=SimpleNamespace(name='fake'),
-                mkt=mkt,
-                shm=shm,
-                timeframe=60,
-                backfill_from_shm_index=1,
-                backfill_from_dt=from_timestamp(180),
-                sampler_stream=Sampler(),
-                backfill_until_dt=from_timestamp(60),
-                storage=storage,
-                write_tsdb=True,
-            )
-
-    trio.run(main)
-    assert shm.pushed[0]['time'].tolist() == [120]
-    assert storage.frames[0]['time'].tolist() == [60, 120]
-    assert events == ['storage', 'shm', 'sampler']
-
-
-def test_latest_frame_is_persisted_before_shm() -> None:
-    '''
-    Startup persists the most-recent provider frame before publication.
-
-    '''
-    frame = np.zeros(
-        2,
-        dtype=np.dtype(def_iohlcv_fields),
-    )
-    frame['time'] = [60, 120]
-    events: list[str] = []
-
-    class Storage:
-        async def update_ohlcv(self, *args) -> None:
-            events.append('storage')
-
-    class Shm:
-        def push(self, array, **kwargs) -> None:
-            events.append('shm')
-
-    async def main() -> None:
-        with trio.fail_after(0.5):
-            await publish_latest_frame(
-                storage=Storage(),
-                mkt=SimpleNamespace(
-                    fqme='x.test',
-                    dst=SimpleNamespace(atype='crypto'),
-                    src=SimpleNamespace(atype='crypto_currency'),
-                    get_fqme=lambda **kwargs: 'x.test',
-                ),
-                shm=Shm(),
-                array=frame,
-                timeframe=60,
-            )
-
-    trio.run(main)
-    assert events == ['storage', 'shm']
-
-
-def test_backfill_notification_timeout_is_bounded(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    '''
-    A wedged sampler can not indefinitely shield actor teardown.
-
-    '''
-    monkeypatch.setattr(
-        'piker.tsp._history._notify_timeout_s',
-        0.01,
-    )
-
-    class Sampler:
-        async def send(self, msg) -> None:
-            await trio.sleep_forever()
-
-    async def main() -> None:
-        with trio.fail_after(0.1):
-            await notify_backfill(
-                Sampler(),
-                SimpleNamespace(fqme='x.test'),
-                60,
-            )
-
-    trio.run(main)

tests/test_storage_nativedb.py

@@ -1,364 +0,0 @@
-'''
-NativeDB durability and history-preservation regressions.
-
-'''
-from fcntl import (
-    flock,
-    LOCK_EX,
-    LOCK_UN,
-)
-import os
-from pathlib import Path
-
-import numpy as np
-import polars as pl
-import pytest
-import trio
-
-from piker import tsp
-from piker.data._source import def_iohlcv_fields
-from piker.storage.nativedb import NativeStorageClient
-
-
-def mk_ohlcv(
-    times: tuple[float, ...],
-    closes: tuple[float, ...] | None = None,
-
-) -> np.ndarray:
-    '''
-    Build a minimal structured OHLCV frame.
-
-    '''
-    array = np.zeros(
-        len(times),
-        dtype=np.dtype(def_iohlcv_fields),
-    )
-    array['index'] = np.arange(len(times))
-    array['time'] = times
-    array['close'] = closes or times
-    return array
-
-
-def run(coro) -> None:
-    '''
-    Run a NativeDB operation with a bounded Trio clock.
-
-    '''
-    async def main() -> None:
-        with trio.fail_after(1):
-            await coro
-
-    trio.run(main)
-
-
-@pytest.mark.parametrize('use_polars', [False, True])
-def test_numpy_and_polars_round_trip(
-    tmp_path: Path,
-    use_polars: bool,
-) -> None:
-    '''
-    Both supported frame types survive durable serialization.
-
-    '''
-    client = NativeStorageClient(tmp_path)
-    array = mk_ohlcv(
-        (60, 120, 180),
-        (1, 2, 3),
-    )
-    payload = tsp.np2pl(array) if use_polars else array
-
-    run(client.write_ohlcv('x.test', payload, 60))
-    loaded = trio.run(client.read_ohlcv, 'x.test', 60)
-
-    assert loaded['time'].tolist() == [60, 120, 180]
-    assert loaded['close'].tolist() == [1, 2, 3]
-
-
-def test_update_preserves_history_and_resolves_conflicts(
-    tmp_path: Path,
-) -> None:
-    '''
-    Incremental writes retain old rows and prefer incoming bars.
-
-    '''
-    client = NativeStorageClient(tmp_path)
-    old = mk_ohlcv(
-        (60, 120, 180),
-        (1, 2, 3),
-    )
-    incoming = mk_ohlcv(
-        (180, 240),
-        (30, 4),
-    )
-
-    run(client.write_ohlcv('x.test', old, 60))
-    run(client.update_ohlcv('x.test', incoming, 60))
-    stored = pl.read_parquet(client.mk_path('x.test', 60))
-
-    assert stored['time'].to_list() == [60, 120, 180, 240]
-    assert stored['close'].to_list() == [1, 2, 30, 4]
-    assert stored['index'].to_list() == [0, 1, 2, 3]
-
-
-def test_write_ohlcv_remains_an_explicit_replacement(
-    tmp_path: Path,
-) -> None:
-    '''
-    Full repair writes can intentionally remove persisted rows.
-
-    '''
-    client = NativeStorageClient(tmp_path)
-    old = mk_ohlcv((60, 120, 180))
-    replacement = mk_ohlcv((120, 180))
-
-    run(client.write_ohlcv('x.test', old, 60))
-    run(client.write_ohlcv('x.test', replacement, 60))
-    stored = pl.read_parquet(client.mk_path('x.test', 60))
-
-    assert stored['time'].to_list() == [120, 180]
-
-
-def test_failed_write_preserves_file_and_cache(
-    tmp_path: Path,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    '''
-    A failed parquet write leaves durable and cached data intact.
-
-    '''
-    client = NativeStorageClient(tmp_path)
-    old = mk_ohlcv((60, 120))
-    incoming = mk_ohlcv((180,))
-    run(client.write_ohlcv('x.test', old, 60))
-
-    def fail_write(
-        df: pl.DataFrame,
-        file: Path,
-        *args,
-        **kwargs,
-    ) -> None:
-        Path(file).write_bytes(b'partial parquet')
-        raise OSError('simulated write failure')
-
-    monkeypatch.setattr(
-        pl.DataFrame,
-        'write_parquet',
-        fail_write,
-    )
-    with pytest.raises(OSError, match='simulated write failure'):
-        run(client.update_ohlcv('x.test', incoming, 60))
-
-    stored = pl.read_parquet(client.mk_path('x.test', 60))
-    cached = trio.run(client.as_df, 'x.test', 60, False)
-    assert stored['time'].to_list() == [60, 120]
-    assert cached['time'].to_list() == [60, 120]
-    assert not list(tmp_path.glob('*.tmp'))
-
-
-def test_invalid_temporary_parquet_is_not_committed(
-    tmp_path: Path,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    '''
-    The reopened candidate must validate before atomic replacement.
-
-    '''
-    client = NativeStorageClient(tmp_path)
-    old = mk_ohlcv((60, 120))
-    run(client.write_ohlcv('x.test', old, 60))
-    real_read = pl.read_parquet
-
-    def corrupt_candidate(source, *args, **kwargs):
-        if Path(source).suffix == '.tmp':
-            return pl.DataFrame({'time': [180]})
-        return real_read(source, *args, **kwargs)
-
-    monkeypatch.setattr(pl, 'read_parquet', corrupt_candidate)
-    with pytest.raises(ValueError, match='missing columns'):
-        run(client.update_ohlcv(
-            'x.test',
-            mk_ohlcv((180,)),
-            60,
-        ))
-
-    path = client.mk_path('x.test', 60)
-    stored = real_read(path)
-    cached = trio.run(client.as_df, 'x.test', 60, False)
-    assert stored['time'].to_list() == [60, 120]
-    assert cached['time'].to_list() == [60, 120]
-    assert not list(tmp_path.glob('*.tmp'))
-
-
-def test_directory_sync_failure_keeps_visible_cache_consistent(
-    tmp_path: Path,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    '''
-    A post-replace durability error leaves cache matching the file.
-
-    '''
-    client = NativeStorageClient(tmp_path)
-    run(client.write_ohlcv('x.test', mk_ohlcv((60,)), 60))
-    real_fsync = os.fsync
-    calls: int = 0
-
-    def fail_directory_sync(fd: int) -> None:
-        nonlocal calls
-        calls += 1
-        if calls == 2:
-            raise OSError('simulated directory sync failure')
-        real_fsync(fd)
-
-    monkeypatch.setattr(os, 'fsync', fail_directory_sync)
-    with pytest.raises(OSError, match='directory sync failure'):
-        run(client.update_ohlcv(
-            'x.test',
-            mk_ohlcv((120,)),
-            60,
-        ))
-
-    stored = pl.read_parquet(client.mk_path('x.test', 60))
-    cached = trio.run(client.as_df, 'x.test', 60, False)
-    assert stored['time'].to_list() == [60, 120]
-    assert cached['time'].to_list() == [60, 120]
-
-
-@pytest.mark.parametrize(
-    'times',
-    [
-        (0, 60),
-        (60, 60),
-        (120, 60),
-    ],
-)
-def test_update_rejects_invalid_timestamps(
-    tmp_path: Path,
-    times: tuple[float, ...],
-) -> None:
-    '''
-    Incremental input must have positive, increasing timestamps.
-
-    '''
-    client = NativeStorageClient(tmp_path)
-    incoming = mk_ohlcv(times)
-
-    with pytest.raises(ValueError):
-        run(client.update_ohlcv('x.test', incoming, 60))
-
-    assert not client.mk_path('x.test', 60).exists()
-
-
-def test_replacement_rejects_invalid_timestamps(
-    tmp_path: Path,
-) -> None:
-    '''
-    Explicit replacement enforces the same durable invariants.
-
-    '''
-    client = NativeStorageClient(tmp_path)
-
-    with pytest.raises(ValueError):
-        run(client.write_ohlcv(
-            'x.test',
-            mk_ohlcv((60, 60)),
-            60,
-        ))
-
-    assert not client.mk_path('x.test', 60).exists()
-
-
-def test_write_rejects_invalid_schema_and_values(
-    tmp_path: Path,
-) -> None:
-    '''
-    Durable OHLCV columns must exist and contain finite numbers.
-
-    '''
-    client = NativeStorageClient(tmp_path)
-    with pytest.raises(ValueError, match='missing columns'):
-        run(client.write_ohlcv(
-            'x.test',
-            pl.DataFrame({'time': [60]}),
-            60,
-        ))
-
-    invalid = mk_ohlcv((60,))
-    invalid['close'] = np.nan
-    with pytest.raises(ValueError, match='finite numeric'):
-        run(client.write_ohlcv('x.test', invalid, 60))
-
-
-def test_series_do_not_interfere(
-    tmp_path: Path,
-) -> None:
-    '''
-    FQME and timeframe keys isolate merge and cache state.
-
-    '''
-    client = NativeStorageClient(tmp_path)
-    run(client.write_ohlcv('x.test', mk_ohlcv((60,)), 60))
-    run(client.write_ohlcv('x.test', mk_ohlcv((1, 2)), 1))
-    run(client.write_ohlcv('y.test', mk_ohlcv((60, 120)), 60))
-    run(client.update_ohlcv('x.test', mk_ohlcv((120,)), 60))
-
-    x_60 = pl.read_parquet(client.mk_path('x.test', 60))
-    x_1 = pl.read_parquet(client.mk_path('x.test', 1))
-    y_60 = pl.read_parquet(client.mk_path('y.test', 60))
-    assert x_60['time'].to_list() == [60, 120]
-    assert x_1['time'].to_list() == [1, 2]
-    assert y_60['time'].to_list() == [60, 120]
-
-
-def test_index_files_ignores_lock_and_stale_temp_files(
-    tmp_path: Path,
-) -> None:
-    '''
-    Crash leftovers and writer locks are not durable series entries.
-
-    '''
-    client = NativeStorageClient(tmp_path)
-    run(client.write_ohlcv('x.test', mk_ohlcv((60,)), 60))
-    (tmp_path / 'x.test.ohlcv60s.parquet.crash.tmp').touch()
-
-    index = client.index_files()
-
-    assert list(index) == ['x.test']
-    assert index['x.test']['period'] == 60
-
-
-def test_contended_file_lock_yields_to_trio(
-    tmp_path: Path,
-) -> None:
-    '''
-    Cross-client lock contention does not block the actor loop.
-
-    '''
-    client = NativeStorageClient(tmp_path)
-    run(client.write_ohlcv('x.test', mk_ohlcv((60,)), 60))
-    path = client.mk_path('x.test', 60)
-    lock_path = path.with_name(f'.{path.name}.lock')
-
-    async def main() -> None:
-        done = trio.Event()
-
-        async def update() -> None:
-            await client.update_ohlcv(
-                'x.test',
-                mk_ohlcv((120,)),
-                60,
-            )
-            done.set()
-
-        with lock_path.open('a+b') as lock_file:
-            flock(lock_file.fileno(), LOCK_EX)
-            async with trio.open_nursery() as nursery:
-                nursery.start_soon(update)
-                await trio.sleep(0.03)
-                assert not done.is_set()
-                flock(lock_file.fileno(), LOCK_UN)
-                with trio.fail_after(0.5):
-                    await done.wait()
-
-    trio.run(main)
-    stored = pl.read_parquet(path)
-    assert stored['time'].to_list() == [60, 120]

tests/test_tsp_analysis.py

@@ -1,330 +0,0 @@
-'''
-Deterministic time-series analysis regressions.
-
-'''
-import numpy as np
-import polars as pl
-import pytest
-
-from piker.data._source import def_iohlcv_fields
-from piker.tsp import _anal
-from piker.tsp._anal import (
-    dedupe,
-    detect_time_gaps,
-    detect_time_ordering_errors,
-    get_null_segs,
-    iter_null_segs,
-    np2pl,
-    with_dts,
-)
-from piker.tsp._dedupe_smart import dedupe_ohlcv_smart
-
-
-def mk_ohlcv(
-    times: list[int],
-    indexes: list[int]|None = None,
-) -> np.ndarray:
-    '''
-    Build a synthetic OHLCV frame from timestamps.
-
-    '''
-    size: int = len(times)
-    frame = np.zeros(
-        size,
-        dtype=np.dtype(def_iohlcv_fields),
-    )
-    frame['index'] = (
-        np.arange(size)
-        if indexes is None
-        else indexes
-    )
-    frame['time'] = times
-    frame['open'] = np.arange(size) + 10
-    frame['high'] = frame['open'] + 2
-    frame['low'] = frame['open'] - 2
-    frame['close'] = frame['open'] + 1
-    frame['volume'] = np.arange(size) + 100
-    return frame
-
-
-def expected_null_segments(
-    null_positions: list[int],
-    size: int,
-    offset: int,
-    margin: int,
-
-) -> list[list[int]]:
-    '''
-    Group relative null positions with a simple state machine.
-
-    '''
-    runs: list[list[int]] = []
-    for pos in null_positions:
-        if (
-            not runs
-            or
-            pos != runs[-1][1] + 1
-        ):
-            runs.append([pos, pos])
-        else:
-            runs[-1][1] = pos
-
-    frame_end: int = offset + size - 1
-    return [
-        [
-            max(offset, offset + start - margin),
-            min(frame_end, offset + end + margin),
-        ]
-        for start, end in runs
-    ]
-
-
-def test_dedupe_recomputes_dts_after_sort() -> None:
-    '''
-    Derived columns describe final sorted, unique rows.
-
-    '''
-    src: pl.DataFrame = np2pl(mk_ohlcv([
-        60,
-        180,
-        120,
-        180,
-    ]))
-
-    _, deduped, diff = dedupe(src)
-
-    assert diff == 1
-    assert deduped['time'].to_list() == [60, 120, 180]
-    assert deduped['time_prev'].to_list() == [None, 60, 120]
-    assert deduped['s_diff'].to_list() == [None, 60, 60]
-    assert detect_time_gaps(
-        deduped,
-        expect_period=60,
-    ).is_empty()
-
-
-def test_negative_delta_is_an_ordering_error(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    '''
-    Reversed timestamps are reported as ordering faults, not gaps.
-
-    '''
-    src = np2pl(mk_ohlcv([120, 60]))
-    wdts: pl.DataFrame = with_dts(src)
-    errors = detect_time_ordering_errors(wdts)
-
-    assert detect_time_gaps(
-        wdts,
-        expect_period=30,
-    ).is_empty()
-    assert errors['time_prev'].to_list() == [120]
-    assert errors['time'].to_list() == [60]
-    assert errors['s_diff'].to_list() == [-60]
-
-    warnings: list[str] = []
-    monkeypatch.setattr(_anal.log, 'warning', warnings.append)
-    dedupe(src)
-    assert len(warnings) == 1
-    assert 'non-positive timestamp step' in warnings[0]
-
-
-def test_single_null_row_is_a_segment() -> None:
-    '''
-    A one-row null segment remains repairable.
-
-    '''
-    frame: np.ndarray = mk_ohlcv([
-        60,
-        120,
-        0,
-        240,
-        300,
-    ])
-
-    nulls = get_null_segs(frame, period=60)
-
-    assert nulls is not None
-    segments, zero_indexes, zero_rows = nulls
-    assert segments == [[1, 3]]
-    assert zero_indexes.tolist() == [2]
-    assert zero_rows['index'].tolist() == [2]
-
-
-@pytest.mark.parametrize(
-    ('times', 'expected'),
-    [
-        ([0, 0, 180, 240, 300, 360], [[10, 12]]),
-        ([60, 0, 0, 240, 0, 360], [[10, 13], [13, 15]]),
-        ([60, 120, 180, 240, 0, 0], [[13, 15]]),
-    ],
-)
-def test_null_segment_boundaries_match_numpy_and_polars(
-    times: list[int],
-    expected: list[list[int]],
-) -> None:
-    '''
-    Segments have exact, inclusive, margin-expanded endpoints.
-
-    '''
-    frame: np.ndarray = mk_ohlcv(
-        times,
-        indexes=list(range(10, 16)),
-    )
-
-    np_nulls = get_null_segs(frame, period=60)
-    pl_nulls = get_null_segs(np2pl(frame), period=60)
-
-    assert np_nulls is not None
-    assert pl_nulls is not None
-    assert np_nulls[0] == expected
-    assert pl_nulls[0] == expected
-    assert np_nulls[1].tolist() == pl_nulls[1].to_list()
-
-
-def test_null_segment_grouping_exhaustive() -> None:
-    '''
-    Every null layout through six rows matches an independent oracle.
-
-    '''
-    offset: int = 10
-    for size in range(1, 7):
-        for mask in range(1, 1 << size):
-            null_positions: list[int] = [
-                pos
-                for pos in range(size)
-                if mask & (1 << pos)
-            ]
-            frame = mk_ohlcv(
-                [
-                    0
-                    if pos in null_positions
-                    else (pos + 1) * 60
-                    for pos in range(size)
-                ],
-                indexes=list(range(offset, offset + size)),
-            )
-
-            for margin in range(3):
-                expected = expected_null_segments(
-                    null_positions,
-                    size,
-                    offset,
-                    margin,
-                )
-                for candidate in (frame, np2pl(frame)):
-                    nulls = get_null_segs(
-                        candidate,
-                        period=60,
-                        imargin=margin,
-                    )
-                    assert nulls is not None
-                    assert nulls[0] == expected
-
-
-def test_null_segment_margin_and_column() -> None:
-    '''
-    The requested column and index margin define each segment.
-
-    '''
-    frame: np.ndarray = mk_ohlcv(
-        [60, 120, 180, 240, 300],
-        indexes=[10, 11, 12, 13, 14],
-    )
-    frame['volume'][1:3] = 0
-
-    for candidate in (frame, np2pl(frame)):
-        no_margin = get_null_segs(
-            candidate,
-            period=60,
-            imargin=0,
-            col='volume',
-        )
-        wide_margin = get_null_segs(
-            candidate,
-            period=60,
-            imargin=2,
-            col='volume',
-        )
-
-        assert no_margin is not None
-        assert wide_margin is not None
-        assert no_margin[0] == [[11, 12]]
-        assert wide_margin[0] == [[10, 14]]
-        assert list(no_margin[1]) == [11, 12]
-
-
-def test_null_segment_input_invariants() -> None:
-    '''
-    Empty results and invalid index/margin inputs are explicit.
-
-    '''
-    frame = mk_ohlcv([60, 120, 180])
-    assert get_null_segs(frame, period=60) is None
-
-    with pytest.raises(ValueError, match='imargin'):
-        get_null_segs(
-            mk_ohlcv([60, 0, 180]),
-            period=60,
-            imargin=-1,
-        )
-
-    with pytest.raises(ValueError, match='contiguous'):
-        get_null_segs(
-            mk_ohlcv(
-                [60, 0, 180],
-                indexes=[10, 12, 13],
-            ),
-            period=60,
-        )
-
-
-def test_iter_null_segs_uses_precomputed_segments(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    '''
-    Iteration preserves inclusive boundary rows without rescanning.
-
-    '''
-    frame = mk_ohlcv([60, 120, 0, 240, 300])
-    nulls = get_null_segs(frame, period=60)
-    assert nulls is not None
-
-    def fail_rescan(*args, **kwargs):
-        raise AssertionError('precomputed segments were ignored')
-
-    monkeypatch.setattr(_anal, 'get_null_segs', fail_rescan)
-    segments = list(iter_null_segs(
-        timeframe=60,
-        frame=frame,
-        null_segs=nulls,
-    ))
-
-    assert len(segments) == 1
-    assert segments[0][:6] == (
-        1,
-        3,
-        1,
-        3,
-        120,
-        240,
-    )
-
-
-def test_smart_dedupe_honors_sort_without_dupes() -> None:
-    '''
-    The sort option applies even when no duplicates exist.
-
-    '''
-    src: pl.DataFrame = np2pl(mk_ohlcv([
-        180,
-        60,
-        120,
-    ]))
-
-    _, deduped, diff, _, issues = dedupe_ohlcv_smart(src)
-
-    assert diff == 0
-    assert issues is None
-    assert deduped['time'].to_list() == [60, 120, 180]
-    assert deduped['s_diff'].to_list() == [None, 60, 60]